Persistence¶

Data catalog and persistence components.

class BarDataWrangler¶

Bases: object

bar_type¶

price_precision¶

process_record_batch_bytes(data)¶

size_precision¶

class CatalogBackend¶

Bases: object

static External(name)¶

Parquet = CatalogBackend.Parquet¶

external_name¶

static from_str(value)¶

name¶

value¶

class DataBackendSession¶

Bases: object

Provides a DataFusion session and registers DataFusion queries.

The session is used to register data sources and make queries on them. A query returns a Chunk of Arrow records. It is decoded and converted into a Vec of data by types that implement [DecodeDataFromRecordBatch].

add_custom_file(type_name, table_name, file_path, sql_query=None)¶

Registers a Parquet file for a custom data type identified by type_name.

The custom data type must have been registered via ensure_custom_data_registered::<T>() before calling this method.

add_file(data_type, table_name, file_path, sql_query=None)¶

Registers a Parquet file and adds a batch stream for decoding.

The caller must specify T to indicate the kind of data expected. table_name is the logical name for queries; file_path is the Parquet path; sql_query defaults to SELECT * FROM {table_name} ORDER BY ts_init if None.

When custom_type_name is Some, it is merged into each batch’s schema metadata before decoding (as type_name). Use this for custom data when Parquet/DataFusion does not preserve schema metadata so the decoder can look up the type in the registry.

The file data must be ordered by the ts_init in ascending order for this to work correctly.

# Errors

Returns an error if parquet registration, SQL planning, stream execution, or data decoding setup fails.

register_object_store_from_uri(uri, storage_options=None)¶

# Errors

Returns an error if the object store URI cannot be normalized or the backend cannot be created.

to_query_result()¶

class DataCatalogConfig¶

Bases: object

Configuration for a catalog available to request-time historical data loading.

catalog_backend¶: Returns the catalog backend implementation to use.

fs_protocol¶: Returns the fsspec file system protocol for the data catalog.

fs_rust_storage_option_keys¶

name¶: Returns the catalog registration name.

params¶: Returns backend-specific catalog parameters.

path¶: Returns the path to the data catalog.

read_only¶: Returns whether the catalog rejects response write-back.

class DataQueryResult¶

Bases: object

to_list()¶

Collects the remaining query records as native Python objects.

# Errors

Returns an error if a query stream fails or a record batch cannot be decoded.

class MacroYieldCurveData¶

Bases: object

YieldCurveData-equivalent custom data type using the macro with Vec<f64> fields.

Tests Vec<f64> / ListFloat64 support. Exposed to Python for roundtrip tests.

curve_name¶

classmethod decode_record_batch_py(metadata, py_batch)¶: Decodes a PyArrow RecordBatch into custom data instances.

encode_record_batch_py(items)¶: Encodes custom data instances into a PyArrow RecordBatch.

classmethod from_json(data)¶

interest_rates¶

tenors¶

to_json()¶

ts_event¶

ts_init¶

class OrderBookDeltaDataWrangler¶

Bases: object

instrument_id¶

price_precision¶

process_record_batch_bytes(data)¶

size_precision¶

class OrderBookDepthDataWrangler¶

Bases: object

instrument_id¶

price_precision¶

process_record_batch_bytes(data)¶

size_precision¶

class ParquetDataCatalog¶

Bases: object

A catalog for writing data to Parquet files.

consolidate_catalog(start=None, end=None, ensure_contiguous_files=None, deduplicate=None)¶

Consolidate all data files in the catalog within the specified time range.

# Parameters

start: Optional start timestamp (nanoseconds since Unix epoch)
end: Optional end timestamp (nanoseconds since Unix epoch)
ensure_contiguous_files: Optional flag to ensure files are contiguous
deduplicate: Optional flag to deduplicate rows when combining files

consolidate_catalog_by_period(period_nanos=None, start=None, end=None, ensure_contiguous_files=None)¶

Consolidate all data files in the catalog by splitting them into fixed time periods.

This method identifies all leaf directories in the catalog that contain parquet files and consolidates them by period. A leaf directory is one that contains files but no subdirectories. This is a convenience method that effectively calls consolidate_data_by_period for all data types and instrument IDs in the catalog.

