Architecture¶

Pipeline stages¶

ob-analytics turns order event streams and authoritative trade records into structured analytics:

Design decisions¶

• DataFrames end to end. Pandas for speed; the column-list constants and validate_* functions in schemas.py document the column contracts.
• The rebuild engine is separate, and holds no pandas. engine/ takes order events as numpy columns and returns book states, order lifecycles, and queue positions -- nothing else. It knows nothing about analytics, plots, sources, or files, and its results carry a row index back into the caller's event table instead of copying columns out, so adding a column to the schema never widens the interface. _engine_frames.py is the one place pandas and the engine meet. This is what lets the inside be replaced with a faster implementation, or fed one event at a time, without anything above it moving.
• Two API levels — Pipeline for one-line runs; individual classes (BitstampLoader, BitstampTradeReader, etc.) for step-by-step control.
• Pluggable everything — any object with the right method signature works; no inheritance required (structural typing via Protocol).
• One Source shape, file or live. Every data source states its level (L2/L3) and feed_type, carries typed settings, and implements one or both capabilities — OfflineSource (replay stored files) and LiveSource (capture a venue). Both register in the one SOURCES registry (register_source), and third-party sources load through the ob_analytics.sources entry-point group.
• Per-run parameters live on RunContext, not Source. Construction-time parameters (typed settings, fixed venue config) belong on the source; per-session parameters (trading date, write-time level cap) live on RunContext and are passed to Pipeline(source=..., ctx=...). This keeps sources reusable across multiple runs without re-instantiation and avoids baking session state into long-lived objects.

Scale envelope¶

ob-analytics keeps the full event, depth, and trade tables in memory (pandas), so the working set scales with the event count. Peak RSS grows roughly linearly at ~1 GiB per 1M events, with the depth stages (price_level_volume → depth_metrics) dominating both memory and time:

(Measured with scripts/bench_scale.py --envelope: the bundled ~314k-event sample tiled to each size, each size run in its own process, peak RSS via getrusage. Slightly conservative — tiling adds some transient overhead.)

Where the time goes¶

Per stage, on the bundled 314k-event sample (scripts/bench_scale.py, one Linux x86_64 machine — the shape holds, the seconds do not travel):

