From 55a1f130d395bd2d305e525160432695e2ae9a08 Mon Sep 17 00:00:00 2001 From: Tibor Date: Tue, 28 Jul 2026 23:30:28 +0200 Subject: [PATCH 1/4] Capture individual Binance trades --- AGENTS.md | 1 + CHANGELOG.md | 4 + docs/architecture.md | 2 +- docs/connectors.md | 44 +- docs/data-guide.md | 8 +- docs/schema.md | 11 + pyproject.toml | 1 + src/ordersim/connectors/binance/__init__.py | 11 +- src/ordersim/connectors/binance/_parsing.py | 24 + .../connectors/binance/_recent_trades.py | 206 +++++++ src/ordersim/connectors/binance/_storage.py | 7 +- src/ordersim/connectors/binance/l2.py | 21 + src/ordersim/connectors/binance/raw_trades.py | 198 ++++++ src/ordersim/connectors/binance/schema.py | 66 ++ src/ordersim/connectors/binance/source.py | 9 + tests/test_binance_raw_trades.py | 562 ++++++++++++++++++ 16 files changed, 1167 insertions(+), 8 deletions(-) create mode 100644 src/ordersim/connectors/binance/_recent_trades.py create mode 100644 src/ordersim/connectors/binance/raw_trades.py create mode 100644 tests/test_binance_raw_trades.py diff --git a/AGENTS.md b/AGENTS.md index df652f0..55c9c4b 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -39,6 +39,7 @@ extraction targets and should not be imported until they exist. | `ordersim/connectors/databento.py` | Databento MBO normalization | Yes | | `ordersim/connectors/parquet.py` | Normalized Parquet `MBOEvent` source | Yes | | `ordersim/connectors/binance/` | Binance capture and typed L2 source; not MBO | Public venue API | +| `ordersim/connectors/binance/raw_trades.py` | Individual trade capture with ID-gap evidence | Public capture 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 01a8053..888f9c4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,10 @@ All notable public changes to `ordersim` are documented here. depth, aggregate-trade, and book-ticker records. - Added snapshot bridging and `pu`/`u` continuity validation for standard Binance diff-depth segments. +- Added rate-budgeted Binance individual-trade capture with overlapping REST + polls, trade-ID deduplication, explicit gap records, and RPI trade flags. +- Added typed `BinanceRawTrade` records alongside aggregate trades so the more + detailed public evidence is available to future reconstruction models. ## 0.1.3 - 2026-05-20 diff --git a/docs/architecture.md b/docs/architecture.md index 5d5f7b4..6caea2b 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -67,7 +67,7 @@ Lower-fidelity venue data takes a longer, explicit path: ```mermaid flowchart LR - venue["Venue L2 + trades"] + venue["Venue L2 + aggregate and individual trades"] capture["Raw capture"] source["Typed L2 source"] model["Named reconstruction model"] diff --git a/docs/connectors.md b/docs/connectors.md index d8e45a2..6374cd6 100644 --- a/docs/connectors.md +++ b/docs/connectors.md @@ -86,12 +86,24 @@ ordersim-binance-capture captures/binance \ --include-rpi ``` +For the most detailed public trade evidence, run the individual-trade recorder +beside the WebSocket capture: + +```bash +ordersim-binance-raw-trades captures/binance-raw-trades \ + --symbol BTCUSDT \ + --symbol ETHUSDT \ + --duration-hours 72 \ + --poll-interval-seconds 2 +``` + 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. | +| Individual trades | REST recent trades with unique trade IDs and RPI flags. | | 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. | @@ -118,6 +130,30 @@ modeled `MBOEvent` rows. Capture files are local research data and must not be committed to the repository. +### Individual Trade Capture + +USD-M's documented WebSocket market stream exposes `aggTrade`, not an +individual-trade stream. The public `/fapi/v1/trades` REST endpoint is more +detailed: each row has its own trade ID and `isRPITrade` flag. + +`ordersim-binance-raw-trades` polls that endpoint with overlapping 1,000-row +windows. It: + +- stores each exact individual trade payload once; +- records request timing and Binance's reported one-minute request weight; +- deduplicates overlapping responses by trade ID; +- writes `raw_trade_gap` whenever the next observed ID is not contiguous; +- writes explicit poll errors rather than silently retrying. + +At Binance's current 25-unit request weight, two symbols polled every two +seconds consume an estimated 1,500 units per minute. Configuration is rejected +when it would exceed the recorder's conservative 1,800-unit budget. This leaves +headroom below Binance's venue limit for snapshots and operational variance. + +The aggregate-trade stream remains valuable as an independent reconciliation +feed. It is not treated as a substitute for individual trades when individual +trade evidence is available. + ### Reading Completed Captures `BinanceCaptureSource` streams completed gzip capture files directly, so a @@ -135,12 +171,16 @@ for event in source.validated_depth_events(): for trade in source.aggregate_trades(): print(trade) + +for trade in source.raw_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. +`BinanceAggregateTrade`, `BinanceRawTrade`, 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 diff --git a/docs/data-guide.md b/docs/data-guide.md index 23cbd7e..525f3ec 100644 --- a/docs/data-guide.md +++ b/docs/data-guide.md @@ -107,9 +107,11 @@ records raw L2 depth, aggregate trades, and integrity metadata, but its output 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: +snapshots, sequence-validated depth updates, aggregate trades, individual +trades, and book tickers. Capture individual trades with +`ordersim-binance-raw-trades`; retain `aggTrade` as a reconciliation feed rather +than using it as a lower-detail substitute. That typed source is the input +boundary for the planned named virtual-L3 reconstruction model: ```text raw capture -> BinanceCaptureSource -> named model -> modeled MBO + manifest diff --git a/docs/schema.md b/docs/schema.md index 49f8b10..4c9c797 100644 --- a/docs/schema.md +++ b/docs/schema.md @@ -32,6 +32,7 @@ quantity at that price; zero means remove the level. | `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. | +| `BinanceRawTrade` | `trade_id`, `price`, `quantity`, `quote_quantity`, `buyer_is_maker`, `is_rpi_trade` | One individually identified public trade. | | `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 @@ -45,6 +46,16 @@ unit and exact conversion rule before producing the canonical integer `MBOEvent.size`. These L2 records are therefore not accepted directly by `Replay`. +Raw-trade capture files also contain audit envelopes: + +- `raw_trade_poll` records request timing, returned ID bounds, and request + weight; +- `raw_trade_gap` records a missing individual trade-ID range; +- `raw_trade_poll_error` records a failed request and the last retained ID. + +These audit rows are available through `envelopes()` and are not emitted by +`raw_trades()`. + ## `MBOEvent` `MBOEvent` represents one Level 3 / market-by-order event. diff --git a/pyproject.toml b/pyproject.toml index aa758e1..490a93f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -66,6 +66,7 @@ parquet = [ [project.scripts] ordersim-binance-capture = "ordersim.connectors.binance.capture:main" +ordersim-binance-raw-trades = "ordersim.connectors.binance.raw_trades:main" [project.urls] Repository = "https://github.com/tradingexpert/ordersim" diff --git a/src/ordersim/connectors/binance/__init__.py b/src/ordersim/connectors/binance/__init__.py index 9c8d3fe..477b1c1 100644 --- a/src/ordersim/connectors/binance/__init__.py +++ b/src/ordersim/connectors/binance/__init__.py @@ -13,9 +13,15 @@ BinanceDepthSnapshot, BinanceDepthUpdate, BinancePriceLevel, + BinanceRawTrade, DepthStreamKind, ) -from ordersim.connectors.binance.schema import BinanceCaptureConfig, CaptureManifest +from ordersim.connectors.binance.raw_trades import capture_binance_raw_trades +from ordersim.connectors.binance.schema import ( + BinanceCaptureConfig, + BinanceRawTradeCaptureConfig, + CaptureManifest, +) from ordersim.connectors.binance.source import ( BinanceCaptureSource, BinanceSequenceError, @@ -31,8 +37,11 @@ "BinanceDepthSnapshot", "BinanceDepthUpdate", "BinancePriceLevel", + "BinanceRawTrade", + "BinanceRawTradeCaptureConfig", "BinanceSequenceError", "CaptureManifest", "DepthStreamKind", "capture_binance", + "capture_binance_raw_trades", ] diff --git a/src/ordersim/connectors/binance/_parsing.py b/src/ordersim/connectors/binance/_parsing.py index 1e4749a..e80fa9c 100644 --- a/src/ordersim/connectors/binance/_parsing.py +++ b/src/ordersim/connectors/binance/_parsing.py @@ -11,6 +11,7 @@ BinanceDepthSnapshot, BinanceDepthUpdate, BinancePriceLevel, + BinanceRawTrade, CaptureKind, CaptureScope, DepthStreamKind, @@ -34,6 +35,10 @@ def parse_envelope(raw: object) -> BinanceCaptureEnvelope: "connection_error", "depth_snapshot", "message", + "raw_trade", + "raw_trade_gap", + "raw_trade_poll", + "raw_trade_poll_error", "sequence_gap", ), ) @@ -125,6 +130,25 @@ def parse_aggregate_trade( ) +def parse_raw_trade(envelope: BinanceCaptureEnvelope) -> BinanceRawTrade: + """Normalize one individually identified REST trade.""" + + payload = envelope.payload + return BinanceRawTrade( + symbol=envelope.symbol, + connection_id=envelope.connection_id, + received_at_ns=envelope.received_at_ns, + received_monotonic_ns=envelope.received_monotonic_ns, + trade_id=required_int(payload, "id"), + price=required_decimal(payload, "price"), + quantity=required_decimal(payload, "qty"), + quote_quantity=required_decimal(payload, "quoteQty"), + trade_time_ns=milliseconds_to_nanoseconds(payload, "time"), + buyer_is_maker=required_bool(payload, "isBuyerMaker"), + is_rpi_trade=required_bool(payload, "isRPITrade"), + ) + + def parse_book_ticker(envelope: BinanceCaptureEnvelope) -> BinanceBookTicker: """Normalize one captured real-time book ticker.""" diff --git a/src/ordersim/connectors/binance/_recent_trades.py b/src/ordersim/connectors/binance/_recent_trades.py new file mode 100644 index 0000000..bb454a9 --- /dev/null +++ b/src/ordersim/connectors/binance/_recent_trades.py @@ -0,0 +1,206 @@ +"""HTTP transport and trade-ID tracking for Binance raw trades.""" + +import asyncio +import json +import time +import urllib.parse +import urllib.request +from collections.abc import Callable, Mapping +from dataclasses import dataclass +from typing import Protocol + +from ordersim.connectors.binance.schema import JsonObject + +RECENT_TRADES_URL = "https://fapi.binance.com/fapi/v1/trades" + + +@dataclass(frozen=True, slots=True) +class RecentTradesBatch: + """One response from the Binance recent-trades endpoint.""" + + trades: tuple[JsonObject, ...] + request_started_at_ns: int + request_finished_at_ns: int + used_weight_1m: int | None + + +class RecentTradesClient(Protocol): + """Fetch individual recent trades for one symbol.""" + + async def recent_trades( + self, + symbol: str, + limit: int, + ) -> RecentTradesBatch: + """Return one recent-trades response.""" + + +FetchUrl = Callable[[str], tuple[bytes, Mapping[str, str]]] + + +class HttpRecentTradesClient: + """Standard-library client for Binance individual recent trades.""" + + def __init__(self, *, fetch_url: FetchUrl | None = None) -> None: + self._fetch_url = fetch_url or _fetch_url + + async def recent_trades( + self, + symbol: str, + limit: int, + ) -> RecentTradesBatch: + """Fetch and validate one recent-trades response.""" + + query = urllib.parse.urlencode({"symbol": symbol, "limit": limit}) + started_at_ns = time.time_ns() + raw, headers = await asyncio.to_thread( + self._fetch_url, + f"{RECENT_TRADES_URL}?{query}", + ) + finished_at_ns = time.time_ns() + payload = json.loads(raw) + trades = _validate_trades(payload) + return RecentTradesBatch( + trades=trades, + request_started_at_ns=started_at_ns, + request_finished_at_ns=finished_at_ns, + used_weight_1m=_optional_header_int(headers, "x-mbx-used-weight-1m"), + ) + + +@dataclass(frozen=True, slots=True) +class RawTradeGap: + """One missing range in the observed individual trade IDs.""" + + expected_trade_id: int + first_received_trade_id: int + + def as_dict(self) -> dict[str, int]: + """Return a JSON-compatible representation.""" + + return { + "expected_trade_id": self.expected_trade_id, + "first_received_trade_id": self.first_received_trade_id, + "missing_count": self.first_received_trade_id - self.expected_trade_id, + } + + +@dataclass(frozen=True, slots=True) +class RawTradeSelection: + """New trades and any gap found in one overlapping response.""" + + trades: tuple[JsonObject, ...] + gap: RawTradeGap | None + + +class RawTradeCursor: + """Deduplicate overlapping responses and expose missing trade IDs.""" + + def __init__(self) -> None: + self._last_trade_id: int | None = None + + @property + def last_trade_id(self) -> int | None: + """Return the greatest trade ID observed so far.""" + + return self._last_trade_id + + def select(self, trades: tuple[JsonObject, ...]) -> RawTradeSelection: + """Return only unseen trades, preserving endpoint order.""" + + trade_ids = tuple(_trade_id(trade) for trade in trades) + adjacent_ids = zip(trade_ids, trade_ids[1:], strict=False) + if any(right <= left for left, right in adjacent_ids): + raise ValueError("Binance recent trades must have increasing IDs") + + previous = self._last_trade_id + if previous is None: + new_trades = trades + else: + new_trades = tuple( + trade + for trade, trade_id in zip(trades, trade_ids, strict=True) + if trade_id > previous + ) + + gap = None + if previous is not None and new_trades: + first_new_id = _trade_id(new_trades[0]) + if first_new_id > previous + 1: + gap = RawTradeGap( + expected_trade_id=previous + 1, + first_received_trade_id=first_new_id, + ) + + if new_trades: + self._last_trade_id = _trade_id(new_trades[-1]) + return RawTradeSelection(trades=new_trades, gap=gap) + + +def _validate_trades(payload: object) -> tuple[JsonObject, ...]: + if not isinstance(payload, list): + raise ValueError("Binance recent-trades response must be a list") + trades: list[JsonObject] = [] + for trade in payload: + if not isinstance(trade, dict): + raise ValueError("Binance recent trade must be a JSON object") + _trade_id(trade) + _required_int(trade, "time") + _required_str(trade, "price") + _required_str(trade, "qty") + _required_str(trade, "quoteQty") + _required_bool(trade, "isBuyerMaker") + _required_bool(trade, "isRPITrade") + trades.append(trade) + return tuple(trades) + + +def _trade_id(trade: Mapping[str, object]) -> int: + return _required_int(trade, "id") + + +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 raw-trade 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): + raise ValueError(f"Binance raw-trade field {field!r} must be a 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 raw-trade field {field!r} must be a boolean") + return value + + +def _optional_header_int( + headers: Mapping[str, str], + field: str, +) -> int | None: + value = next( + ( + header_value + for header_name, header_value in headers.items() + if header_name.lower() == field.lower() + ), + None, + ) + if value is None: + return None + try: + return int(value) + except ValueError as exc: + raise ValueError(f"Binance header {field!r} must be an integer") from exc + + +def _fetch_url(url: str) -> tuple[bytes, Mapping[str, str]]: + request = urllib.request.Request(url, headers={"User-Agent": "ordersim"}) + with urllib.request.urlopen(request, timeout=20) as response: + return response.read(), dict(response.headers.items()) diff --git a/src/ordersim/connectors/binance/_storage.py b/src/ordersim/connectors/binance/_storage.py index c8cecb8..e4df7be 100644 --- a/src/ordersim/connectors/binance/_storage.py +++ b/src/ordersim/connectors/binance/_storage.py @@ -13,6 +13,7 @@ from ordersim.connectors.binance.schema import ( CAPTURE_SCHEMA_VERSION, BinanceCaptureConfig, + BinanceRawTradeCaptureConfig, CaptureManifest, JsonObject, ) @@ -21,7 +22,10 @@ class RawCaptureSink: """Serialize capture envelopes to hourly gzip JSONL files.""" - def __init__(self, config: BinanceCaptureConfig) -> None: + def __init__( + self, + config: BinanceCaptureConfig | BinanceRawTradeCaptureConfig, + ) -> None: self._config = config self._run_id = uuid.uuid4().hex self._started_at_ns = time.time_ns() @@ -79,6 +83,7 @@ def close(self) -> CaptureManifest: include_rpi=self._config.include_rpi, counts=dict(sorted(self._counts.items())), files=tuple(self._files), + capture_type=self._config.capture_type, ) manifest_path = self._config.output_dir / f"manifest-{self._run_id}.json" manifest_path.write_text( diff --git a/src/ordersim/connectors/binance/l2.py b/src/ordersim/connectors/binance/l2.py index 14f6713..87b6f7a 100644 --- a/src/ordersim/connectors/binance/l2.py +++ b/src/ordersim/connectors/binance/l2.py @@ -11,6 +11,10 @@ "connection_error", "depth_snapshot", "message", + "raw_trade", + "raw_trade_gap", + "raw_trade_poll", + "raw_trade_poll_error", "sequence_gap", ] CaptureScope: TypeAlias = Literal["public", "market"] @@ -95,6 +99,23 @@ class BinanceAggregateTrade: buyer_is_maker: bool +@dataclass(frozen=True, slots=True) +class BinanceRawTrade: + """One individual trade returned by Binance USD-M.""" + + symbol: str + connection_id: str + received_at_ns: int + received_monotonic_ns: int + trade_id: int + price: Decimal + quantity: Decimal + quote_quantity: Decimal + trade_time_ns: int + buyer_is_maker: bool + is_rpi_trade: bool + + @dataclass(frozen=True, slots=True) class BinanceBookTicker: """One real-time Binance best-bid/ask observation.""" diff --git a/src/ordersim/connectors/binance/raw_trades.py b/src/ordersim/connectors/binance/raw_trades.py new file mode 100644 index 0000000..582654f --- /dev/null +++ b/src/ordersim/connectors/binance/raw_trades.py @@ -0,0 +1,198 @@ +"""Capture individual Binance USD-M trades through overlapping REST polls.""" + +import argparse +import asyncio +import uuid +from collections.abc import Sequence +from pathlib import Path + +from ordersim.connectors.binance._recent_trades import ( + HttpRecentTradesClient, + RawTradeCursor, + RecentTradesBatch, + RecentTradesClient, +) +from ordersim.connectors.binance._storage import RawCaptureSink +from ordersim.connectors.binance.schema import ( + BinanceRawTradeCaptureConfig, + CaptureManifest, +) + +RAW_TRADES_STREAM = "rest:/fapi/v1/trades" + + +async def capture_binance_raw_trades( + config: BinanceRawTradeCaptureConfig, + *, + client: RecentTradesClient | None = None, +) -> CaptureManifest: + """Capture individual trades until duration expiry or interruption.""" + + active_client = client or HttpRecentTradesClient() + sink = RawCaptureSink(config) + interrupted = False + tasks = [ + asyncio.create_task( + _poll_symbol( + config=config, + sink=sink, + client=active_client, + symbol=symbol, + ) + ) + for symbol in config.symbols + ] + 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 _poll_symbol( + *, + config: BinanceRawTradeCaptureConfig, + sink: RawCaptureSink, + client: RecentTradesClient, + symbol: str, +) -> None: + connection_id = uuid.uuid4().hex + cursor = RawTradeCursor() + while True: + try: + batch = await client.recent_trades(symbol, config.request_limit) + await _record_batch( + sink=sink, + symbol=symbol, + connection_id=connection_id, + cursor=cursor, + batch=batch, + ) + except asyncio.CancelledError: + raise + except Exception as exc: + await sink.write( + kind="raw_trade_poll_error", + scope="market", + symbol=symbol, + connection_id=connection_id, + stream=RAW_TRADES_STREAM, + payload={ + "error_type": type(exc).__name__, + "message": str(exc), + "last_trade_id": cursor.last_trade_id, + }, + ) + await asyncio.sleep(config.retry_delay_seconds) + continue + await asyncio.sleep(config.poll_interval_seconds) + + +async def _record_batch( + *, + sink: RawCaptureSink, + symbol: str, + connection_id: str, + cursor: RawTradeCursor, + batch: RecentTradesBatch, +) -> None: + selection = cursor.select(batch.trades) + first_trade_id = batch.trades[0]["id"] if batch.trades else None + last_trade_id = batch.trades[-1]["id"] if batch.trades else None + await sink.write( + kind="raw_trade_poll", + scope="market", + symbol=symbol, + connection_id=connection_id, + stream=RAW_TRADES_STREAM, + payload={ + "request_started_at_ns": batch.request_started_at_ns, + "request_finished_at_ns": batch.request_finished_at_ns, + "used_weight_1m": batch.used_weight_1m, + "returned_count": len(batch.trades), + "new_count": len(selection.trades), + "first_trade_id": first_trade_id, + "last_trade_id": last_trade_id, + }, + ) + if selection.gap is not None: + await sink.write( + kind="raw_trade_gap", + scope="market", + symbol=symbol, + connection_id=connection_id, + stream=RAW_TRADES_STREAM, + payload=selection.gap.as_dict(), + ) + for trade in selection.trades: + await sink.write( + kind="raw_trade", + scope="market", + symbol=symbol, + connection_id=connection_id, + stream=RAW_TRADES_STREAM, + payload=trade, + ) + + +def _parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description="Record individual Binance USD-M trades from REST." + ) + 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( + "--poll-interval-seconds", + type=float, + default=2.0, + help="Seconds between requests per symbol (default: 2).", + ) + return parser + + +def main(argv: Sequence[str] | None = None) -> None: + """Run the `ordersim-binance-raw-trades` command.""" + + args = _parser().parse_args(argv) + duration_seconds = ( + None if args.duration_hours is None else args.duration_hours * 60 * 60 + ) + config = BinanceRawTradeCaptureConfig( + output_dir=args.output_dir, + symbols=tuple(args.symbols), + duration_seconds=duration_seconds, + poll_interval_seconds=args.poll_interval_seconds, + ) + try: + manifest = asyncio.run(capture_binance_raw_trades(config)) + except KeyboardInterrupt: + return + print( + f"raw-trade capture {manifest.run_id} complete: " + f"{manifest.counts.get('raw_trade', 0)} trades" + ) + + +if __name__ == "__main__": + main() diff --git a/src/ordersim/connectors/binance/schema.py b/src/ordersim/connectors/binance/schema.py index 7736054..c4272ea 100644 --- a/src/ordersim/connectors/binance/schema.py +++ b/src/ordersim/connectors/binance/schema.py @@ -3,15 +3,20 @@ from collections.abc import Mapping from dataclasses import dataclass from pathlib import Path +from typing import ClassVar CAPTURE_SCHEMA_VERSION = 1 JsonObject = dict[str, object] +RECENT_TRADES_REQUEST_WEIGHT = 25 +RAW_TRADE_WEIGHT_BUDGET_PER_MINUTE = 1_800 @dataclass(frozen=True, slots=True) class BinanceCaptureConfig: """Configuration for one raw Binance USD-M futures capture.""" + capture_type: ClassVar[str] = "websocket" + output_dir: Path symbols: tuple[str, ...] duration_seconds: float | None = None @@ -55,6 +60,65 @@ def market_streams(self, symbol: str) -> tuple[str, ...]: return (f"{symbol.lower()}@aggTrade",) +@dataclass(frozen=True, slots=True) +class BinanceRawTradeCaptureConfig: + """Configuration for polling individual Binance USD-M trades.""" + + capture_type: ClassVar[str] = "raw_trades" + + output_dir: Path + symbols: tuple[str, ...] + duration_seconds: float | None = None + poll_interval_seconds: float = 2.0 + request_limit: int = 1000 + retry_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.poll_interval_seconds <= 0: + raise ValueError("poll_interval_seconds must be positive") + if not 1 <= self.request_limit <= 1000: + raise ValueError("request_limit must be between 1 and 1000") + if self.retry_delay_seconds < 0: + raise ValueError("retry_delay_seconds must be non-negative") + if ( + self.estimated_request_weight_per_minute + > RAW_TRADE_WEIGHT_BUDGET_PER_MINUTE + ): + raise ValueError( + "raw-trade polling would exceed the conservative " + f"{RAW_TRADE_WEIGHT_BUDGET_PER_MINUTE} weight/minute budget" + ) + + object.__setattr__(self, "output_dir", Path(self.output_dir)) + object.__setattr__(self, "symbols", symbols) + + @property + def include_rpi(self) -> bool: + """Return false because this recorder does not capture depth.""" + + return False + + @property + def estimated_request_weight_per_minute(self) -> float: + """Return the configured recent-trades request weight per minute.""" + + return ( + len(self.symbols) + * RECENT_TRADES_REQUEST_WEIGHT + * 60 + / self.poll_interval_seconds + ) + + @dataclass(frozen=True, slots=True) class DepthSequenceGap: """One discontinuity in a Binance diff-depth stream.""" @@ -114,6 +178,7 @@ class CaptureManifest: include_rpi: bool counts: dict[str, int] files: tuple[str, ...] + capture_type: str = "websocket" def as_dict(self) -> JsonObject: """Return a JSON-compatible representation.""" @@ -125,6 +190,7 @@ def as_dict(self) -> JsonObject: "ended_at_ns": self.ended_at_ns, "symbols": list(self.symbols), "include_rpi": self.include_rpi, + "capture_type": self.capture_type, "counts": self.counts, "files": list(self.files), } diff --git a/src/ordersim/connectors/binance/source.py b/src/ordersim/connectors/binance/source.py index 43dc9b0..84e44eb 100644 --- a/src/ordersim/connectors/binance/source.py +++ b/src/ordersim/connectors/binance/source.py @@ -14,6 +14,7 @@ parse_depth_snapshot, parse_depth_update, parse_envelope, + parse_raw_trade, required_int, ) from ordersim.connectors.binance.l2 import ( @@ -23,6 +24,7 @@ BinanceDepthEvent, BinanceDepthSnapshot, BinanceDepthUpdate, + BinanceRawTrade, DepthStreamKind, ) from ordersim.connectors.binance.schema import CAPTURE_SCHEMA_VERSION @@ -151,6 +153,13 @@ def aggregate_trades(self) -> Iterator[BinanceAggregateTrade]: if is_stream(envelope, "@aggTrade"): yield parse_aggregate_trade(envelope) + def raw_trades(self) -> Iterator[BinanceRawTrade]: + """Yield individually identified REST trades.""" + + for envelope in self.envelopes(): + if envelope.kind == "raw_trade": + yield parse_raw_trade(envelope) + def book_tickers(self) -> Iterator[BinanceBookTicker]: """Yield normalized best-bid/ask messages.""" diff --git a/tests/test_binance_raw_trades.py b/tests/test_binance_raw_trades.py new file mode 100644 index 0000000..ed04209 --- /dev/null +++ b/tests/test_binance_raw_trades.py @@ -0,0 +1,562 @@ +import asyncio +import gzip +import json +import runpy +import sys +from collections.abc import Mapping +from decimal import Decimal +from pathlib import Path +from types import SimpleNamespace + +import pytest + +import ordersim.connectors.binance._recent_trades as recent_trades_module +import ordersim.connectors.binance.raw_trades as raw_trades_module +from ordersim.connectors.binance._recent_trades import ( + HttpRecentTradesClient, + RawTradeCursor, + RecentTradesBatch, + _fetch_url, +) +from ordersim.connectors.binance._storage import RawCaptureSink +from ordersim.connectors.binance.l2 import BinanceRawTrade +from ordersim.connectors.binance.raw_trades import ( + _poll_symbol, + _record_batch, + capture_binance_raw_trades, + main, +) +from ordersim.connectors.binance.schema import BinanceRawTradeCaptureConfig +from ordersim.connectors.binance.source import BinanceCaptureSource + + +def raw_trade(trade_id: int) -> dict[str, object]: + return { + "id": trade_id, + "price": "63893.80", + "qty": "0.001", + "quoteQty": "63.89", + "time": 1_785_273_611_946 + trade_id, + "isBuyerMaker": False, + "isRPITrade": trade_id % 2 == 0, + } + + +def batch( + *trade_ids: int, + used_weight_1m: int | None = 25, +) -> RecentTradesBatch: + return RecentTradesBatch( + trades=tuple(raw_trade(trade_id) for trade_id in trade_ids), + request_started_at_ns=100, + request_finished_at_ns=200, + used_weight_1m=used_weight_1m, + ) + + +def read_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_raw_trade_config_normalizes_symbols_and_budgets_weight( + tmp_path: Path, +) -> None: + config = BinanceRawTradeCaptureConfig( + output_dir=tmp_path, + symbols=("btcusdt", "ETHUSDT"), + duration_seconds=60, + poll_interval_seconds=2, + ) + + assert config.symbols == ("BTCUSDT", "ETHUSDT") + assert config.include_rpi is False + assert config.estimated_request_weight_per_minute == 1500 + + +@pytest.mark.parametrize( + ("symbols", "duration", "interval", "limit", "delay", "message"), + [ + ((), None, 2, 1000, 2, "at least one symbol"), + (("BTC-USDT",), None, 2, 1000, 2, "letters and numbers"), + (("BTCUSDT", "btcusdt"), None, 2, 1000, 2, "unique"), + (("BTCUSDT",), 0, 2, 1000, 2, "positive"), + (("BTCUSDT",), None, 0, 1000, 2, "positive"), + (("BTCUSDT",), None, 2, 0, 2, "between 1 and 1000"), + (("BTCUSDT",), None, 2, 1001, 2, "between 1 and 1000"), + (("BTCUSDT",), None, 2, 1000, -1, "non-negative"), + ( + ("BTCUSDT", "ETHUSDT", "BNBUSDT"), + None, + 2, + 1000, + 2, + "weight/minute budget", + ), + ], +) +def test_raw_trade_config_rejects_invalid_values( + tmp_path: Path, + symbols: tuple[str, ...], + duration: float | None, + interval: float, + limit: int, + delay: float, + message: str, +) -> None: + with pytest.raises(ValueError, match=message): + BinanceRawTradeCaptureConfig( + output_dir=tmp_path, + symbols=symbols, + duration_seconds=duration, + poll_interval_seconds=interval, + request_limit=limit, + retry_delay_seconds=delay, + ) + + +def test_http_client_preserves_individual_trade_payloads() -> None: + requested_urls: list[str] = [] + + def fake_fetch(url: str) -> tuple[bytes, Mapping[str, str]]: + requested_urls.append(url) + return ( + json.dumps([raw_trade(10), raw_trade(11)]).encode(), + {"X-MBX-USED-WEIGHT-1M": "125"}, + ) + + response = asyncio.run( + HttpRecentTradesClient(fetch_url=fake_fetch).recent_trades("BTCUSDT", 1000) + ) + + assert "symbol=BTCUSDT" in requested_urls[0] + assert "limit=1000" in requested_urls[0] + assert response.trades == (raw_trade(10), raw_trade(11)) + assert response.used_weight_1m == 125 + assert response.request_finished_at_ns >= response.request_started_at_ns + + +@pytest.mark.parametrize( + ("payload", "message"), + [ + ({}, "must be a list"), + ([[]], "JSON object"), + ([{"id": "1"}], "'id' must be an integer"), + ([{**raw_trade(1), "time": True}], "'time' must be an integer"), + ([{**raw_trade(1), "price": 1}], "'price' must be a string"), + ([{**raw_trade(1), "isRPITrade": "false"}], "must be a boolean"), + ], +) +def test_http_client_rejects_malformed_trade_payloads( + payload: object, + message: str, +) -> None: + client = HttpRecentTradesClient( + fetch_url=lambda url: (json.dumps(payload).encode(), {}) + ) + + with pytest.raises(ValueError, match=message): + asyncio.run(client.recent_trades("BTCUSDT", 1000)) + + +def test_http_client_rejects_malformed_weight_header() -> None: + client = HttpRecentTradesClient( + fetch_url=lambda url: ( + json.dumps([raw_trade(1)]).encode(), + {"x-mbx-used-weight-1m": "many"}, + ) + ) + + with pytest.raises(ValueError, match="header"): + asyncio.run(client.recent_trades("BTCUSDT", 1000)) + + +def test_http_client_allows_missing_weight_header() -> None: + client = HttpRecentTradesClient( + fetch_url=lambda url: (json.dumps([raw_trade(1)]).encode(), {}) + ) + + response = asyncio.run(client.recent_trades("BTCUSDT", 1000)) + + assert response.used_weight_1m is None + + +def test_fetch_url_sets_user_agent_and_returns_headers( + monkeypatch: pytest.MonkeyPatch, +) -> None: + class FakeResponse: + headers = {"x-mbx-used-weight-1m": "25"} + + def __enter__(self) -> "FakeResponse": + return self + + def __exit__(self, *args: object) -> None: + return None + + def read(self) -> bytes: + return b"[]" + + def fake_urlopen(request: object, *, timeout: int) -> FakeResponse: + assert request.get_header("User-agent") == "ordersim" + assert timeout == 20 + return FakeResponse() + + monkeypatch.setattr( + recent_trades_module.urllib.request, + "urlopen", + fake_urlopen, + ) + + body, headers = _fetch_url("https://example.test") + + assert body == b"[]" + assert headers["x-mbx-used-weight-1m"] == "25" + + +def test_cursor_deduplicates_overlap_and_reports_gaps() -> None: + cursor = RawTradeCursor() + + first = cursor.select((raw_trade(10), raw_trade(11), raw_trade(12))) + overlap = cursor.select((raw_trade(11), raw_trade(12), raw_trade(13))) + gap = cursor.select((raw_trade(15), raw_trade(16))) + + assert [trade["id"] for trade in first.trades] == [10, 11, 12] + assert [trade["id"] for trade in overlap.trades] == [13] + assert overlap.gap is None + assert gap.gap is not None + assert gap.gap.as_dict() == { + "expected_trade_id": 14, + "first_received_trade_id": 15, + "missing_count": 1, + } + assert cursor.last_trade_id == 16 + + +def test_cursor_handles_empty_and_fully_stale_responses() -> None: + cursor = RawTradeCursor() + + assert cursor.select(()).trades == () + cursor.select((raw_trade(10), raw_trade(11))) + stale = cursor.select((raw_trade(8), raw_trade(9))) + + assert stale.trades == () + assert stale.gap is None + assert cursor.last_trade_id == 11 + + +def test_cursor_rejects_non_increasing_trade_ids() -> None: + cursor = RawTradeCursor() + + with pytest.raises(ValueError, match="increasing IDs"): + cursor.select((raw_trade(10), raw_trade(10))) + + +def test_record_batch_writes_poll_gap_and_exact_new_trades(tmp_path: Path) -> None: + config = BinanceRawTradeCaptureConfig( + output_dir=tmp_path, + symbols=("BTCUSDT",), + ) + sink = RawCaptureSink(config) + cursor = RawTradeCursor() + + async def exercise() -> None: + await _record_batch( + sink=sink, + symbol="BTCUSDT", + connection_id="raw-1", + cursor=cursor, + batch=batch(10, 11), + ) + await _record_batch( + sink=sink, + symbol="BTCUSDT", + connection_id="raw-1", + cursor=cursor, + batch=batch(11, 13), + ) + + asyncio.run(exercise()) + manifest = sink.close() + records = read_records(tmp_path) + + assert [record["kind"] for record in records] == [ + "raw_trade_poll", + "raw_trade", + "raw_trade", + "raw_trade_poll", + "raw_trade_gap", + "raw_trade", + ] + assert records[-1]["payload"] == raw_trade(13) + assert records[3]["payload"]["new_count"] == 1 + assert manifest.capture_type == "raw_trades" + assert manifest.counts == { + "raw_trade": 3, + "raw_trade_gap": 1, + "raw_trade_poll": 2, + } + + +def test_capture_source_normalizes_individual_raw_trades(tmp_path: Path) -> None: + config = BinanceRawTradeCaptureConfig( + output_dir=tmp_path, + symbols=("BTCUSDT",), + ) + sink = RawCaptureSink(config) + + asyncio.run( + _record_batch( + sink=sink, + symbol="BTCUSDT", + connection_id="raw-1", + cursor=RawTradeCursor(), + batch=batch(10), + ) + ) + manifest = sink.close() + + trade = next( + BinanceCaptureSource.from_manifest( + tmp_path / f"manifest-{manifest.run_id}.json" + ).raw_trades() + ) + + assert trade == BinanceRawTrade( + symbol="BTCUSDT", + connection_id="raw-1", + received_at_ns=trade.received_at_ns, + received_monotonic_ns=trade.received_monotonic_ns, + trade_id=10, + price=Decimal("63893.80"), + quantity=Decimal("0.001"), + quote_quantity=Decimal("63.89"), + trade_time_ns=1_785_273_611_956_000_000, + buyer_is_maker=False, + is_rpi_trade=True, + ) + + +def test_capture_source_handles_file_without_raw_trades(tmp_path: Path) -> None: + capture_path = tmp_path / "empty.jsonl.gz" + with gzip.open(capture_path, mode="wt", encoding="utf-8"): + pass + + assert tuple(BinanceCaptureSource((capture_path,)).raw_trades()) == () + + +class FakeRecentTradesClient: + def __init__(self, responses: dict[str, RecentTradesBatch]) -> None: + self.responses = responses + self.calls: list[tuple[str, int]] = [] + + async def recent_trades( + self, + symbol: str, + limit: int, + ) -> RecentTradesBatch: + self.calls.append((symbol, limit)) + return self.responses[symbol] + + +def test_finite_capture_polls_each_symbol_and_writes_manifest( + tmp_path: Path, +) -> None: + config = BinanceRawTradeCaptureConfig( + output_dir=tmp_path, + symbols=("BTCUSDT", "ETHUSDT"), + duration_seconds=0.01, + ) + client = FakeRecentTradesClient( + { + "BTCUSDT": batch(10), + "ETHUSDT": batch(20), + } + ) + + manifest = asyncio.run(capture_binance_raw_trades(config, client=client)) + + assert sorted(client.calls) == [("BTCUSDT", 1000), ("ETHUSDT", 1000)] + assert manifest.counts == {"raw_trade": 2, "raw_trade_poll": 2} + assert next(tmp_path.glob("manifest-*.json")).exists() + + +def test_poll_loop_records_errors_before_retry( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + config = BinanceRawTradeCaptureConfig( + output_dir=tmp_path, + symbols=("BTCUSDT",), + retry_delay_seconds=0, + ) + sink = RawCaptureSink(config) + second_attempt = asyncio.Event() + attempts = 0 + + class FailingClient: + async def recent_trades( + self, + symbol: str, + limit: int, + ) -> RecentTradesBatch: + nonlocal attempts + attempts += 1 + if attempts == 1: + raise ConnectionError("test failure") + second_attempt.set() + await asyncio.Event().wait() + return batch() + + monkeypatch.setattr( + raw_trades_module.uuid, + "uuid4", + lambda: SimpleNamespace(hex="raw-1"), + ) + + async def exercise() -> None: + task = asyncio.create_task( + _poll_symbol( + config=config, + sink=sink, + client=FailingClient(), + symbol="BTCUSDT", + ) + ) + await second_attempt.wait() + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + asyncio.run(exercise()) + sink.close() + record = read_records(tmp_path)[0] + + assert record["kind"] == "raw_trade_poll_error" + assert record["connection_id"] == "raw-1" + assert record["payload"] == { + "error_type": "ConnectionError", + "message": "test failure", + "last_trade_id": None, + } + + +def test_cancelled_capture_still_writes_manifest(tmp_path: Path) -> None: + config = BinanceRawTradeCaptureConfig( + output_dir=tmp_path, + symbols=("BTCUSDT",), + ) + + class BlockingClient: + def __init__(self) -> None: + self.started = asyncio.Event() + + async def recent_trades( + self, + symbol: str, + limit: int, + ) -> RecentTradesBatch: + self.started.set() + await asyncio.Event().wait() + return batch() + + client = BlockingClient() + + async def exercise() -> None: + task = asyncio.create_task( + capture_binance_raw_trades(config, client=client) + ) + await client.started.wait() + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + asyncio.run(exercise()) + + assert next(tmp_path.glob("manifest-*.json")).exists() + + +def test_main_builds_config_and_reports_trade_count( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + observed: list[BinanceRawTradeCaptureConfig] = [] + + async def fake_capture( + config: BinanceRawTradeCaptureConfig, + ) -> object: + observed.append(config) + return type( + "Manifest", + (), + {"run_id": "run-1", "counts": {"raw_trade": 7}}, + )() + + monkeypatch.setattr(raw_trades_module, "capture_binance_raw_trades", fake_capture) + + main( + [ + str(tmp_path), + "--symbol", + "BTCUSDT", + "--duration-hours", + "1.5", + "--poll-interval-seconds", + "2.5", + ] + ) + + assert observed[0].duration_seconds == 5_400 + assert observed[0].poll_interval_seconds == 2.5 + assert "7 trades" in capsys.readouterr().out + + +def test_main_handles_keyboard_interrupt( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + async def interrupted(config: BinanceRawTradeCaptureConfig) -> object: + raise KeyboardInterrupt + + monkeypatch.setattr(raw_trades_module, "capture_binance_raw_trades", interrupted) + + main([str(tmp_path), "--symbol", "BTCUSDT"]) + + +@pytest.mark.filterwarnings( + "ignore:.*found in sys.modules.*:RuntimeWarning" +) +def test_module_entry_point_runs_main( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + manifest = type( + "Manifest", + (), + {"run_id": "run-1", "counts": {"raw_trade": 3}}, + )() + + def fake_run(coroutine: object) -> object: + coroutine.close() + return manifest + + monkeypatch.setattr(asyncio, "run", fake_run) + monkeypatch.setattr( + sys, + "argv", + [ + "ordersim-binance-raw-trades", + str(tmp_path), + "--symbol", + "BTCUSDT", + ], + ) + + runpy.run_module( + "ordersim.connectors.binance.raw_trades", + run_name="__main__", + ) + + assert "3 trades" in capsys.readouterr().out From 825604789c91c4e9502450dadd7e58abf91c6d7d Mon Sep 17 00:00:00 2001 From: Tibor Date: Tue, 28 Jul 2026 23:38:38 +0200 Subject: [PATCH 2/4] Tolerate late Binance trade visibility --- CHANGELOG.md | 3 +- docs/connectors.md | 4 +- .../connectors/binance/_recent_trades.py | 76 ++++++++++++------- tests/test_binance_raw_trades.py | 45 ++++++++--- 4 files changed, 88 insertions(+), 40 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 888f9c4..c3ebbe7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,7 +13,8 @@ All notable public changes to `ordersim` are documented here. - Added snapshot bridging and `pu`/`u` continuity validation for standard Binance diff-depth segments. - Added rate-budgeted Binance individual-trade capture with overlapping REST - polls, trade-ID deduplication, explicit gap records, and RPI trade flags. + polls, late-ID tolerance, trade-ID deduplication, explicit gap records, and + RPI trade flags. - Added typed `BinanceRawTrade` records alongside aggregate trades so the more detailed public evidence is available to future reconstruction models. diff --git a/docs/connectors.md b/docs/connectors.md index 6374cd6..dc8128b 100644 --- a/docs/connectors.md +++ b/docs/connectors.md @@ -142,7 +142,9 @@ windows. It: - stores each exact individual trade payload once; - records request timing and Binance's reported one-minute request weight; - deduplicates overlapping responses by trade ID; -- writes `raw_trade_gap` whenever the next observed ID is not contiguous; +- accepts temporarily late IDs while they remain inside the overlap window; +- writes `raw_trade_gap` only after the 1,000-row endpoint window has moved + beyond a missing ID; - writes explicit poll errors rather than silently retrying. At Binance's current 25-unit request weight, two symbols polled every two diff --git a/src/ordersim/connectors/binance/_recent_trades.py b/src/ordersim/connectors/binance/_recent_trades.py index bb454a9..a71dd9f 100644 --- a/src/ordersim/connectors/binance/_recent_trades.py +++ b/src/ordersim/connectors/binance/_recent_trades.py @@ -94,47 +94,69 @@ class RawTradeSelection: class RawTradeCursor: - """Deduplicate overlapping responses and expose missing trade IDs.""" + """Deduplicate overlap and defer gaps until recovery is impossible.""" def __init__(self) -> None: - self._last_trade_id: int | None = None + self._next_expected_trade_id: int | None = None + self._greatest_observed_trade_id: int | None = None + self._pending_trade_ids: set[int] = set() @property def last_trade_id(self) -> int | None: """Return the greatest trade ID observed so far.""" - return self._last_trade_id + return self._greatest_observed_trade_id def select(self, trades: tuple[JsonObject, ...]) -> RawTradeSelection: - """Return only unseen trades, preserving endpoint order.""" + """Return unseen trades and any newly unrecoverable ID range.""" trade_ids = tuple(_trade_id(trade) for trade in trades) adjacent_ids = zip(trade_ids, trade_ids[1:], strict=False) if any(right <= left for left, right in adjacent_ids): raise ValueError("Binance recent trades must have increasing IDs") - - previous = self._last_trade_id - if previous is None: - new_trades = trades - else: - new_trades = tuple( - trade - for trade, trade_id in zip(trades, trade_ids, strict=True) - if trade_id > previous - ) - - gap = None - if previous is not None and new_trades: - first_new_id = _trade_id(new_trades[0]) - if first_new_id > previous + 1: - gap = RawTradeGap( - expected_trade_id=previous + 1, - first_received_trade_id=first_new_id, - ) - - if new_trades: - self._last_trade_id = _trade_id(new_trades[-1]) - return RawTradeSelection(trades=new_trades, gap=gap) + if not trade_ids: + return RawTradeSelection(trades=(), gap=None) + + if self._next_expected_trade_id is None: + self._next_expected_trade_id = trade_ids[0] + + gap = self._unrecoverable_gap(trade_ids[0]) + new_trades: list[JsonObject] = [] + for trade, trade_id in zip(trades, trade_ids, strict=True): + if ( + trade_id < self._next_expected_trade_id + or trade_id in self._pending_trade_ids + ): + continue + self._pending_trade_ids.add(trade_id) + new_trades.append(trade) + if ( + self._greatest_observed_trade_id is None + or trade_id > self._greatest_observed_trade_id + ): + self._greatest_observed_trade_id = trade_id + + while self._next_expected_trade_id in self._pending_trade_ids: + self._pending_trade_ids.remove(self._next_expected_trade_id) + self._next_expected_trade_id += 1 + + return RawTradeSelection(trades=tuple(new_trades), gap=gap) + + def _unrecoverable_gap(self, oldest_returned_trade_id: int) -> RawTradeGap | None: + expected = self._next_expected_trade_id + if expected is None or oldest_returned_trade_id <= expected: + return None + gap = RawTradeGap( + expected_trade_id=expected, + first_received_trade_id=oldest_returned_trade_id, + ) + self._next_expected_trade_id = oldest_returned_trade_id + self._pending_trade_ids = { + trade_id + for trade_id in self._pending_trade_ids + if trade_id >= oldest_returned_trade_id + } + return gap def _validate_trades(payload: object) -> tuple[JsonObject, ...]: diff --git a/tests/test_binance_raw_trades.py b/tests/test_binance_raw_trades.py index ed04209..1c2159e 100644 --- a/tests/test_binance_raw_trades.py +++ b/tests/test_binance_raw_trades.py @@ -214,23 +214,37 @@ def fake_urlopen(request: object, *, timeout: int) -> FakeResponse: assert headers["x-mbx-used-weight-1m"] == "25" -def test_cursor_deduplicates_overlap_and_reports_gaps() -> None: +def test_cursor_deduplicates_overlap_and_accepts_late_ids() -> None: cursor = RawTradeCursor() first = cursor.select((raw_trade(10), raw_trade(11), raw_trade(12))) - overlap = cursor.select((raw_trade(11), raw_trade(12), raw_trade(13))) - gap = cursor.select((raw_trade(15), raw_trade(16))) + delayed = cursor.select((raw_trade(11), raw_trade(12), raw_trade(14))) + recovered = cursor.select( + (raw_trade(12), raw_trade(13), raw_trade(14), raw_trade(15)) + ) assert [trade["id"] for trade in first.trades] == [10, 11, 12] - assert [trade["id"] for trade in overlap.trades] == [13] - assert overlap.gap is None + assert [trade["id"] for trade in delayed.trades] == [14] + assert delayed.gap is None + assert [trade["id"] for trade in recovered.trades] == [13, 15] + assert recovered.gap is None + assert cursor.last_trade_id == 15 + + +def test_cursor_reports_gap_after_endpoint_window_moves_past_it() -> None: + cursor = RawTradeCursor() + + cursor.select((raw_trade(10), raw_trade(11), raw_trade(13))) + gap = cursor.select((raw_trade(13), raw_trade(14))) + assert gap.gap is not None assert gap.gap.as_dict() == { - "expected_trade_id": 14, - "first_received_trade_id": 15, + "expected_trade_id": 12, + "first_received_trade_id": 13, "missing_count": 1, } - assert cursor.last_trade_id == 16 + assert [trade["id"] for trade in gap.trades] == [14] + assert cursor.last_trade_id == 14 def test_cursor_handles_empty_and_fully_stale_responses() -> None: @@ -275,6 +289,13 @@ async def exercise() -> None: cursor=cursor, batch=batch(11, 13), ) + await _record_batch( + sink=sink, + symbol="BTCUSDT", + connection_id="raw-1", + cursor=cursor, + batch=batch(13, 14), + ) asyncio.run(exercise()) manifest = sink.close() @@ -285,16 +306,18 @@ async def exercise() -> None: "raw_trade", "raw_trade", "raw_trade_poll", + "raw_trade", + "raw_trade_poll", "raw_trade_gap", "raw_trade", ] - assert records[-1]["payload"] == raw_trade(13) + assert records[-1]["payload"] == raw_trade(14) assert records[3]["payload"]["new_count"] == 1 assert manifest.capture_type == "raw_trades" assert manifest.counts == { - "raw_trade": 3, + "raw_trade": 4, "raw_trade_gap": 1, - "raw_trade_poll": 2, + "raw_trade_poll": 3, } From f2605841ad9d6d220203f16e35d95b093cf8cb04 Mon Sep 17 00:00:00 2001 From: Tibor Date: Tue, 28 Jul 2026 23:45:03 +0200 Subject: [PATCH 3/4] Poll Binance raw trades at burst-safe cadence --- docs/connectors.md | 12 +++++++----- src/ordersim/connectors/binance/raw_trades.py | 4 ++-- src/ordersim/connectors/binance/schema.py | 4 ++-- tests/test_binance_raw_trades.py | 6 +++--- 4 files changed, 14 insertions(+), 12 deletions(-) diff --git a/docs/connectors.md b/docs/connectors.md index dc8128b..8f99d9d 100644 --- a/docs/connectors.md +++ b/docs/connectors.md @@ -94,7 +94,7 @@ ordersim-binance-raw-trades captures/binance-raw-trades \ --symbol BTCUSDT \ --symbol ETHUSDT \ --duration-hours 72 \ - --poll-interval-seconds 2 + --poll-interval-seconds 0.5 ``` The recorder uses Binance's USD-M futures sources: @@ -147,10 +147,12 @@ windows. It: beyond a missing ID; - writes explicit poll errors rather than silently retrying. -At Binance's current 25-unit request weight, two symbols polled every two -seconds consume an estimated 1,500 units per minute. Configuration is rejected -when it would exceed the recorder's conservative 1,800-unit budget. This leaves -headroom below Binance's venue limit for snapshots and operational variance. +At Binance's current 5-unit request weight, two symbols polled every 500 ms +consume an estimated 1,200 units per minute. Configuration is rejected when it +would exceed the recorder's conservative 1,800-unit budget. This leaves +headroom below Binance's venue limit for snapshots and operational variance +while keeping the 1,000-trade overlap window ahead of observed BTCUSDT and +ETHUSDT bursts. The aggregate-trade stream remains valuable as an independent reconciliation feed. It is not treated as a substitute for individual trades when individual diff --git a/src/ordersim/connectors/binance/raw_trades.py b/src/ordersim/connectors/binance/raw_trades.py index 582654f..ac8ee06 100644 --- a/src/ordersim/connectors/binance/raw_trades.py +++ b/src/ordersim/connectors/binance/raw_trades.py @@ -165,8 +165,8 @@ def _parser() -> argparse.ArgumentParser: parser.add_argument( "--poll-interval-seconds", type=float, - default=2.0, - help="Seconds between requests per symbol (default: 2).", + default=0.5, + help="Seconds between requests per symbol (default: 0.5).", ) return parser diff --git a/src/ordersim/connectors/binance/schema.py b/src/ordersim/connectors/binance/schema.py index c4272ea..2192f92 100644 --- a/src/ordersim/connectors/binance/schema.py +++ b/src/ordersim/connectors/binance/schema.py @@ -7,7 +7,7 @@ CAPTURE_SCHEMA_VERSION = 1 JsonObject = dict[str, object] -RECENT_TRADES_REQUEST_WEIGHT = 25 +RECENT_TRADES_REQUEST_WEIGHT = 5 RAW_TRADE_WEIGHT_BUDGET_PER_MINUTE = 1_800 @@ -69,7 +69,7 @@ class BinanceRawTradeCaptureConfig: output_dir: Path symbols: tuple[str, ...] duration_seconds: float | None = None - poll_interval_seconds: float = 2.0 + poll_interval_seconds: float = 0.5 request_limit: int = 1000 retry_delay_seconds: float = 2.0 diff --git a/tests/test_binance_raw_trades.py b/tests/test_binance_raw_trades.py index 1c2159e..44fe288 100644 --- a/tests/test_binance_raw_trades.py +++ b/tests/test_binance_raw_trades.py @@ -67,12 +67,12 @@ def test_raw_trade_config_normalizes_symbols_and_budgets_weight( output_dir=tmp_path, symbols=("btcusdt", "ETHUSDT"), duration_seconds=60, - poll_interval_seconds=2, + poll_interval_seconds=0.5, ) assert config.symbols == ("BTCUSDT", "ETHUSDT") assert config.include_rpi is False - assert config.estimated_request_weight_per_minute == 1500 + assert config.estimated_request_weight_per_minute == 1200 @pytest.mark.parametrize( @@ -89,7 +89,7 @@ def test_raw_trade_config_normalizes_symbols_and_budgets_weight( ( ("BTCUSDT", "ETHUSDT", "BNBUSDT"), None, - 2, + 0.25, 1000, 2, "weight/minute budget", From cea0861fd6a1fca6f16891513c116d3b64362749 Mon Sep 17 00:00:00 2001 From: Tibor Date: Tue, 28 Jul 2026 23:54:28 +0200 Subject: [PATCH 4/4] Capture individual Binance trade stream --- CHANGELOG.md | 3 + docs/connectors.md | 45 +++++---- docs/data-guide.md | 14 +-- docs/schema.md | 4 +- src/ordersim/connectors/binance/__init__.py | 2 + src/ordersim/connectors/binance/_parsing.py | 23 +++++ src/ordersim/connectors/binance/_transport.py | 1 + src/ordersim/connectors/binance/capture.py | 58 ++++++++---- src/ordersim/connectors/binance/l2.py | 19 +++- src/ordersim/connectors/binance/schema.py | 44 ++++++++- src/ordersim/connectors/binance/source.py | 11 ++- tests/test_binance_capture.py | 92 ++++++++++++++++++- tests/test_binance_l2_source.py | 43 +++++++++ 13 files changed, 313 insertions(+), 46 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c3ebbe7..3d87fa6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,9 @@ All notable public changes to `ordersim` are documented here. - Added rate-budgeted Binance individual-trade capture with overlapping REST polls, late-ID tolerance, trade-ID deduplication, explicit gap records, and RPI trade flags. +- Added the real-time Binance individual `@trade` stream to the main capture, + with explicit trade-ID discontinuity records; aggregate trades remain + reconciliation evidence. - Added typed `BinanceRawTrade` records alongside aggregate trades so the more detailed public evidence is available to future reconstruction models. diff --git a/docs/connectors.md b/docs/connectors.md index 8f99d9d..c832c8b 100644 --- a/docs/connectors.md +++ b/docs/connectors.md @@ -75,8 +75,8 @@ Install the optional WebSocket dependency: pip install "ordersim[binance]" ``` -Record three days of standard depth, aggregate trades, real-time top-of-book, -and the optional RPI depth stream: +Record three days of standard depth, individual and aggregate trades, +real-time top-of-book, and the optional RPI depth stream: ```bash ordersim-binance-capture captures/binance \ @@ -86,8 +86,9 @@ ordersim-binance-capture captures/binance \ --include-rpi ``` -For the most detailed public trade evidence, run the individual-trade recorder -beside the WebSocket capture: +The capture command records Binance's real-time individual `@trade` stream. +Run the REST recorder beside it to retain the endpoint's additional +`isRPITrade` classification and overlapping-window audit evidence: ```bash ordersim-binance-raw-trades captures/binance-raw-trades \ @@ -102,8 +103,9 @@ The recorder uses Binance's USD-M futures sources: | Evidence | Source behavior | |---|---| | Diff depth | Absolute price-level quantities at up to 100 ms updates. | +| Individual trade stream | Real-time trades with one unique trade ID per message. | | Aggregate trades | Trades grouped by price and taking side over 100 ms. | -| Individual trades | REST recent trades with unique trade IDs and RPI flags. | +| REST individual trades | Recent trades with unique trade IDs and RPI flags. | | 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. | @@ -119,7 +121,8 @@ Each raw exchange payload is preserved inside a gzip JSONL envelope with: 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. +connection segment and obtains a new REST snapshot. A `trade_gap` row records +any non-consecutive trade ID observed within one individual-trade connection. These files are intentionally not canonical replay data. Binance depth has no stable public order IDs, and individual additions and cancellations inside an @@ -132,9 +135,15 @@ repository. ### Individual Trade Capture -USD-M's documented WebSocket market stream exposes `aggTrade`, not an -individual-trade stream. The public `/fapi/v1/trades` REST endpoint is more -detailed: each row has its own trade ID and `isRPITrade` flag. +Binance USD-M currently emits an individual `@trade` WebSocket message for +each trade. This is the primary real-time trade source: it avoids the +information loss in `aggTrade` and avoids a rolling REST window during bursts. +The raw payload is preserved without assigning semantics to fields that +Binance has not documented publicly. + +The public `/fapi/v1/trades` REST endpoint complements the stream. Each row has +the same trade ID plus an `isRPITrade` flag that is not present in the +WebSocket message. `ordersim-binance-raw-trades` polls that endpoint with overlapping 1,000-row windows. It: @@ -151,12 +160,11 @@ At Binance's current 5-unit request weight, two symbols polled every 500 ms consume an estimated 1,200 units per minute. Configuration is rejected when it would exceed the recorder's conservative 1,800-unit budget. This leaves headroom below Binance's venue limit for snapshots and operational variance -while keeping the 1,000-trade overlap window ahead of observed BTCUSDT and -ETHUSDT bursts. +for normal operation. The recorder is supplemental, not a claim that polling +alone can remain ahead of every 1,000-trade burst. -The aggregate-trade stream remains valuable as an independent reconciliation -feed. It is not treated as a substitute for individual trades when individual -trade evidence is available. +The aggregate-trade stream remains valuable only as an independent +reconciliation feed. It is not treated as a substitute for individual trades. ### Reading Completed Captures @@ -176,15 +184,18 @@ for event in source.validated_depth_events(): for trade in source.aggregate_trades(): print(trade) +for trade in source.individual_trades(): + print(trade) + for trade in source.raw_trades(): print(trade) ``` This is a typed Binance source, not the canonical `DataSource` protocol. It emits `BinanceDepthSnapshot`, `BinanceDepthUpdate`, -`BinanceAggregateTrade`, `BinanceRawTrade`, and `BinanceBookTicker` records -rather than `MBOEvent`. Passing it directly to `Replay` is intentionally -unsupported. +`BinanceIndividualTrade`, `BinanceAggregateTrade`, `BinanceRawTrade`, 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 diff --git a/docs/data-guide.md b/docs/data-guide.md index 525f3ec..283b303 100644 --- a/docs/data-guide.md +++ b/docs/data-guide.md @@ -103,15 +103,15 @@ 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. +records raw L2 depth, individual and aggregate trades, and integrity metadata, +but its output 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, individual -trades, and book tickers. Capture individual trades with -`ordersim-binance-raw-trades`; retain `aggTrade` as a reconciliation feed rather -than using it as a lower-detail substitute. That typed source is the input -boundary for the planned named virtual-L3 reconstruction model: +snapshots, sequence-validated depth updates, trades, and book tickers. The main +capture preserves Binance's individual `@trade` stream. Run +`ordersim-binance-raw-trades` beside it for REST reconciliation and RPI trade +flags; retain `aggTrade` only as another reconciliation feed. That typed source +is the input boundary for the planned named virtual-L3 reconstruction model: ```text raw capture -> BinanceCaptureSource -> named model -> modeled MBO + manifest diff --git a/docs/schema.md b/docs/schema.md index 4c9c797..15c5aac 100644 --- a/docs/schema.md +++ b/docs/schema.md @@ -32,7 +32,8 @@ quantity at that price; zero means remove the level. | `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. | -| `BinanceRawTrade` | `trade_id`, `price`, `quantity`, `quote_quantity`, `buyer_is_maker`, `is_rpi_trade` | One individually identified public trade. | +| `BinanceIndividualTrade` | `trade_id`, `price`, `quantity`, `buyer_is_maker` | One real-time individually identified WebSocket trade. | +| `BinanceRawTrade` | `trade_id`, `price`, `quantity`, `quote_quantity`, `buyer_is_maker`, `is_rpi_trade` | One individually identified REST trade. | | `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 @@ -48,6 +49,7 @@ unit and exact conversion rule before producing the canonical integer Raw-trade capture files also contain audit envelopes: +- `trade_gap` records a non-consecutive individual WebSocket trade ID; - `raw_trade_poll` records request timing, returned ID bounds, and request weight; - `raw_trade_gap` records a missing individual trade-ID range; diff --git a/src/ordersim/connectors/binance/__init__.py b/src/ordersim/connectors/binance/__init__.py index 477b1c1..938722b 100644 --- a/src/ordersim/connectors/binance/__init__.py +++ b/src/ordersim/connectors/binance/__init__.py @@ -12,6 +12,7 @@ BinanceDepthEvent, BinanceDepthSnapshot, BinanceDepthUpdate, + BinanceIndividualTrade, BinancePriceLevel, BinanceRawTrade, DepthStreamKind, @@ -36,6 +37,7 @@ "BinanceDepthEvent", "BinanceDepthSnapshot", "BinanceDepthUpdate", + "BinanceIndividualTrade", "BinancePriceLevel", "BinanceRawTrade", "BinanceRawTradeCaptureConfig", diff --git a/src/ordersim/connectors/binance/_parsing.py b/src/ordersim/connectors/binance/_parsing.py index e80fa9c..7b902d2 100644 --- a/src/ordersim/connectors/binance/_parsing.py +++ b/src/ordersim/connectors/binance/_parsing.py @@ -10,6 +10,7 @@ BinanceCaptureEnvelope, BinanceDepthSnapshot, BinanceDepthUpdate, + BinanceIndividualTrade, BinancePriceLevel, BinanceRawTrade, CaptureKind, @@ -40,6 +41,7 @@ def parse_envelope(raw: object) -> BinanceCaptureEnvelope: "raw_trade_poll", "raw_trade_poll_error", "sequence_gap", + "trade_gap", ), ) scope = required_choice(raw, "scope", ("public", "market")) @@ -130,6 +132,27 @@ def parse_aggregate_trade( ) +def parse_individual_trade( + envelope: BinanceCaptureEnvelope, +) -> BinanceIndividualTrade: + """Normalize one individually identified WebSocket trade.""" + + payload = envelope.payload + check_payload_symbol(envelope) + return BinanceIndividualTrade( + 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, + trade_id=required_int(payload, "t"), + price=required_decimal(payload, "p"), + quantity=required_decimal(payload, "q"), + buyer_is_maker=required_bool(payload, "m"), + ) + + def parse_raw_trade(envelope: BinanceCaptureEnvelope) -> BinanceRawTrade: """Normalize one individually identified REST trade.""" diff --git a/src/ordersim/connectors/binance/_transport.py b/src/ordersim/connectors/binance/_transport.py index 8a97298..8e7ad59 100644 --- a/src/ordersim/connectors/binance/_transport.py +++ b/src/ordersim/connectors/binance/_transport.py @@ -11,6 +11,7 @@ PUBLIC_STREAM_URL = "wss://fstream.binance.com/public/stream?streams=" MARKET_STREAM_URL = "wss://fstream.binance.com/market/stream?streams=" +INDIVIDUAL_TRADE_STREAM_URL = "wss://fstream.binance.com/stream?streams=" DEPTH_SNAPSHOT_URL = "https://fapi.binance.com/fapi/v1/depth" diff --git a/src/ordersim/connectors/binance/capture.py b/src/ordersim/connectors/binance/capture.py index 3385826..e907775 100644 --- a/src/ordersim/connectors/binance/capture.py +++ b/src/ordersim/connectors/binance/capture.py @@ -1,9 +1,10 @@ """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. +connection identifiers, and sequence-gap records. Individual and aggregate +trades are both retained. The recorder deliberately does not turn aggregated +depth into `MBOEvent` rows; that inference belongs to a named reconstruction +model. """ import argparse @@ -14,6 +15,7 @@ from ordersim.connectors.binance._storage import RawCaptureSink from ordersim.connectors.binance._transport import ( + INDIVIDUAL_TRADE_STREAM_URL, MARKET_STREAM_URL, PUBLIC_STREAM_URL, Transport, @@ -23,6 +25,7 @@ BinanceCaptureConfig, CaptureManifest, DepthSequenceTracker, + TradeSequenceTracker, ) @@ -47,7 +50,7 @@ async def capture_binance( ) ) for symbol in config.symbols - for scope in ("public", "market") + for scope in ("public", "market", "individual") ] try: if config.duration_seconds is None: @@ -74,6 +77,7 @@ async def _capture_with_reconnect( symbol: str, scope: str, ) -> None: + capture_scope = "public" if scope == "public" else "market" while True: connection_id = uuid.uuid4().hex try: @@ -90,7 +94,7 @@ async def _capture_with_reconnect( except Exception as exc: await sink.write( kind="connection_error", - scope=scope, + scope=capture_scope, symbol=symbol, connection_id=connection_id, stream=None, @@ -111,17 +115,27 @@ async def _capture_connection( 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 + if scope == "public": + streams = config.public_streams(symbol) + base_url = PUBLIC_STREAM_URL + capture_scope = "public" + elif scope == "market": + streams = config.market_streams(symbol) + base_url = MARKET_STREAM_URL + capture_scope = "market" + elif scope == "individual": + streams = config.individual_trade_streams(symbol) + base_url = INDIVIDUAL_TRADE_STREAM_URL + capture_scope = "market" + else: + raise ValueError(f"unsupported Binance capture scope {scope!r}") + trackers: dict[str, DepthSequenceTracker] = {} + trade_tracker = TradeSequenceTracker() async with transport.connect(base_url + "/".join(streams)) as messages: await sink.write( kind="connection_open", - scope=scope, + scope=capture_scope, symbol=symbol, connection_id=connection_id, stream=None, @@ -131,7 +145,7 @@ async def _capture_connection( snapshot = await transport.depth_snapshot(symbol, config.snapshot_limit) await sink.write( kind="depth_snapshot", - scope=scope, + scope=capture_scope, symbol=symbol, connection_id=connection_id, stream=None, @@ -141,12 +155,24 @@ async def _capture_connection( async for stream, payload in messages: await sink.write( kind="message", - scope=scope, + scope=capture_scope, symbol=symbol, connection_id=connection_id, stream=stream, payload=payload, ) + if stream.endswith("@trade"): + gap = trade_tracker.observe(payload) + if gap is not None: + await sink.write( + kind="trade_gap", + scope=capture_scope, + symbol=symbol, + connection_id=connection_id, + stream=stream, + payload=gap.as_dict(), + ) + continue if "@depth@" not in stream and "@rpiDepth@" not in stream: continue tracker = trackers.setdefault(stream, DepthSequenceTracker()) @@ -154,7 +180,7 @@ async def _capture_connection( if gap is not None: await sink.write( kind="sequence_gap", - scope=scope, + scope=capture_scope, symbol=symbol, connection_id=connection_id, stream=stream, @@ -164,7 +190,7 @@ async def _capture_connection( def _parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( - description="Record Binance USD-M depth and aggregate-trade evidence." + description="Record Binance USD-M depth and trade evidence." ) parser.add_argument("output_dir", type=Path) parser.add_argument( diff --git a/src/ordersim/connectors/binance/l2.py b/src/ordersim/connectors/binance/l2.py index 87b6f7a..0f03f7c 100644 --- a/src/ordersim/connectors/binance/l2.py +++ b/src/ordersim/connectors/binance/l2.py @@ -16,6 +16,7 @@ "raw_trade_poll", "raw_trade_poll_error", "sequence_gap", + "trade_gap", ] CaptureScope: TypeAlias = Literal["public", "market"] @@ -99,9 +100,25 @@ class BinanceAggregateTrade: buyer_is_maker: bool +@dataclass(frozen=True, slots=True) +class BinanceIndividualTrade: + """One real-time, individually identified Binance trade.""" + + symbol: str + connection_id: str + event_time_ns: int + trade_time_ns: int + received_at_ns: int + received_monotonic_ns: int + trade_id: int + price: Decimal + quantity: Decimal + buyer_is_maker: bool + + @dataclass(frozen=True, slots=True) class BinanceRawTrade: - """One individual trade returned by Binance USD-M.""" + """One individual trade returned by Binance USD-M REST.""" symbol: str connection_id: str diff --git a/src/ordersim/connectors/binance/schema.py b/src/ordersim/connectors/binance/schema.py index 2192f92..08a286b 100644 --- a/src/ordersim/connectors/binance/schema.py +++ b/src/ordersim/connectors/binance/schema.py @@ -55,10 +55,15 @@ def public_streams(self, symbol: str) -> tuple[str, ...]: return tuple(streams) def market_streams(self, symbol: str) -> tuple[str, ...]: - """Return the trade streams captured for one symbol.""" + """Return the aggregate-trade streams captured for one symbol.""" return (f"{symbol.lower()}@aggTrade",) + def individual_trade_streams(self, symbol: str) -> tuple[str, ...]: + """Return the individual-trade streams captured for one symbol.""" + + return (f"{symbol.lower()}@trade",) + @dataclass(frozen=True, slots=True) class BinanceRawTradeCaptureConfig: @@ -166,6 +171,43 @@ def observe(self, payload: Mapping[str, object]) -> DepthSequenceGap | None: ) +@dataclass(frozen=True, slots=True) +class TradeSequenceGap: + """One discontinuity in an individual-trade WebSocket stream.""" + + expected_trade_id: int + received_trade_id: int + + def as_dict(self) -> dict[str, int]: + """Return a JSON-compatible representation.""" + + return { + "expected_trade_id": self.expected_trade_id, + "received_trade_id": self.received_trade_id, + "missing_count": max(0, self.received_trade_id - self.expected_trade_id), + } + + +class TradeSequenceTracker: + """Validate individual trade-ID continuity within one connection.""" + + def __init__(self) -> None: + self._previous_trade_id: int | None = None + + def observe(self, payload: Mapping[str, object]) -> TradeSequenceGap | None: + """Observe one trade payload and return a discontinuity when present.""" + + trade_id = _required_int(payload, "t") + previous = self._previous_trade_id + self._previous_trade_id = trade_id + if previous is None or trade_id == previous + 1: + return None + return TradeSequenceGap( + expected_trade_id=previous + 1, + received_trade_id=trade_id, + ) + + @dataclass(frozen=True, slots=True) class CaptureManifest: """Summary of one recorder process.""" diff --git a/src/ordersim/connectors/binance/source.py b/src/ordersim/connectors/binance/source.py index 84e44eb..2f483f0 100644 --- a/src/ordersim/connectors/binance/source.py +++ b/src/ordersim/connectors/binance/source.py @@ -14,6 +14,7 @@ parse_depth_snapshot, parse_depth_update, parse_envelope, + parse_individual_trade, parse_raw_trade, required_int, ) @@ -24,6 +25,7 @@ BinanceDepthEvent, BinanceDepthSnapshot, BinanceDepthUpdate, + BinanceIndividualTrade, BinanceRawTrade, DepthStreamKind, ) @@ -153,8 +155,15 @@ def aggregate_trades(self) -> Iterator[BinanceAggregateTrade]: if is_stream(envelope, "@aggTrade"): yield parse_aggregate_trade(envelope) + def individual_trades(self) -> Iterator[BinanceIndividualTrade]: + """Yield real-time, individually identified WebSocket trades.""" + + for envelope in self.envelopes(): + if is_stream(envelope, "@trade"): + yield parse_individual_trade(envelope) + def raw_trades(self) -> Iterator[BinanceRawTrade]: - """Yield individually identified REST trades.""" + """Yield individually identified REST trades with RPI flags.""" for envelope in self.envelopes(): if envelope.kind == "raw_trade": diff --git a/tests/test_binance_capture.py b/tests/test_binance_capture.py index 9782fab..e29ebc8 100644 --- a/tests/test_binance_capture.py +++ b/tests/test_binance_capture.py @@ -14,6 +14,7 @@ import ordersim.connectors.binance.capture as capture_module from ordersim.connectors.binance._storage import RawCaptureSink from ordersim.connectors.binance._transport import ( + INDIVIDUAL_TRADE_STREAM_URL, MARKET_STREAM_URL, PUBLIC_STREAM_URL, WebSocketTransport, @@ -31,6 +32,7 @@ BinanceCaptureConfig, CaptureManifest, DepthSequenceTracker, + TradeSequenceTracker, ) @@ -51,6 +53,7 @@ def test_capture_config_names_the_highest_resolution_standard_streams( "btcusdt@rpiDepth@500ms", ) assert config.market_streams("BTCUSDT") == ("btcusdt@aggTrade",) + assert config.individual_trade_streams("BTCUSDT") == ("btcusdt@trade",) @pytest.mark.parametrize( @@ -107,6 +110,33 @@ def test_depth_sequence_tracker_rejects_malformed_payloads() -> None: tracker.observe({"U": 12, "u": 10, "pu": 9}) +def test_trade_sequence_tracker_reports_missing_and_repeated_ids() -> None: + tracker = TradeSequenceTracker() + + assert tracker.observe({"t": 100}) is None + assert tracker.observe({"t": 101}) is None + missing = tracker.observe({"t": 104}) + repeated = tracker.observe({"t": 104}) + + assert missing is not None + assert missing.as_dict() == { + "expected_trade_id": 102, + "received_trade_id": 104, + "missing_count": 2, + } + assert repeated is not None + assert repeated.as_dict() == { + "expected_trade_id": 105, + "received_trade_id": 104, + "missing_count": 0, + } + + +def test_trade_sequence_tracker_rejects_malformed_payload() -> None: + with pytest.raises(ValueError, match="must be an integer"): + TradeSequenceTracker().observe({"t": "100"}) + + 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"}}' @@ -285,6 +315,64 @@ def test_market_connection_records_trades_without_snapshot(tmp_path: Path) -> No } +def test_individual_trade_connection_records_ids_and_gap(tmp_path: Path) -> None: + config = BinanceCaptureConfig(output_dir=tmp_path, symbols=("BTCUSDT",)) + transport = FakeTransport( + [ + ("btcusdt@trade", {"t": 100, "p": "100.1", "q": "0.25"}), + ("btcusdt@trade", {"t": 103, "p": "100.2", "q": "0.50"}), + ] + ) + sink = RawCaptureSink(config) + + asyncio.run( + _capture_connection( + config=config, + sink=sink, + transport=transport, + symbol="BTCUSDT", + scope="individual", + connection_id="individual-1", + ) + ) + sink.close() + records = _read_capture_records(tmp_path) + + assert transport.connected_urls == [ + INDIVIDUAL_TRADE_STREAM_URL + "btcusdt@trade" + ] + assert transport.snapshot_calls == [] + assert [record["kind"] for record in records] == [ + "connection_open", + "message", + "message", + "trade_gap", + ] + assert records[-1]["payload"] == { + "expected_trade_id": 101, + "received_trade_id": 103, + "missing_count": 2, + } + + +def test_capture_connection_rejects_unknown_scope(tmp_path: Path) -> None: + config = BinanceCaptureConfig(output_dir=tmp_path, symbols=("BTCUSDT",)) + sink = RawCaptureSink(config) + + with pytest.raises(ValueError, match="unsupported Binance capture scope"): + asyncio.run( + _capture_connection( + config=config, + sink=sink, + transport=FakeTransport([]), + symbol="BTCUSDT", + scope="unknown", + connection_id="unknown-1", + ) + ) + sink.close() + + def test_websocket_transport_uses_injected_network_functions() -> None: calls: list[tuple[str, int, int]] = [] @@ -453,9 +541,9 @@ def test_finite_capture_closes_connections_and_writes_manifest( manifest = asyncio.run(capture_binance(config, transport=transport)) - assert manifest.counts["connection_open"] == 2 + assert manifest.counts["connection_open"] == 3 assert manifest.counts["depth_snapshot"] == 1 - assert len(transport.connected_urls) == 2 + assert len(transport.connected_urls) == 3 assert next(tmp_path.glob("manifest-*.json")).exists() diff --git a/tests/test_binance_l2_source.py b/tests/test_binance_l2_source.py index 4aac109..d21fa00 100644 --- a/tests/test_binance_l2_source.py +++ b/tests/test_binance_l2_source.py @@ -9,6 +9,7 @@ BinanceCaptureSource, BinanceDepthSnapshot, BinanceDepthUpdate, + BinanceIndividualTrade, BinancePriceLevel, BinanceSequenceError, ) @@ -198,6 +199,7 @@ def test_source_normalizes_exact_depth_trade_and_ticker_records( depth = tuple(source.depth_updates()) rpi_depth = tuple(source.depth_updates(stream_kind="rpi_depth")) trades = tuple(source.aggregate_trades()) + individual_trades = tuple(source.individual_trades()) tickers = tuple(source.book_tickers()) assert snapshots == ( @@ -230,6 +232,7 @@ def test_source_normalizes_exact_depth_trade_and_ticker_records( assert trades[0].quantity == Decimal("1.250") assert trades[0].normal_quantity == Decimal("1.000") assert trades[0].buyer_is_maker is True + assert individual_trades == () assert tickers[0].bid_price == Decimal("100.10") assert tickers[0].ask_quantity == Decimal("2.250") @@ -272,6 +275,46 @@ def test_aggregate_trade_preserves_missing_normal_quantity(tmp_path: Path) -> No assert trade.normal_quantity is None +def test_source_normalizes_individual_websocket_trade(tmp_path: Path) -> None: + row = capture_row( + kind="message", + scope="market", + connection_id="individual-1", + stream="btcusdt@trade", + payload={ + "e": "trade", + "E": 20, + "T": 19, + "s": "BTCUSDT", + "t": 700, + "p": "100.20", + "q": "0.125", + "m": False, + "X": "MARKET", + "st": 1, + }, + received_at_ns=21_000, + ) + capture_path, _ = write_capture(tmp_path, [row]) + + trades = tuple(BinanceCaptureSource((capture_path,)).individual_trades()) + + assert trades == ( + BinanceIndividualTrade( + symbol="BTCUSDT", + connection_id="individual-1", + event_time_ns=20_000_000, + trade_time_ns=19_000_000, + received_at_ns=21_000, + received_monotonic_ns=21_100, + trade_id=700, + price=Decimal("100.20"), + quantity=Decimal("0.125"), + buyer_is_maker=False, + ), + ) + + def test_validated_depth_rejects_gap_after_bridge(tmp_path: Path) -> None: rows = standard_rows() payload = rows[4]["payload"]