# Parameters

period_nanos: Optional period duration for consolidation in nanoseconds. Default is 1 day (86400000000000). Examples: 3600000000000 (1 hour), 604800000000000 (7 days), 1800000000000 (30 minutes)
start: Optional start timestamp for the consolidation range (nanoseconds since Unix epoch)
end: Optional end timestamp for the consolidation range (nanoseconds since Unix epoch)
ensure_contiguous_files: Optional flag to control file naming strategy

consolidate_data(data_type, instrument_id=None, start=None, end=None, ensure_contiguous_files=None, deduplicate=None)¶

Consolidate data files for a specific data type within the specified time range.

# Parameters

data_type: The stored family to target (data type, record type, or instrument type).
instrument_id: Optional instrument ID filter
start: Optional start timestamp (nanoseconds since Unix epoch)
end: Optional end timestamp (nanoseconds since Unix epoch)
ensure_contiguous_files: Optional flag to ensure files are contiguous
deduplicate: Optional flag to deduplicate rows when combining files

consolidate_data_by_period(data_type, identifier=None, period_nanos=None, start=None, end=None, ensure_contiguous_files=None)¶

Consolidate data files by splitting them into fixed time periods.

This method queries data by period and writes consolidated files immediately, using efficient period-based consolidation logic. When start/end boundaries intersect existing files, the function automatically splits those files to preserve all data.

# Parameters

data_type: The stored family to target (data type, record type, or instrument type).
identifier: Optional instrument ID to consolidate. If None, consolidates all instruments
period_nanos: Optional period duration for consolidation in nanoseconds. Default is 1 day (86400000000000). Examples: 3600000000000 (1 hour), 604800000000000 (7 days), 1800000000000 (30 minutes)
start: Optional start timestamp for consolidation range (nanoseconds since Unix epoch)
end: Optional end timestamp for consolidation range (nanoseconds since Unix epoch)
ensure_contiguous_files: Optional flag to control file naming strategy

convert_stream_to_data(instance_id, data_type, subdirectory=None, identifiers=None, use_ts_event_for_ts_init=False)¶

Convert stream data from feather files to parquet files.

This method reads data from feather files generated during a backtest or live run and writes it to the catalog in parquet format. It’s useful for converting temporary stream data into a more permanent and queryable format.

# Parameters

instance_id: The ID of the backtest or live run instance
data_type: The stored family to convert (data type or record type).
subdirectory: Optional subdirectory containing the feather files. Either “backtest” or “live” (default: “backtest”)
identifiers: Optional list of identifiers to filter by (instrument IDs or bar types)
use_ts_event_for_ts_init: If true, replaces the ts_init column with ts_event column values before deserializing

# Returns

Returns nothing on success.

# Examples

