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 DataCatalogConfig

Bases: object

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

batch_size

Returns the number of rows per batch the catalog reads and writes.

catalog_backend

Returns the catalog backend implementation to use.

compression

Returns the compression codec of written data files.

fs_protocol

Returns the fsspec file system protocol for the data catalog.

fs_rust_storage_option_keys
max_row_group_size

Returns the maximum number of rows per written row group.

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 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, identifier=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).

  • identifier: Optional instrument ID or bar type 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, environment=Ellipsis, 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).

  • environment: The environment of the run, which names the folder holding its feather files (default: Environment.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, environment=Environment.BACKTEST

)

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

“instance-456”, NautilusDataType.TradeTick, environment=Environment.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, identifier=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).

  • identifier: Optional instrument ID or bar type filter

  • start: Start timestamp (nanoseconds since Unix epoch)

  • end: End timestamp (nanoseconds since Unix epoch)

get_intervals(data_type, identifier=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).

  • identifier: Optional instrument ID or bar type filter

# Returns

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

get_missing_intervals_for_request(start, end, data_type, identifier=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).

  • identifier: Optional instrument ID or bar type 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 the data, record, and instrument types stored 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, identifier)

List all Parquet files in the catalog for a given data type and identifier.

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(data_type, 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, identifier=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).

  • identifier: Optional instrument ID or bar type 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, identifier=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).

  • identifier: Optional instrument ID or bar type 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, identifier=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).

  • identifier: Optional instrument ID or bar type 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, timezone=Ellipsis)
static size(max_size)
timezone
class RotationMode

Bases: object

The rotation policy of a streaming writer, without its parameters.

INTERVAL = RotationMode.INTERVAL
NO_ROTATION = RotationMode.NO_ROTATION
SCHEDULED_DATES = RotationMode.SCHEDULED_DATES
SIZE = RotationMode.SIZE
classmethod from_str(data)
name
value
classmethod variants()
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.

The writer appends Feather files under the local writer_path, in one {backtest|sandbox|live}/{instance_id} directory per run. With a catalog, the writer for that catalog’s backend also promotes the files into it; without one, the Feather files are the only output.

catalog
data_types
delete_feather_after_promotion
flush_interval_ms
instrument_types
params
promote_on_close
promotion_interval_ms
record_filters
record_types
replace_existing
rotation_config
use_ts_event_for_ts_init
writer_backend

Feather without a catalog, otherwise the catalog’s backend.

Type:

Returns the writer backend

writer_path
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()

Seals all open files so each complete stream is visible as a .feather file.

Writes after closing open new files.

flush()

Flushes buffered bytes of every open file to disk.

This is called automatically based on flush_interval_ms if configured, but can also be called manually by the client. Files stay open, so flushing never starts a new file.

get_current_file_info()

Returns the current file size and path of each open file, keyed by the type it stages.

get_next_rotation_time(data_type)

Returns the next rotation time of a type’s file, or None if not set.

Pass a NautilusInstrumentType for the file of one instrument class.

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.

Appends Feather files under a local directory. With a catalog, the writer for that catalog’s backend (Parquet or a registered name) also promotes the files into it; without one, the Feather writer keeps only the Feather files.

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