From 7f26fa185c8520249b76a55a54f7f2cc2027b09a Mon Sep 17 00:00:00 2001 From: Tibor Date: Tue, 28 Jul 2026 13:23:23 +0200 Subject: [PATCH] Add typed Binance L2 capture source --- AGENTS.md | 2 +- CHANGELOG.md | 4 + docs/architecture.md | 7 +- docs/connectors.md | 55 ++- docs/data-guide.md | 16 +- docs/schema.md | 28 ++ src/ordersim/connectors/binance/__init__.py | 24 ++ src/ordersim/connectors/binance/_parsing.py | 252 ++++++++++++ src/ordersim/connectors/binance/l2.py | 115 ++++++ src/ordersim/connectors/binance/source.py | 159 ++++++++ tests/test_binance_l2_source.py | 423 ++++++++++++++++++++ 11 files changed, 1077 insertions(+), 8 deletions(-) create mode 100644 src/ordersim/connectors/binance/_parsing.py create mode 100644 src/ordersim/connectors/binance/l2.py create mode 100644 src/ordersim/connectors/binance/source.py create mode 100644 tests/test_binance_l2_source.py diff --git a/AGENTS.md b/AGENTS.md index 14f2bf7..df652f0 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -38,7 +38,7 @@ extraction targets and should not be imported until they exist. | `ordersim/connectors/csv.py` | Normalized CSV `MBOEvent` source | Yes | | `ordersim/connectors/databento.py` | Databento MBO normalization | Yes | | `ordersim/connectors/parquet.py` | Normalized Parquet `MBOEvent` source | Yes | -| `ordersim/connectors/binance/` | Binance L2 capture and integrity boundary | Capture API only | +| `ordersim/connectors/binance/` | Binance capture and typed L2 source; not MBO | Public venue API | | `ordersim/latency.py` | Latency model contracts and reference models | Yes | | `ordersim/replay/simulator.py` | Replay orchestration and `run_many` | Yes | | `ordersim/testing/` | Public helpers for extension tests | Public | diff --git a/CHANGELOG.md b/CHANGELOG.md index c2be564..01a8053 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,10 @@ All notable public changes to `ordersim` are documented here. aggregate-trade, book-ticker, snapshot, and RPI evidence. - Added connection manifests and explicit diff-depth sequence-gap records, while keeping lower-fidelity capture separate from modeled MBO replay. +- Added a typed, streaming reader for completed Binance captures with exact + depth, aggregate-trade, and book-ticker records. +- Added snapshot bridging and `pu`/`u` continuity validation for standard + Binance diff-depth segments. ## 0.1.3 - 2026-05-20 diff --git a/docs/architecture.md b/docs/architecture.md index 202eb55..5d5f7b4 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -69,19 +69,22 @@ Lower-fidelity venue data takes a longer, explicit path: flowchart LR venue["Venue L2 + trades"] capture["Raw capture"] + source["Typed L2 source"] model["Named reconstruction model"] modeled["Modeled MBOEvent stream"] parquet["Canonical Parquet + model manifest"] replay["Replay"] - venue --> capture --> model --> modeled --> parquet --> replay + venue --> capture --> source --> model --> modeled --> parquet --> replay ``` Capture code may live beside connectors because it owns venue I/O and source schemas. Capture alone is not a `DataSource`: observed L2 rows must not be presented as exchange-native MBO. The reconstruction model owns that lower-fidelity assumption and must preserve a manifest describing how its -events were inferred. +events were inferred. For Binance, `BinanceCaptureSource` is the typed, +sequence-validated L2 boundary between the raw evidence and that future model; +it does not implement the canonical MBO `DataSource` protocol. ## One Replay Run diff --git a/docs/connectors.md b/docs/connectors.md index 47bc0ca..d8e45a2 100644 --- a/docs/connectors.md +++ b/docs/connectors.md @@ -111,13 +111,62 @@ connection segment and obtains a new REST snapshot. These files are intentionally not canonical replay data. Binance depth has no stable public order IDs, and individual additions and cancellations inside an -update window are not observable. A future named L2-to-virtual-L3 model will -consume the capture, document the inference policy, and only then emit modeled -`MBOEvent` rows. +update window are not observable. A named L2-to-virtual-L3 model must consume +the typed capture records, document the inference policy, and only then emit +modeled `MBOEvent` rows. Capture files are local research data and must not be committed to the repository. +### Reading Completed Captures + +`BinanceCaptureSource` streams completed gzip capture files directly, so a +multi-day capture does not need to be loaded into memory: + +```python +from ordersim.connectors.binance import BinanceCaptureSource + +source = BinanceCaptureSource.from_manifest( + "captures/binance/manifest-20260728T105500Z.json" +) + +for event in source.validated_depth_events(): + print(event) + +for trade in source.aggregate_trades(): + print(trade) +``` + +This is a typed Binance source, not the canonical `DataSource` protocol. It +emits `BinanceDepthSnapshot`, `BinanceDepthUpdate`, +`BinanceAggregateTrade`, and `BinanceBookTicker` records rather than +`MBOEvent`. Passing it directly to `Replay` is intentionally unsupported. + +The reader preserves prices and quantities as exact `Decimal` values. Binance +exchange event and transaction timestamps (`E` and `T`) are milliseconds since +the Unix epoch and are normalized to UTC nanoseconds. Local UTC receive and +monotonic receive timestamps from the capture envelope remain available +separately. + +`validated_depth_events()` applies Binance's USD-M synchronization rules per +connection: + +1. Require a REST snapshot before standard diff-depth updates. +2. Discard buffered updates where `u < lastUpdateId`. +3. Require the first retained update to satisfy + `U <= lastUpdateId <= u`. +4. Require each later update's `pu` to equal the preceding update's `u`. + +A broken segment raises `BinanceSequenceError`; it is never repaired silently. +Depth quantities are absolute, and a zero quantity means remove that price +level. RPI depth is available separately through +`depth_updates(stream_kind="rpi_depth")`; it is not merged into standard depth +by the source. + +Aggregate trades preserve Binance's optional `nq` field as +`normal_quantity`. When present, it is the quantity excluding trades involving +RPI orders. When absent, `normal_quantity` is `None`, not an inferred value. + For the user-facing decision guide, see `docs/data-guide.md`. ## In-Memory Sources diff --git a/docs/data-guide.md b/docs/data-guide.md index 36ce41f..23cbd7e 100644 --- a/docs/data-guide.md +++ b/docs/data-guide.md @@ -104,8 +104,20 @@ the connector and decide whether the connector is valid for the research task. Binance USD-M depth is one such lower-fidelity source. The Binance capture tool records raw L2 depth, aggregate trades, and integrity metadata, but its output -is not accepted by `Replay` as observed MBO. See `docs/connectors.md` for the -capture boundary and the planned modeled reconstruction path. +is not accepted by `Replay` as observed MBO. + +After a capture completes, use `BinanceCaptureSource` to stream exact typed +snapshots, sequence-validated depth updates, aggregate trades, and book +tickers. That typed source is the input boundary for the planned named +virtual-L3 reconstruction model: + +```text +raw capture -> BinanceCaptureSource -> named model -> modeled MBO + manifest +``` + +There is deliberately no direct +`BinanceCaptureSource -> Replay` path. See `docs/connectors.md` for the capture +and validation contract. ## Related Docs diff --git a/docs/schema.md b/docs/schema.md index 3b11692..49f8b10 100644 --- a/docs/schema.md +++ b/docs/schema.md @@ -17,6 +17,34 @@ before replay sees the event. The replay layer operates only on normalized integer timestamps. It does not carry Python timezone objects or infer timezone rules after normalization. +## Binance L2 Records + +The Binance connector exposes typed records before the modeled-MBO boundary. +These records describe observed aggregated depth and trades; they do not claim +to contain stable exchange order IDs. + +`BinancePriceLevel` holds an exact positive `Decimal` price and a non-negative +`Decimal` quantity. In a depth update, the quantity is the new absolute +quantity at that price; zero means remove the level. + +| Record | Important fields | Meaning | +|---|---|---| +| `BinanceDepthSnapshot` | `last_update_id`, `bids`, `asks` | REST depth state anchoring one connection. | +| `BinanceDepthUpdate` | `first_update_id`, `final_update_id`, `previous_update_id`, `bids`, `asks` | One standard or RPI absolute-quantity diff-depth message. | +| `BinanceAggregateTrade` | `aggregate_trade_id`, `price`, `quantity`, `normal_quantity`, `buyer_is_maker` | Trades aggregated by price and taking side. | +| `BinanceBookTicker` | `update_id`, bid and ask price/quantity | Real-time best bid and ask observation. | + +All records include `symbol`, `connection_id`, UTC receive nanoseconds, and +local monotonic receive nanoseconds. Stream messages also include exchange +event and transaction/trade times normalized from Binance milliseconds to UTC +nanoseconds. + +Binance contract quantity can be fractional, so the connector preserves it as +`Decimal`. A future virtual-L3 reconstruction model must declare its quantity +unit and exact conversion rule before producing the canonical integer +`MBOEvent.size`. These L2 records are therefore not accepted directly by +`Replay`. + ## `MBOEvent` `MBOEvent` represents one Level 3 / market-by-order event. diff --git a/src/ordersim/connectors/binance/__init__.py b/src/ordersim/connectors/binance/__init__.py index 0e1e345..9c8d3fe 100644 --- a/src/ordersim/connectors/binance/__init__.py +++ b/src/ordersim/connectors/binance/__init__.py @@ -5,10 +5,34 @@ """ from ordersim.connectors.binance.capture import capture_binance +from ordersim.connectors.binance.l2 import ( + BinanceAggregateTrade, + BinanceBookTicker, + BinanceCaptureEnvelope, + BinanceDepthEvent, + BinanceDepthSnapshot, + BinanceDepthUpdate, + BinancePriceLevel, + DepthStreamKind, +) from ordersim.connectors.binance.schema import BinanceCaptureConfig, CaptureManifest +from ordersim.connectors.binance.source import ( + BinanceCaptureSource, + BinanceSequenceError, +) __all__ = [ + "BinanceAggregateTrade", + "BinanceBookTicker", "BinanceCaptureConfig", + "BinanceCaptureEnvelope", + "BinanceCaptureSource", + "BinanceDepthEvent", + "BinanceDepthSnapshot", + "BinanceDepthUpdate", + "BinancePriceLevel", + "BinanceSequenceError", "CaptureManifest", + "DepthStreamKind", "capture_binance", ] diff --git a/src/ordersim/connectors/binance/_parsing.py b/src/ordersim/connectors/binance/_parsing.py new file mode 100644 index 0000000..1e4749a --- /dev/null +++ b/src/ordersim/connectors/binance/_parsing.py @@ -0,0 +1,252 @@ +"""Mechanical conversion from Binance JSON fields to typed records.""" + +from collections.abc import Mapping +from decimal import Decimal, InvalidOperation +from typing import cast + +from ordersim.connectors.binance.l2 import ( + BinanceAggregateTrade, + BinanceBookTicker, + BinanceCaptureEnvelope, + BinanceDepthSnapshot, + BinanceDepthUpdate, + BinancePriceLevel, + CaptureKind, + CaptureScope, + DepthStreamKind, +) +from ordersim.connectors.binance.schema import CAPTURE_SCHEMA_VERSION + + +def parse_envelope(raw: object) -> BinanceCaptureEnvelope: + """Convert one decoded JSON value into a validated capture envelope.""" + + if not isinstance(raw, dict): + raise ValueError("capture row must be a JSON object") + version = required_int(raw, "schema_version") + if version != CAPTURE_SCHEMA_VERSION: + raise ValueError(f"unsupported capture schema_version {version}") + kind = required_choice( + raw, + "kind", + ( + "connection_open", + "connection_error", + "depth_snapshot", + "message", + "sequence_gap", + ), + ) + scope = required_choice(raw, "scope", ("public", "market")) + stream = raw.get("stream") + if stream is not None and not isinstance(stream, str): + raise ValueError("capture field 'stream' must be a string or null") + payload = raw.get("payload") + if not isinstance(payload, dict): + raise ValueError("capture field 'payload' must be an object") + return BinanceCaptureEnvelope( + schema_version=version, + received_at_ns=required_int(raw, "received_at_ns"), + received_monotonic_ns=required_int(raw, "received_monotonic_ns"), + kind=cast(CaptureKind, kind), + scope=cast(CaptureScope, scope), + symbol=required_str(raw, "symbol"), + connection_id=required_str(raw, "connection_id"), + stream=stream, + payload=payload, + ) + + +def parse_depth_snapshot( + envelope: BinanceCaptureEnvelope, +) -> BinanceDepthSnapshot: + """Normalize one captured REST depth snapshot.""" + + return BinanceDepthSnapshot( + symbol=envelope.symbol, + connection_id=envelope.connection_id, + received_at_ns=envelope.received_at_ns, + received_monotonic_ns=envelope.received_monotonic_ns, + last_update_id=required_int(envelope.payload, "lastUpdateId"), + bids=price_levels(envelope.payload, "bids"), + asks=price_levels(envelope.payload, "asks"), + ) + + +def parse_depth_update( + envelope: BinanceCaptureEnvelope, + *, + stream_kind: DepthStreamKind, +) -> BinanceDepthUpdate: + """Normalize one captured standard or RPI depth update.""" + + payload = envelope.payload + check_payload_symbol(envelope) + return BinanceDepthUpdate( + symbol=envelope.symbol, + connection_id=envelope.connection_id, + stream_kind=stream_kind, + event_time_ns=milliseconds_to_nanoseconds(payload, "E"), + transaction_time_ns=milliseconds_to_nanoseconds(payload, "T"), + received_at_ns=envelope.received_at_ns, + received_monotonic_ns=envelope.received_monotonic_ns, + first_update_id=required_int(payload, "U"), + final_update_id=required_int(payload, "u"), + previous_update_id=required_int(payload, "pu"), + bids=price_levels(payload, "b"), + asks=price_levels(payload, "a"), + ) + + +def parse_aggregate_trade( + envelope: BinanceCaptureEnvelope, +) -> BinanceAggregateTrade: + """Normalize one captured aggregate trade.""" + + payload = envelope.payload + check_payload_symbol(envelope) + normal_quantity = ( + required_decimal(payload, "nq") if payload.get("nq") is not None else None + ) + return BinanceAggregateTrade( + symbol=envelope.symbol, + connection_id=envelope.connection_id, + event_time_ns=milliseconds_to_nanoseconds(payload, "E"), + trade_time_ns=milliseconds_to_nanoseconds(payload, "T"), + received_at_ns=envelope.received_at_ns, + received_monotonic_ns=envelope.received_monotonic_ns, + aggregate_trade_id=required_int(payload, "a"), + price=required_decimal(payload, "p"), + quantity=required_decimal(payload, "q"), + normal_quantity=normal_quantity, + first_trade_id=required_int(payload, "f"), + last_trade_id=required_int(payload, "l"), + buyer_is_maker=required_bool(payload, "m"), + ) + + +def parse_book_ticker(envelope: BinanceCaptureEnvelope) -> BinanceBookTicker: + """Normalize one captured real-time book ticker.""" + + payload = envelope.payload + check_payload_symbol(envelope) + return BinanceBookTicker( + symbol=envelope.symbol, + connection_id=envelope.connection_id, + event_time_ns=milliseconds_to_nanoseconds(payload, "E"), + transaction_time_ns=milliseconds_to_nanoseconds(payload, "T"), + received_at_ns=envelope.received_at_ns, + received_monotonic_ns=envelope.received_monotonic_ns, + update_id=required_int(payload, "u"), + bid_price=required_decimal(payload, "b"), + bid_quantity=required_decimal(payload, "B"), + ask_price=required_decimal(payload, "a"), + ask_quantity=required_decimal(payload, "A"), + ) + + +def depth_stream_kind( + envelope: BinanceCaptureEnvelope, +) -> DepthStreamKind | None: + """Identify a standard or RPI depth message.""" + + if envelope.kind != "message" or envelope.stream is None: + return None + if "@rpiDepth@" in envelope.stream: + return "rpi_depth" + if "@depth@" in envelope.stream: + return "depth" + return None + + +def is_stream(envelope: BinanceCaptureEnvelope, suffix: str) -> bool: + """Return whether an envelope is a message for the requested stream.""" + + return ( + envelope.kind == "message" + and envelope.stream is not None + and envelope.stream.endswith(suffix) + ) + + +def check_payload_symbol(envelope: BinanceCaptureEnvelope) -> None: + payload_symbol = required_str(envelope.payload, "s") + if payload_symbol != envelope.symbol: + raise ValueError( + f"payload symbol {payload_symbol!r} does not match " + f"envelope symbol {envelope.symbol!r}" + ) + + +def price_levels( + payload: Mapping[str, object], + field: str, +) -> tuple[BinancePriceLevel, ...]: + rows = payload.get(field) + if not isinstance(rows, list): + raise ValueError(f"Binance field {field!r} must be a list") + levels: list[BinancePriceLevel] = [] + for row in rows: + if not isinstance(row, list) or len(row) < 2: + raise ValueError(f"Binance field {field!r} has an invalid level") + levels.append( + BinancePriceLevel( + price=decimal_value(row[0], field), + quantity=decimal_value(row[1], field), + ) + ) + return tuple(levels) + + +def milliseconds_to_nanoseconds( + payload: Mapping[str, object], + field: str, +) -> int: + return required_int(payload, field) * 1_000_000 + + +def required_int(payload: Mapping[str, object], field: str) -> int: + value = payload.get(field) + if isinstance(value, bool) or not isinstance(value, int): + raise ValueError(f"Binance field {field!r} must be an integer") + return value + + +def required_str(payload: Mapping[str, object], field: str) -> str: + value = payload.get(field) + if not isinstance(value, str) or not value: + raise ValueError(f"Binance field {field!r} must be a non-empty string") + return value + + +def required_bool(payload: Mapping[str, object], field: str) -> bool: + value = payload.get(field) + if not isinstance(value, bool): + raise ValueError(f"Binance field {field!r} must be a boolean") + return value + + +def required_decimal(payload: Mapping[str, object], field: str) -> Decimal: + return decimal_value(payload.get(field), field) + + +def decimal_value(value: object, field: str) -> Decimal: + if not isinstance(value, str): + raise ValueError(f"Binance field {field!r} must contain decimal strings") + try: + return Decimal(value) + except InvalidOperation as exc: + raise ValueError( + f"Binance field {field!r} contains an invalid decimal" + ) from exc + + +def required_choice( + payload: Mapping[str, object], + field: str, + choices: tuple[str, ...], +) -> str: + value = required_str(payload, field) + if value not in choices: + raise ValueError(f"Binance field {field!r} has unsupported value {value!r}") + return value diff --git a/src/ordersim/connectors/binance/l2.py b/src/ordersim/connectors/binance/l2.py new file mode 100644 index 0000000..14f6713 --- /dev/null +++ b/src/ordersim/connectors/binance/l2.py @@ -0,0 +1,115 @@ +"""Typed Binance Level 2 market-data records.""" + +from collections.abc import Mapping +from dataclasses import dataclass +from decimal import Decimal +from typing import Literal, TypeAlias + +DepthStreamKind: TypeAlias = Literal["depth", "rpi_depth"] +CaptureKind: TypeAlias = Literal[ + "connection_open", + "connection_error", + "depth_snapshot", + "message", + "sequence_gap", +] +CaptureScope: TypeAlias = Literal["public", "market"] + + +@dataclass(frozen=True, slots=True) +class BinanceCaptureEnvelope: + """One raw capture row with local receive metadata.""" + + schema_version: int + received_at_ns: int + received_monotonic_ns: int + kind: CaptureKind + scope: CaptureScope + symbol: str + connection_id: str + stream: str | None + payload: Mapping[str, object] + + +@dataclass(frozen=True, slots=True) +class BinancePriceLevel: + """One absolute price and quantity pair from Binance.""" + + price: Decimal + quantity: Decimal + + def __post_init__(self) -> None: + if self.price <= 0: + raise ValueError("price must be positive") + if self.quantity < 0: + raise ValueError("quantity must be non-negative") + + +@dataclass(frozen=True, slots=True) +class BinanceDepthSnapshot: + """One REST snapshot anchoring a diff-depth connection.""" + + symbol: str + connection_id: str + received_at_ns: int + received_monotonic_ns: int + last_update_id: int + bids: tuple[BinancePriceLevel, ...] + asks: tuple[BinancePriceLevel, ...] + + +@dataclass(frozen=True, slots=True) +class BinanceDepthUpdate: + """One absolute-quantity diff-depth update.""" + + symbol: str + connection_id: str + stream_kind: DepthStreamKind + event_time_ns: int + transaction_time_ns: int + received_at_ns: int + received_monotonic_ns: int + first_update_id: int + final_update_id: int + previous_update_id: int + bids: tuple[BinancePriceLevel, ...] + asks: tuple[BinancePriceLevel, ...] + + +@dataclass(frozen=True, slots=True) +class BinanceAggregateTrade: + """Trades aggregated by Binance over price and taking side.""" + + symbol: str + connection_id: str + event_time_ns: int + trade_time_ns: int + received_at_ns: int + received_monotonic_ns: int + aggregate_trade_id: int + price: Decimal + quantity: Decimal + normal_quantity: Decimal | None + first_trade_id: int + last_trade_id: int + buyer_is_maker: bool + + +@dataclass(frozen=True, slots=True) +class BinanceBookTicker: + """One real-time Binance best-bid/ask observation.""" + + symbol: str + connection_id: str + event_time_ns: int + transaction_time_ns: int + received_at_ns: int + received_monotonic_ns: int + update_id: int + bid_price: Decimal + bid_quantity: Decimal + ask_price: Decimal + ask_quantity: Decimal + + +BinanceDepthEvent: TypeAlias = BinanceDepthSnapshot | BinanceDepthUpdate diff --git a/src/ordersim/connectors/binance/source.py b/src/ordersim/connectors/binance/source.py new file mode 100644 index 0000000..43dc9b0 --- /dev/null +++ b/src/ordersim/connectors/binance/source.py @@ -0,0 +1,159 @@ +"""Stream typed records from completed raw Binance captures.""" + +import gzip +import json +from collections.abc import Iterator +from dataclasses import dataclass +from pathlib import Path + +from ordersim.connectors.binance._parsing import ( + depth_stream_kind, + is_stream, + parse_aggregate_trade, + parse_book_ticker, + parse_depth_snapshot, + parse_depth_update, + parse_envelope, + required_int, +) +from ordersim.connectors.binance.l2 import ( + BinanceAggregateTrade, + BinanceBookTicker, + BinanceCaptureEnvelope, + BinanceDepthEvent, + BinanceDepthSnapshot, + BinanceDepthUpdate, + DepthStreamKind, +) +from ordersim.connectors.binance.schema import CAPTURE_SCHEMA_VERSION + + +class BinanceSequenceError(ValueError): + """Raised when a captured depth segment cannot be replayed continuously.""" + + +@dataclass(frozen=True, slots=True) +class BinanceCaptureSource: + """Stream normalized records from completed gzip capture files.""" + + files: tuple[Path, ...] + + def __post_init__(self) -> None: + paths = tuple(Path(path) for path in self.files) + if not paths: + raise ValueError("at least one capture file is required") + missing = [str(path) for path in paths if not path.is_file()] + if missing: + raise FileNotFoundError(f"capture files do not exist: {missing}") + object.__setattr__(self, "files", paths) + + @classmethod + def from_manifest(cls, manifest_path: str | Path) -> "BinanceCaptureSource": + """Build a source from one completed capture manifest.""" + + path = Path(manifest_path) + raw = json.loads(path.read_text(encoding="utf-8")) + if not isinstance(raw, dict): + raise ValueError("capture manifest must be a JSON object") + if required_int(raw, "schema_version") != CAPTURE_SCHEMA_VERSION: + raise ValueError("unsupported capture manifest schema_version") + names = raw.get("files") + if not isinstance(names, list) or not all( + isinstance(name, str) for name in names + ): + raise ValueError("capture manifest files must be a list of names") + return cls(tuple(path.parent / name for name in names)) + + def envelopes(self) -> Iterator[BinanceCaptureEnvelope]: + """Yield validated raw envelopes in capture-file order.""" + + for path in self.files: + with gzip.open(path, mode="rt", encoding="utf-8") as rows: + for line_number, line in enumerate(rows, start=1): + try: + raw = json.loads(line) + yield parse_envelope(raw) + except (TypeError, ValueError) as exc: + raise ValueError( + f"invalid capture row {path}:{line_number}: {exc}" + ) from exc + + def depth_snapshots(self) -> Iterator[BinanceDepthSnapshot]: + """Yield every captured REST depth snapshot.""" + + for envelope in self.envelopes(): + if envelope.kind == "depth_snapshot": + yield parse_depth_snapshot(envelope) + + def depth_updates( + self, + *, + stream_kind: DepthStreamKind = "depth", + ) -> Iterator[BinanceDepthUpdate]: + """Yield raw standard or RPI diff-depth updates.""" + + for envelope in self.envelopes(): + kind = depth_stream_kind(envelope) + if kind == stream_kind: + yield parse_depth_update(envelope, stream_kind=kind) + + def validated_depth_events(self) -> Iterator[BinanceDepthEvent]: + """Yield snapshot-anchored, sequence-continuous standard depth.""" + + snapshots: dict[str, BinanceDepthSnapshot] = {} + started: set[str] = set() + previous_update_ids: dict[str, int] = {} + + for envelope in self.envelopes(): + if envelope.kind == "depth_snapshot": + snapshot = parse_depth_snapshot(envelope) + snapshots[envelope.connection_id] = snapshot + yield snapshot + continue + if depth_stream_kind(envelope) != "depth": + continue + + update = parse_depth_update(envelope, stream_kind="depth") + connection_id = envelope.connection_id + snapshot = snapshots.get(connection_id) + if snapshot is None: + raise BinanceSequenceError( + f"depth update before snapshot for connection {connection_id}" + ) + if connection_id not in started: + if update.final_update_id < snapshot.last_update_id: + continue + if not ( + update.first_update_id + <= snapshot.last_update_id + <= update.final_update_id + ): + raise BinanceSequenceError( + "first depth update does not bridge snapshot " + f"{snapshot.last_update_id}: " + f"U={update.first_update_id}, u={update.final_update_id}" + ) + started.add(connection_id) + else: + expected = previous_update_ids[connection_id] + if update.previous_update_id != expected: + raise BinanceSequenceError( + f"depth sequence gap for connection {connection_id}: " + f"expected pu={expected}, got {update.previous_update_id}" + ) + previous_update_ids[connection_id] = update.final_update_id + yield update + + def aggregate_trades(self) -> Iterator[BinanceAggregateTrade]: + """Yield normalized aggregate-trade messages.""" + + for envelope in self.envelopes(): + if is_stream(envelope, "@aggTrade"): + yield parse_aggregate_trade(envelope) + + def book_tickers(self) -> Iterator[BinanceBookTicker]: + """Yield normalized best-bid/ask messages.""" + + for envelope in self.envelopes(): + if is_stream(envelope, "@bookTicker"): + yield parse_book_ticker(envelope) diff --git a/tests/test_binance_l2_source.py b/tests/test_binance_l2_source.py new file mode 100644 index 0000000..4aac109 --- /dev/null +++ b/tests/test_binance_l2_source.py @@ -0,0 +1,423 @@ +import gzip +import json +from decimal import Decimal +from pathlib import Path + +import pytest + +from ordersim.connectors.binance import ( + BinanceCaptureSource, + BinanceDepthSnapshot, + BinanceDepthUpdate, + BinancePriceLevel, + BinanceSequenceError, +) + + +def capture_row( + *, + kind: str, + scope: str = "public", + symbol: str = "BTCUSDT", + connection_id: str = "public-1", + stream: str | None = None, + payload: object, + received_at_ns: int = 1_000, +) -> dict[str, object]: + return { + "schema_version": 1, + "received_at_ns": received_at_ns, + "received_monotonic_ns": received_at_ns + 100, + "kind": kind, + "scope": scope, + "symbol": symbol, + "connection_id": connection_id, + "stream": stream, + "payload": payload, + } + + +def standard_rows() -> list[dict[str, object]]: + return [ + capture_row( + kind="connection_open", + payload={"streams": ["btcusdt@depth@100ms"]}, + ), + capture_row( + kind="depth_snapshot", + payload={ + "lastUpdateId": 100, + "bids": [["100.10", "2.500"]], + "asks": [["100.20", "3.750"]], + }, + received_at_ns=2_000, + ), + capture_row( + kind="message", + stream="btcusdt@depth@100ms", + payload={ + "e": "depthUpdate", + "E": 10, + "T": 9, + "s": "BTCUSDT", + "U": 90, + "u": 99, + "pu": 89, + "b": [["100.10", "2.000"]], + "a": [], + }, + received_at_ns=3_000, + ), + capture_row( + kind="message", + stream="btcusdt@depth@100ms", + payload={ + "e": "depthUpdate", + "E": 11, + "T": 10, + "s": "BTCUSDT", + "U": 99, + "u": 101, + "pu": 99, + "b": [["100.10", "1.500"], ["100.00", "0"]], + "a": [["100.20", "3.000"]], + }, + received_at_ns=4_000, + ), + capture_row( + kind="message", + stream="btcusdt@depth@100ms", + payload={ + "e": "depthUpdate", + "E": 12, + "T": 11, + "s": "BTCUSDT", + "U": 102, + "u": 103, + "pu": 101, + "b": [], + "a": [["100.20", "2.250"]], + }, + received_at_ns=5_000, + ), + capture_row( + kind="message", + stream="btcusdt@bookTicker", + payload={ + "e": "bookTicker", + "E": 12, + "T": 11, + "s": "BTCUSDT", + "u": 103, + "b": "100.10", + "B": "1.500", + "a": "100.20", + "A": "2.250", + }, + received_at_ns=5_100, + ), + capture_row( + kind="message", + stream="btcusdt@rpiDepth@500ms", + payload={ + "e": "depthUpdate", + "E": 12, + "T": 11, + "s": "BTCUSDT", + "U": 200, + "u": 201, + "pu": 199, + "b": [["100.10", "0.250"]], + "a": [], + }, + received_at_ns=5_200, + ), + capture_row( + kind="connection_open", + scope="market", + connection_id="market-1", + payload={"streams": ["btcusdt@aggTrade"]}, + ), + capture_row( + kind="message", + scope="market", + connection_id="market-1", + stream="btcusdt@aggTrade", + payload={ + "e": "aggTrade", + "E": 13, + "T": 12, + "s": "BTCUSDT", + "a": 500, + "p": "100.20", + "q": "1.250", + "nq": "1.000", + "f": 700, + "l": 702, + "m": True, + }, + received_at_ns=6_000, + ), + ] + + +def write_capture( + tmp_path: Path, + rows: list[object], + *, + with_manifest: bool = False, +) -> tuple[Path, Path | None]: + capture_path = tmp_path / "binance-run-20260728T100000Z.jsonl.gz" + with gzip.open(capture_path, mode="wt", encoding="utf-8") as file: + for row in rows: + file.write(json.dumps(row, separators=(",", ":")) + "\n") + + if not with_manifest: + return capture_path, None + manifest_path = tmp_path / "manifest-run.json" + manifest_path.write_text( + json.dumps( + { + "schema_version": 1, + "run_id": "run", + "files": [capture_path.name], + } + ), + encoding="utf-8", + ) + return capture_path, manifest_path + + +def test_source_normalizes_exact_depth_trade_and_ticker_records( + tmp_path: Path, +) -> None: + capture_path, _ = write_capture(tmp_path, standard_rows()) + source = BinanceCaptureSource((capture_path,)) + + snapshots = tuple(source.depth_snapshots()) + depth = tuple(source.depth_updates()) + rpi_depth = tuple(source.depth_updates(stream_kind="rpi_depth")) + trades = tuple(source.aggregate_trades()) + tickers = tuple(source.book_tickers()) + + assert snapshots == ( + BinanceDepthSnapshot( + symbol="BTCUSDT", + connection_id="public-1", + received_at_ns=2_000, + received_monotonic_ns=2_100, + last_update_id=100, + bids=( + BinancePriceLevel( + price=Decimal("100.10"), + quantity=Decimal("2.500"), + ), + ), + asks=( + BinancePriceLevel( + price=Decimal("100.20"), + quantity=Decimal("3.750"), + ), + ), + ), + ) + assert len(depth) == 3 + assert depth[1].event_time_ns == 11_000_000 + assert depth[1].transaction_time_ns == 10_000_000 + assert depth[1].bids[1].quantity == Decimal("0") + assert rpi_depth[0].stream_kind == "rpi_depth" + assert trades[0].price == Decimal("100.20") + assert trades[0].quantity == Decimal("1.250") + assert trades[0].normal_quantity == Decimal("1.000") + assert trades[0].buyer_is_maker is True + assert tickers[0].bid_price == Decimal("100.10") + assert tickers[0].ask_quantity == Decimal("2.250") + + +def test_validated_depth_events_drop_stale_updates_and_bridge_snapshot( + tmp_path: Path, +) -> None: + capture_path, _ = write_capture(tmp_path, standard_rows()) + source = BinanceCaptureSource((capture_path,)) + + events = tuple(source.validated_depth_events()) + + assert isinstance(events[0], BinanceDepthSnapshot) + assert [event.final_update_id for event in events[1:]] == [101, 103] + assert all(isinstance(event, BinanceDepthUpdate) for event in events[1:]) + + +def test_source_can_be_built_from_completed_manifest(tmp_path: Path) -> None: + capture_path, manifest_path = write_capture( + tmp_path, + standard_rows(), + with_manifest=True, + ) + + source = BinanceCaptureSource.from_manifest(manifest_path) + + assert source.files == (capture_path,) + assert len(tuple(source.envelopes())) == len(standard_rows()) + + +def test_aggregate_trade_preserves_missing_normal_quantity(tmp_path: Path) -> None: + rows = standard_rows() + trade_payload = rows[-1]["payload"] + assert isinstance(trade_payload, dict) + trade_payload.pop("nq") + capture_path, _ = write_capture(tmp_path, rows) + + trade = next(BinanceCaptureSource((capture_path,)).aggregate_trades()) + + assert trade.normal_quantity is None + + +def test_validated_depth_rejects_gap_after_bridge(tmp_path: Path) -> None: + rows = standard_rows() + payload = rows[4]["payload"] + assert isinstance(payload, dict) + payload["pu"] = 999 + capture_path, _ = write_capture(tmp_path, rows) + + with pytest.raises(BinanceSequenceError, match="expected pu=101, got 999"): + tuple(BinanceCaptureSource((capture_path,)).validated_depth_events()) + + +def test_validated_depth_rejects_update_before_snapshot(tmp_path: Path) -> None: + rows = standard_rows() + rows.pop(1) + capture_path, _ = write_capture(tmp_path, rows) + + with pytest.raises(BinanceSequenceError, match="before snapshot"): + tuple(BinanceCaptureSource((capture_path,)).validated_depth_events()) + + +def test_validated_depth_rejects_update_that_does_not_bridge_snapshot( + tmp_path: Path, +) -> None: + rows = standard_rows() + payload = rows[3]["payload"] + assert isinstance(payload, dict) + payload["U"] = 101 + capture_path, _ = write_capture(tmp_path, rows) + + with pytest.raises(BinanceSequenceError, match="does not bridge"): + tuple(BinanceCaptureSource((capture_path,)).validated_depth_events()) + + +def test_capture_source_rejects_missing_or_empty_file_lists(tmp_path: Path) -> None: + with pytest.raises(ValueError, match="at least one"): + BinanceCaptureSource(()) + with pytest.raises(FileNotFoundError, match="do not exist"): + BinanceCaptureSource((tmp_path / "missing.jsonl.gz",)) + + +@pytest.mark.parametrize( + ("manifest", "message"), + [ + ([], "JSON object"), + ({"schema_version": 2, "files": []}, "schema_version"), + ({"schema_version": 1, "files": "one"}, "list of names"), + ], +) +def test_manifest_validation( + tmp_path: Path, + manifest: object, + message: str, +) -> None: + path = tmp_path / "manifest.json" + path.write_text(json.dumps(manifest), encoding="utf-8") + + with pytest.raises(ValueError, match=message): + BinanceCaptureSource.from_manifest(path) + + +@pytest.mark.parametrize( + ("change", "message"), + [ + ({"schema_version": 2}, "schema_version"), + ({"received_at_ns": "1"}, "received_at_ns"), + ({"kind": "unknown"}, "unsupported value"), + ({"scope": "unknown"}, "unsupported value"), + ({"stream": 1}, "stream"), + ({"payload": []}, "payload"), + ], +) +def test_capture_envelope_validation_reports_file_and_line( + tmp_path: Path, + change: dict[str, object], + message: str, +) -> None: + row = standard_rows()[0] + row.update(change) + capture_path, _ = write_capture(tmp_path, [row]) + + with pytest.raises(ValueError, match=rf":1:.*{message}"): + tuple(BinanceCaptureSource((capture_path,)).envelopes()) + + +def test_capture_envelope_must_be_an_object(tmp_path: Path) -> None: + capture_path, _ = write_capture(tmp_path, [[]]) + + with pytest.raises(ValueError, match="JSON object"): + tuple(BinanceCaptureSource((capture_path,)).envelopes()) + + +def test_normalization_rejects_payload_symbol_mismatch(tmp_path: Path) -> None: + rows = standard_rows() + payload = rows[-1]["payload"] + assert isinstance(payload, dict) + payload["s"] = "ETHUSDT" + capture_path, _ = write_capture(tmp_path, rows) + + with pytest.raises(ValueError, match="does not match"): + tuple(BinanceCaptureSource((capture_path,)).aggregate_trades()) + + +@pytest.mark.parametrize( + ("row_index", "field", "value", "reader", "message"), + [ + (1, "bids", "not-levels", "snapshot", "must be a list"), + (1, "bids", [["100"]], "snapshot", "invalid level"), + (-1, "s", "", "trade", "non-empty string"), + (-1, "m", "true", "trade", "boolean"), + (-1, "p", 100, "trade", "decimal strings"), + (-1, "p", "not-a-decimal", "trade", "invalid decimal"), + ], +) +def test_normalization_rejects_malformed_vendor_fields( + tmp_path: Path, + row_index: int, + field: str, + value: object, + reader: str, + message: str, +) -> None: + rows = standard_rows() + payload = rows[row_index]["payload"] + assert isinstance(payload, dict) + payload[field] = value + capture_path, _ = write_capture(tmp_path, rows) + source = BinanceCaptureSource((capture_path,)) + + records = ( + source.depth_snapshots() if reader == "snapshot" else source.aggregate_trades() + ) + with pytest.raises(ValueError, match=message): + tuple(records) + + +@pytest.mark.parametrize( + ("price", "quantity", "message"), + [ + (Decimal("0"), Decimal("1"), "price"), + (Decimal("1"), Decimal("-1"), "quantity"), + ], +) +def test_price_level_rejects_invalid_values( + price: Decimal, + quantity: Decimal, + message: str, +) -> None: + with pytest.raises(ValueError, match=message): + BinancePriceLevel(price=price, quantity=quantity)