```python # Convert backtest stream data to parquet catalog.convert_stream_to_data(

“instance-123”, NautilusDataType.QuoteTick, subdirectory=”backtest”

)

# Convert live run data with identifier filtering catalog.convert_stream_to_data(

“instance-456”, NautilusDataType.TradeTick, subdirectory=”live”, identifiers=[“EUR/USD.SIM”]

)¶

delete_catalog_range(start=None, end=None)¶

Delete data within a specified time range across the entire catalog.

This method identifies all leaf directories in the catalog that contain parquet files and deletes data within the specified time range from each directory. A leaf directory is one that contains files but no subdirectories. This is a convenience method that effectively calls delete_data_range for all data types and instrument IDs in the catalog.

# Parameters

start: Optional start timestamp for the deletion range (nanoseconds since Unix epoch)
end: Optional end timestamp for the deletion range (nanoseconds since Unix epoch)

# Notes

This operation permanently removes data and cannot be undone
The deletion process handles file intersections intelligently by splitting files when they partially overlap with the deletion range
Files completely within the deletion range are removed entirely
Files partially overlapping the deletion range are split to preserve data outside the range
This method is useful for bulk data cleanup operations across the entire catalog
Empty directories are not automatically removed after deletion

delete_data_range(data_type, identifier=None, start=None, end=None)¶

Delete data within a specified time range for a specific data type and instrument.

This method identifies all parquet files that intersect with the specified time range and handles them appropriately: - Files completely within the range are deleted - Files partially overlapping the range are split to preserve data outside the range - The original intersecting files are removed after processing

# Parameters

data_type: The data type to delete from.
identifier: Optional identifier to delete data for. If None, deletes data across all identifiers
start: Optional start timestamp for the deletion range (nanoseconds since Unix epoch)
end: Optional end timestamp for the deletion range (nanoseconds since Unix epoch)

# Notes

This operation permanently removes data and cannot be undone
Files that partially overlap the deletion range are split to preserve data outside the range
The method ensures data integrity by using atomic operations where possible
Empty directories are not automatically removed after deletion

extend_file_name(data_type, instrument_id=None, *, start, end)¶

Extend file names in the catalog with additional timestamp information.

# Parameters

data_type: The stored family to target (data type, record type, or instrument type).
instrument_id: Optional instrument ID filter
start: Start timestamp (nanoseconds since Unix epoch)
end: End timestamp (nanoseconds since Unix epoch)

get_intervals(data_type, instrument_id=None)¶

Get time intervals covered by data for a specific data class and instrument.

# Parameters

data_type: The stored family to target (data type, record type, or instrument type).
instrument_id: Optional instrument ID filter

# Returns

Returns a list of (start, end) timestamp tuples representing covered intervals.

get_missing_intervals_for_request(start, end, data_type, instrument_id=None)¶

Get missing time intervals for a data request.

# Parameters

start: Start timestamp (nanoseconds since Unix epoch)
end: End timestamp (nanoseconds since Unix epoch)
data_type: The stored family to target (data type, record type, or instrument type).
instrument_id: Optional instrument ID filter

# Returns

Returns a list of (start, end) timestamp tuples representing missing intervals.

instruments(instrument_ids=None, start=None, end=None, where_clause=None, instrument_type=None)¶

Query instruments from the catalog.

# Parameters

instrument_ids: Optional list of instrument IDs to filter by. If None, returns all instruments.
start: Optional inclusive lower bound for ts_init filtering.
end: Optional inclusive upper bound for ts_init filtering.
where_clause: Optional SQL WHERE clause for additional filtering.

# Returns

Returns a list of instrument objects (e.g. CurrencyPair, Equity).

list_backtest_runs()¶

List all backtest run IDs available in the catalog.

# Returns

Returns a list of backtest run IDs (as directory stems) in the catalog.

list_backtests()¶: List all backtest run instances available in the catalog.

list_data_types()¶

List all data types available in the catalog.

# Returns

Returns a list of data type names (as directory stems) in the catalog.

list_instruments(data_type)¶: List all instrument IDs available in the catalog for a given catalog type.

list_live_runs()¶

List all live run IDs available in the catalog.

# Returns

Returns a list of live run IDs (as directory stems) in the catalog.

list_parquet_files(data_type, instrument_id)¶: List all Parquet files in the catalog for a given data type and instrument.

migrate_from_legacy_parquet_path(parquet_path, storage_options=None, dry_run=False)¶: Rewrites a legacy Parquet catalog into this current Parquet catalog.

query(data_type, identifiers=None, start=None, end=None, where_clause=None, files=None, optimize_file_loading=True)¶: Queries one data family and returns the decoded Python objects.

query_bars(identifiers=None, start=None, end=None, where_clause=None)¶

Query bar data from Parquet files.

# Parameters

identifiers: Optional list of identifiers to filter by. Can be instrument_id strings (e.g., “EUR/USD.SIM”) or bar_type strings (e.g., “EUR/USD.SIM-1-MINUTE-LAST-EXTERNAL”). For bars, partial matching is supported (e.g., “EUR/USD.SIM” will match all bar types for that instrument).
start: Optional start timestamp (nanoseconds since Unix epoch)
end: Optional end timestamp (nanoseconds since Unix epoch)
where_clause: Optional SQL WHERE clause for additional filtering

# Returns

Returns a vector of Bar objects matching the query criteria.

query_custom_data(type_name, identifiers=None, start=None, end=None, where_clause=None)¶: Query custom data from Parquet files.

query_data_arrow_bytes(data_type, identifiers=None, start=None, end=None, where_clause=None, display=True, as_of=None)¶: Query catalog data as display-friendly Arrow IPC stream bytes.

query_data_arrow_stream(data_type, identifiers=None, start=None, end=None, where_clause=None, display=True, as_of=None)¶: Query catalog data as an Arrow C stream PyCapsule.

query_files(data_type, identifiers=None, start=None, end=None)¶

Query files in the catalog matching the specified criteria.

# Parameters

data_type: The stored family to target (data type, record type, or instrument type).
identifiers: Optional list of identifiers to filter by. Can be instrument_id strings (e.g., “EUR/USD.SIM”) or bar_type strings (e.g., “EUR/USD.SIM-1-MINUTE-LAST-EXTERNAL”). For bars, partial matching is supported.
start: Optional start timestamp (nanoseconds since Unix epoch)
end: Optional end timestamp (nanoseconds since Unix epoch)

# Returns

Returns a list of file paths matching the criteria.

query_first_timestamp(data_type, instrument_id=None)¶

Query the first timestamp for a specific data class and instrument.

# Parameters

data_type: The stored family to target (data type, record type, or instrument type).
instrument_id: Optional instrument ID filter

# Returns

Returns the first timestamp as nanoseconds since Unix epoch, or None if no data exists.

query_index_price_updates(identifiers=None, start=None, end=None, where_clause=None)¶

Query index price update data from Parquet files.

# Parameters

identifiers: Optional list of identifiers to filter by.
start: Optional start timestamp (nanoseconds since Unix epoch)
end: Optional end timestamp (nanoseconds since Unix epoch)
where_clause: Optional SQL WHERE clause for additional filtering

# Returns

Returns a vector of IndexPriceUpdate objects matching the query criteria.

query_instrument_arrow_bytes(instrument_ids=None, start=None, end=None, where_clause=None, instrument_type=None)¶: Query instruments as display-friendly Arrow IPC stream bytes.

query_instrument_arrow_stream(instrument_ids=None, start=None, end=None, where_clause=None, instrument_type=None)¶: Query instruments as an Arrow C stream PyCapsule.

query_last_timestamp(data_type, instrument_id=None)¶

Query the last timestamp for a specific data class and instrument.

# Parameters

data_type: The stored family to target (data type, record type, or instrument type).
instrument_id: Optional instrument ID filter

# Returns

Returns the last timestamp as nanoseconds since Unix epoch, or None if no data exists.

query_mark_price_updates(identifiers=None, start=None, end=None, where_clause=None)¶

Query mark price update data from Parquet files.

# Parameters

identifiers: Optional list of identifiers to filter by.
start: Optional start timestamp (nanoseconds since Unix epoch)
end: Optional end timestamp (nanoseconds since Unix epoch)
where_clause: Optional SQL WHERE clause for additional filtering

# Returns

Returns a vector of MarkPriceUpdate objects matching the query criteria.

query_metadata(data_type, identifiers=None, start=None, end=None, where_clause=None)¶: Query catalog metadata keyed by the first timestamp where each metadata is used.

query_option_greeks(identifiers=None, start=None, end=None, where_clause=None)¶: Query option greeks data from Parquet files.

query_order_book_deltas(identifiers=None, start=None, end=None, where_clause=None)¶

Query order book delta data from Parquet files.

# Parameters

identifiers: Optional list of identifiers to filter by. Can be instrument_id strings (e.g., “EUR/USD.SIM”) or bar_type strings (e.g., “EUR/USD.SIM-1-MINUTE-LAST-EXTERNAL”). For bars, partial matching is supported.
start: Optional start timestamp (nanoseconds since Unix epoch)
end: Optional end timestamp (nanoseconds since Unix epoch)
where_clause: Optional SQL WHERE clause for additional filtering

# Returns

Returns a vector of OrderBookDelta objects matching the query criteria.

query_order_book_depths(identifiers=None, start=None, end=None, where_clause=None)¶

Query order book depth data from Parquet files.

# Parameters

identifiers: Optional list of identifiers to filter by.
start: Optional start timestamp (nanoseconds since Unix epoch)
end: Optional end timestamp (nanoseconds since Unix epoch)
where_clause: Optional SQL WHERE clause for additional filtering

# Returns

Returns a vector of OrderBookDepth objects matching the query criteria.

query_quote_ticks(identifiers=None, start=None, end=None, where_clause=None)¶

Query quote tick data from Parquet files.

# Parameters

identifiers: Optional list of identifiers to filter by. Can be instrument_id strings (e.g., “EUR/USD.SIM”) or bar_type strings (e.g., “EUR/USD.SIM-1-MINUTE-LAST-EXTERNAL”). For bars, partial matching is supported.
start: Optional start timestamp (nanoseconds since Unix epoch)
end: Optional end timestamp (nanoseconds since Unix epoch)
where_clause: Optional SQL WHERE clause for additional filtering

# Returns

Returns a vector of QuoteTick objects matching the query criteria.

query_record_arrow_bytes(record_type, identifier=None, start=None, end=None, where_clause=None, display=True, as_of=None)¶: Query catalog records and return Arrow IPC stream bytes.

query_record_arrow_stream(record_type, identifier=None, start=None, end=None, where_clause=None, display=True, as_of=None)¶: Query catalog records as an Arrow C stream PyCapsule.

query_trade_ticks(identifiers=None, start=None, end=None, where_clause=None)¶

Query trade tick data from Parquet files.

# Parameters

identifiers: Optional list of identifiers to filter by. Can be instrument_id strings (e.g., “EUR/USD.SIM”) or bar_type strings (e.g., “EUR/USD.SIM-1-MINUTE-LAST-EXTERNAL”). For bars, partial matching is supported.
start: Optional start timestamp (nanoseconds since Unix epoch)
end: Optional end timestamp (nanoseconds since Unix epoch)
where_clause: Optional SQL WHERE clause for additional filtering

# Returns

Returns a vector of TradeTick objects matching the query criteria.

read_backtest(instance_id)¶

Read data from a backtest run instance.

# Parameters

instance_id: The ID of the backtest run instance

# Returns

Returns a list of data objects from the backtest run, sorted by timestamp.

read_live_run(instance_id)¶

Read data from a live run instance.

# Parameters

instance_id: The ID of the live run instance

# Returns

Returns a list of data objects from the live run, sorted by timestamp.

reset_all_file_names()¶: Reset all catalog file names to their canonical form.

reset_data_file_names(data_type, instrument_id=None)¶

Reset data file names for a specific data class to their canonical form.

# Parameters

data_type: The stored family to target (data type, record type, or instrument type).
instrument_id: Optional instrument ID filter

write_bars(data, start=None, end=None, skip_disjoint_check=False)¶

Write bar data to Parquet files.

# Parameters

data: Vector of bars to write
start: Optional start timestamp override (nanoseconds since Unix epoch)
end: Optional end timestamp override (nanoseconds since Unix epoch)

# Returns

Returns the path of the created file as a string.

write_custom_data(data, start=None, end=None, skip_disjoint_check=False)¶

Writes custom data to Parquet files.

Requires CustomData wrappers. Callers must wrap raw custom objects in CustomData(data_type=DataType(cls, metadata=…), data=…) before writing.

The registered Arrow schema must contain ts_init. Any ts_event or ts_init fields must use timestamp(“ns”, tz=”UTC”); incompatible schemas fail before writing.

write_index_price_updates(data, start=None, end=None, skip_disjoint_check=False)¶

Write index price update data to Parquet files.

# Parameters

data: Vector of index price updates to write
start: Optional start timestamp override (nanoseconds since Unix epoch)
end: Optional end timestamp override (nanoseconds since Unix epoch)

# Returns

Returns the path of the created file as a string.

write_instrument_closes(data, start=None, end=None, skip_disjoint_check=False)¶: Write instrument close data to Parquet files.

write_instrument_statuses(data, start=None, end=None, skip_disjoint_check=False)¶: Write instrument status data to Parquet files.

write_instruments(instruments)¶

Write instruments to Parquet files in the catalog.

Instruments are stored under data/instruments/{instrument_id}/ using timestamp-ranged parquet file names, allowing multiple historical versions of the same instrument to be written across separate calls.

# Parameters

instruments: A Python list of instrument objects (e.g. CurrencyPair, Equity).

# Returns

Returns a list of written file paths.

write_mark_price_updates(data, start=None, end=None, skip_disjoint_check=False)¶

Write mark price update data to Parquet files.

# Parameters

data: Vector of mark price updates to write
start: Optional start timestamp override (nanoseconds since Unix epoch)
end: Optional end timestamp override (nanoseconds since Unix epoch)

# Returns

Returns the path of the created file as a string.

write_option_greeks(data, start=None, end=None, skip_disjoint_check=False)¶

Write option greeks data to Parquet files.

# Parameters

data: Vector of option greeks to write
start: Optional start timestamp override (nanoseconds since Unix epoch)
end: Optional end timestamp override (nanoseconds since Unix epoch)

# Returns

Returns the path of the created file as a string.

write_order_book_deltas(data, start=None, end=None, skip_disjoint_check=False)¶

Write order book delta data to Parquet files.

# Parameters

data: Vector of order book deltas to write
start: Optional start timestamp override (nanoseconds since Unix epoch)
end: Optional end timestamp override (nanoseconds since Unix epoch)

# Returns

Returns the path of the created file as a string.

write_order_book_depths(data, start=None, end=None, skip_disjoint_check=False)¶

Write order book depth data to Parquet files.

# Parameters

data: Vector of order book depths to write
start: Optional start timestamp override (nanoseconds since Unix epoch)
end: Optional end timestamp override (nanoseconds since Unix epoch)

# Returns

Returns the path of the created file as a string.

write_quote_ticks(data, start=None, end=None, skip_disjoint_check=False)¶

Write quote tick data to Parquet files.

# Parameters

data: Vector of quote ticks to write
start: Optional start timestamp override (nanoseconds since Unix epoch)
end: Optional end timestamp override (nanoseconds since Unix epoch)

# Returns

Returns the path of the created file as a string.

write_record_arrow_bytes(record_type, data, identifier=None, params=None)¶: Write catalog records from Arrow IPC stream bytes.

write_trade_ticks(data, start=None, end=None, skip_disjoint_check=False)¶

Write trade tick data to Parquet files.

# Parameters

data: Vector of trade ticks to write
start: Optional start timestamp override (nanoseconds since Unix epoch)
end: Optional end timestamp override (nanoseconds since Unix epoch)

# Returns

Returns the path of the created file as a string.

class QuoteTickDataWrangler¶

Bases: object

instrument_id¶

price_precision¶

process_record_batch_bytes(data)¶

size_precision¶

class RotationConfig¶

Bases: object

static interval(interval_ns)¶

interval_ns¶

max_size¶

mode¶

static no_rotation()¶

schedule_ns¶

static scheduled_dates(interval_ns, schedule_ns)¶

static size(max_size)¶

class RustTestCustomData¶

Bases: object

A simple Rust custom data type for roundtrip testing.

Used in persistence integration tests (test_catalog.rs) and Python roundtrip tests. Tests call ensure_custom_data_registered::<RustTestCustomData>() before using the catalog.

classmethod decode_record_batch_py(metadata, py_batch)¶: Decodes a PyArrow RecordBatch into custom data instances.

encode_record_batch_py(items)¶: Encodes custom data instances into a PyArrow RecordBatch.

flag¶

classmethod from_json(data)¶

instrument_id¶

to_json()¶

ts_event¶

ts_init¶

value¶

class RustTestFixedCustomData¶

Bases: object

Rust custom data type with native fixed-point Arrow fields.

aggressor_side¶

classmethod decode_record_batch_py(metadata, py_batch)¶: Decodes a PyArrow RecordBatch into custom data instances.

encode_record_batch_py(items)¶: Encodes custom data instances into a PyArrow RecordBatch.

classmethod from_json(data)¶

instrument_id¶

notional¶

nullable_aggressor_side¶

nullable_notional¶

price¶

quantity¶

to_json()¶

ts_event¶

ts_init¶

class RustTestParamsCustomData¶

Bases: object

Rust custom data type that exercises Params field support in the macro.

classmethod decode_record_batch_py(metadata, py_batch)¶: Decodes a PyArrow RecordBatch into custom data instances.

encode_record_batch_py(items)¶: Encodes custom data instances into a PyArrow RecordBatch.

classmethod from_json(data)¶

name¶

params¶

to_json()¶

ts_event¶

ts_init¶

class RustTestPriceMapCustomData¶

Bases: object

Rust custom data type that exercises typed map field support in the macro.

classmethod decode_record_batch_py(metadata, py_batch)¶: Decodes a PyArrow RecordBatch into custom data instances.

encode_record_batch_py(items)¶: Encodes custom data instances into a PyArrow RecordBatch.

classmethod from_json(data)¶

name¶

prices¶

to_json()¶

ts_event¶

ts_init¶

class RustTestTypedMapCustomData¶

Bases: object

Rust custom data type that exercises typed JSON map values across PyO3-supported types.

account_ids¶

bar_types¶

booleans¶

currencies¶

classmethod decode_record_batch_py(metadata, py_batch)¶: Decodes a PyArrow RecordBatch into custom data instances.

encode_record_batch_py(items)¶: Encodes custom data instances into a PyArrow RecordBatch.

floats_32¶

floats_64¶

classmethod from_json(data)¶

hash_prices_by_instrument¶

instrument_ids¶

integers_i32¶

integers_i64¶

integers_u32¶

integers_u64¶

monies¶

monies_by_currency¶

name¶

prices¶

prices_by_bar_type¶

prices_by_instrument¶

quantities¶

quantities_by_account¶

strings¶

to_json()¶

ts_event¶

ts_init¶

class StreamingConfig¶

Bases: object

Configuration streaming live or backtest runs to a persistence writer.

catalog_path¶

data_types¶

flush_interval_ms¶

fs_protocol¶

instrument_types¶

max_file_size¶

params¶

record_filters¶

record_types¶

replace_existing¶

rotation_config¶

rotation_interval_ns¶

rotation_mode¶

schedule_ns¶

writer_backend¶

class StreamingFeatherWriter¶

Bases: object

Python binding for the Rust FeatherWriter.

This provides a streaming writer of Nautilus objects into feather files with rotation capabilities, matching the interface of Python’s StreamingFeatherWriter.

close()¶

Closes all writers by flushing and removing them.

After calling this, no further writes should be performed.

flush()¶

Flushes all active buffers by writing any remaining buffered bytes to the object store.

This is called automatically based on flush_interval_ms if configured, but can also be called manually by the client.

get_current_file_info()¶

Returns information about the current files being written.

Returns a dictionary mapping writer keys to (size, path) tuples.

get_next_rotation_time(type_str, instrument_id=None)¶: Returns the next rotation time for a writer, or None if not set.

is_closed¶: Returns whether the writer has been closed (no active writers).

subscribe()¶

Subscribes to all messages on the message bus (pattern “*”).

This matches the behavior of Python’s StreamingFeatherWriter when subscribed via trader.subscribe(“*”, writer.write).

unsubscribe()¶: Unsubscribes from the message bus.

write(data)¶

Writes a data object to the stream.

# Parameters

data: The data object to write (must be a Nautilus data type from pyo3).

class StreamingWriter¶

Bases: object

Python binding for the backend-selected streaming writer.

Resolves the writer through the persistence writer-factory registry, mirroring writer_backend selection: Feather, Parquet, or a registered name.

backend¶: Returns the resolved writer backend name.

close()¶: Closes the writer after flushing buffered data.

flush()¶: Flushes buffered data to durable storage.

write(data)¶: Writes a single Nautilus data value.

class TradeTickDataWrangler¶

Bases: object

instrument_id¶

price_precision¶

process_record_batch_bytes(data)¶

size_precision¶