Data I/O¶

Parquet serialization and writer registry.

load_data ¶

load_data(path: str | Path) -> dict[str, pd.DataFrame]

Load pre-processed pipeline data from a Parquet directory or pickle file.

Parameters:

Name	Type	Description	Default
`path`	`str or Path`	If path is a directory, each `.parquet` file inside is loaded as a DataFrame keyed by its stem (`events.parquet` → `"events"`). If path is a single file with a `.pkl` / `.pickle` extension, it is loaded via :func:`pandas.read_pickle` for backward compatibility (not recommended for untrusted data).	required

Returns:

Type	Description
`dict of str to pandas.DataFrame`

Raises:

Type	Description
`ConfigError`	If a Parquet file declares a schema version this build does not support. A file written with no version key (legacy data, or the bundled sample) loads with a warning — see :func:`ob_analytics.schemas.check_schema_version`.

save_data ¶

save_data(
    lob_data: dict[str, DataFrame],
    path: str | Path,
    *,
    fmt: str = "parquet",
    writer: DataWriter | None = None,
    config: Any = None,
    ctx: Any = None,
    **write_kwargs: Any,
) -> None

Save pipeline data to disk.

Parameters:

Name	Type	Description	Default
`lob_data`	`dict of str to pandas.DataFrame`	The DataFrames to save (keys become file stems).	required
`path`	`str or Path`	Destination directory (Parquet) or file (pickle).	required
`fmt`	`str`	Serialisation format. Built-in values are `"parquet"` (default) and `"pickle"`. The `"parquet"` path writes one file per key and tags each with :data:`SCHEMA_VERSION` in its metadata (checked by :func:`load_data`). A source name (e.g. `"bitstamp"`, `"lobster"`) round-trips through that source's own writer (its `create_writer` capability, so no separate writer registration is needed); a source that needs run state — LOBSTER's `trading_date` — reads it from ctx. A generic, source-independent writer registered via :func:`register_writer` is also resolved by name.	`'parquet'`
`writer`	`DataWriter`	A pre-constructed writer instance. When provided, fmt is ignored and the writer is used directly. This is the preferred path when saving from a :class:`Pipeline` that already holds a configured writer.	`None`
`config`	`Any`	Forwarded to a registered writer factory when `fmt` names one. `ctx` defaults to an empty :class:`~ob_analytics.protocols.RunContext`.	`None`
`ctx`	`Any`	Forwarded to a registered writer factory when `fmt` names one. `ctx` defaults to an empty :class:`~ob_analytics.protocols.RunContext`.	`None`
`**write_kwargs`	`Any`	Extra keyword arguments forwarded to `writer.write()`.	`{}`

register_writer ¶

register_writer(name: str, factory: WriterFactory) -> None

Register a writer factory under name for use with save_data(fmt=name, ctx=...).

The factory is called as factory(config, ctx) and must return a :class:DataWriter. This is what lets format-specific writers (e.g. :class:~ob_analytics.lobster.LobsterWriter, which needs trading_date) participate in the registry — they pull required parameters from the :class:~ob_analytics.protocols.RunContext.

list_writers ¶

list_writers() -> list[str]

Return a sorted list of registered writer names.