From 37a3bdba011d0d66c852fac02448d316134efe58 Mon Sep 17 00:00:00 2001 From: Tibor Date: Tue, 28 Jul 2026 13:02:03 +0200 Subject: [PATCH] Add Binance L2 capture tooling --- .gitignore | 1 + AGENTS.md | 6 + CHANGELOG.md | 7 + docs/architecture.md | 22 + docs/assumptions.md | 5 + docs/connectors.md | 55 ++ docs/data-guide.md | 5 + pyproject.toml | 6 + src/ordersim/connectors/binance/__init__.py | 14 + src/ordersim/connectors/binance/__main__.py | 5 + src/ordersim/connectors/binance/_storage.py | 106 ++++ src/ordersim/connectors/binance/_transport.py | 84 +++ src/ordersim/connectors/binance/capture.py | 210 +++++++ src/ordersim/connectors/binance/schema.py | 137 ++++ tests/test_binance_capture.py | 588 ++++++++++++++++++ 15 files changed, 1251 insertions(+) create mode 100644 src/ordersim/connectors/binance/__init__.py create mode 100644 src/ordersim/connectors/binance/__main__.py create mode 100644 src/ordersim/connectors/binance/_storage.py create mode 100644 src/ordersim/connectors/binance/_transport.py create mode 100644 src/ordersim/connectors/binance/capture.py create mode 100644 src/ordersim/connectors/binance/schema.py create mode 100644 tests/test_binance_capture.py diff --git a/.gitignore b/.gitignore index 9323949..0377ba0 100644 --- a/.gitignore +++ b/.gitignore @@ -26,6 +26,7 @@ venv/ model_cache*/ runs/ artifacts/ +captures/ # OS/editor .DS_Store diff --git a/AGENTS.md b/AGENTS.md index bfaaf7d..14f2bf7 100644 --- a/AGENTS.md +++ b/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`. diff --git a/CHANGELOG.md b/CHANGELOG.md index b935e86..c2be564 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/docs/architecture.md b/docs/architecture.md index c8d49c6..202eb55 100644 --- a/docs/architecture.md +++ b/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 | diff --git a/docs/assumptions.md b/docs/assumptions.md index 304d0e6..51238f1 100644 --- a/docs/assumptions.md +++ b/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 diff --git a/docs/connectors.md b/docs/connectors.md index 1eb803f..47bc0ca 100644 --- a/docs/connectors.md +++ b/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 diff --git a/docs/data-guide.md b/docs/data-guide.md index 4dd541c..36ce41f 100644 --- a/docs/data-guide.md +++ b/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` diff --git a/pyproject.toml b/pyproject.toml index 1052309..aa758e1 100644 --- a/pyproject.toml +++ b/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" diff --git a/src/ordersim/connectors/binance/__init__.py b/src/ordersim/connectors/binance/__init__.py new file mode 100644 index 0000000..0e1e345 --- /dev/null +++ b/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", +] diff --git a/src/ordersim/connectors/binance/__main__.py b/src/ordersim/connectors/binance/__main__.py new file mode 100644 index 0000000..168986a --- /dev/null +++ b/src/ordersim/connectors/binance/__main__.py @@ -0,0 +1,5 @@ +"""Run the Binance capture command.""" + +from ordersim.connectors.binance.capture import main + +main() diff --git a/src/ordersim/connectors/binance/_storage.py b/src/ordersim/connectors/binance/_storage.py new file mode 100644 index 0000000..c8cecb8 --- /dev/null +++ b/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) diff --git a/src/ordersim/connectors/binance/_transport.py b/src/ordersim/connectors/binance/_transport.py new file mode 100644 index 0000000..8a97298 --- /dev/null +++ b/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() diff --git a/src/ordersim/connectors/binance/capture.py b/src/ordersim/connectors/binance/capture.py new file mode 100644 index 0000000..3385826 --- /dev/null +++ b/src/ordersim/connectors/binance/capture.py @@ -0,0 +1,210 @@ +"""Record raw Binance depth and trade evidence for modeled replay. + +The recorder preserves exchange payloads and adds local receive timestamps, +connection identifiers, and sequence-gap records. It deliberately does not +turn aggregated depth into `MBOEvent` rows; that inference belongs to a named +reconstruction model. +""" + +import argparse +import asyncio +import uuid +from collections.abc import Sequence +from pathlib import Path + +from ordersim.connectors.binance._storage import RawCaptureSink +from ordersim.connectors.binance._transport import ( + MARKET_STREAM_URL, + PUBLIC_STREAM_URL, + Transport, + WebSocketTransport, +) +from ordersim.connectors.binance.schema import ( + BinanceCaptureConfig, + CaptureManifest, + DepthSequenceTracker, +) + + +async def capture_binance( + config: BinanceCaptureConfig, + *, + transport: Transport | None = None, +) -> CaptureManifest: + """Capture raw Binance evidence until duration expiry or interruption.""" + + active_transport = transport or WebSocketTransport() + sink = RawCaptureSink(config) + interrupted = False + tasks = [ + asyncio.create_task( + _capture_with_reconnect( + config=config, + sink=sink, + transport=active_transport, + symbol=symbol, + scope=scope, + ) + ) + for symbol in config.symbols + for scope in ("public", "market") + ] + try: + if config.duration_seconds is None: + await asyncio.Event().wait() + else: + await asyncio.sleep(config.duration_seconds) + except asyncio.CancelledError: + interrupted = True + finally: + for task in tasks: + task.cancel() + await asyncio.gather(*tasks, return_exceptions=True) + manifest = sink.close() + if interrupted: + raise asyncio.CancelledError + return manifest + + +async def _capture_with_reconnect( + *, + config: BinanceCaptureConfig, + sink: RawCaptureSink, + transport: Transport, + symbol: str, + scope: str, +) -> None: + while True: + connection_id = uuid.uuid4().hex + try: + await _capture_connection( + config=config, + sink=sink, + transport=transport, + symbol=symbol, + scope=scope, + connection_id=connection_id, + ) + except asyncio.CancelledError: + raise + except Exception as exc: + await sink.write( + kind="connection_error", + scope=scope, + symbol=symbol, + connection_id=connection_id, + stream=None, + payload={ + "error_type": type(exc).__name__, + "message": str(exc), + }, + ) + await asyncio.sleep(config.reconnect_delay_seconds) + + +async def _capture_connection( + *, + config: BinanceCaptureConfig, + sink: RawCaptureSink, + transport: Transport, + symbol: str, + scope: str, + connection_id: str, +) -> None: + streams = ( + config.public_streams(symbol) + if scope == "public" + else config.market_streams(symbol) + ) + base_url = PUBLIC_STREAM_URL if scope == "public" else MARKET_STREAM_URL + trackers: dict[str, DepthSequenceTracker] = {} + async with transport.connect(base_url + "/".join(streams)) as messages: + await sink.write( + kind="connection_open", + scope=scope, + symbol=symbol, + connection_id=connection_id, + stream=None, + payload={"streams": list(streams)}, + ) + if scope == "public": + snapshot = await transport.depth_snapshot(symbol, config.snapshot_limit) + await sink.write( + kind="depth_snapshot", + scope=scope, + symbol=symbol, + connection_id=connection_id, + stream=None, + payload=snapshot, + ) + + async for stream, payload in messages: + await sink.write( + kind="message", + scope=scope, + symbol=symbol, + connection_id=connection_id, + stream=stream, + payload=payload, + ) + if "@depth@" not in stream and "@rpiDepth@" not in stream: + continue + tracker = trackers.setdefault(stream, DepthSequenceTracker()) + gap = tracker.observe(payload) + if gap is not None: + await sink.write( + kind="sequence_gap", + scope=scope, + symbol=symbol, + connection_id=connection_id, + stream=stream, + payload=gap.as_dict(), + ) + + +def _parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description="Record Binance USD-M depth and aggregate-trade evidence." + ) + parser.add_argument("output_dir", type=Path) + parser.add_argument( + "--symbol", + action="append", + required=True, + dest="symbols", + help="USD-M symbol; repeat for more than one (for example BTCUSDT).", + ) + parser.add_argument( + "--duration-hours", + type=float, + help="Stop after this many hours; otherwise run until interrupted.", + ) + parser.add_argument( + "--include-rpi", + action="store_true", + help="Also record the 500 ms RPI depth stream.", + ) + return parser + + +def main(argv: Sequence[str] | None = None) -> None: + """Run the `ordersim-binance-capture` command.""" + + args = _parser().parse_args(argv) + duration_seconds = ( + None if args.duration_hours is None else args.duration_hours * 60 * 60 + ) + config = BinanceCaptureConfig( + output_dir=args.output_dir, + symbols=tuple(args.symbols), + duration_seconds=duration_seconds, + include_rpi=args.include_rpi, + ) + try: + manifest = asyncio.run(capture_binance(config)) + except KeyboardInterrupt: + return + print( + f"capture {manifest.run_id} complete: " + f"{sum(manifest.counts.values())} records" + ) diff --git a/src/ordersim/connectors/binance/schema.py b/src/ordersim/connectors/binance/schema.py new file mode 100644 index 0000000..7736054 --- /dev/null +++ b/src/ordersim/connectors/binance/schema.py @@ -0,0 +1,137 @@ +"""Schemas and integrity checks for Binance market-data capture.""" + +from collections.abc import Mapping +from dataclasses import dataclass +from pathlib import Path + +CAPTURE_SCHEMA_VERSION = 1 +JsonObject = dict[str, object] + + +@dataclass(frozen=True, slots=True) +class BinanceCaptureConfig: + """Configuration for one raw Binance USD-M futures capture.""" + + output_dir: Path + symbols: tuple[str, ...] + duration_seconds: float | None = None + include_rpi: bool = False + snapshot_limit: int = 1000 + reconnect_delay_seconds: float = 2.0 + + def __post_init__(self) -> None: + symbols = tuple(symbol.strip().upper() for symbol in self.symbols) + if not symbols: + raise ValueError("at least one symbol is required") + if any(not symbol.isalnum() for symbol in symbols): + raise ValueError("symbols must contain only letters and numbers") + if len(set(symbols)) != len(symbols): + raise ValueError("symbols must be unique") + if self.duration_seconds is not None and self.duration_seconds <= 0: + raise ValueError("duration_seconds must be positive") + if self.snapshot_limit not in (5, 10, 20, 50, 100, 500, 1000): + raise ValueError("unsupported Binance snapshot_limit") + if self.reconnect_delay_seconds < 0: + raise ValueError("reconnect_delay_seconds must be non-negative") + + object.__setattr__(self, "output_dir", Path(self.output_dir)) + object.__setattr__(self, "symbols", symbols) + + def public_streams(self, symbol: str) -> tuple[str, ...]: + """Return the public-book streams captured for one symbol.""" + + prefix = symbol.lower() + streams = [ + f"{prefix}@depth@100ms", + f"{prefix}@bookTicker", + ] + if self.include_rpi: + streams.append(f"{prefix}@rpiDepth@500ms") + return tuple(streams) + + def market_streams(self, symbol: str) -> tuple[str, ...]: + """Return the trade streams captured for one symbol.""" + + return (f"{symbol.lower()}@aggTrade",) + + +@dataclass(frozen=True, slots=True) +class DepthSequenceGap: + """One discontinuity in a Binance diff-depth stream.""" + + expected_previous_update_id: int + reported_previous_update_id: int + first_update_id: int + final_update_id: int + + def as_dict(self) -> dict[str, int]: + """Return a JSON-compatible representation.""" + + return { + "expected_previous_update_id": self.expected_previous_update_id, + "reported_previous_update_id": self.reported_previous_update_id, + "first_update_id": self.first_update_id, + "final_update_id": self.final_update_id, + } + + +class DepthSequenceTracker: + """Validate `pu` continuity within one WebSocket connection.""" + + def __init__(self) -> None: + self._previous_final_update_id: int | None = None + + def observe(self, payload: Mapping[str, object]) -> DepthSequenceGap | None: + """Observe one depth payload and return a gap when continuity breaks.""" + + first_update_id = _required_int(payload, "U") + final_update_id = _required_int(payload, "u") + previous_update_id = _required_int(payload, "pu") + if final_update_id < first_update_id: + raise ValueError("depth payload has u < U") + + expected = self._previous_final_update_id + self._previous_final_update_id = final_update_id + if expected is None or previous_update_id == expected: + return None + return DepthSequenceGap( + expected_previous_update_id=expected, + reported_previous_update_id=previous_update_id, + first_update_id=first_update_id, + final_update_id=final_update_id, + ) + + +@dataclass(frozen=True, slots=True) +class CaptureManifest: + """Summary of one recorder process.""" + + schema_version: int + run_id: str + started_at_ns: int + ended_at_ns: int + symbols: tuple[str, ...] + include_rpi: bool + counts: dict[str, int] + files: tuple[str, ...] + + def as_dict(self) -> JsonObject: + """Return a JSON-compatible representation.""" + + return { + "schema_version": self.schema_version, + "run_id": self.run_id, + "started_at_ns": self.started_at_ns, + "ended_at_ns": self.ended_at_ns, + "symbols": list(self.symbols), + "include_rpi": self.include_rpi, + "counts": self.counts, + "files": list(self.files), + } + + +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 diff --git a/tests/test_binance_capture.py b/tests/test_binance_capture.py new file mode 100644 index 0000000..9782fab --- /dev/null +++ b/tests/test_binance_capture.py @@ -0,0 +1,588 @@ +import asyncio +import builtins +import gzip +import json +import runpy +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager +from pathlib import Path +from types import SimpleNamespace + +import pytest + +import ordersim.connectors.binance._transport as transport_module +import ordersim.connectors.binance.capture as capture_module +from ordersim.connectors.binance._storage import RawCaptureSink +from ordersim.connectors.binance._transport import ( + MARKET_STREAM_URL, + PUBLIC_STREAM_URL, + WebSocketTransport, + _read_url, + combined_messages, + decode_combined_message, +) +from ordersim.connectors.binance.capture import ( + _capture_connection, + _capture_with_reconnect, + capture_binance, + main, +) +from ordersim.connectors.binance.schema import ( + BinanceCaptureConfig, + CaptureManifest, + DepthSequenceTracker, +) + + +def test_capture_config_names_the_highest_resolution_standard_streams( + tmp_path: Path, +) -> None: + config = BinanceCaptureConfig( + output_dir=tmp_path, + symbols=("btcusdt", "ETHUSDT"), + duration_seconds=10, + include_rpi=True, + ) + + assert config.symbols == ("BTCUSDT", "ETHUSDT") + assert config.public_streams("BTCUSDT") == ( + "btcusdt@depth@100ms", + "btcusdt@bookTicker", + "btcusdt@rpiDepth@500ms", + ) + assert config.market_streams("BTCUSDT") == ("btcusdt@aggTrade",) + + +@pytest.mark.parametrize( + ("symbols", "duration", "limit", "delay", "message"), + [ + ((), None, 1000, 2, "at least one symbol"), + (("BTC-USDT",), None, 1000, 2, "letters and numbers"), + (("BTCUSDT", "btcusdt"), None, 1000, 2, "unique"), + (("BTCUSDT",), 0, 1000, 2, "positive"), + (("BTCUSDT",), None, 200, 2, "snapshot_limit"), + (("BTCUSDT",), None, 1000, -1, "non-negative"), + ], +) +def test_capture_config_rejects_invalid_values( + tmp_path: Path, + symbols: tuple[str, ...], + duration: float | None, + limit: int, + delay: float, + message: str, +) -> None: + with pytest.raises(ValueError, match=message): + BinanceCaptureConfig( + output_dir=tmp_path, + symbols=symbols, + duration_seconds=duration, + snapshot_limit=limit, + reconnect_delay_seconds=delay, + ) + + +def test_depth_sequence_tracker_reports_discontinuity() -> None: + tracker = DepthSequenceTracker() + + assert tracker.observe({"U": 10, "u": 12, "pu": 9}) is None + assert tracker.observe({"U": 13, "u": 15, "pu": 12}) is None + gap = tracker.observe({"U": 20, "u": 22, "pu": 19}) + + assert gap is not None + assert gap.as_dict() == { + "expected_previous_update_id": 15, + "reported_previous_update_id": 19, + "first_update_id": 20, + "final_update_id": 22, + } + + +def test_depth_sequence_tracker_rejects_malformed_payloads() -> None: + tracker = DepthSequenceTracker() + + with pytest.raises(ValueError, match="must be an integer"): + tracker.observe({"U": "10", "u": 12, "pu": 9}) + with pytest.raises(ValueError, match="u < U"): + tracker.observe({"U": 12, "u": 10, "pu": 9}) + + +def test_combined_message_decoder_preserves_exact_strings() -> None: + stream, payload = decode_combined_message( + b'{"stream":"btcusdt@aggTrade","data":{"p":"123.4500","q":"0.010"}}' + ) + + assert stream == "btcusdt@aggTrade" + assert payload == {"p": "123.4500", "q": "0.010"} + + +@pytest.mark.parametrize( + "raw", + [ + "[]", + '{"stream":1,"data":{}}', + '{"stream":"btcusdt@depth@100ms","data":[]}', + ], +) +def test_combined_message_decoder_rejects_non_envelopes(raw: str) -> None: + with pytest.raises(ValueError): + decode_combined_message(raw) + + +class FakeWebSocket: + def __init__(self, messages: list[str]) -> None: + self._messages = messages + + def __aiter__(self) -> AsyncIterator[str]: + return self._iterate() + + async def _iterate(self) -> AsyncIterator[str]: + for message in self._messages: + yield message + + +def test_combined_messages_iterates_decoded_records() -> None: + websocket = FakeWebSocket( + ['{"stream":"btcusdt@bookTicker","data":{"b":"100","a":"101"}}'] + ) + + async def collect() -> list[tuple[str, dict[str, object]]]: + return [item async for item in combined_messages(websocket)] + + assert asyncio.run(collect()) == [ + ("btcusdt@bookTicker", {"b": "100", "a": "101"}) + ] + + +class FakeTransport: + def __init__(self, messages: list[tuple[str, dict[str, object]]]) -> None: + self.messages = messages + self.connected_urls: list[str] = [] + self.snapshot_calls: list[tuple[str, int]] = [] + + @asynccontextmanager + async def connect( + self, url: str + ) -> AsyncIterator[AsyncIterator[tuple[str, dict[str, object]]]]: + self.connected_urls.append(url) + + async def iterate() -> AsyncIterator[tuple[str, dict[str, object]]]: + for message in self.messages: + yield message + + yield iterate() + + async def depth_snapshot(self, symbol: str, limit: int) -> dict[str, object]: + self.snapshot_calls.append((symbol, limit)) + return { + "lastUpdateId": 9, + "bids": [["100.0", "2.0"]], + "asks": [["101.0", "3.0"]], + } + + +class BlockingTransport(FakeTransport): + def __init__(self) -> None: + super().__init__([]) + self.connected = asyncio.Event() + + @asynccontextmanager + async def connect( + self, url: str + ) -> AsyncIterator[AsyncIterator[tuple[str, dict[str, object]]]]: + self.connected_urls.append(url) + self.connected.set() + + async def wait_forever() -> AsyncIterator[tuple[str, dict[str, object]]]: + await asyncio.Event().wait() + yield "", {} + + yield wait_forever() + + +def _read_capture_records(output_dir: Path) -> list[dict[str, object]]: + capture_file = next(output_dir.glob("*.jsonl.gz")) + with gzip.open(capture_file, mode="rt", encoding="utf-8") as file: + return [json.loads(line) for line in file] + + +def test_public_connection_records_snapshot_messages_and_gap(tmp_path: Path) -> None: + config = BinanceCaptureConfig( + output_dir=tmp_path, + symbols=("BTCUSDT",), + include_rpi=True, + ) + transport = FakeTransport( + [ + ("btcusdt@depth@100ms", {"U": 10, "u": 12, "pu": 9}), + ("btcusdt@bookTicker", {"b": "100", "a": "101"}), + ("btcusdt@depth@100ms", {"U": 20, "u": 22, "pu": 19}), + ] + ) + sink = RawCaptureSink(config) + + asyncio.run( + _capture_connection( + config=config, + sink=sink, + transport=transport, + symbol="BTCUSDT", + scope="public", + connection_id="public-1", + ) + ) + manifest = sink.close() + records = _read_capture_records(tmp_path) + + assert transport.connected_urls == [ + PUBLIC_STREAM_URL + + "btcusdt@depth@100ms/btcusdt@bookTicker/" + + "btcusdt@rpiDepth@500ms" + ] + assert transport.snapshot_calls == [("BTCUSDT", 1000)] + assert [record["kind"] for record in records] == [ + "connection_open", + "depth_snapshot", + "message", + "message", + "message", + "sequence_gap", + ] + assert records[2]["payload"] == {"U": 10, "u": 12, "pu": 9} + assert manifest.counts == { + "connection_open": 1, + "depth_snapshot": 1, + "message": 3, + "sequence_gap": 1, + } + + +def test_market_connection_records_trades_without_snapshot(tmp_path: Path) -> None: + config = BinanceCaptureConfig(output_dir=tmp_path, symbols=("BTCUSDT",)) + transport = FakeTransport( + [("btcusdt@aggTrade", {"p": "100.1", "q": "0.25", "m": True})] + ) + sink = RawCaptureSink(config) + + asyncio.run( + _capture_connection( + config=config, + sink=sink, + transport=transport, + symbol="BTCUSDT", + scope="market", + connection_id="market-1", + ) + ) + sink.close() + + assert transport.connected_urls == [MARKET_STREAM_URL + "btcusdt@aggTrade"] + assert transport.snapshot_calls == [] + assert _read_capture_records(tmp_path)[1]["payload"] == { + "p": "100.1", + "q": "0.25", + "m": True, + } + + +def test_websocket_transport_uses_injected_network_functions() -> None: + calls: list[tuple[str, int, int]] = [] + + @asynccontextmanager + async def fake_connect( + url: str, *, open_timeout: int, ping_timeout: int + ) -> AsyncIterator[FakeWebSocket]: + calls.append((url, open_timeout, ping_timeout)) + yield FakeWebSocket( + ['{"stream":"btcusdt@aggTrade","data":{"p":"100","q":"1"}}'] + ) + + def fake_read_url(url: str) -> bytes: + assert "symbol=BTCUSDT" in url + assert "limit=1000" in url + return b'{"lastUpdateId":9,"bids":[],"asks":[]}' + + transport = WebSocketTransport(connect=fake_connect, read_url=fake_read_url) + + async def exercise() -> tuple[list[tuple[str, dict[str, object]]], object]: + async with transport.connect("wss://example") as messages: + received = [message async for message in messages] + snapshot = await transport.depth_snapshot("BTCUSDT", 1000) + return received, snapshot + + messages, snapshot = asyncio.run(exercise()) + + assert calls == [("wss://example", 20, 20)] + assert messages == [("btcusdt@aggTrade", {"p": "100", "q": "1"})] + assert snapshot == {"lastUpdateId": 9, "bids": [], "asks": []} + + +def test_websocket_transport_rejects_non_object_snapshot() -> None: + transport = WebSocketTransport( + connect=lambda *args, **kwargs: None, + read_url=lambda url: b"[]", + ) + + with pytest.raises(ValueError, match="JSON object"): + asyncio.run(transport.depth_snapshot("BTCUSDT", 1000)) + + +def test_websocket_transport_loads_optional_dependency( + monkeypatch: pytest.MonkeyPatch, +) -> None: + original_import = builtins.__import__ + sentinel_connect = object() + + def fake_import( + name: str, + globals: object = None, + locals: object = None, + fromlist: tuple[str, ...] = (), + level: int = 0, + ) -> object: + if name == "websockets.asyncio.client": + return SimpleNamespace(connect=sentinel_connect) + return original_import(name, globals, locals, fromlist, level) + + monkeypatch.setattr(builtins, "__import__", fake_import) + + transport = WebSocketTransport() + + assert transport._connect is sentinel_connect + + +def test_websocket_transport_explains_missing_optional_dependency( + monkeypatch: pytest.MonkeyPatch, +) -> None: + original_import = builtins.__import__ + + def fake_import( + name: str, + globals: object = None, + locals: object = None, + fromlist: tuple[str, ...] = (), + level: int = 0, + ) -> object: + if name == "websockets.asyncio.client": + raise ImportError("not installed") + return original_import(name, globals, locals, fromlist, level) + + monkeypatch.setattr(builtins, "__import__", fake_import) + + with pytest.raises(RuntimeError, match=r"ordersim\[binance\]"): + WebSocketTransport() + + +def test_read_url_uses_ordersim_user_agent(monkeypatch: pytest.MonkeyPatch) -> None: + class FakeResponse: + def __enter__(self) -> "FakeResponse": + return self + + def __exit__(self, *args: object) -> None: + return None + + def read(self) -> bytes: + return b'{"ok":true}' + + def fake_urlopen(request: object, *, timeout: int) -> FakeResponse: + assert request.get_header("User-agent") == "ordersim" + assert timeout == 20 + return FakeResponse() + + monkeypatch.setattr(transport_module.urllib.request, "urlopen", fake_urlopen) + + assert _read_url("https://example.test") == b'{"ok":true}' + + +def test_sink_writes_manifest_with_capture_files(tmp_path: Path) -> None: + config = BinanceCaptureConfig(output_dir=tmp_path, symbols=("BTCUSDT",)) + sink = RawCaptureSink(config) + + asyncio.run( + sink.write( + kind="message", + scope="market", + symbol="BTCUSDT", + connection_id="one", + stream="btcusdt@aggTrade", + payload={"p": "100.00000000", "q": "1.500"}, + ) + ) + manifest = sink.close() + manifest_file = next(tmp_path.glob("manifest-*.json")) + stored_manifest = json.loads(manifest_file.read_text(encoding="utf-8")) + + assert manifest.symbols == ("BTCUSDT",) + assert manifest.counts == {"message": 1} + assert stored_manifest["files"] == list(manifest.files) + assert _read_capture_records(tmp_path)[0]["payload"] == { + "p": "100.00000000", + "q": "1.500", + } + + +def test_sink_can_close_without_any_messages(tmp_path: Path) -> None: + config = BinanceCaptureConfig(output_dir=tmp_path, symbols=("BTCUSDT",)) + + manifest = RawCaptureSink(config).close() + + assert manifest.counts == {} + assert manifest.files == () + + +def test_sink_rotates_at_hour_boundary(tmp_path: Path) -> None: + config = BinanceCaptureConfig(output_dir=tmp_path, symbols=("BTCUSDT",)) + sink = RawCaptureSink(config) + + sink._rotate(0) + sink._rotate(3_600_000_000_000) + manifest = sink.close() + + assert len(manifest.files) == 2 + + +def test_finite_capture_closes_connections_and_writes_manifest( + tmp_path: Path, +) -> None: + config = BinanceCaptureConfig( + output_dir=tmp_path, + symbols=("BTCUSDT",), + duration_seconds=0.01, + ) + transport = BlockingTransport() + + manifest = asyncio.run(capture_binance(config, transport=transport)) + + assert manifest.counts["connection_open"] == 2 + assert manifest.counts["depth_snapshot"] == 1 + assert len(transport.connected_urls) == 2 + assert next(tmp_path.glob("manifest-*.json")).exists() + + +def test_cancelled_capture_still_writes_manifest(tmp_path: Path) -> None: + config = BinanceCaptureConfig(output_dir=tmp_path, symbols=("BTCUSDT",)) + transport = BlockingTransport() + + async def exercise() -> None: + task = asyncio.create_task(capture_binance(config, transport=transport)) + await transport.connected.wait() + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + asyncio.run(exercise()) + + assert next(tmp_path.glob("manifest-*.json")).exists() + + +def test_reconnect_loop_records_connection_error( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + config = BinanceCaptureConfig( + output_dir=tmp_path, + symbols=("BTCUSDT",), + reconnect_delay_seconds=0, + ) + sink = RawCaptureSink(config) + attempts = 0 + second_attempt = asyncio.Event() + + async def fake_capture_connection(**kwargs: object) -> None: + nonlocal attempts + attempts += 1 + if attempts == 1: + raise ConnectionError("test disconnect") + second_attempt.set() + await asyncio.Event().wait() + + monkeypatch.setattr( + capture_module, + "_capture_connection", + fake_capture_connection, + ) + + async def exercise() -> None: + task = asyncio.create_task( + _capture_with_reconnect( + config=config, + sink=sink, + transport=FakeTransport([]), + symbol="BTCUSDT", + scope="public", + ) + ) + await second_attempt.wait() + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + asyncio.run(exercise()) + sink.close() + records = _read_capture_records(tmp_path) + + assert records[0]["kind"] == "connection_error" + assert records[0]["payload"] == { + "error_type": "ConnectionError", + "message": "test disconnect", + } + + +def test_main_builds_capture_config_and_reports_result( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + received: list[BinanceCaptureConfig] = [] + + async def fake_capture(config: BinanceCaptureConfig) -> CaptureManifest: + received.append(config) + return CaptureManifest( + schema_version=1, + run_id="run-one", + started_at_ns=1, + ended_at_ns=2, + symbols=config.symbols, + include_rpi=config.include_rpi, + counts={"message": 3}, + files=("one.jsonl.gz",), + ) + + monkeypatch.setattr(capture_module, "capture_binance", fake_capture) + + main( + [ + str(tmp_path), + "--symbol", + "BTCUSDT", + "--duration-hours", + "2", + "--include-rpi", + ] + ) + + assert received[0].duration_seconds == 7200 + assert received[0].include_rpi is True + assert capsys.readouterr().out == "capture run-one complete: 3 records\n" + + +def test_main_handles_keyboard_interrupt( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + def interrupt(coroutine: object) -> None: + coroutine.close() + raise KeyboardInterrupt + + monkeypatch.setattr(capture_module.asyncio, "run", interrupt) + + assert main([str(tmp_path), "--symbol", "BTCUSDT"]) is None + + +def test_module_entry_point_calls_main(monkeypatch: pytest.MonkeyPatch) -> None: + called: list[bool] = [] + monkeypatch.setattr(capture_module, "main", lambda: called.append(True)) + + runpy.run_module("ordersim.connectors.binance.__main__", run_name="__main__") + + assert called == [True]