Capture individual Binance trades

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 |

CHANGELOG.md

@@ -12,6 +12,14 @@ 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, 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.
 
 ## 0.1.3 - 2026-05-20
 

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"]

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,12 +86,26 @@ ordersim-binance-capture captures/binance \
   --include-rpi
 ```
 
+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 \
+  --symbol BTCUSDT \
+  --symbol ETHUSDT \
+  --duration-hours 72 \
+  --poll-interval-seconds 0.5
+```
+
 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. |
+| 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. |
@@ -107,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
@@ -118,6 +133,39 @@ modeled `MBOEvent` rows.
 Capture files are local research data and must not be committed to the
 repository.
 
+### Individual Trade Capture
+
+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:
+
+- stores each exact individual trade payload once;
+- records request timing and Binance's reported one-minute request weight;
+- deduplicates overlapping responses by trade ID;
+- 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 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
+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 only as an independent
+reconciliation feed. It is not treated as a substitute for individual trades.
+
 ### Reading Completed Captures
 
 `BinanceCaptureSource` streams completed gzip capture files directly, so a
@@ -135,12 +183,19 @@ 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`, 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

docs/data-guide.md

@@ -103,13 +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, and book
-tickers. 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

docs/schema.md

@@ -32,6 +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. |
+| `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
@@ -45,6 +47,17 @@ 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:
+
+- `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;
+- `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.

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"

src/ordersim/connectors/binance/__init__.py

@@ -12,10 +12,17 @@
     BinanceDepthEvent,
     BinanceDepthSnapshot,
     BinanceDepthUpdate,
+    BinanceIndividualTrade,
     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,
@@ -30,9 +37,13 @@
     "BinanceDepthEvent",
     "BinanceDepthSnapshot",
     "BinanceDepthUpdate",
+    "BinanceIndividualTrade",
     "BinancePriceLevel",
+    "BinanceRawTrade",
+    "BinanceRawTradeCaptureConfig",
     "BinanceSequenceError",
     "CaptureManifest",
     "DepthStreamKind",
     "capture_binance",
+    "capture_binance_raw_trades",
 ]

src/ordersim/connectors/binance/_parsing.py

@@ -10,7 +10,9 @@
     BinanceCaptureEnvelope,
     BinanceDepthSnapshot,
     BinanceDepthUpdate,
+    BinanceIndividualTrade,
     BinancePriceLevel,
+    BinanceRawTrade,
     CaptureKind,
     CaptureScope,
     DepthStreamKind,
@@ -34,7 +36,12 @@ 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",
+            "trade_gap",
         ),
     )
     scope = required_choice(raw, "scope", ("public", "market"))
@@ -125,6 +132,46 @@ 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."""
+
+    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."""