Pipeline¶

The main orchestrator. Runs the full sequence: load → build trades → classify → depth. Use Pipeline(source=...) for LOBSTER or other registered sources, or pass individual components (loader=, trade_source=) to override specific stages. Flow-toxicity metrics are computed after the run by calling compute_vpin / compute_kyle_lambda / order_flow_imbalance on result.trades.

The source registry (register_source / list_sources / get_source) lives in ob_analytics.sources.

Pipeline ¶

Pipeline(
    config: PipelineConfig | None = None,
    *,
    source: Source | None = None,
    loader: EventLoader | None = None,
    trade_source: TradeSource | None = None,
    ctx: RunContext | None = None,
)

Configurable, composable order book analytics pipeline.

Each processing stage is handled by a pluggable component that satisfies the corresponding protocol. Pass your own implementations to override any stage.

Parameters:

Name	Type	Description	Default
`config`	`PipelineConfig`	Central configuration. Passed to default components when they are not explicitly provided.	`None`
`source`	`OfflineSource`	A source descriptor that provides the default loader, trade source, writer, and config overrides. Defaults to :class:`~ob_analytics.bitstamp.BitstampSource`. Explicit component arguments take precedence over the source's factories.	`None`
`loader`	`EventLoader`	Loads raw events from a data source. Overrides the source's loader.	`None`
`trade_source`	`TradeSource`	Builds the trades DataFrame. Overrides the source's trade source.	`None`

Notes

A live capture records the instrument's tick size and lot size in its meta.json (see :mod:ob_analytics.capture_record). :meth:run and :meth:run_windows use them for that capture when config does not set them, so a capture replays on its own price and size grid, as ob-analytics process reads it. A tick size you set (or a price_decimals) keeps the recorded tick size out, and a lot size (or a volume_decimals) the recorded lot size. When you pass your own loader or trade_source, the record is not used: those components hold their own config.

writer `property` ¶

writer: DataWriter | None

The source-provided writer, if any.

It is built from the pipeline's own config. When a run took the tick and lot size from a capture's record, build a writer from that run's config instead: source.create_writer(result.config, ctx).

from_source `classmethod` ¶

from_source(
    name: str,
    *,
    ctx: RunContext | None = None,
    **kwargs: Any,
) -> Pipeline

Create a pipeline from a registered source name.

Parameters:

Name	Type	Description	Default
`name`	`str`	Registered source name (case-insensitive), e.g. `"bitstamp"` or `"lobster"`.	required
`ctx`	`RunContext`	Per-run parameters (e.g. `trading_date`) forwarded to `OfflineSource.create_*` factories.	`None`
`**kwargs`	`Any`	Passed to the :class:`~ob_analytics.protocols.Source` constructor.	`{}`

run ¶

run(
    source: Any, *, ctx: RunContext | None = None
) -> PipelineResult

Execute the full pipeline on source and return results.

Parameters:

Name	Type	Description	Default
`source`	`Any`	Data source for the loader (typically a file path).	required
`ctx`	`RunContext`	Override the pipeline's default :class:`RunContext` for this single call. When `None`, the `ctx` provided at construction (or the default empty context) is used.	`None`

Returns:

Type	Description
`PipelineResult`	Frozen dataclass with `events`, `trades`, `depth`, `depth_summary`, `config`, and `level`.

Steps (L3 / per-order feeds)

Load events (EventLoader.load)
Build trades (TradeSource.load)
Classify order types
Compute price-level depth
Compute depth metrics
Compute order aggressiveness

For an :attr:~ob_analytics.protocols.Level.L2 source the run takes the price-level path instead (see :meth:_run_l2): the loader yields the depth frame directly, depth metrics and trade signs are computed on it, and the per-order stages (3, 6) are skipped.

run_windows ¶

run_windows(
    source: Any,
    boundaries: Iterable[Any],
    output: str | Path,
    *,
    carry: bool = True,
) -> Path

Run the pipeline one time window at a time and write the result.

Cuts source at each of the boundaries, runs the depth stages on one window at a time, and writes each window's rows to output as soon as the window is done. The depth stages are what makes a run large, so their peak memory is set by the largest window rather than by the whole input. The output is one Parquet file per table, the same folder a single run saved with :func:~ob_analytics.data.save_data gives, and :func:~ob_analytics.data.load_data reads it back.

With carry on, each window starts from the book the previous window ended with, so the cuts do not show in the output: every table matches a single run row for row, though events comes out in window order. One column can differ. aggressiveness_bps reads the quote standing before each order by event_id, so on a source whose event_id is not in time order (Bitstamp numbers its events by order) it can read a quote from another window, and the value differs from a single run's. With carry off, each window starts from an empty book, as if it were a separate input: the orders resting at a cut are missing from the next window's depth.

Parameters:

Name	Type	Description	Default
`source`	`Any`	Data source for the loader (typically a file path), as for :meth:`run`.	required
`boundaries`	`time or iterable of times`	Where to cut, anything :class:`pandas.Timestamp` accepts; one time on its own is one cut. A time with no zone is read as UTC. n cuts make n + 1 windows that cover the whole input: a window holds the rows at or after its start and before its end.	required
`output`	`str or Path`	The folder to write `events`, `trades`, `depth` and `depth_summary` to, created when missing. Files already there under those names are replaced, but only once every window is done: a run that fails part-way leaves the folder as it was.	required
`carry`	`bool`	Start each window from the previous window's book. Default `True`.	`True`

Returns:

Type	Description
`Path`	The output folder.

