Skip to content

fix(collector): verify complete market tape coverage #293

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
proerror77 merged 1 commit into from
Jul 23, 2026
Merged

fix(collector): verify complete market tape coverage #293

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
132 changes: 127 additions & 5 deletions rust_hft/data-pipelines/core/src/binance_lob_replay.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,7 @@ struct ReplaySequenceState {
session_id: String,
last_update_id: u64,
bridged: bool,
continuity_complete: bool,
bids: HashMap<String, String>,
asks: HashMap<String, String>,
}
Expand Down Expand Up @@ -130,11 +131,36 @@ impl ReplaySequenceValidator {
event_type: &str,
raw: &serde_json::Map<String, Value>,
received_at_ns: u64,
) -> Result<Vec<ReplaySequenceEvent>> {
self.observe_inner(event_type, raw, received_at_ns, false)
}

pub fn observe_verified_stream_coverage_checkpoint(
&mut self,
raw: &serde_json::Map<String, Value>,
received_at_ns: u64,
) -> Result<Vec<ReplaySequenceEvent>> {
if raw.get("stream_coverage_verified").and_then(Value::as_bool) != Some(true) {
anyhow::bail!("checkpoint has no verified stream coverage");
}
self.observe_inner("checkpoint", raw, received_at_ns, true)
}

