Skip to content

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

writer property 

writer: DataWriter | None

The source-provided writer, if any.

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)
  1. Load events (EventLoader.load)
  2. Build trades (TradeSource.load)
  3. Classify order types
  4. Compute price-level depth
  5. Compute depth metrics
  6. 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.

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 and the run's tick size (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) -> 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

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).