diff --git a/AGENTS.md b/AGENTS.md index 55c9c4b..7d01807 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -40,6 +40,8 @@ extraction targets and should not be imported until they exist. | `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/connectors/binance/reconstruction.py` | Named virtual-MBO reconstruction policies | Public model API | +| `ordersim/connectors/binance/reconstruction_study.py` | Streaming alignment and evidence report | Public research CLI | | `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 3d87fa6..b87d3df 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,12 @@ All notable public changes to `ordersim` are documented here. reconciliation evidence. - Added typed `BinanceRawTrade` records alongside aggregate trades so the more detailed public evidence is available to future reconstruction models. +- Added deterministic Binance L2-to-virtual-MBO reconstruction with explicit + queue-conservative and queue-optimistic policies, exact quantity scaling, + and canonical `MBOEvent` output. +- Added a streaming reconstruction study that aligns individual trades to + depth intervals, validates joinable book-ticker states, preserves reconnect + boundaries, and reports inferred flow and model sensitivity. ## 0.1.3 - 2026-05-20 diff --git a/README.md b/README.md index fe371dc..17770d1 100644 --- a/README.md +++ b/README.md @@ -30,6 +30,8 @@ backtesting, and market microstructure research. strategy's orders, position, and portfolio state isolated. - Exposes a small, regular Python API that is easy to read, debug, test, and extend. +- Captures Binance L2 and individual-trade evidence and reconstructs modeled + MBO under explicit conservative and optimistic queue assumptions. ## What It Is Not @@ -129,6 +131,7 @@ Optional data integrations and file formats are installed separately as extras: ```bash pip install "ordersim[databento]" pip install "ordersim[parquet]" +pip install "ordersim[binance]" ``` Normalized CSV input works without optional dependencies: @@ -265,7 +268,8 @@ same input. `0.1.x` is live on PyPI. The current public line includes the Python reference engine, packaged C++ default, canonical connector -> Parquet -> replay workflow, -latency models, economics, and public execution-equivalence fixtures. +latency models, economics, public execution-equivalence fixtures, and an +evidence-first Binance L2-to-virtual-MBO research path. Planned next milestones: @@ -284,6 +288,7 @@ Planned next milestones: - Architecture: `docs/architecture.md` - Data guide: `docs/data-guide.md` - Connectors: `docs/connectors.md` +- Binance reconstruction study: `docs/binance-reconstruction-study.md` - Releasing: `docs/releasing.md` - Engineering standards: `docs/engineering-standards.md` - Benchmarks: `docs/benchmarks.md` diff --git a/docs/architecture.md b/docs/architecture.md index 6caea2b..c49d177 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -70,7 +70,7 @@ flowchart LR venue["Venue L2 + aggregate and individual trades"] capture["Raw capture"] source["Typed L2 source"] - model["Named reconstruction model"] + model["BinanceMBOReconstructor
named queue policy"] modeled["Modeled MBOEvent stream"] parquet["Canonical Parquet + model manifest"] replay["Replay"] @@ -81,10 +81,11 @@ flowchart LR Capture code may live beside connectors because it owns venue I/O and source schemas. Capture alone is not a `DataSource`: observed L2 rows must not be presented as exchange-native MBO. The reconstruction model owns that -lower-fidelity assumption and must preserve a manifest describing how its +lower-fidelity assumption and preserves a study manifest describing how its events were inferred. For Binance, `BinanceCaptureSource` is the typed, -sequence-validated L2 boundary between the raw evidence and that future model; -it does not implement the canonical MBO `DataSource` protocol. +sequence-validated L2 boundary and `BinanceMBOReconstructor` emits modeled +canonical events for one snapshot-anchored segment. The capture source itself +does not implement the canonical MBO `DataSource` protocol. ## One Replay Run diff --git a/docs/assumptions.md b/docs/assumptions.md index 51238f1..0d241cd 100644 --- a/docs/assumptions.md +++ b/docs/assumptions.md @@ -36,6 +36,25 @@ must preserve the observed price-level updates and trades. Reconstruction must name its cancellation-allocation, event-ordering, and randomness assumptions, and modeled events must not be presented as exchange-native MBO. +The Binance virtual-MBO model uses the minimum-flow identity at each price +level and depth interval: + +```text +ending quantity = starting quantity + adds - cancels - traded quantity +``` + +It infers the smallest non-negative add and cancel quantities that satisfy that +identity. This determines aggregate flow, but it does not determine where a +cancel occurred inside the queue or exactly when inferred additions appeared +inside Binance's update window. + +The recommended default is `queue-conservative`: additions needed by an +interval are placed before its first trade and cancellations remove the newest +modeled liquidity. This avoids selecting the assumptions most favorable to a +passive strategy. `queue-optimistic` adds only when required and removes the +oldest liquidity. Run both when queue position materially affects the result; +their difference is model sensitivity, not measurement error. + ## Queue Assumptions When the strategy places a resting limit order, the default model assumes the diff --git a/docs/binance-reconstruction-study.md b/docs/binance-reconstruction-study.md new file mode 100644 index 0000000..e5e1d29 --- /dev/null +++ b/docs/binance-reconstruction-study.md @@ -0,0 +1,120 @@ +# Binance L2 Reconstruction Study + +This note records the first empirical validation of `ordersim`'s Binance +L2-to-virtual-MBO path. It is evidence for a named model, not a claim that +aggregated depth reveals Binance's true FIFO order queue. + +## Capture + +The study used locally captured Binance USD-M futures evidence from +2026-08-08 through 2026-08-10: + +- standard full diff depth at 100 ms; +- individual `@trade` messages; +- real-time book ticker; +- REST snapshots anchoring every depth connection; +- aggregate trades, RPI depth, and REST individual trades retained as + supplemental audit evidence. + +The primary continuous window ends at `2026-08-10T17:12:57Z`, immediately +before a machine-wide network outage. Later recovery records remain in the raw +archive but were excluded. Raw capture files are research data and are not +committed to this repository. + +## Method + +Individual trades are aligned to consecutive depth endpoints using Binance +transaction timestamps. A 60-second local receive-time buffer allows the +independent depth and trade connections to arrive out of order. This is an +offline evidence-alignment window, not simulated exchange latency. + +The first full ETH pass showed why this must be measured: a two-second buffer +left 13,422 trades late, and a ten-second buffer left 71. The 60-second pass +left none; its maximum executable-trade receive delay was 29.121 seconds. Each +study report records both maximum receive delay and any residual late-trade +lag so the buffer can be reassessed for a different capture environment. + +For each side and price level: + +```text +ending quantity = starting quantity + adds - cancels - traded quantity +``` + +The model infers the smallest non-negative add and cancel quantities satisfying +that identity. It then applies two queue assumptions: + +- `queue-conservative`: infer additions before the first trade at the level and + cancel newest modeled liquidity first; +- `queue-optimistic`: add only as needed and cancel oldest modeled liquidity + first. + +Both policies must reproduce every observed depth endpoint. Joinable +book-ticker rows provide an independent check of top-of-book price and +quantity. Every reconnect starts a separate snapshot-anchored segment. + +## Results + +| Measure | BTCUSDT | ETHUSDT | +|---|---:|---:| +| Snapshot-anchored segments | 4 | 3 | +| Depth endpoints reconstructed | 1,668,215 | 1,659,917 | +| Executable individual trades aligned | 2,847,061 | 4,498,386 | +| Broken depth segments | 0 | 0 | +| Late trades after validated buffer | 0 | 0 | +| Exact book-ticker matches | 44,277 / 44,277 | 72,104 / 72,104 | +| Zero-value `@trade` messages excluded | 11,850 | 12,741 | +| Boundary-unassigned trades | 2,052 | 91 | +| Observed trade quantity | 158,708.331 BTC | 4,524,196.266 ETH | +| Required within-window replenishment | 17,306.673 BTC | 662,705.196 ETH | +| Replenishment / trade quantity | 10.90% | 14.65% | + +The book-ticker denominator includes only rows whose update ID exactly matches +a processed depth endpoint. It is not the count of all captured book-ticker +messages. + +Zero-price, zero-quantity `@trade` messages are preserved in raw capture. Their +raw payload includes undocumented fields such as `X=NA`; the study counts them +but does not assign execution semantics or emit invalid zero-sized MBO rows. + +## Interpretation + +The results support the exchange-time alignment and minimum-flow accounting: + +- all depth segments remained sequence-continuous; +- no executable trade arrived too late for the validated alignment buffer; +- every independently joinable top-of-book state matched exactly; +- the required replenishment ratio is measurable rather than hidden. + +They do not identify the true order-level queue. Both named policies can match +the same L2 endpoints while producing different queue-ahead paths for a +hypothetical resting order. Strategy conclusions that change materially +between the two policies should be reported as model-sensitive. + +## Reproduce + +Run the study against a completed raw capture directory: + +```bash +ordersim-binance-reconstruction-study captures/binance \ + --symbol BTCUSDT \ + --quantity-step 0.001 \ + --reorder-buffer-ms 60000 \ + --until-received-at-ns 1786381977000000000 \ + --output reports/btcusdt-reconstruction.json +``` + +The JSON report is a model manifest: it records the policy names, quantity +unit, cutoff, segment identifiers, alignment exceptions, inferred flow, and +book-ticker checks. + +## Open Validation Questions + +Useful contributions include: + +- cancellation-allocation models supported by published microstructure work; +- comparisons against private MBO for a venue or interval where both L2 and L3 + are available; +- passive-fill sensitivity experiments across the conservative and optimistic + bounds; +- evidence about Binance's zero-value `@trade` messages without relying on + undocumented fields as stable production contracts. diff --git a/docs/connectors.md b/docs/connectors.md index c832c8b..4ee0811 100644 --- a/docs/connectors.md +++ b/docs/connectors.md @@ -126,9 +126,8 @@ 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 -update window are not observable. A named L2-to-virtual-L3 model must consume -the typed capture records, document the inference policy, and only then emit -modeled `MBOEvent` rows. +update window are not observable. `BinanceMBOReconstructor` consumes aligned +typed records under a named policy and only then emits modeled `MBOEvent` rows. Capture files are local research data and must not be committed to the repository. @@ -222,6 +221,51 @@ Aggregate trades preserve Binance's optional `nq` field as `normal_quantity`. When present, it is the quantity excluding trades involving RPI orders. When absent, `normal_quantity` is `None`, not an inferred value. +### Studying L2-to-Virtual-MBO Reconstruction + +Run the evidence study before materializing modeled MBO for a new symbol or +capture: + +```bash +ordersim-binance-reconstruction-study captures/binance \ + --symbol BTCUSDT \ + --quantity-step 0.001 \ + --output reports/btcusdt-reconstruction.json +``` + +The study streams raw files in receive order and aligns individual trades to +depth intervals using Binance transaction timestamps. A configurable +receive-time buffer permits messages from the independent depth and trade +connections to arrive in a different local order. The default is 60 seconds, +chosen from the public validation capture; every report includes maximum trade +receive delay and residual late-trade lag so another environment can verify the +margin. It reports: + +- broken or stale depth segments; +- trades arriving too late for their exchange-time interval; +- maximum executable-trade receive delay; +- zero-price, zero-quantity `@trade` messages excluded from execution flow; +- minimally inferred add and cancel quantities; +- visible replenishment required to support observed trades; +- exact book-ticker matches where update IDs can be joined; +- metrics for both named queue policies. + +`queue-conservative` is the recommended default for fill research because it +does not choose the cancellation and addition assumptions most favorable to a +resting strategy. `queue-optimistic` is a sensitivity bound. If the strategy's +result changes materially between them, the capture does not support one +precise queue-fill claim. + +The minimum-flow identity preserves every observed L2 endpoint. Endpoint +equality therefore validates implementation, not the hidden queue hypothesis. +Book-ticker joins, sequence continuity, late-trade counts, and required +replenishment provide the independent evidence in the report. + +Each fresh REST snapshot begins a new reconstruction segment. Do not concatenate +segments as if the reconnect were an observed continuous MBO stream; write +separate canonical files or preserve an explicit segment boundary in the +research workflow. + For the user-facing decision guide, see `docs/data-guide.md`. ## In-Memory Sources diff --git a/docs/data-guide.md b/docs/data-guide.md index 283b303..0ebd18e 100644 --- a/docs/data-guide.md +++ b/docs/data-guide.md @@ -111,12 +111,19 @@ 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: +is the input boundary for the named virtual-L3 reconstruction model: ```text raw capture -> BinanceCaptureSource -> named model -> modeled MBO + manifest ``` +Use `BinanceMBOReconstructor` to produce canonical `MBOEvent` rows for one +snapshot-anchored connection segment. Use +`ordersim-binance-reconstruction-study` first on a new capture or symbol to +measure alignment, inferred flow, required replenishment, and the difference +between the named queue policies. Reconnect segments remain separate because +canonical MBO has no implicit clear-book event. + There is deliberately no direct `BinanceCaptureSource -> Replay` path. See `docs/connectors.md` for the capture and validation contract. diff --git a/docs/schema.md b/docs/schema.md index 15c5aac..298e9bd 100644 --- a/docs/schema.md +++ b/docs/schema.md @@ -42,10 +42,15 @@ event and transaction/trade times normalized from Binance milliseconds to UTC nanoseconds. Binance contract quantity can be fractional, so the connector preserves it as -`Decimal`. A future virtual-L3 reconstruction model must declare its quantity -unit and exact conversion rule before producing the canonical integer -`MBOEvent.size`. These L2 records are therefore not accepted directly by -`Replay`. +`Decimal`. `BinanceReconstructionConfig.quantity_step` declares the exact unit +used to produce canonical integer `MBOEvent.size` values. A quantity that is +not exactly divisible by that unit is rejected rather than rounded. The typed +L2 records are not accepted directly by `Replay`. + +The real-time `@trade` stream can contain zero-price, zero-quantity messages +whose untouched raw payload reports `X=NA`. The reconstruction study counts +these separately as `zero_value_trade_messages`; it does not present them as +executions or emit zero-sized canonical rows. Raw-trade capture files also contain audit envelopes: diff --git a/pyproject.toml b/pyproject.toml index 490a93f..77164ec 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -67,6 +67,7 @@ parquet = [ [project.scripts] ordersim-binance-capture = "ordersim.connectors.binance.capture:main" ordersim-binance-raw-trades = "ordersim.connectors.binance.raw_trades:main" +ordersim-binance-reconstruction-study = "ordersim.connectors.binance.reconstruction_study: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 938722b..4edec79 100644 --- a/src/ordersim/connectors/binance/__init__.py +++ b/src/ordersim/connectors/binance/__init__.py @@ -13,11 +13,19 @@ BinanceDepthSnapshot, BinanceDepthUpdate, BinanceIndividualTrade, + BinanceObservedEvent, BinancePriceLevel, BinanceRawTrade, DepthStreamKind, ) from ordersim.connectors.binance.raw_trades import capture_binance_raw_trades +from ordersim.connectors.binance.reconstruction import ( + BinanceMBOReconstructor, + BinanceReconstructionConfig, + BinanceReconstructionMetrics, + BinanceReconstructionPolicy, + BinanceReconstructionStep, +) from ordersim.connectors.binance.schema import ( BinanceCaptureConfig, BinanceRawTradeCaptureConfig, @@ -38,9 +46,15 @@ "BinanceDepthSnapshot", "BinanceDepthUpdate", "BinanceIndividualTrade", + "BinanceMBOReconstructor", + "BinanceObservedEvent", "BinancePriceLevel", "BinanceRawTrade", "BinanceRawTradeCaptureConfig", + "BinanceReconstructionConfig", + "BinanceReconstructionMetrics", + "BinanceReconstructionPolicy", + "BinanceReconstructionStep", "BinanceSequenceError", "CaptureManifest", "DepthStreamKind", diff --git a/src/ordersim/connectors/binance/l2.py b/src/ordersim/connectors/binance/l2.py index 0f03f7c..60e0fd7 100644 --- a/src/ordersim/connectors/binance/l2.py +++ b/src/ordersim/connectors/binance/l2.py @@ -151,3 +151,9 @@ class BinanceBookTicker: BinanceDepthEvent: TypeAlias = BinanceDepthSnapshot | BinanceDepthUpdate +BinanceObservedEvent: TypeAlias = ( + BinanceDepthSnapshot + | BinanceDepthUpdate + | BinanceIndividualTrade + | BinanceBookTicker +) diff --git a/src/ordersim/connectors/binance/reconstruction.py b/src/ordersim/connectors/binance/reconstruction.py new file mode 100644 index 0000000..35e1b10 --- /dev/null +++ b/src/ordersim/connectors/binance/reconstruction.py @@ -0,0 +1,520 @@ +"""Deterministic virtual-MBO reconstruction from Binance L2 and trades. + +The model preserves observed depth endpoints and individual trades. It does +not claim to recover exchange-native order IDs or the true placement of +cancellations inside a price-level queue. +""" + +from collections import deque +from collections.abc import Iterable +from dataclasses import dataclass +from decimal import Decimal +from typing import Literal, TypeAlias + +from ordersim.connectors.binance.l2 import ( + BinanceDepthSnapshot, + BinanceDepthUpdate, + BinanceIndividualTrade, + BinancePriceLevel, +) +from ordersim.types import BookSide, MBOEvent + +BinanceReconstructionPolicy: TypeAlias = Literal[ + "queue-conservative", + "queue-optimistic", +] + + +@dataclass(frozen=True, slots=True) +class BinanceReconstructionConfig: + """Configuration for one deterministic virtual-MBO reconstruction. + + `quantity_step` maps Binance's exact decimal contract quantity to the + integer sizes required by `MBOEvent`. + + `queue-conservative` places all minimally required additions before the + first trade at a level and removes cancellations from the newest visible + liquidity. `queue-optimistic` adds liquidity only when needed and removes + cancellations from the oldest visible liquidity. The two policies bound + queue-position effects; neither is presented as an observed FIFO queue. + + Set `emit_events=False` only for aggregate evidence studies. The same + endpoint and flow calculations run, but returned `events` tuples are empty + and explicit virtual-order queues are not retained. + """ + + quantity_step: Decimal + policy: BinanceReconstructionPolicy = "queue-conservative" + emit_events: bool = True + + def __post_init__(self) -> None: + if self.quantity_step <= 0: + raise ValueError("quantity_step must be positive") + if self.policy not in ("queue-conservative", "queue-optimistic"): + raise ValueError(f"unsupported reconstruction policy {self.policy!r}") + + +@dataclass(frozen=True, slots=True) +class BinanceReconstructionMetrics: + """Evidence and inferred flow produced by one reconstruction step.""" + + depth_updates: int = 0 + levels_checked: int = 0 + trade_count: int = 0 + trade_units: int = 0 + snapshot_add_units: int = 0 + inferred_add_units: int = 0 + inferred_cancel_units: int = 0 + required_replenishment_units: int = 0 + pre_trade_add_units: int = 0 + + +@dataclass(frozen=True, slots=True) +class BinanceReconstructionStep: + """Canonical events and metrics for one depth endpoint.""" + + events: tuple[MBOEvent, ...] + metrics: BinanceReconstructionMetrics + + +@dataclass(slots=True) +class _VirtualOrder: + order_id: int + size: int + + +class BinanceMBOReconstructor: + """Incrementally reconstruct one snapshot-anchored Binance depth segment. + + Call `bootstrap(snapshot, bridge_update)` once, then call `apply_update` + with each later standard depth update and the individual trades aligned to + that update's exchange-time interval. + """ + + def __init__(self, config: BinanceReconstructionConfig) -> None: + self.config = config + self._symbol: str | None = None + self._connection_id: str | None = None + self._previous_transaction_time_ns: int | None = None + self._queues: dict[tuple[BookSide, Decimal], deque[_VirtualOrder]] = {} + self._level_units: dict[tuple[BookSide, Decimal], int] = {} + self._next_order_id = 1 + self._best_bid: Decimal | None = None + self._best_ask: Decimal | None = None + + @property + def symbol(self) -> str | None: + """The bootstrapped symbol, or `None` before bootstrap.""" + + return self._symbol + + @property + def previous_transaction_time_ns(self) -> int | None: + """Exchange transaction time of the last applied depth update.""" + + return self._previous_transaction_time_ns + + def bootstrap( + self, + snapshot: BinanceDepthSnapshot, + bridge_update: BinanceDepthUpdate, + ) -> BinanceReconstructionStep: + """Anchor the virtual book and apply the first snapshot-bridging update.""" + + if self._symbol is not None: + raise RuntimeError("reconstructor is already bootstrapped") + if bridge_update.stream_kind != "depth": + raise ValueError("bridge_update must come from standard depth") + if snapshot.symbol != bridge_update.symbol: + raise ValueError("snapshot and bridge update symbols differ") + if snapshot.connection_id != bridge_update.connection_id: + raise ValueError("snapshot and bridge update connections differ") + if not ( + bridge_update.first_update_id + <= snapshot.last_update_id + <= bridge_update.final_update_id + ): + raise ValueError("bridge update does not span snapshot last_update_id") + + self._symbol = snapshot.symbol + self._connection_id = snapshot.connection_id + ts_ns = bridge_update.transaction_time_ns + events: list[MBOEvent] = [] + snapshot_units = 0 + + for side, levels in (("bid", snapshot.bids), ("ask", snapshot.asks)): + for level in self._sorted_levels(side, levels): + size = self._to_units(level.quantity) + if size == 0: + continue + self._append_public(side, level.price, size) + self._record_add(events, ts_ns, side, level.price, size) + snapshot_units += size + + inferred_add, inferred_cancel, checked = self._apply_endpoint( + bridge_update, + events, + ) + self._previous_transaction_time_ns = bridge_update.transaction_time_ns + return BinanceReconstructionStep( + events=tuple(events), + metrics=BinanceReconstructionMetrics( + depth_updates=1, + levels_checked=checked, + snapshot_add_units=snapshot_units, + inferred_add_units=inferred_add, + inferred_cancel_units=inferred_cancel, + ), + ) + + def apply_update( + self, + update: BinanceDepthUpdate, + trades: Iterable[BinanceIndividualTrade] = (), + ) -> BinanceReconstructionStep: + """Apply one later depth endpoint and its aligned individual trades.""" + + previous_time = self._require_compatible_update(update) + aligned_trades = tuple( + sorted(trades, key=lambda trade: (trade.trade_time_ns, trade.trade_id)) + ) + for trade in aligned_trades: + if trade.symbol != update.symbol: + raise ValueError("trade and depth update symbols differ") + if trade.price <= 0 or trade.quantity <= 0: + raise ValueError("trade price and quantity must be positive") + if not previous_time < trade.trade_time_ns <= update.transaction_time_ns: + raise ValueError("trade is outside the depth update interval") + + endpoint_targets = self._endpoint_targets(update) + trade_units_by_level: dict[tuple[BookSide, Decimal], int] = {} + for trade in aligned_trades: + key = (self._resting_side(trade), trade.price) + trade_units_by_level[key] = ( + trade_units_by_level.get(key, 0) + self._to_units(trade.quantity) + ) + + keys = set(endpoint_targets) | set(trade_units_by_level) + targets = { + key: endpoint_targets.get(key, self._level_units.get(key, 0)) + for key in keys + } + minimum_adds: dict[tuple[BookSide, Decimal], int] = {} + required_replenishment = 0 + for key in keys: + before = self._level_units.get(key, 0) + traded = trade_units_by_level.get(key, 0) + target = targets.get(key, before) + minimum_adds[key] = max(0, target + traded - before) + required_replenishment += max(0, traded - before) + + events: list[MBOEvent] = [] + preadded: set[tuple[BookSide, Decimal]] = set() + inferred_add = 0 + pre_trade_add = 0 + trade_units = 0 + + for trade in aligned_trades: + side = self._resting_side(trade) + key = (side, trade.price) + size = self._to_units(trade.quantity) + if self.config.policy == "queue-conservative" and key not in preadded: + addition = minimum_adds[key] + if addition: + self._append_public(side, trade.price, addition) + self._record_add( + events, + trade.trade_time_ns, + side, + trade.price, + addition, + ) + inferred_add += addition + pre_trade_add += addition + preadded.add(key) + elif self.config.policy == "queue-optimistic": + shortfall = max(0, size - self._level_units.get(key, 0)) + if shortfall: + self._append_public(side, trade.price, shortfall) + self._record_add( + events, + trade.trade_time_ns, + side, + trade.price, + shortfall, + ) + inferred_add += shortfall + pre_trade_add += shortfall + + self._consume_public(side, trade.price, size) + if self.config.emit_events: + events.append( + MBOEvent( + ts_ns=trade.trade_time_ns, + action="trade", + side=side, + price=trade.price, + size=size, + order_id=trade.trade_id, + ) + ) + trade_units += size + + endpoint_add, inferred_cancel, checked = self._apply_endpoint( + update, + events, + targets=targets, + ) + inferred_add += endpoint_add + self._previous_transaction_time_ns = update.transaction_time_ns + return BinanceReconstructionStep( + events=tuple(events), + metrics=BinanceReconstructionMetrics( + depth_updates=1, + levels_checked=checked, + trade_count=len(aligned_trades), + trade_units=trade_units, + inferred_add_units=inferred_add, + inferred_cancel_units=inferred_cancel, + required_replenishment_units=required_replenishment, + pre_trade_add_units=pre_trade_add, + ), + ) + + def level_quantity(self, side: BookSide, price: Decimal) -> int: + """Return reconstructed public quantity in integer quantity steps.""" + + return self._level_units.get((side, price), 0) + + def book_top(self) -> tuple[Decimal | None, Decimal | None]: + """Return the reconstructed public best bid and ask.""" + + return self._best_bid, self._best_ask + + def _require_compatible_update(self, update: BinanceDepthUpdate) -> int: + if self._symbol is None or self._previous_transaction_time_ns is None: + raise RuntimeError("bootstrap must be called before apply_update") + if update.symbol != self._symbol: + raise ValueError("depth update symbol differs from bootstrap") + if update.connection_id != self._connection_id: + raise ValueError("depth update connection differs from bootstrap") + if update.stream_kind != "depth": + raise ValueError("only standard depth updates can be reconstructed") + if update.transaction_time_ns < self._previous_transaction_time_ns: + raise ValueError("depth transaction time moved backwards") + return self._previous_transaction_time_ns + + def _apply_endpoint( + self, + update: BinanceDepthUpdate, + events: list[MBOEvent], + *, + targets: dict[tuple[BookSide, Decimal], int] | None = None, + ) -> tuple[int, int, int]: + endpoint_targets = ( + self._endpoint_targets(update) if targets is None else targets + ) + + inferred_add = 0 + inferred_cancel = 0 + for side, price in sorted(endpoint_targets, key=self._level_sort_key): + target = endpoint_targets[(side, price)] + current = self._level_units.get((side, price), 0) + if current < target: + addition = target - current + self._append_public(side, price, addition) + self._record_add( + events, + update.transaction_time_ns, + side, + price, + addition, + ) + inferred_add += addition + elif current > target: + cancellation = current - target + events.extend( + self._cancel_public( + side, + price, + cancellation, + update.transaction_time_ns, + ) + ) + inferred_cancel += cancellation + if self._level_units.get((side, price), 0) != target: + raise AssertionError("reconstruction did not reach depth endpoint") + return inferred_add, inferred_cancel, len(endpoint_targets) + + def _endpoint_targets( + self, + update: BinanceDepthUpdate, + ) -> dict[tuple[BookSide, Decimal], int]: + targets: dict[tuple[BookSide, Decimal], int] = {} + for side, levels in (("bid", update.bids), ("ask", update.asks)): + for level in levels: + targets[(side, level.price)] = self._to_units(level.quantity) + return targets + + def _append_public(self, side: BookSide, price: Decimal, size: int) -> None: + key = (side, price) + if self.config.emit_events: + self._queues.setdefault(key, deque()).append( + _VirtualOrder(order_id=self._next_order_id, size=size) + ) + self._next_order_id += 1 + self._level_units[key] = self._level_units.get(key, 0) + size + if side == "bid" and (self._best_bid is None or price > self._best_bid): + self._best_bid = price + elif side == "ask" and (self._best_ask is None or price < self._best_ask): + self._best_ask = price + + def _consume_public(self, side: BookSide, price: Decimal, size: int) -> None: + key = (side, price) + if not self.config.emit_events: + if self._level_units.get(key, 0) < size: + raise AssertionError("trade exceeded reconstructed public quantity") + self._level_units[key] -= size + self._drop_empty_level(key) + return + + queue = self._queues.get(key) + remaining = size + while remaining and queue: + order = queue[0] + consumed = min(order.size, remaining) + order.size -= consumed + remaining -= consumed + self._level_units[key] -= consumed + if order.size == 0: + queue.popleft() + if remaining: + raise AssertionError("trade exceeded reconstructed public quantity") + self._drop_empty_level(key) + + def _cancel_public( + self, + side: BookSide, + price: Decimal, + size: int, + ts_ns: int, + ) -> list[MBOEvent]: + key = (side, price) + if not self.config.emit_events: + if self._level_units.get(key, 0) < size: + raise AssertionError("cancel exceeded reconstructed public quantity") + self._level_units[key] -= size + self._drop_empty_level(key) + return [] + + queue = self._queues.get(key) + remaining = size + events: list[MBOEvent] = [] + while remaining and queue: + order = queue[0] if self.config.policy == "queue-optimistic" else queue[-1] + cancelled = min(order.size, remaining) + if self.config.emit_events: + events.append( + MBOEvent( + ts_ns=ts_ns, + action="cancel", + side=side, + price=price, + size=cancelled, + order_id=order.order_id, + ) + ) + order.size -= cancelled + remaining -= cancelled + self._level_units[key] -= cancelled + if order.size == 0: + if self.config.policy == "queue-optimistic": + queue.popleft() + else: + queue.pop() + if remaining: + raise AssertionError("cancel exceeded reconstructed public quantity") + self._drop_empty_level(key) + return events + + def _drop_empty_level(self, key: tuple[BookSide, Decimal]) -> None: + if self._level_units.get(key) == 0: + self._level_units.pop(key, None) + self._queues.pop(key, None) + side, price = key + if side == "bid" and price == self._best_bid: + self._best_bid = max( + ( + level_price + for level_side, level_price in self._level_units + if level_side == "bid" + ), + default=None, + ) + elif side == "ask" and price == self._best_ask: + self._best_ask = min( + ( + level_price + for level_side, level_price in self._level_units + if level_side == "ask" + ), + default=None, + ) + + def _add_event( + self, + ts_ns: int, + side: BookSide, + price: Decimal, + size: int, + ) -> MBOEvent: + order_id = self._queues[(side, price)][-1].order_id + return MBOEvent( + ts_ns=ts_ns, + action="add", + side=side, + price=price, + size=size, + order_id=order_id, + ) + + def _record_add( + self, + events: list[MBOEvent], + ts_ns: int, + side: BookSide, + price: Decimal, + size: int, + ) -> None: + if self.config.emit_events: + events.append(self._add_event(ts_ns, side, price, size)) + + def _to_units(self, quantity: Decimal) -> int: + units = quantity / self.config.quantity_step + integral = units.to_integral_value() + if units != integral: + raise ValueError( + f"quantity {quantity} is not divisible by quantity_step " + f"{self.config.quantity_step}" + ) + return int(integral) + + @staticmethod + def _resting_side(trade: BinanceIndividualTrade) -> BookSide: + return "bid" if trade.buyer_is_maker else "ask" + + @staticmethod + def _sorted_levels( + side: BookSide, + levels: tuple[BinancePriceLevel, ...], + ) -> tuple[BinancePriceLevel, ...]: + return tuple( + sorted(levels, key=lambda level: level.price, reverse=side == "bid") + ) + + @staticmethod + def _level_sort_key( + key: tuple[BookSide, Decimal], + ) -> tuple[int, Decimal]: + side, price = key + return (0 if side == "bid" else 1, -price if side == "bid" else price) diff --git a/src/ordersim/connectors/binance/reconstruction_study.py b/src/ordersim/connectors/binance/reconstruction_study.py new file mode 100644 index 0000000..6fe9b4d --- /dev/null +++ b/src/ordersim/connectors/binance/reconstruction_study.py @@ -0,0 +1,546 @@ +"""Streaming evidence study for Binance L2-to-virtual-MBO reconstruction.""" + +import argparse +import json +from collections import deque +from collections.abc import Iterable, Mapping, Sequence +from dataclasses import dataclass +from decimal import Decimal +from pathlib import Path + +from ordersim.connectors.binance.l2 import ( + BinanceBookTicker, + BinanceDepthSnapshot, + BinanceDepthUpdate, + BinanceIndividualTrade, + BinanceObservedEvent, +) +from ordersim.connectors.binance.reconstruction import ( + BinanceMBOReconstructor, + BinanceReconstructionConfig, + BinanceReconstructionMetrics, + BinanceReconstructionPolicy, +) +from ordersim.connectors.binance.source import BinanceCaptureSource + +MODEL_NAME = "binance-virtual-mbo-minimum-flow-v1" +ALIGNMENT_NAME = "exchange-transaction-time" + + +@dataclass(frozen=True, slots=True) +class BinanceReconstructionStudyConfig: + """Configuration for an offline reconstruction study over raw evidence.""" + + symbol: str + quantity_step: Decimal + policies: tuple[BinanceReconstructionPolicy, ...] = ( + "queue-conservative", + "queue-optimistic", + ) + reorder_buffer_ns: int = 60_000_000_000 + until_received_at_ns: int | None = None + + def __post_init__(self) -> None: + symbol = self.symbol.strip().upper() + if not symbol: + raise ValueError("symbol must not be empty") + if self.quantity_step <= 0: + raise ValueError("quantity_step must be positive") + if not self.policies: + raise ValueError("at least one reconstruction policy is required") + if len(set(self.policies)) != len(self.policies): + raise ValueError("reconstruction policies must be unique") + if self.reorder_buffer_ns < 0: + raise ValueError("reorder_buffer_ns must be non-negative") + object.__setattr__(self, "symbol", symbol) + + +@dataclass(frozen=True, slots=True) +class BinanceReconstructionSegmentReport: + """Evidence summary for one snapshot-anchored depth connection.""" + + connection_id: str + snapshot_last_update_id: int + first_update_id: int + last_update_id: int + first_transaction_time_ns: int + last_transaction_time_ns: int + metrics_by_policy: Mapping[str, BinanceReconstructionMetrics] + + def as_dict(self) -> dict[str, object]: + """Return a JSON-serializable segment manifest.""" + + return { + "connection_id": self.connection_id, + "snapshot_last_update_id": self.snapshot_last_update_id, + "first_update_id": self.first_update_id, + "last_update_id": self.last_update_id, + "first_transaction_time_ns": self.first_transaction_time_ns, + "last_transaction_time_ns": self.last_transaction_time_ns, + "metrics_by_policy": { + policy: _metrics_as_dict(metrics) + for policy, metrics in self.metrics_by_policy.items() + }, + } + + +@dataclass(frozen=True, slots=True) +class BinanceReconstructionStudyReport: + """JSON-ready evidence report for one symbol and capture boundary.""" + + symbol: str + quantity_step: Decimal + policies: tuple[BinanceReconstructionPolicy, ...] + reorder_buffer_ns: int + until_received_at_ns: int | None + observations: int + depth_snapshots: int + stale_depth_updates: int + broken_depth_segments: int + duplicate_trades: int + zero_value_trade_messages: int + max_trade_receive_delay_ns: int + late_trades: int + max_late_trade_lag_ns: int + unassigned_trades: int + exact_book_ticker_matches: int + exact_book_ticker_mismatches: int + segments: tuple[BinanceReconstructionSegmentReport, ...] + totals_by_policy: Mapping[str, BinanceReconstructionMetrics] + + def as_dict(self) -> dict[str, object]: + """Return a JSON-serializable model and evidence manifest.""" + + return { + "schema_version": 1, + "model": MODEL_NAME, + "alignment": ALIGNMENT_NAME, + "symbol": self.symbol, + "quantity_step": str(self.quantity_step), + "policies": list(self.policies), + "reorder_buffer_ns": self.reorder_buffer_ns, + "until_received_at_ns": self.until_received_at_ns, + "observations": self.observations, + "depth_snapshots": self.depth_snapshots, + "stale_depth_updates": self.stale_depth_updates, + "broken_depth_segments": self.broken_depth_segments, + "duplicate_trades": self.duplicate_trades, + "zero_value_trade_messages": self.zero_value_trade_messages, + "max_trade_receive_delay_ns": self.max_trade_receive_delay_ns, + "late_trades": self.late_trades, + "max_late_trade_lag_ns": self.max_late_trade_lag_ns, + "unassigned_trades": self.unassigned_trades, + "exact_book_ticker_matches": self.exact_book_ticker_matches, + "exact_book_ticker_mismatches": self.exact_book_ticker_mismatches, + "segments": [segment.as_dict() for segment in self.segments], + "totals_by_policy": { + policy: _metrics_as_dict(metrics) + for policy, metrics in self.totals_by_policy.items() + }, + } + + +@dataclass(slots=True) +class _MetricsAccumulator: + depth_updates: int = 0 + levels_checked: int = 0 + trade_count: int = 0 + trade_units: int = 0 + snapshot_add_units: int = 0 + inferred_add_units: int = 0 + inferred_cancel_units: int = 0 + required_replenishment_units: int = 0 + pre_trade_add_units: int = 0 + + def add(self, metrics: BinanceReconstructionMetrics) -> None: + self.depth_updates += metrics.depth_updates + self.levels_checked += metrics.levels_checked + self.trade_count += metrics.trade_count + self.trade_units += metrics.trade_units + self.snapshot_add_units += metrics.snapshot_add_units + self.inferred_add_units += metrics.inferred_add_units + self.inferred_cancel_units += metrics.inferred_cancel_units + self.required_replenishment_units += metrics.required_replenishment_units + self.pre_trade_add_units += metrics.pre_trade_add_units + + def freeze(self) -> BinanceReconstructionMetrics: + return BinanceReconstructionMetrics( + depth_updates=self.depth_updates, + levels_checked=self.levels_checked, + trade_count=self.trade_count, + trade_units=self.trade_units, + snapshot_add_units=self.snapshot_add_units, + inferred_add_units=self.inferred_add_units, + inferred_cancel_units=self.inferred_cancel_units, + required_replenishment_units=self.required_replenishment_units, + pre_trade_add_units=self.pre_trade_add_units, + ) + + +@dataclass(slots=True) +class _ActiveSegment: + snapshot: BinanceDepthSnapshot + models: dict[BinanceReconstructionPolicy, BinanceMBOReconstructor] + metrics: dict[BinanceReconstructionPolicy, _MetricsAccumulator] + pending_updates: deque[BinanceDepthUpdate] + book_tickers: deque[BinanceBookTicker] + previous_final_update_id: int + first_update_id: int + last_update_id: int + first_transaction_time_ns: int + last_transaction_time_ns: int + + +class _StudyRunner: + def __init__(self, config: BinanceReconstructionStudyConfig) -> None: + self.config = config + self.observations = 0 + self.depth_snapshots = 0 + self.stale_depth_updates = 0 + self.broken_depth_segments = 0 + self.duplicate_trades = 0 + self.zero_value_trade_messages = 0 + self.max_trade_receive_delay_ns = 0 + self.late_trades = 0 + self.max_late_trade_lag_ns = 0 + self.unassigned_trades = 0 + self.exact_book_ticker_matches = 0 + self.exact_book_ticker_mismatches = 0 + self._last_trade_id: int | None = None + self._trades: list[BinanceIndividualTrade] = [] + self._snapshot: BinanceDepthSnapshot | None = None + self._segment: _ActiveSegment | None = None + self._segments: list[BinanceReconstructionSegmentReport] = [] + self._totals = { + policy: _MetricsAccumulator() for policy in config.policies + } + + def run( + self, + observations: Iterable[BinanceObservedEvent], + ) -> BinanceReconstructionStudyReport: + for observation in observations: + if observation.symbol != self.config.symbol: + continue + self.observations += 1 + self._accept(observation) + watermark = observation.received_at_ns - self.config.reorder_buffer_ns + self._flush_ready(watermark) + + self._finish_segment() + self.unassigned_trades += len(self._trades) + self._trades.clear() + return BinanceReconstructionStudyReport( + symbol=self.config.symbol, + quantity_step=self.config.quantity_step, + policies=self.config.policies, + reorder_buffer_ns=self.config.reorder_buffer_ns, + until_received_at_ns=self.config.until_received_at_ns, + observations=self.observations, + depth_snapshots=self.depth_snapshots, + stale_depth_updates=self.stale_depth_updates, + broken_depth_segments=self.broken_depth_segments, + duplicate_trades=self.duplicate_trades, + zero_value_trade_messages=self.zero_value_trade_messages, + max_trade_receive_delay_ns=self.max_trade_receive_delay_ns, + late_trades=self.late_trades, + max_late_trade_lag_ns=self.max_late_trade_lag_ns, + unassigned_trades=self.unassigned_trades, + exact_book_ticker_matches=self.exact_book_ticker_matches, + exact_book_ticker_mismatches=self.exact_book_ticker_mismatches, + segments=tuple(self._segments), + totals_by_policy={ + policy: totals.freeze() for policy, totals in self._totals.items() + }, + ) + + def _accept(self, observation: BinanceObservedEvent) -> None: + if isinstance(observation, BinanceDepthSnapshot): + self._start_snapshot(observation) + elif isinstance(observation, BinanceDepthUpdate): + self._accept_depth(observation) + elif isinstance(observation, BinanceIndividualTrade): + self._accept_trade(observation) + elif isinstance(observation, BinanceBookTicker): + self._accept_book_ticker(observation) + + def _start_snapshot(self, snapshot: BinanceDepthSnapshot) -> None: + self.depth_snapshots += 1 + self._finish_segment() + self.unassigned_trades += len(self._trades) + self._trades.clear() + self._snapshot = snapshot + + def _accept_depth(self, update: BinanceDepthUpdate) -> None: + if self._snapshot is None: + return + if update.connection_id != self._snapshot.connection_id: + return + if self._segment is None: + if update.final_update_id < self._snapshot.last_update_id: + self.stale_depth_updates += 1 + return + if not ( + update.first_update_id + <= self._snapshot.last_update_id + <= update.final_update_id + ): + self.broken_depth_segments += 1 + self._snapshot = None + return + self._bootstrap(update) + return + if update.previous_update_id != self._segment.previous_final_update_id: + self.broken_depth_segments += 1 + self._finish_segment() + self._snapshot = None + return + self._segment.pending_updates.append(update) + self._segment.previous_final_update_id = update.final_update_id + + def _bootstrap(self, update: BinanceDepthUpdate) -> None: + assert self._snapshot is not None + models = { + policy: BinanceMBOReconstructor( + BinanceReconstructionConfig( + quantity_step=self.config.quantity_step, + policy=policy, + emit_events=False, + ) + ) + for policy in self.config.policies + } + metrics = {policy: _MetricsAccumulator() for policy in self.config.policies} + for policy, model in models.items(): + step = model.bootstrap(self._snapshot, update) + metrics[policy].add(step.metrics) + self._totals[policy].add(step.metrics) + self._segment = _ActiveSegment( + snapshot=self._snapshot, + models=models, + metrics=metrics, + pending_updates=deque(), + book_tickers=deque(), + previous_final_update_id=update.final_update_id, + first_update_id=update.final_update_id, + last_update_id=update.final_update_id, + first_transaction_time_ns=update.transaction_time_ns, + last_transaction_time_ns=update.transaction_time_ns, + ) + before = len(self._trades) + self._trades = [ + trade + for trade in self._trades + if trade.trade_time_ns > update.transaction_time_ns + ] + self.unassigned_trades += before - len(self._trades) + + def _accept_trade(self, trade: BinanceIndividualTrade) -> None: + if self._last_trade_id is not None and trade.trade_id <= self._last_trade_id: + self.duplicate_trades += 1 + return + self._last_trade_id = trade.trade_id + if trade.price <= 0 or trade.quantity <= 0: + self.zero_value_trade_messages += 1 + return + self.max_trade_receive_delay_ns = max( + self.max_trade_receive_delay_ns, + trade.received_at_ns - trade.trade_time_ns, + ) + if self._segment is not None: + previous_time = next( + iter(self._segment.models.values()) + ).previous_transaction_time_ns + if previous_time is not None and trade.trade_time_ns <= previous_time: + self.late_trades += 1 + self.max_late_trade_lag_ns = max( + self.max_late_trade_lag_ns, + previous_time - trade.trade_time_ns, + ) + return + self._trades.append(trade) + + def _accept_book_ticker(self, ticker: BinanceBookTicker) -> None: + segment = self._segment + if segment is None or ticker.connection_id != segment.snapshot.connection_id: + return + segment.book_tickers.append(ticker) + + def _flush_ready(self, watermark_received_at_ns: int) -> None: + segment = self._segment + if segment is None: + return + while ( + segment.pending_updates + and segment.pending_updates[0].received_at_ns <= watermark_received_at_ns + ): + self._flush_update(segment.pending_updates.popleft()) + + def _flush_update(self, update: BinanceDepthUpdate) -> None: + segment = self._segment + if segment is None: + return + previous_time = next(iter(segment.models.values())).previous_transaction_time_ns + assert previous_time is not None + aligned: list[BinanceIndividualTrade] = [] + retained: list[BinanceIndividualTrade] = [] + for trade in self._trades: + if trade.trade_time_ns <= previous_time: + self.late_trades += 1 + self.max_late_trade_lag_ns = max( + self.max_late_trade_lag_ns, + previous_time - trade.trade_time_ns, + ) + elif trade.trade_time_ns <= update.transaction_time_ns: + aligned.append(trade) + else: + retained.append(trade) + self._trades = retained + + for policy, model in segment.models.items(): + step = model.apply_update(update, aligned) + segment.metrics[policy].add(step.metrics) + self._totals[policy].add(step.metrics) + + self._compare_book_ticker(update) + segment.last_update_id = update.final_update_id + segment.last_transaction_time_ns = update.transaction_time_ns + + def _compare_book_ticker(self, update: BinanceDepthUpdate) -> None: + segment = self._segment + if segment is None: + return + tickers: list[BinanceBookTicker] = [] + while ( + segment.book_tickers + and segment.book_tickers[0].update_id <= update.final_update_id + ): + ticker = segment.book_tickers.popleft() + if ticker.update_id == update.final_update_id: + tickers.append(ticker) + if not tickers: + return + model = next(iter(segment.models.values())) + bid_price, ask_price = model.book_top() + for ticker in tickers: + matches = ( + bid_price == ticker.bid_price + and ask_price == ticker.ask_price + and bid_price is not None + and ask_price is not None + and model.level_quantity("bid", bid_price) + == _quantity_units(ticker.bid_quantity, self.config.quantity_step) + and model.level_quantity("ask", ask_price) + == _quantity_units(ticker.ask_quantity, self.config.quantity_step) + ) + if matches: + self.exact_book_ticker_matches += 1 + else: + self.exact_book_ticker_mismatches += 1 + + def _finish_segment(self) -> None: + segment = self._segment + if segment is None: + return + while segment.pending_updates: + self._flush_update(segment.pending_updates.popleft()) + self._segments.append( + BinanceReconstructionSegmentReport( + connection_id=segment.snapshot.connection_id, + snapshot_last_update_id=segment.snapshot.last_update_id, + first_update_id=segment.first_update_id, + last_update_id=segment.last_update_id, + first_transaction_time_ns=segment.first_transaction_time_ns, + last_transaction_time_ns=segment.last_transaction_time_ns, + metrics_by_policy={ + policy: metrics.freeze() + for policy, metrics in segment.metrics.items() + }, + ) + ) + self._segment = None + + +def run_binance_reconstruction_study( + source: BinanceCaptureSource, + config: BinanceReconstructionStudyConfig, +) -> BinanceReconstructionStudyReport: + """Run the reconstruction study without loading the capture into memory.""" + + return _StudyRunner(config).run( + source.observations( + until_received_at_ns=config.until_received_at_ns, + symbol=config.symbol, + ) + ) + + +def _quantity_units(quantity: Decimal, quantity_step: Decimal) -> int: + units = quantity / quantity_step + integral = units.to_integral_value() + if units != integral: + raise ValueError( + f"quantity {quantity} is not divisible by quantity_step {quantity_step}" + ) + return int(integral) + + +def _metrics_as_dict(metrics: BinanceReconstructionMetrics) -> dict[str, int]: + return { + "depth_updates": metrics.depth_updates, + "levels_checked": metrics.levels_checked, + "trade_count": metrics.trade_count, + "trade_units": metrics.trade_units, + "snapshot_add_units": metrics.snapshot_add_units, + "inferred_add_units": metrics.inferred_add_units, + "inferred_cancel_units": metrics.inferred_cancel_units, + "required_replenishment_units": metrics.required_replenishment_units, + "pre_trade_add_units": metrics.pre_trade_add_units, + } + + +def _parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description="Study Binance L2 and individual trades as virtual MBO." + ) + parser.add_argument("capture_dir", type=Path) + parser.add_argument("--symbol", required=True) + parser.add_argument("--quantity-step", type=Decimal, required=True) + parser.add_argument("--until-received-at-ns", type=int) + parser.add_argument("--reorder-buffer-ms", type=int, default=60_000) + parser.add_argument( + "--policy", + action="append", + choices=("queue-conservative", "queue-optimistic"), + dest="policies", + ) + parser.add_argument("--output", type=Path) + return parser + + +def main(argv: Sequence[str] | None = None) -> None: + """Run the `ordersim-binance-reconstruction-study` command.""" + + args = _parser().parse_args(argv) + policies = tuple(args.policies or ("queue-conservative", "queue-optimistic")) + config = BinanceReconstructionStudyConfig( + symbol=args.symbol, + quantity_step=args.quantity_step, + policies=policies, + reorder_buffer_ns=args.reorder_buffer_ms * 1_000_000, + until_received_at_ns=args.until_received_at_ns, + ) + report = run_binance_reconstruction_study( + BinanceCaptureSource.from_directory(args.capture_dir), + config, + ) + rendered = json.dumps(report.as_dict(), indent=2, sort_keys=True) + "\n" + if args.output is None: + print(rendered, end="") + else: + args.output.parent.mkdir(parents=True, exist_ok=True) + args.output.write_text(rendered, encoding="utf-8") + + +if __name__ == "__main__": + main() diff --git a/src/ordersim/connectors/binance/source.py b/src/ordersim/connectors/binance/source.py index 2f483f0..70a7be1 100644 --- a/src/ordersim/connectors/binance/source.py +++ b/src/ordersim/connectors/binance/source.py @@ -26,6 +26,7 @@ BinanceDepthSnapshot, BinanceDepthUpdate, BinanceIndividualTrade, + BinanceObservedEvent, BinanceRawTrade, DepthStreamKind, ) @@ -68,7 +69,19 @@ def from_manifest(cls, manifest_path: str | Path) -> "BinanceCaptureSource": raise ValueError("capture manifest files must be a list of names") return cls(tuple(path.parent / name for name in names)) - def envelopes(self) -> Iterator[BinanceCaptureEnvelope]: + @classmethod + def from_directory(cls, directory: str | Path) -> "BinanceCaptureSource": + """Build a source from hourly capture files in chronological order.""" + + path = Path(directory) + return cls(tuple(sorted(path.glob("binance-*.jsonl.gz")))) + + def envelopes( + self, + *, + until_received_at_ns: int | None = None, + symbol: str | None = None, + ) -> Iterator[BinanceCaptureEnvelope]: """Yield validated raw envelopes in capture-file order.""" for path in self.files: @@ -76,12 +89,47 @@ def envelopes(self) -> Iterator[BinanceCaptureEnvelope]: for line_number, line in enumerate(rows, start=1): try: raw = json.loads(line) - yield parse_envelope(raw) + if not isinstance(raw, dict): + raise ValueError("capture envelope must be a JSON object") + received_at_ns = raw.get("received_at_ns") + if ( + until_received_at_ns is not None + and isinstance(received_at_ns, int) + and received_at_ns > until_received_at_ns + ): + return + if symbol is not None and raw.get("symbol") != symbol: + continue + envelope = parse_envelope(raw) + yield envelope except (TypeError, ValueError) as exc: raise ValueError( f"invalid capture row {path}:{line_number}: {exc}" ) from exc + def observations( + self, + *, + until_received_at_ns: int | None = None, + symbol: str | None = None, + ) -> Iterator[BinanceObservedEvent]: + """Yield depth, individual-trade, and book-ticker evidence in receive order.""" + + for envelope in self.envelopes( + until_received_at_ns=until_received_at_ns, + symbol=symbol, + ): + if envelope.kind == "depth_snapshot": + yield parse_depth_snapshot(envelope) + continue + stream_kind = depth_stream_kind(envelope) + if stream_kind == "depth": + yield parse_depth_update(envelope, stream_kind=stream_kind) + elif is_stream(envelope, "@trade"): + yield parse_individual_trade(envelope) + elif is_stream(envelope, "@bookTicker"): + yield parse_book_ticker(envelope) + def depth_snapshots(self) -> Iterator[BinanceDepthSnapshot]: """Yield every captured REST depth snapshot.""" diff --git a/tests/test_binance_reconstruction.py b/tests/test_binance_reconstruction.py new file mode 100644 index 0000000..e167b23 --- /dev/null +++ b/tests/test_binance_reconstruction.py @@ -0,0 +1,353 @@ +from dataclasses import replace +from decimal import Decimal +from typing import cast + +import pytest + +from ordersim.connectors.binance import ( + BinanceDepthSnapshot, + BinanceDepthUpdate, + BinanceIndividualTrade, + BinanceMBOReconstructor, + BinancePriceLevel, + BinanceReconstructionConfig, + BinanceReconstructionPolicy, +) +from ordersim.sim import ( + CppMatchingEngine, + MatchingEngine, + cpp_execution_engine_available, +) + + +def level(price: str, quantity: str) -> BinancePriceLevel: + return BinancePriceLevel(price=Decimal(price), quantity=Decimal(quantity)) + + +def snapshot(*, bid_quantity: str = "10") -> BinanceDepthSnapshot: + return BinanceDepthSnapshot( + symbol="BTCUSDT", + connection_id="depth-1", + received_at_ns=90, + received_monotonic_ns=900, + last_update_id=100, + bids=(level("100", bid_quantity),), + asks=(level("101", "8"),), + ) + + +def update( + update_id: int, + *, + bids: tuple[BinancePriceLevel, ...] = (), + asks: tuple[BinancePriceLevel, ...] = (), +) -> BinanceDepthUpdate: + return BinanceDepthUpdate( + symbol="BTCUSDT", + connection_id="depth-1", + stream_kind="depth", + event_time_ns=update_id * 10, + transaction_time_ns=update_id * 10, + received_at_ns=update_id * 10 + 1, + received_monotonic_ns=update_id * 100, + first_update_id=update_id, + final_update_id=update_id, + previous_update_id=update_id - 1, + bids=bids, + asks=asks, + ) + + +def trade( + trade_id: int, + *, + quantity: str, + buyer_is_maker: bool = True, + price: str = "100", +) -> BinanceIndividualTrade: + return BinanceIndividualTrade( + symbol="BTCUSDT", + connection_id="trades-1", + event_time_ns=trade_id * 10, + trade_time_ns=trade_id * 10, + received_at_ns=trade_id * 10 + 2, + received_monotonic_ns=trade_id * 100, + trade_id=trade_id, + price=Decimal(price), + quantity=Decimal(quantity), + buyer_is_maker=buyer_is_maker, + ) + + +def model( + policy: BinanceReconstructionPolicy = "queue-conservative", +) -> BinanceMBOReconstructor: + return BinanceMBOReconstructor( + BinanceReconstructionConfig( + quantity_step=Decimal("1"), + policy=policy, + ) + ) + + +def test_reconstruction_reaches_every_observed_depth_endpoint() -> None: + reconstructor = model() + bootstrap = reconstructor.bootstrap(snapshot(), update(100)) + + step = reconstructor.apply_update( + update(110, bids=(level("100", "9"),)), + trades=(trade(105, quantity="4"),), + ) + + assert reconstructor.level_quantity("bid", Decimal("100")) == 9 + assert reconstructor.level_quantity("ask", Decimal("101")) == 8 + assert step.metrics.trade_units == 4 + assert step.metrics.inferred_add_units == 3 + assert step.metrics.inferred_cancel_units == 0 + assert bootstrap.metrics.snapshot_add_units == 18 + + +def test_trade_at_unchanged_level_requires_visible_replenishment() -> None: + reconstructor = model("queue-optimistic") + reconstructor.bootstrap(snapshot(bid_quantity="2"), update(100)) + + step = reconstructor.apply_update( + update(110), + trades=(trade(105, quantity="3"),), + ) + + assert reconstructor.level_quantity("bid", Decimal("100")) == 2 + assert [event.action for event in step.events] == ["add", "trade", "add"] + assert step.metrics.required_replenishment_units == 1 + assert step.metrics.inferred_add_units == 3 + + +def test_reconstructed_events_run_in_the_reference_matching_engine() -> None: + reconstructor = model() + engine = MatchingEngine() + + bootstrap = reconstructor.bootstrap(snapshot(), update(100)) + step = reconstructor.apply_update( + update( + 110, + bids=(level("100", "7"),), + asks=(level("101", "6"),), + ), + trades=( + trade(104, quantity="2"), + trade( + 106, + quantity="1", + buyer_is_maker=False, + price="101", + ), + ), + ) + + for event in bootstrap.events + step.events: + engine.apply_event(event) + + bids, asks = engine.book_depth(1) + assert [(row.price, row.size) for row in bids] == [(Decimal("100"), 7)] + assert [(row.price, row.size) for row in asks] == [(Decimal("101"), 6)] + + +@pytest.mark.skipif( + not cpp_execution_engine_available(), + reason="optional C++ execution engine is not built", +) +def test_reconstructed_events_have_equivalent_python_and_cpp_execution() -> None: + reconstructor = model() + bootstrap = reconstructor.bootstrap(snapshot(), update(100)) + step = reconstructor.apply_update( + update(110, bids=(level("100", "5"),)), + trades=(trade(105, quantity="11"),), + ) + python_engine = MatchingEngine() + cpp_engine = CppMatchingEngine(tick_size=Decimal("1")) + + for event in bootstrap.events: + python_engine.apply_event(event) + cpp_engine.apply_event(event) + python_order = python_engine.place_limit("buy", Decimal("100"), 1) + cpp_order = cpp_engine.place_limit("buy", Decimal("100"), 1) + + python_fills = [ + fill for event in step.events for fill in python_engine.apply_event(event) + ] + cpp_fills = [ + fill for event in step.events for fill in cpp_engine.apply_event(event) + ] + + assert python_order.order_id == cpp_order.order_id + assert python_fills == cpp_fills + assert python_engine.book_depth(1) == cpp_engine.book_depth(1) + + +def test_named_policies_bound_cancellation_effect_on_queue_position() -> None: + queue_ahead: dict[str, int] = {} + + for policy in ("queue-conservative", "queue-optimistic"): + reconstructor = model(policy) + engine = MatchingEngine() + bootstrap = reconstructor.bootstrap(snapshot(), update(100)) + for event in bootstrap.events: + engine.apply_event(event) + + own_order_id = engine.place_limit("buy", Decimal("100"), 1).order_id + assert own_order_id is not None + + addition = reconstructor.apply_update( + update(110, bids=(level("100", "15"),)) + ) + cancellation = reconstructor.apply_update( + update(120, bids=(level("100", "10"),)) + ) + for event in addition.events + cancellation.events: + engine.apply_event(event) + + own_order = next( + order for order in engine.own_orders() if order.order_id == own_order_id + ) + queue_ahead[policy] = own_order.queue_ahead_size + + assert queue_ahead == { + "queue-conservative": 10, + "queue-optimistic": 5, + } + + +def test_reconstruction_rejects_invalid_bridge_and_quantity_step() -> None: + invalid_bridge = update(101) + + with pytest.raises(ValueError, match="does not span"): + model().bootstrap(snapshot(), invalid_bridge) + + fractional = BinanceMBOReconstructor( + BinanceReconstructionConfig(quantity_step=Decimal("0.01")) + ) + with pytest.raises(ValueError, match="not divisible"): + fractional.bootstrap(snapshot(bid_quantity="1.005"), update(100)) + + +def test_trade_must_belong_to_the_depth_interval() -> None: + reconstructor = model() + reconstructor.bootstrap(snapshot(), update(100)) + + with pytest.raises(ValueError, match="outside"): + reconstructor.apply_update( + update(110), + trades=(trade(99, quantity="1"),), + ) + + +def test_configuration_and_bootstrap_boundaries_are_explicit() -> None: + with pytest.raises(ValueError, match="quantity_step"): + BinanceReconstructionConfig(quantity_step=Decimal("0")) + with pytest.raises(ValueError, match="unsupported"): + BinanceReconstructionConfig( + quantity_step=Decimal("1"), + policy=cast(BinanceReconstructionPolicy, "unknown"), + ) + + reconstructor = model() + assert reconstructor.symbol is None + with pytest.raises(RuntimeError, match="bootstrap"): + reconstructor.apply_update(update(101)) + with pytest.raises(ValueError, match="standard depth"): + reconstructor.bootstrap( + snapshot(), + replace(update(100), stream_kind="rpi_depth"), + ) + with pytest.raises(ValueError, match="symbols differ"): + reconstructor.bootstrap( + snapshot(), + replace(update(100), symbol="ETHUSDT"), + ) + with pytest.raises(ValueError, match="connections differ"): + reconstructor.bootstrap( + snapshot(), + replace(update(100), connection_id="depth-2"), + ) + + reconstructor.bootstrap(snapshot(), update(100)) + assert reconstructor.symbol == "BTCUSDT" + with pytest.raises(RuntimeError, match="already"): + reconstructor.bootstrap(snapshot(), update(100)) + + +@pytest.mark.parametrize( + ("changed", "message"), + [ + ({"symbol": "ETHUSDT"}, "symbol differs"), + ({"connection_id": "depth-2"}, "connection differs"), + ({"stream_kind": "rpi_depth"}, "standard depth"), + ({"transaction_time_ns": 999}, "moved backwards"), + ], +) +def test_later_depth_updates_must_remain_in_one_segment( + changed: dict[str, object], + message: str, +) -> None: + reconstructor = model() + reconstructor.bootstrap(snapshot(), update(100)) + + with pytest.raises(ValueError, match=message): + reconstructor.apply_update(replace(update(110), **changed)) + + +def test_trade_validation_and_best_price_removal() -> None: + reconstructor = model() + reconstructor.bootstrap(snapshot(), update(100)) + + with pytest.raises(ValueError, match="symbols differ"): + reconstructor.apply_update( + update(110), + trades=(replace(trade(105, quantity="1"), symbol="ETHUSDT"),), + ) + with pytest.raises(ValueError, match="must be positive"): + reconstructor.apply_update( + update(110), + trades=(replace(trade(105, quantity="1"), quantity=Decimal("0")),), + ) + + reconstructor.apply_update( + update( + 110, + bids=(level("100", "0"),), + asks=(level("101", "0"),), + ) + ) + assert reconstructor.book_top() == (None, None) + + +def test_zero_snapshot_levels_and_metrics_only_mode() -> None: + empty_bid = snapshot(bid_quantity="0") + reconstructor = BinanceMBOReconstructor( + BinanceReconstructionConfig( + quantity_step=Decimal("1"), + emit_events=False, + ) + ) + + bootstrap = reconstructor.bootstrap(empty_bid, update(100)) + step = reconstructor.apply_update( + update(110, bids=(level("100", "2"),)) + ) + traded = reconstructor.apply_update( + update(120), + trades=(trade(115, quantity="3"),), + ) + removed = reconstructor.apply_update( + update( + 130, + bids=(level("100", "0"),), + asks=(level("101", "0"),), + ) + ) + + assert bootstrap.events == () + assert step.events == () + assert traded.events == () + assert removed.events == () + assert reconstructor.book_top() == (None, None) diff --git a/tests/test_binance_reconstruction_study.py b/tests/test_binance_reconstruction_study.py new file mode 100644 index 0000000..ce0e976 --- /dev/null +++ b/tests/test_binance_reconstruction_study.py @@ -0,0 +1,494 @@ +import gzip +import json +from decimal import Decimal +from pathlib import Path + +import pytest + +from ordersim.connectors.binance import ( + BinanceCaptureSource, +) +from ordersim.connectors.binance.reconstruction_study import ( + BinanceReconstructionStudyConfig, + main, + run_binance_reconstruction_study, +) + + +def row( + *, + kind: str, + payload: dict[str, object], + received_at_ns: int, + stream: str | None = None, + scope: str = "public", + connection_id: str = "depth-1", + symbol: str = "BTCUSDT", +) -> dict[str, object]: + return { + "schema_version": 1, + "received_at_ns": received_at_ns, + "received_monotonic_ns": received_at_ns + 1, + "kind": kind, + "scope": scope, + "symbol": symbol, + "connection_id": connection_id, + "stream": stream, + "payload": payload, + } + + +def study_rows() -> list[dict[str, object]]: + return [ + row( + kind="message", + stream="btcusdt@depth@100ms", + received_at_ns=80_000_000, + payload={ + "e": "depthUpdate", + "E": 80, + "T": 80, + "s": "BTCUSDT", + "U": 80, + "u": 80, + "pu": 79, + "b": [], + "a": [], + }, + ), + row( + kind="depth_snapshot", + received_at_ns=90_000_000, + payload={ + "lastUpdateId": 100, + "bids": [["100", "10"]], + "asks": [["101", "8"]], + }, + ), + row( + kind="message", + stream="btcusdt@depth@100ms", + received_at_ns=95_000_000, + payload={ + "e": "depthUpdate", + "E": 95, + "T": 95, + "s": "BTCUSDT", + "U": 99, + "u": 99, + "pu": 98, + "b": [], + "a": [], + }, + ), + row( + kind="message", + stream="btcusdt@depth@100ms", + received_at_ns=101_000_000, + payload={ + "e": "depthUpdate", + "E": 100, + "T": 100, + "s": "BTCUSDT", + "U": 100, + "u": 100, + "pu": 99, + "b": [], + "a": [], + }, + ), + row( + kind="message", + scope="market", + connection_id="trades-1", + stream="btcusdt@trade", + received_at_ns=106_000_000, + payload={ + "e": "trade", + "E": 105, + "T": 105, + "s": "BTCUSDT", + "t": 700, + "p": "100", + "q": "4", + "m": True, + }, + ), + row( + kind="message", + scope="market", + connection_id="trades-1", + stream="btcusdt@trade", + received_at_ns=107_000_000, + payload={ + "e": "trade", + "E": 106, + "T": 106, + "s": "BTCUSDT", + "t": 701, + "p": "0", + "q": "0", + "m": False, + "X": "NA", + }, + ), + row( + kind="message", + scope="market", + connection_id="trades-1", + stream="btcusdt@trade", + received_at_ns=108_000_000, + payload={ + "e": "trade", + "E": 107, + "T": 107, + "s": "BTCUSDT", + "t": 700, + "p": "100", + "q": "4", + "m": True, + }, + ), + row( + kind="message", + stream="btcusdt@depth@100ms", + received_at_ns=111_000_000, + payload={ + "e": "depthUpdate", + "E": 110, + "T": 110, + "s": "BTCUSDT", + "U": 101, + "u": 101, + "pu": 100, + "b": [["100", "9"]], + "a": [], + }, + ), + row( + kind="message", + stream="btcusdt@bookTicker", + received_at_ns=112_000_000, + payload={ + "e": "bookTicker", + "E": 110, + "T": 110, + "s": "BTCUSDT", + "u": 101, + "b": "100", + "B": "9", + "a": "101", + "A": "8", + }, + ), + ] + + +def write_capture(tmp_path: Path, rows: list[dict[str, object]]) -> Path: + tmp_path.mkdir(parents=True, exist_ok=True) + path = tmp_path / "binance-run-20260808T170000Z.jsonl.gz" + with gzip.open(path, mode="wt", encoding="utf-8") as file: + for capture_row in rows: + file.write(json.dumps(capture_row) + "\n") + return path + + +def test_study_aligns_trades_and_compares_named_reconstructions( + tmp_path: Path, +) -> None: + capture = write_capture(tmp_path, study_rows()) + source = BinanceCaptureSource((capture,)) + + report = run_binance_reconstruction_study( + source, + BinanceReconstructionStudyConfig( + symbol="btcusdt", + quantity_step=Decimal("1"), + ), + ) + + assert report.symbol == "BTCUSDT" + assert report.depth_snapshots == 1 + assert len(report.segments) == 1 + assert report.exact_book_ticker_matches == 1 + assert report.exact_book_ticker_mismatches == 0 + assert report.stale_depth_updates == 1 + assert report.zero_value_trade_messages == 1 + assert report.duplicate_trades == 1 + assert report.max_trade_receive_delay_ns == 1_000_000 + assert report.late_trades == 0 + assert report.unassigned_trades == 0 + + conservative = report.totals_by_policy["queue-conservative"] + optimistic = report.totals_by_policy["queue-optimistic"] + assert conservative.depth_updates == optimistic.depth_updates == 2 + assert conservative.trade_units == optimistic.trade_units == 4 + assert conservative.inferred_add_units == optimistic.inferred_add_units == 3 + assert conservative.pre_trade_add_units == 3 + assert optimistic.pre_trade_add_units == 0 + + +def test_study_report_is_an_explicit_model_manifest(tmp_path: Path) -> None: + write_capture(tmp_path, study_rows()) + report = run_binance_reconstruction_study( + BinanceCaptureSource.from_directory(tmp_path), + BinanceReconstructionStudyConfig( + symbol="BTCUSDT", + quantity_step=Decimal("1"), + policies=("queue-conservative",), + until_received_at_ns=112_000_000, + ), + ) + + manifest = report.as_dict() + + assert manifest["model"] == "binance-virtual-mbo-minimum-flow-v1" + assert manifest["alignment"] == "exchange-transaction-time" + assert manifest["quantity_step"] == "1" + assert manifest["until_received_at_ns"] == 112_000_000 + assert manifest["policies"] == ["queue-conservative"] + + +def test_capture_source_cutoff_stops_before_later_rows(tmp_path: Path) -> None: + capture = write_capture(tmp_path, study_rows()) + source = BinanceCaptureSource((capture,)) + + observations = tuple(source.observations(until_received_at_ns=106_000_000)) + + assert len(observations) == 5 + assert observations[-1].received_at_ns == 106_000_000 + assert tuple(source.observations(symbol="ETHUSDT")) == () + + +@pytest.mark.parametrize( + ("changed", "message"), + [ + ({"symbol": " "}, "symbol"), + ({"quantity_step": Decimal("0")}, "quantity_step"), + ({"policies": ()}, "at least one"), + ( + {"policies": ("queue-conservative", "queue-conservative")}, + "unique", + ), + ({"reorder_buffer_ns": -1}, "non-negative"), + ], +) +def test_study_configuration_rejects_ambiguous_inputs( + changed: dict[str, object], + message: str, +) -> None: + values: dict[str, object] = { + "symbol": "BTCUSDT", + "quantity_step": Decimal("1"), + } + values.update(changed) + + with pytest.raises(ValueError, match=message): + BinanceReconstructionStudyConfig(**values) # type: ignore[arg-type] + + +def test_study_marks_sequence_breaks_and_ticker_mismatches(tmp_path: Path) -> None: + rows = study_rows() + depth_payload = rows[-2]["payload"] + assert isinstance(depth_payload, dict) + depth_payload["pu"] = 999 + capture = write_capture(tmp_path / "broken", rows) + + broken = run_binance_reconstruction_study( + BinanceCaptureSource((capture,)), + BinanceReconstructionStudyConfig( + symbol="BTCUSDT", + quantity_step=Decimal("1"), + ), + ) + + assert broken.broken_depth_segments == 1 + + rows = study_rows() + ticker_payload = rows[-1]["payload"] + assert isinstance(ticker_payload, dict) + ticker_payload["B"] = "10" + capture = write_capture(tmp_path / "mismatch", rows) + mismatch = run_binance_reconstruction_study( + BinanceCaptureSource((capture,)), + BinanceReconstructionStudyConfig( + symbol="BTCUSDT", + quantity_step=Decimal("1"), + ), + ) + + assert mismatch.exact_book_ticker_mismatches == 1 + + +def test_study_ignores_other_symbols_and_depth_connections(tmp_path: Path) -> None: + rows = study_rows() + rows.insert( + 0, + row( + kind="message", + symbol="ETHUSDT", + connection_id="eth-depth", + stream="ethusdt@bookTicker", + received_at_ns=70_000_000, + payload={ + "e": "bookTicker", + "E": 70, + "T": 70, + "s": "ETHUSDT", + "u": 1, + "b": "10", + "B": "1", + "a": "11", + "A": "1", + }, + ), + ) + rows.insert( + 3, + row( + kind="message", + connection_id="other-depth", + stream="btcusdt@depth@100ms", + received_at_ns=92_000_000, + payload={ + "e": "depthUpdate", + "E": 92, + "T": 92, + "s": "BTCUSDT", + "U": 100, + "u": 100, + "pu": 99, + "b": [], + "a": [], + }, + ), + ) + capture = write_capture(tmp_path, rows) + + report = run_binance_reconstruction_study( + BinanceCaptureSource((capture,)), + BinanceReconstructionStudyConfig( + symbol="BTCUSDT", + quantity_step=Decimal("1"), + ), + ) + + assert report.observations == len(rows) - 1 + assert len(report.segments) == 1 + + +def test_study_rejects_an_update_that_cannot_bridge_snapshot(tmp_path: Path) -> None: + rows = study_rows() + bridge_payload = rows[3]["payload"] + assert isinstance(bridge_payload, dict) + bridge_payload["U"] = 101 + bridge_payload["u"] = 101 + capture = write_capture(tmp_path, rows) + + report = run_binance_reconstruction_study( + BinanceCaptureSource((capture,)), + BinanceReconstructionStudyConfig( + symbol="BTCUSDT", + quantity_step=Decimal("1"), + ), + ) + + assert report.broken_depth_segments == 1 + assert report.segments == () + + +def test_study_rejects_ticker_quantity_that_cannot_be_scaled(tmp_path: Path) -> None: + rows = study_rows() + ticker_payload = rows[-1]["payload"] + assert isinstance(ticker_payload, dict) + ticker_payload["B"] = "9.5" + capture = write_capture(tmp_path, rows) + + with pytest.raises(ValueError, match="not divisible"): + run_binance_reconstruction_study( + BinanceCaptureSource((capture,)), + BinanceReconstructionStudyConfig( + symbol="BTCUSDT", + quantity_step=Decimal("1"), + ), + ) + + +def test_study_counts_late_and_unassigned_trades(tmp_path: Path) -> None: + rows = study_rows() + rows.append( + row( + kind="message", + scope="market", + connection_id="trades-1", + stream="btcusdt@trade", + received_at_ns=120_000_000, + payload={ + "e": "trade", + "E": 109, + "T": 109, + "s": "BTCUSDT", + "t": 702, + "p": "100", + "q": "1", + "m": True, + }, + ) + ) + rows.append( + row( + kind="message", + scope="market", + connection_id="trades-1", + stream="btcusdt@trade", + received_at_ns=121_000_000, + payload={ + "e": "trade", + "E": 115, + "T": 115, + "s": "BTCUSDT", + "t": 703, + "p": "100", + "q": "1", + "m": True, + }, + ) + ) + capture = write_capture(tmp_path, rows) + + report = run_binance_reconstruction_study( + BinanceCaptureSource((capture,)), + BinanceReconstructionStudyConfig( + symbol="BTCUSDT", + quantity_step=Decimal("1"), + reorder_buffer_ns=0, + ), + ) + + assert report.late_trades == 1 + assert report.max_late_trade_lag_ns == 1_000_000 + assert report.max_trade_receive_delay_ns == 11_000_000 + assert report.unassigned_trades == 1 + + +def test_study_cli_writes_json_or_stdout( + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + write_capture(tmp_path, study_rows()) + output = tmp_path / "reports" / "study.json" + common = [ + str(tmp_path), + "--symbol", + "BTCUSDT", + "--quantity-step", + "1", + "--policy", + "queue-conservative", + ] + + main(common) + assert '"model": "binance-virtual-mbo-minimum-flow-v1"' in capsys.readouterr().out + + main([*common, "--output", str(output)]) + assert json.loads(output.read_text(encoding="utf-8"))["symbol"] == "BTCUSDT"