Add Binance L2 capture tooling

.gitignore

@@ -26,6 +26,7 @@ venv/
 model_cache*/
 runs/
 artifacts/
+captures/
 
 # OS/editor
 .DS_Store

AGENTS.md

@@ -38,6 +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/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 |
@@ -110,6 +111,11 @@ For repeated research workflows, prefer:
 Direct connector replay is still useful for smoke tests, one-off inspection, and
 connector development.
 
+Lower-fidelity acquisition is a separate path. A venue recorder may live under
+`ordersim/connectors/` because it owns vendor I/O and schemas, but it is not a
+`DataSource` until a named model can emit valid `MBOEvent` rows. Preserve raw
+evidence separately from inferred order-level events.
+
 ### Add an Instrument Spec
 
 1. Add a plain `InstrumentSpec`.

CHANGELOG.md

@@ -2,6 +2,13 @@
 
 All notable public changes to `ordersim` are documented here.
 
+## Unreleased
+
+- Added an optional Binance USD-M recorder for synchronized raw depth,
+  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.
+
 ## 0.1.3 - 2026-05-20
 
 - Improved default replay throughput by precompiling canonical event streams

docs/architecture.md

@@ -63,6 +63,26 @@ Direct connector replay is supported, but repeated research should normally
 materialize canonical Parquet once and replay from it thereafter. See
 `docs/data-guide.md`.
 
+Lower-fidelity venue data takes a longer, explicit path:
+
+```mermaid
+flowchart LR
+    venue["Venue L2 + trades"]
+    capture["Raw capture"]
+    model["Named reconstruction model"]
+    modeled["Modeled MBOEvent stream"]
+    parquet["Canonical Parquet + model manifest"]
+    replay["Replay"]
+
+    venue --> capture --> 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.
+
 ## One Replay Run
 
 ```mermaid
@@ -154,6 +174,8 @@ inside `run_many(...)` should match running it alone on the same input.
 | Goal | Extension point |
 |---|---|
 | Support a new vendor format | add a connector under `ordersim/connectors/` |
+| Record lower-fidelity venue evidence | add a capture tool beside the venue connector |
+| Infer order-level events from L2 | add a named reconstruction model and manifest |
 | Add a latency assumption | implement `LatencyModel` |
 | Add an execution implementation | implement `ExecutionEngine` and prove equivalence |
 | Change strategy logic | use only the `OrderGateway` surface |

docs/assumptions.md

@@ -31,6 +31,11 @@ MBP source provides aggregated size by price level, not individual queue
 members. Queue-position behavior inferred from MBP is necessarily a model, not
 an observation.
 
+Raw L2 capture and L2-to-L3 reconstruction are separate operations. Capture
+must preserve the observed price-level updates and trades. Reconstruction must
+name its cancellation-allocation, event-ordering, and randomness assumptions,
+and modeled events must not be presented as exchange-native MBO.
+
 ## Queue Assumptions
 
 When the strategy places a resting limit order, the default model assumes the

docs/connectors.md

@@ -63,6 +63,61 @@ Direct connector replay remains useful for smoke tests, one-off inspection, and
 connector development. CSV remains useful for tiny examples and reviewable
 fixtures.
 
+## Binance L2 Capture
+
+Binance USD-M futures publishes aggregated price-level depth rather than stable
+market-by-order identifiers. The capture tool therefore records source evidence
+but does not expose it as `MBOEvent` data.
+
+Install the optional WebSocket dependency:
+
+```bash
+pip install "ordersim[binance]"
+```
+
+Record three days of standard depth, aggregate trades, real-time top-of-book,
+and the optional RPI depth stream:
+
+```bash
+ordersim-binance-capture captures/binance \
+  --symbol BTCUSDT \
+  --symbol ETHUSDT \
+  --duration-hours 72 \
+  --include-rpi
+```
+
+The recorder uses Binance's USD-M futures sources:
+
+| Evidence | Source behavior |
+|---|---|
+| Diff depth | Absolute price-level quantities at up to 100 ms updates. |
+| Aggregate trades | Trades grouped by price and taking side over 100 ms. |
+| Book ticker | Real-time best bid and ask for integrity checks. |
+| RPI depth | Optional 500 ms depth including RPI orders. |
+| REST snapshot | Initial visible book, requested after the depth stream opens. |
+
+Each raw exchange payload is preserved inside a gzip JSONL envelope with:
+
+- UTC receive time in nanoseconds;
+- local monotonic receive time in nanoseconds;
+- source scope and stream name;
+- a connection identifier;
+- the untouched JSON payload.
+
+The recorder writes hourly files and one manifest per process. It also records
+a `sequence_gap` row whenever a depth event's `pu` value does not equal the
+prior event's `u` value within the same connection. A reconnect begins a new
+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.
+
+Capture files are local research data and must not be committed to the
+repository.
+
 For the user-facing decision guide, see `docs/data-guide.md`.
 
 ## In-Memory Sources

