feat(data): validate Binance market tape - #104
Conversation
|
You have reached your Codex usage limits for code reviews. You can see your limits in the Codex usage dashboard. |
📝 WalkthroughWalkthroughAdds a public Binance market-tape module with schema allowlists, depth and aggregate-trade parsers, clock checks, per-symbol sequence validators, helper functions, and comprehensive unit tests. ChangesBinance market tape
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant ArchivedEvent
participant DepthSourceClock
participant DepthSourceClockSequenceValidator
ArchivedEvent->>DepthSourceClock: Parse and validate depth event
DepthSourceClock->>DepthSourceClockSequenceValidator: Observe symbol update
DepthSourceClockSequenceValidator-->>DepthSourceClock: Accept or reject sequence
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
There was a problem hiding this comment.
Actionable comments posted: 3
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@rust_hft/data-pipelines/core/src/binance_market_tape.rs`:
- Around line 221-248: The validator around the last-trade state must also track
each trade’s previous last_trade_id and validate the next first_trade_id is
exactly previous last_trade_id + 1, rejecting both gaps and rollbacks while
preserving aggregate-ID and timestamp checks. Extend the state stored by the
last map and add a targeted counterexample test with consecutive aggregate IDs
but discontinuous underlying trade ranges.
- Around line 130-145: Update the state handling in the clock validation method
around the last-entry insertion so an omitted transaction time does not replace
the previously observed Some value; retain the last known transaction timestamp
for subsequent rollback checks while preserving None when no prior timestamp
exists. Add a targeted counterexample test covering Some(100) → None → Some(99),
which must reject the rollback.
- Around line 55-59: Update the archived constructors from_archived_event and
the corresponding archived event constructor to validate the envelope’s schema
and event type against the exact frame type before calling from_frame. Reject
unsupported schemas and mismatched envelope/frame identities, and add targeted
counterexample tests covering both unsupported schemas and mislabeled
event/frame pairs.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: b8e3cdec-9f46-4704-8909-35856693d51f
📒 Files selected for processing (2)
rust_hft/data-pipelines/core/src/binance_market_tape.rsrust_hft/data-pipelines/core/src/lib.rs
| pub fn from_archived_event(raw: &Map<String, Value>, received_at_ns: u64) -> Result<Self> { | ||
| Self::from_frame( | ||
| raw.get("frame").context("depth event has no frame")?, | ||
| received_at_ns, | ||
| ) |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Bind archived schema and event type to the parsed frame.
Both archived constructors extract only frame, so an unsupported schema or mislabeled event (snapshot containing a depth update, for example) is accepted. Validate the archived schema/event pair and require the exact expected event type before parsing.
Add counterexamples for unsupported schemas and mismatched envelope/frame identities. As per coding guidelines, “Every safety boundary must have a targeted counterexample test.”
Also applies to: 165-170
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@rust_hft/data-pipelines/core/src/binance_market_tape.rs` around lines 55 -
59, Update the archived constructors from_archived_event and the corresponding
archived event constructor to validate the envelope’s schema and event type
against the exact frame type before calling from_frame. Reject unsupported
schemas and mismatched envelope/frame identities, and add targeted
counterexample tests covering both unsupported schemas and mislabeled
event/frame pairs.
Source: Coding guidelines
| if clock.event_time_ms < previous_event_time | ||
| || previous_transaction_time | ||
| .zip(clock.transaction_time_ms) | ||
| .is_some_and(|(previous, current)| current < previous) | ||
| { | ||
| anyhow::bail!("{} depth source-time rollback", clock.symbol); | ||
| } | ||
| } | ||
| self.last.insert( | ||
| clock.symbol.clone(), | ||
| ( | ||
| clock.event_time_ms, | ||
| clock.transaction_time_ms, | ||
| clock.final_update_id, | ||
| ), | ||
| ); |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Preserve the last known transaction clock when T is omitted.
A Some(100) → None → Some(99) sequence passes because Line 142 replaces the prior timestamp with None. Compare future values against the last observed transaction time and retain that value across frames without T.
Add this omission/rollback sequence as a counterexample test. As per coding guidelines, “Every safety boundary must have a targeted counterexample test.”
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@rust_hft/data-pipelines/core/src/binance_market_tape.rs` around lines 130 -
145, Update the state handling in the clock validation method around the
last-entry insertion so an omitted transaction time does not replace the
previously observed Some value; retain the last known transaction timestamp for
subsequent rollback checks while preserving None when no prior timestamp exists.
Add a targeted counterexample test covering Some(100) → None → Some(99), which
must reject the rollback.
Source: Coding guidelines
| if let Some((previous_id, previous_event_time, previous_trade_time)) = | ||
| self.last.get(&trade.symbol).copied() | ||
| { | ||
| let expected = previous_id | ||
| .checked_add(1) | ||
| .context("aggregate trade id overflow")?; | ||
| if trade.aggregate_trade_id != expected { | ||
| anyhow::bail!( | ||
| "{} aggregate trade gap expected={} received={}", | ||
| trade.symbol, | ||
| expected, | ||
| trade.aggregate_trade_id | ||
| ); | ||
| } | ||
| if trade.event_time_ms < previous_event_time | ||
| || trade.trade_time_ms < previous_trade_time | ||
| { | ||
| anyhow::bail!("{} aggregate trade source-time rollback", trade.symbol); | ||
| } | ||
| } | ||
| self.last.insert( | ||
| trade.symbol.clone(), | ||
| ( | ||
| trade.aggregate_trade_id, | ||
| trade.event_time_ms, | ||
| trade.trade_time_ms, | ||
| ), | ||
| ); |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Validate continuity of the underlying trade-ID ranges.
The validator tracks only aggregate_trade_id; consecutive aggregate IDs with first_trade_id jumping beyond the previous last_trade_id + 1 are accepted, silently losing trade-level sequence evidence. Store the previous last_trade_id and reject gaps or rollbacks in first_trade_id.
Add a counterexample with consecutive aggregate IDs but discontinuous trade ranges. As per coding guidelines, “Every safety boundary must have a targeted counterexample test.”
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@rust_hft/data-pipelines/core/src/binance_market_tape.rs` around lines 221 -
248, The validator around the last-trade state must also track each trade’s
previous last_trade_id and validate the next first_trade_id is exactly previous
last_trade_id + 1, rejecting both gaps and rollbacks while preserving
aggregate-ID and timestamp checks. Extend the state stored by the last map and
add a targeted counterexample test with consecutive aggregate IDs but
discontinuous underlying trade ranges.
Source: Coding guidelines
Change contract: Add one shared typed Binance market-tape contract that validates schema/event identity, bounded source clocks, diff-depth U/u/pu continuity, and aggregate-trade numeric and sequence invariants before downstream research consumers can accept a frame.
Out of scope: Collector wiring; immutable artifact manifest or snapshot compilation; backtest consumer changes; DuckDB or PostgreSQL changes; Polymarket; L3/order-id data; queue, impact, fill, or capacity modeling; live execution.
Dependency or merge order: None. This PR is based directly on main SHA a3e879f and can merge or roll back independently. The later OSS artifact verifier will consume this contract in a separate PR.
Focused validation:
Rollout or rollback impact: This exports a shared validation module but does not wire it into a production collector or execution path, so runtime behavior is unchanged until a later consumer PR. Revert this PR to remove the contract; no data migration, database schema, deployment, or live-trading change is involved.
Summary by CodeRabbit