One numpy kernel is four fifths of the run. The stateful FIFO loop is 7%, and everything the loader and the frames do together is under 15% — which is the number ADR 0002 asks for before a second dataframe library could pay for itself, and it falls well short of the 30% gate that record sets. CI checks these numbers on every pull request (issue #144).

Comfortable ceiling ≈ 5M events (~5 GiB) on a typical 16 GB machine — i.e. session-scale data, a few hours of a single liquid instrument. For larger inputs (a full NASDAQ MBO day is 10–100M events, well past this), the recommended pattern is to pre-slice by time window and process each slice independently, concatenating the per-slice outputs. ob-analytics deliberately ships no streaming / out-of-core machinery; pre-slicing keeps the in-memory model simple and predictable (a chunked-run helper could automate it later if real workloads demand it).

Class diagram¶

The package combines protocol-based components with source descriptors that bundle venue-specific defaults and capabilities.

classDiagram
    class Pipeline {
        +config: PipelineConfig
        +loader: EventLoader
        +trade_source: TradeSource
        +writer: DataWriter | None
        +run(source) PipelineResult
        +from_source(name, **kwargs)$ Pipeline
    }

    class PipelineConfig {
        +price_decimals: int
        +volume_decimals: int
        +price_divisor: int
        +timestamp_unit: str
        +depth_bps: int
        +depth_bins: int
    }

    class Source {
        <<Protocol>>
        +name: str
        +level: Level
        +feed_type: FeedType
        +settings: SourceSettings
    }
    class OfflineSource {
        <<Protocol>>
        +create_loader(config, ctx) EventLoader
        +create_trade_source(config, ctx) TradeSource
        +create_writer(config, ctx) DataWriter | None
        +compute_depth(events, config, source, ctx) tuple | None
        +config_defaults() dict
    }
    class LiveSource {
        <<Protocol>>
        +snapshot(config) AsyncIterator
        +stream(config) AsyncIterator
        +shutdown_synthetic_events() AsyncIterator
    }

    class RunContext {
        +trading_date: object | None
    }

    class BitstampSource
    class LobsterSource

    class EventLoader {
        <<Protocol>>
        +load(source) DataFrame
    }
    class TradeSource {
        <<Protocol>>
        +load(events, source) DataFrame
    }
    class DataWriter {
        <<Protocol>>
        +write(data, dest, **kwargs)
    }

    class PipelineResult {
        +events: DataFrame
        +trades: DataFrame
        +depth: DataFrame
        +depth_summary: DataFrame
        +config: PipelineConfig
        +level: Level
    }

    Pipeline --> PipelineConfig
    Pipeline --> EventLoader
    Pipeline --> TradeSource
    Pipeline --> DataWriter
    Pipeline --> PipelineResult
    Pipeline ..> OfflineSource : source=
    Pipeline ..> RunContext : per-run params
    Source <|-- OfflineSource
    Source <|-- LiveSource
    OfflineSource <|.. BitstampSource
    LiveSource <|.. BitstampSource
    OfflineSource <|.. LobsterSource

Data formats¶

The bundled sample under ob_analytics/_sample_data/ is a modern BTC/USD capture (orders.csv + trades.csv).

L2 vs L3: the resolution axis¶

A source declares a level (Level.L2 / Level.L3), orthogonal to its FeedType crossing invariant:

• L3 (market-by-order) — the per-order model above. Events carry order IDs, so set_order_types, order_aggressiveness, and queue reconstruction run.
• L2 (market-by-price) — price-level feeds (Binance, Kalshi, Polymarket, most CCXT sources) publish [price, quantity] levels and diffs with no order IDs. The loader (a DepthSource) yields the depth frame directly and the pipeline skips the per-order stages: PipelineResult.events comes back empty (schema-valid) and PipelineResult.level is Level.L2. DepthMetricsEngine already consumes an absolute price-level book, so depth / spread / trade analytics run unchanged. See the Process L2 feeds how-to.

Module map¶

ob_analytics/
├── __init__.py           # Public API surface + source registration + sample_csv_path()
├── _sample_data/         # Bundled Bitstamp sample (orders.csv + trades.csv)
├── pipeline.py           # Pipeline, PipelineResult
├── sources.py            # SOURCES registry: register_source, list_sources, get_source, entry-point discovery
├── config.py             # PipelineConfig, SourceSettings (frozen Pydantic models)
├── protocols.py          # EventLoader, TradeSource, DataWriter, Source, OfflineSource
├── schemas.py            # column constants + validators (validate_events_df, …)
├── exceptions.py         # ObAnalyticsError hierarchy
├── cli.py                # CLI entry point (process, gallery, bitstamp-demo, lobster-demo, capture, sources)
│
├── bitstamp.py           # BitstampLoader, BitstampTradeReader, BitstampWriter, BitstampSource (offline + live)
├── lobster.py            # LobsterLoader, LobsterTradeReader, LobsterWriter, LobsterSource
├── depth_l2.py           # L2DepthLoader, L2TradeReader, DepthCsvWriter, DepthCsvSource (price-level)
├── engine/               # Order-book engine: events in, book states + lifecycles out
│   ├── __init__.py       # the interface: book_state, order_lifecycles, queue_positions, queue_age_grid
│   ├── _events.py        # OrderEvents -- the shared schema as numpy columns
│   ├── _book.py          # point-in-time rebuild + crossed-book eviction
│   ├── _lifecycles.py    # placement -> outcome, one row per order
│   └── _queue.py         # FIFO queue position and touch composition
├── _engine_frames.py     # pandas <-> engine arrays (the only place the two meet)
├── analytics.py          # order_aggressiveness, trade_impacts, set_order_types, order_book
├── queue.py              # frame faces for the engine's queue reconstruction
├── depth.py              # DepthMetricsEngine, price_level_volume, depth_metrics, get_spread
├── data.py               # save_data, load_data, writer registry
├── flow_toxicity.py      # compute_vpin, compute_kyle_lambda, order_flow_imbalance, KyleLambdaResult
├── _utils.py             # Validation, numerics, timestamp conversion helpers
│
├── live/                 # Optional live-capture machinery ([live] / [ccxt] / [cryptofeed] extras)
│   ├── __init__.py       # live-source registration (ccxt, cryptofeed) + re-exports
│   ├── _base.py          # LiveSource protocol, CaptureConfig, CaptureResult, CaptureSink
│   ├── _runner.py        # Generic asyncio driver + FileCaptureSink
│   ├── bitstamp.py       # Bitstamp WebSocket engine (driven by BitstampSource)
│   ├── ccxt_source.py    # CcxtSource, CcxtSettings (any CCXT venue, L2)
│   └── cryptofeed_source.py  # CryptofeedSource, CryptofeedSettings (L2 or native L3)
│
└── visualization/        # Plotting subsystem
    ├── __init__.py       # plot() dispatcher + RENDERERS registry, PlotTheme, save_figure
    ├── gallery.py        # HTML gallery generation
    ├── _data.py          # Shared data prep for plot backends
    ├── _matplotlib.py    # Matplotlib renderers
    └── _plotly.py        # Plotly renderers

Live capture is optional (install with pip install "ob-analytics[live]", [ccxt], or [cryptofeed]) and writes into the same CSV schema the pipeline reads. Give your Source the live capability -- the three LiveSource async-iterator methods (snapshot, stream, shutdown_synthetic_events) -- and register_source it; a source can add the offline factories too and do both. The runner (run_capturer) handles persistence, raw-frame archival, signal handling, and meta.json finalisation so source authors only write the per-venue parser.