Raises:

Type	Description
`ConfigError`	If no boundary is given, if the boundaries do not strictly increase, or if no window produced rows for one of the tables.

Notes

The loader still reads the whole input, and the events and trades tables are held for the whole run: order types and trade signs are decided over every row, as in a single run. Those two tables are a small part of a run's memory; depth_summary alone is more than twice the size of events.

Where the loader already warns that the depth is off (a Databento modify that carries a fill and also moves the order), the carried order goes onto its new price level, so the windowed depth can differ from a single run's by the same amount.

A source's own depth is not used. LOBSTER's order book file states the book after each message of the whole session, so it cannot be cut; a windowed LOBSTER run rebuilds its depth from the messages instead.

PipelineResult `dataclass` ¶

PipelineResult(
    events: DataFrame,
    trades: DataFrame,
    depth: DataFrame,
    depth_summary: DataFrame,
    config: PipelineConfig,
    level: Level = Level.L3,
)

Immutable container for the core outputs of a pipeline run.

Analytic outputs (VPIN, OFI, Kyle's λ) are intentionally not stored here — compute them post-pipeline from trades and append them to the gallery model's analytics (build panels with the *_panel helpers).

Attributes:

Name	Type	Description
`events, trades, depth, depth_summary`	`DataFrame`	Core pipeline tables. For an :attr:`~ob_analytics.protocols.Level.L2` run `events` is empty (a schema-valid zero-row frame): a price-level feed has no per-order identity, so the per-order stages do not run — read `depth` / `depth_summary` / `trades` instead.
`config`	`PipelineConfig`	The configuration used for the run.
`level`	`Level`	The order-book resolution the run was produced at (:attr:`~ob_analytics.protocols.Level.L3` by default, :attr:`~ob_analytics.protocols.Level.L2` for price-level feeds). Downstream code (the gallery, data-quality) reads it to decide which per-order faces / metrics apply.

to_arrow ¶

to_arrow() -> dict[str, pa.Table]

Return the run's four core tables as Arrow tables.

Each table carries the same key-value metadata a canonical Parquet file carries — the schema version, the run's tick and lot size, and its config (see :mod:ob_analytics.schemas) — so a reader handed these tables in memory is no worse off than one reading the files.

Returns:

Type	Description
`dict of str to pyarrow.Table`	`events`, `trades`, `depth` and `depth_summary`.

Examples:

>>> from ob_analytics import Pipeline, sample_csv_path
>>> tables = Pipeline().run(sample_csv_path()).to_arrow()
>>> sorted(tables)
['depth', 'depth_summary', 'events', 'trades']

to_polars ¶

to_polars() -> dict[str, Any]

Return the run's four core tables as Polars DataFrames.

Polars is not a dependency of ob-analytics (see adr/0002-dataframe-library.md): the public API takes and returns pandas, and this accessor is a convenience for users who already have Polars installed. Install it yourself with pip install polars.

Returns:

Type	Description
`dict of str to polars.DataFrame`	The same keys as :meth:`to_arrow`.

Raises:

Type	Description
`ImportError`	When polars is not installed.

Notes

Polars keeps no schema-level key-value metadata, so the schema version and tick size that :meth:to_arrow attaches do not survive this conversion. Read the tick size from result.config.tick_size, or use :meth:to_arrow when the metadata has to travel with the tables.

metric ¶

metric(name: str, **settings: Any) -> pd.DataFrame

Compute the registered metric name over this result.

Metrics are computed on demand, not stored: a run pays for a metric only when it is asked for, and a third-party metric that raises cannot break :meth:Pipeline.run.

Parameters:

Name	Type	Description	Default
`name`	`str`	Registered metric name (case-insensitive), e.g. `"amihud"`. See :func:`~ob_analytics.metrics.list_metrics`.	required
`**settings`	`Any`	Keyword arguments for the metric's :meth:`~ob_analytics.protocols.Metric.compute`, e.g. `result.metric("vpin", n_buckets=20)`. Leave them out to use the metric's defaults.	`{}`

Returns:

Type	Description
`DataFrame`	The metric's own table, as its :meth:`~ob_analytics.protocols.Metric.compute` returns it.

Raises:

Type	Description
`KeyError`	If no metric is registered under name; the message lists the registered names.

metrics ¶

metrics() -> dict[str, pd.DataFrame]

Compute every registered metric that applies to this run.

A metric declares the resolutions it applies to ( :attr:~ob_analytics.protocols.Metric.levels), so an L3-only metric is skipped on an L2 run rather than failing on its empty events table.

Returns:

Type	Description
`dict of str to pandas.DataFrame`	Metric name → its table, for the metrics whose `levels` include this run's :attr:`level`.

plot ¶

plot(
    concept: str,
    level: Any = None,
    *,
    backend: str = "matplotlib",
    volume_scale: float | None = None,
    **overrides: Any,
) -> Any

Render one plot concept from this result in a single call.

Thin convenience wrapper over :func:ob_analytics.visualization.plot_result, e.g. result.plot("depth_heatmap", col_bias=0.1). See :func:~ob_analytics.visualization.available_concepts for what a given result can plot (it varies by format).

Pipeline¶

Pipeline ¶

writer property ¶

from_source classmethod ¶

run ¶

run_windows ¶

PipelineResult dataclass ¶

to_arrow ¶

to_polars ¶

metric ¶

metrics ¶

plot ¶

writer `property` ¶

from_source `classmethod` ¶

PipelineResult `dataclass` ¶