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. The fields it sets are laid over the source's instrument grid (`source.grid`), and the result is what each run's loader and trade source are built from.	`None`
`source`	`OfflineSource`	A source descriptor that builds each run's loader and trade source, and gives the instrument grid the config starts from (its `grid`). 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, and is used for every run.	`None`
`trade_source`	`TradeSource`	Builds the trades DataFrame. Overrides the source's trade source, and is used for every run.	`None`
`ctx`	`RunContext`	The per-run parameters (such as LOBSTER's `trading_date`) every run uses unless :meth:`run` is given its own.	`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.

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.	`{}`

Raises:

Type	Description
`UnknownNameError`	If no source is registered under name; the message names the registered names closest to it.

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`, `level` and `declarations`.

Steps (L3 / per-order feeds)

Load events (EventLoader.load)
Build trades (TradeSource.load)
Classify order types
Compute price-level depth
Compute depth metrics, and sign the trades the venue left unlabelled
Compute order aggressiveness

Steps 3 to 6 are STAGES in this module. For an :attr:~ob_analytics.protocols.Level.L2 source 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,
    ctx: RunContext | None = None,
) -> 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`
`ctx`	`RunContext`	The context for this run, as for :meth:`run`.	`None`

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

A loader that can read one window at a time (a :class:~ob_analytics.databento.DatabentoLoader can) does so when the source built both the loader and the trade source, which then builds the trades from what the loader kept. The run then never holds more than one window's events, and goes over the windows twice. The first time it reads each window, keeps a few values per order and the trade records, and saves the window's events to a working folder inside output. Order types and trade signs are decided over the whole run, as in a single run, so the second time it reads each saved window back and runs the depth stages on it. carry does not change how such a loader reads: an order's size and price always carry from one window to the next, as they do when the input is read whole.

Any other loader reads the whole input, and the events and trades tables are held for the whole run. Those two tables are a small part of a run's memory; depth_summary alone is more than twice the size of events.

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,
    declarations: FeedDeclarations = FeedDeclarations(),
)

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.
`declarations`	`FeedDeclarations`	What the run's data is declared to show: its source's declarations, with what a live capture recorded in their place (see :func:`~ob_analytics.capture_record.run_declarations`). :meth:`audit` holds the run to them. The defaults declare nothing.

tables ¶

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

Return the run's four core tables keyed by name.

This is what :func:~ob_analytics.data.save_data saves for a result, and what :meth:to_arrow converts.

Returns:

Type	Description
`dict of str to pandas.DataFrame`	`events`, `trades`, `depth` and `depth_summary`. The keys are the same for every run: on an :attr:`~ob_analytics.protocols.Level.L2` run `events` is a schema-valid zero-row frame rather than a missing key.

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, its config and its declarations (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.

audit ¶

audit(
    *, venue_book: BookLevels | None = None
) -> DataQualitySummary

Check this run's data quality against its :attr:declarations.

The same checks as ob-analytics audit: this calls :func:~ob_analytics.analytics.data_quality_summary with the run's tables, its faithful depth, its tick size and its declarations.

Parameters:

Name	Type	Description	Default
`venue_book`	`BookLevels`	The venue's own book after every message, when the source supplies one (LOBSTER's orderbook file, read by :meth:`~ob_analytics.lobster.LobsterSource.venue_book`). The book rebuilt from the events is then compared with it.	`None`

Returns:

Type	Description
`DataQualitySummary`

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
`UnknownNameError`	If no metric is registered under name; the message names the registered names closest to it.

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) and the tables it needs (:attr:~ob_analytics.protocols.Metric.needs), so an L3-only metric is skipped on an L2 run, and a trade metric on a run with no trades, rather than failing on an empty table.

Returns:

Type	Description
`dict of str to pandas.DataFrame`	Metric name → its table, for the metrics that apply to this run (see :func:`~ob_analytics.metrics.skip_reason`).

plot ¶

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

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

The method form of :func:ob_analytics.visualization.plot_result, which takes the same arguments, e.g. result.plot("depth_heatmap", col_bias=0.1). Pass level when this result has the concept at both L2 and L3. See :func:~ob_analytics.visualization.available_concepts for what a given result can plot (it varies by format).

Pipeline¶

Pipeline ¶

from_source classmethod ¶

run ¶

run_windows ¶

PipelineResult dataclass ¶

tables ¶

to_arrow ¶

to_polars ¶

audit ¶

metric ¶

metrics ¶

plot ¶

from_source `classmethod` ¶

PipelineResult `dataclass` ¶