fn observe_inner(
&mut self,
event_type: &str,
raw: &serde_json::Map<String, Value>,
received_at_ns: u64,
allow_verified_static_checkpoint: bool,
) -> Result<Vec<ReplaySequenceEvent>> {
let mut events = Vec::new();
match event_type {
"snapshot" if required_string(raw, "symbol")? == self.symbol => {
if self.state.as_ref().is_some_and(|state| !state.bridged) {
if self
.state
.as_ref()
.is_some_and(|state| !state.continuity_complete)
{
anyhow::bail!("snapshot replaced an unbridged replay series");
}
let snapshot = raw
Expand All @@ -148,6 +174,7 @@ impl ReplaySequenceValidator {
.and_then(Value::as_u64)
.ok_or_else(|| anyhow::anyhow!("snapshot payload has no lastUpdateId"))?,
bridged: false,
continuity_complete: false,
bids: parse_snapshot_side(snapshot.get("bids"))?,
asks: parse_snapshot_side(snapshot.get("asks"))?,
});
Expand Down Expand Up @@ -184,7 +211,12 @@ impl ReplaySequenceValidator {
self.pending.push(diff);
} else if self.state.as_ref().expect("checked state").session_id != diff.session_id
{
if !self.state.as_ref().expect("checked state").bridged {
if !self
.state
.as_ref()
.expect("checked state")
.continuity_complete
{
anyhow::bail!("diff replaced an unbridged replay series");
}
self.state = None;
Expand All @@ -208,6 +240,12 @@ impl ReplaySequenceValidator {
.get("bridged")
.and_then(Value::as_bool)
.ok_or_else(|| anyhow::anyhow!("checkpoint has no bridged state"))?,
continuity_complete: raw
.get("continuity_complete")
.and_then(Value::as_bool)
.unwrap_or_else(|| {
raw.get("bridged").and_then(Value::as_bool) == Some(true)
}),
bids: parse_snapshot_side(raw.get("bids"))?,
asks: parse_snapshot_side(raw.get("asks"))?,
};
Expand All @@ -220,15 +258,29 @@ impl ReplaySequenceValidator {
events.push(replay_checkpoint_seed(raw, received_at_ns)?);
}
Some(state) if state.session_id != checkpoint.session_id => {
if !state.bridged {
if !state.continuity_complete {
anyhow::bail!("checkpoint replaced an unbridged replay series");
}
self.state = Some(checkpoint);
events.push(replay_checkpoint_seed(raw, received_at_ns)?);
}
Some(state)
if allow_verified_static_checkpoint
&& !state.bridged
&& !checkpoint.bridged
&& checkpoint.continuity_complete
&& raw.get("stream_coverage_verified").and_then(Value::as_bool)
== Some(true)
&& state.last_update_id == checkpoint.last_update_id
&& state.bids == checkpoint.bids
&& state.asks == checkpoint.asks =>
{
self.state = Some(checkpoint);
}
Some(state)
if state.last_update_id != checkpoint.last_update_id
|| state.bridged != checkpoint.bridged
|| state.continuity_complete != checkpoint.continuity_complete
|| state.bids != checkpoint.bids
|| state.asks != checkpoint.asks =>
{
Expand Down Expand Up @@ -274,6 +326,7 @@ impl ReplaySequenceValidator {
}
state.last_update_id = update.diff.final_update_id;
state.bridged = true;
state.continuity_complete = true;
update_side(&mut state.bids, &update.diff.bids);
update_side(&mut state.asks, &update.diff.asks);
Ok(true)
Expand All @@ -282,7 +335,10 @@ impl ReplaySequenceValidator {
pub fn finish(&self) -> Result<()> {
if !self.pending.is_empty()
|| self.state.is_none()
|| self.state.as_ref().is_some_and(|state| !state.bridged)
|| self
.state
.as_ref()
.is_some_and(|state| !state.continuity_complete)
{
anyhow::bail!("collector replay did not finish in a bridged state");
}
Expand All @@ -293,7 +349,7 @@ impl ReplaySequenceValidator {
let state = self
.state
.as_ref()
.filter(|state| state.bridged)
.filter(|state| state.continuity_complete)
.ok_or_else(|| anyhow::anyhow!("collector replay has no bridged book snapshot"))?;
Ok(ReplayBookSnapshot {
last_update_id: state.last_update_id,
Expand Down Expand Up @@ -466,6 +522,72 @@ mod tests {
assert!(replay.book_snapshot().is_err());
}

#[test]
fn verified_stream_coverage_bridges_an_unchanged_snapshot() {
let mut replay = ReplaySequenceValidator::new(Market::Spot, "BTCUSDT").unwrap();
let snapshot = json!({
"symbol": "BTCUSDT",
"session_id": "session-1",
"snapshot": {
"lastUpdateId": 100,
"bids": [["100", "1"]],
"asks": [["101", "1"]]
}
});
replay
.observe("snapshot", snapshot.as_object().unwrap(), 100)
.unwrap();
let checkpoint = json!({
"symbol": "BTCUSDT",
"session_id": "session-1",
"last_update_id": 100,
"bridged": false,
"continuity_complete": true,
"stream_coverage_verified": true,
"bids": [["100", "1"]],
"asks": [["101", "1"]]
});

replay
.observe_verified_stream_coverage_checkpoint(checkpoint.as_object().unwrap(), 101)
.unwrap();
replay.finish().unwrap();
}

#[test]
fn generic_replay_does_not_adopt_collector_stream_coverage_semantics() {
let mut replay = ReplaySequenceValidator::new(Market::Spot, "BTCUSDT").unwrap();
let snapshot = json!({
"symbol": "BTCUSDT",
"session_id": "session-1",
"snapshot": {
"lastUpdateId": 100,
"bids": [["100", "1"]],
"asks": [["101", "1"]]
}
});
replay
.observe("snapshot", snapshot.as_object().unwrap(), 100)
.unwrap();
let checkpoint = json!({
"symbol": "BTCUSDT",
"session_id": "session-1",
"last_update_id": 100,
"bridged": false,
"continuity_complete": true,
"stream_coverage_verified": true,
"bids": [["100", "1"]],
"asks": [["101", "1"]]
});

let error = replay
.observe("checkpoint", checkpoint.as_object().unwrap(), 101)
.unwrap_err();
assert!(error
.to_string()
.contains("checkpoint does not match replayed update state"));
}

#[test]
fn replay_book_snapshot_is_full_and_price_sorted() {
let mut replay = ReplaySequenceValidator::new(Market::Spot, "BTCUSDT").unwrap();
Expand Down
7 changes: 6 additions & 1 deletion rust_hft/data-pipelines/core/src/binance_market_tape.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,8 @@ pub struct SymbolLobContinuitySummary {
pub snapshot_seed_count: u64,
pub diff_count: u64,
pub checkpoint_count: u64,
#[serde(default)]
pub stream_coverage_verified: bool,
pub first_update_id: Option<u64>,
pub last_update_id: Option<u64>,
pub first_source_time_ms: Option<u64>,
Expand Down Expand Up @@ -153,6 +155,8 @@ impl LobContinuitySummaryBuilder {
.checkpoint_count
.checked_add(1)
.context("LOB checkpoint count overflow")?;
summary.stream_coverage_verified |=
raw.get("stream_coverage_verified").and_then(Value::as_bool) == Some(true);
summary.observe_received_at(received_at_ns);
summary.observe_depth(bid_levels, ask_levels);
}
Expand Down Expand Up @@ -186,8 +190,8 @@ impl LobContinuitySummaryBuilder {
.iter()
.filter(|(_, summary)| {
summary.snapshot_seed_count == 0
|| summary.diff_count == 0
|| summary.checkpoint_count == 0
|| (summary.diff_count == 0 && !summary.stream_coverage_verified)
})
.map(|(symbol, _)| symbol.clone())
.collect::<Vec<_>>();
Expand Down Expand Up @@ -301,6 +305,7 @@ pub fn event_type_allowed(schema: &str, event_type: &str) -> bool {
MARKET_TAPE_SCHEMA => matches!(
event_type,
"session_start"
| "stream_coverage"
| "snapshot"
| "diff"
| "checkpoint"
Expand Down
Loading
Loading