Skip to content

fix(data): enforce Binance raw trade range continuity #122

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

Closed
proerror77 wants to merge 2 commits into from
Closed

fix(data): enforce Binance raw trade range continuity #122

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
30b56e9
fix(data): enforce Binance raw trade range continuity
Jul 17, 2026
2e03658
fix(data): remove needless process_state borrows
Jul 17, 2026
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
63 changes: 60 additions & 3 deletions 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 @@ -213,13 +213,17 @@ impl AggregateTrade {

#[derive(Debug, Default)]
pub struct AggregateTradeSequenceValidator {
last: HashMap<String, (u64, u64, u64)>,
last: HashMap<String, (u64, u64, u64, u64)>,
}

impl AggregateTradeSequenceValidator {
pub fn observe(&mut self, trade: &AggregateTrade) -> Result<()> {
if let Some((previous_id, previous_event_time, previous_trade_time)) =
self.last.get(&trade.symbol).copied()
if let Some((
previous_id,
previous_last_trade_id,
previous_event_time,
previous_trade_time,
)) = self.last.get(&trade.symbol).copied()
{
let expected = previous_id
.checked_add(1)
Expand All @@ -232,6 +236,17 @@ impl AggregateTradeSequenceValidator {
trade.aggregate_trade_id
);
}
let expected_first_trade_id = previous_last_trade_id
.checked_add(1)
.context("aggregate trade raw trade id overflow")?;
if trade.first_trade_id != expected_first_trade_id {

@chatgpt-codex-connector chatgpt-codex-connector Bot Jul 17, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Avoid rejecting valid futures aggTrade raw-ID skips

When MARKET=usdm, Config::stream_urls subscribes to Binance's futures @aggTrade stream, whose docs say only market trades are aggregated and insurance-fund/ADL trades are not returned (https://developers.binance.com/docs/derivatives/usds-margined-futures/websocket-market-streams/All-Market-Mini-Tickers-Stream#aggregate-trade-streams). If one of those omitted executions consumes a raw trade ID, the next valid aggregate trade can have f > previous_l + 1; this equality then writes aggregate_trade_gap, returns an error, and the archiver keeps reconnecting from the stale validator state rather than accepting the live futures stream.

Useful? React with 👍 / 👎.

anyhow::bail!(
"{} aggregate trade raw trade gap expected={} received={}",
trade.symbol,
expected_first_trade_id,
trade.first_trade_id
);
}
if trade.event_time_ms < previous_event_time
|| trade.trade_time_ms < previous_trade_time
{
Expand All @@ -242,6 +257,7 @@ impl AggregateTradeSequenceValidator {
trade.symbol.clone(),
(
trade.aggregate_trade_id,
trade.last_trade_id,
trade.event_time_ms,
trade.trade_time_ms,
),
Expand Down Expand Up @@ -629,4 +645,45 @@ mod tests {
.to_string()
.contains("rollback"));
}

#[test]
fn aggregate_trade_sequence_rejects_raw_trade_range_gap() {
let mut first_frame = frame(10, 1_700_000_000_000, 1_700_000_000_000);
first_frame["data"]["f"] = json!(100);
first_frame["data"]["l"] = json!(101);
let mut next_frame = frame(11, 1_700_000_000_001, 1_700_000_000_001);
next_frame["data"]["f"] = json!(103);
next_frame["data"]["l"] = json!(103);
let first = AggregateTrade::from_frame(&first_frame, 1_700_000_000_100_000_000).unwrap();
let next = AggregateTrade::from_frame(&next_frame, 1_700_000_000_100_000_000).unwrap();

let mut sequence = AggregateTradeSequenceValidator::default();
sequence.observe(&first).unwrap();
assert!(sequence
.observe(&next)
.unwrap_err()
.to_string()
.contains("gap"));
}

#[test]
fn aggregate_trade_sequence_fails_closed_on_raw_trade_id_overflow() {
let mut first_frame = frame(10, 1_700_000_000_000, 1_700_000_000_000);
first_frame["data"]["f"] = json!(u64::MAX);
first_frame["data"]["l"] = json!(u64::MAX);
let first = AggregateTrade::from_frame(&first_frame, 1_700_000_000_100_000_000).unwrap();
let next = AggregateTrade::from_frame(
&frame(11, 1_700_000_000_001, 1_700_000_000_001),
1_700_000_000_100_000_000,
)
.unwrap();

let mut sequence = AggregateTradeSequenceValidator::default();
sequence.observe(&first).unwrap();
assert!(sequence
.observe(&next)
.unwrap_err()
.to_string()
.contains("overflow"));
}
}
6 changes: 3 additions & 3 deletions rust_hft/tools/collector/src/bin/binance-lob-archiver.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -603,7 +603,7 @@ async fn run_session(
config: Arc<Config>,
mut shutdown: watch::Receiver<bool>,
watchdog: ProcessWatchdog,
mut process_state: &mut ProcessState,
process_state: &mut ProcessState,
) -> anyhow::Result<()> {
let session_id = format!("{:x}-{}", now_ns()?, std::process::id());
let active_symbols = config.active_symbols();
Expand Down Expand Up @@ -665,7 +665,7 @@ async fn run_session(
Some(Ok(Ok(TaskExit::Stopped(Some(event))))) => {
pending_action = process_event(
&config, &mut segment, &mut states, &mut budget, &session_id, event,
&mut process_state,
process_state,
)?;
}
Some(Ok(Ok(TaskExit::SnapshotComplete))) => {},
Expand All @@ -690,7 +690,7 @@ async fn run_session(
Some(event) => {
match process_event(
&config, &mut segment, &mut states, &mut budget, &session_id, event,
&mut process_state,
process_state,
) {
Ok(action) => pending_action = action,
Err(error) => {
Expand Down
Loading