docs/data-guide.md

@@ -102,6 +102,11 @@ All canonical sources must preserve the public `MBOEvent` contract:
 If a vendor source cannot preserve one of those properties, document the loss in
 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.
+
 ## Related Docs
 
 - Connector details: `docs/connectors.md`

pyproject.toml

@@ -44,6 +44,9 @@ classifiers = [
 dependencies = []
 
 [project.optional-dependencies]
+binance = [
+  "websockets>=14",
+]
 databento = [
   "databento",
 ]
@@ -61,6 +64,9 @@ parquet = [
   "pyarrow>=23",
 ]
 
+[project.scripts]
+ordersim-binance-capture = "ordersim.connectors.binance.capture:main"
+
 [project.urls]
 Repository = "https://github.com/tradingexpert/ordersim"
 Issues = "https://github.com/tradingexpert/ordersim/issues"

src/ordersim/connectors/binance/__init__.py

@@ -0,0 +1,14 @@
+"""Binance market-data capture tools.
+
+This package records observed Level 2 data and trades. It does not claim to
+recover exchange-native Level 3 events.
+"""
+
+from ordersim.connectors.binance.capture import capture_binance
+from ordersim.connectors.binance.schema import BinanceCaptureConfig, CaptureManifest
+
+__all__ = [
+    "BinanceCaptureConfig",
+    "CaptureManifest",
+    "capture_binance",
+]

src/ordersim/connectors/binance/__main__.py

@@ -0,0 +1,5 @@
+"""Run the Binance capture command."""
+
+from ordersim.connectors.binance.capture import main
+
+main()

src/ordersim/connectors/binance/_storage.py

@@ -0,0 +1,106 @@
+"""Local storage for raw Binance capture envelopes."""
+
+import asyncio
+import gzip
+import json
+import time
+import uuid
+from collections import Counter
+from collections.abc import Mapping
+from datetime import UTC, datetime
+from typing import Any
+
+from ordersim.connectors.binance.schema import (
+    CAPTURE_SCHEMA_VERSION,
+    BinanceCaptureConfig,
+    CaptureManifest,
+    JsonObject,
+)
+
+
+class RawCaptureSink:
+    """Serialize capture envelopes to hourly gzip JSONL files."""
+
+    def __init__(self, config: BinanceCaptureConfig) -> None:
+        self._config = config
+        self._run_id = uuid.uuid4().hex
+        self._started_at_ns = time.time_ns()
+        self._counts: Counter[str] = Counter()
+        self._files: list[str] = []
+        self._current_hour: str | None = None
+        self._file: Any = None
+        self._lock = asyncio.Lock()
+        config.output_dir.mkdir(parents=True, exist_ok=True)
+
+    async def write(
+        self,
+        *,
+        kind: str,
+        scope: str,
+        symbol: str,
+        connection_id: str,
+        stream: str | None,
+        payload: Mapping[str, object],
+    ) -> None:
+        """Write one raw payload with local capture metadata."""
+
+        received_at_ns = time.time_ns()
+        envelope: JsonObject = {
+            "schema_version": CAPTURE_SCHEMA_VERSION,
+            "received_at_ns": received_at_ns,
+            "received_monotonic_ns": time.monotonic_ns(),
+            "kind": kind,
+            "scope": scope,
+            "symbol": symbol,
+            "connection_id": connection_id,
+            "stream": stream,
+            "payload": dict(payload),
+        }
+        async with self._lock:
+            self._rotate(received_at_ns)
+            self._file.write(
+                json.dumps(envelope, separators=(",", ":"), ensure_ascii=True)
+            )
+            self._file.write("\n")
+            self._counts[kind] += 1
+
+    def close(self) -> CaptureManifest:
+        """Close the active file and write the run manifest."""
+
+        if self._file is not None:
+            self._file.close()
+            self._file = None
+        manifest = CaptureManifest(
+            schema_version=CAPTURE_SCHEMA_VERSION,
+            run_id=self._run_id,
+            started_at_ns=self._started_at_ns,
+            ended_at_ns=time.time_ns(),
+            symbols=self._config.symbols,
+            include_rpi=self._config.include_rpi,
+            counts=dict(sorted(self._counts.items())),
+            files=tuple(self._files),
+        )
+        manifest_path = self._config.output_dir / f"manifest-{self._run_id}.json"
+        manifest_path.write_text(
+            json.dumps(manifest.as_dict(), indent=2, sort_keys=True) + "\n",
+            encoding="utf-8",
+        )
+        return manifest
+
+    def _rotate(self, received_at_ns: int) -> None:
+        hour = datetime.fromtimestamp(received_at_ns / 1_000_000_000, UTC).strftime(
+            "%Y%m%dT%H0000Z"
+        )
+        if hour == self._current_hour:
+            return
+        if self._file is not None:
+            self._file.close()
+        filename = f"binance-{self._run_id}-{hour}.jsonl.gz"
+        self._file = gzip.open(
+            self._config.output_dir / filename,
+            mode="at",
+            encoding="utf-8",
+            compresslevel=1,
+        )
+        self._current_hour = hour
+        self._files.append(filename)

src/ordersim/connectors/binance/_transport.py

@@ -0,0 +1,84 @@
+"""Network transport for Binance public market-data endpoints."""
+
+import asyncio
+import json
+import urllib.request
+from collections.abc import AsyncIterator, Callable
+from contextlib import AbstractAsyncContextManager, asynccontextmanager
+from typing import Any, Protocol
+
+from ordersim.connectors.binance.schema import JsonObject
+
+PUBLIC_STREAM_URL = "wss://fstream.binance.com/public/stream?streams="
+MARKET_STREAM_URL = "wss://fstream.binance.com/market/stream?streams="
+DEPTH_SNAPSHOT_URL = "https://fapi.binance.com/fapi/v1/depth"
+
+
+class Transport(Protocol):
+    def connect(
+        self, url: str
+    ) -> AbstractAsyncContextManager[AsyncIterator[tuple[str, JsonObject]]]:
+        """Open one combined Binance WebSocket stream."""
+
+    async def depth_snapshot(self, symbol: str, limit: int) -> JsonObject:
+        """Fetch one Binance depth snapshot."""
+
+
+class WebSocketTransport:
+    """Optional-dependency transport for Binance public endpoints."""
+
+    def __init__(
+        self,
+        *,
+        connect: Callable[..., AbstractAsyncContextManager[Any]] | None = None,
+        read_url: Callable[[str], bytes] | None = None,
+    ) -> None:
+        if connect is None:
+            try:
+                from websockets.asyncio.client import connect as websocket_connect
+            except ImportError as exc:
+                raise RuntimeError(
+                    'Binance capture requires: pip install "ordersim[binance]"'
+                ) from exc
+            connect = websocket_connect
+        self._connect = connect
+        self._read_url = read_url or _read_url
+
+    @asynccontextmanager
+    async def connect(
+        self, url: str
+    ) -> AsyncIterator[AsyncIterator[tuple[str, JsonObject]]]:
+        async with self._connect(url, open_timeout=20, ping_timeout=20) as websocket:
+            yield combined_messages(websocket)
+
+    async def depth_snapshot(self, symbol: str, limit: int) -> JsonObject:
+        query = f"?symbol={symbol}&limit={limit}"
+        raw = await asyncio.to_thread(self._read_url, DEPTH_SNAPSHOT_URL + query)
+        payload = json.loads(raw)
+        if not isinstance(payload, dict):
+            raise ValueError("Binance depth snapshot must be a JSON object")
+        return payload
+
+
+async def combined_messages(
+    websocket: AsyncIterator[str | bytes],
+) -> AsyncIterator[tuple[str, JsonObject]]:
+    async for raw in websocket:
+        yield decode_combined_message(raw)
+
+
+def decode_combined_message(raw: str | bytes) -> tuple[str, JsonObject]:
+    payload = json.loads(raw)
+    if not isinstance(payload, dict):
+        raise ValueError("Binance stream message must be a JSON object")
+    stream = payload.get("stream")
+    data = payload.get("data")
+    if not isinstance(stream, str) or not isinstance(data, dict):
+        raise ValueError("expected a combined Binance stream envelope")
+    return stream, data
+
+
+def _read_url(url: str) -> bytes:
+    request = urllib.request.Request(url, headers={"User-Agent": "ordersim"})
+    with urllib.request.urlopen(request, timeout=20) as response:
+        return response.read()