From 6c1ab88507b355578221ffe9e3e7d0d48d7dc72a Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Sun, 10 May 2026 10:11:07 +0200 Subject: [PATCH] fix(mbp10): filter front-month per file in mbp10_to_imbalance_bars MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Pre-fix: mbp10_to_imbalance_bars loaded ALL trades from MBP-10 .dbn files without filtering by instrument_id, then fed them through ImbalanceBarSampler. When the dominant ES contract rolls over (ESZ24 → ESH25 → ESM25), price jumps appear at every rollover, failing the precompute_features.rs:371-376 price-continuity gate (3.5% jump rate vs 1% threshold). First observed on workflow train-multi-seed-crb2k yesterday. Fix: mirror the per-file front-month pattern from precompute_features.rs:347-365 (which uses filter_front_month on DbnTrade slices). Same logic applied to Mbp10Trade — but Mbp10Trade didn't carry instrument_id, so: - Added `instrument_id: u32` field to Mbp10Trade. - Populated from `mbp10.hd.instrument_id` in extract_trades_from_dbn_file (the production extraction path). - Defaulted to 0 in extract_trades_from_snapshots (snapshot-diff path used by tests; no per-record id available; backward-compatible — filter is a no-op when all ids are 0). - New filter_front_month_mbp10() helper that mirrors trades_loader's filter_front_month exactly, just operating on Mbp10Trade. - mbp10_to_imbalance_bars now applies filter_front_month_mbp10 per-file in the extraction loop, identical to precompute_features.rs:347-365. Compiles clean. Replaces tracing::debug! with tracing::info! at the per-file log site to make the raw-vs-filtered count visible at production INFO level. The wgdc8 commit 1aaf94306 (EWMA bypass via ImbalanceBarSampler::new) is preserved — fixed-threshold semantics remain correct per pearl_imbalance_bar_ewma_washes_out_configured_threshold. This fix is orthogonal: front-month filter at the trade-tape level, EWMA bypass at the sampler level. Both are needed for imbalance bars to produce sensible output. Co-Authored-By: Claude Opus 4.7 (1M context) --- crates/ml-features/src/mbp10_loader.rs | 55 +++++++++++++++++++++++--- 1 file changed, 50 insertions(+), 5 deletions(-) diff --git a/crates/ml-features/src/mbp10_loader.rs b/crates/ml-features/src/mbp10_loader.rs index de1ff6641..e9bbe29bc 100644 --- a/crates/ml-features/src/mbp10_loader.rs +++ b/crates/ml-features/src/mbp10_loader.rs @@ -343,6 +343,14 @@ pub struct Mbp10Trade { pub timestamp: DateTime, /// True if buyer-initiated (side == 'B'), false otherwise pub is_buy: bool, + /// Databento `instrument_id` for the contract this trade belongs to. + /// Used by `mbp10_to_imbalance_bars` to filter to the front-month + /// (highest-volume) contract per file, mirroring the per-file pattern + /// in `precompute_features.rs:354`. Populated from `mbp10.hd.instrument_id` + /// in `extract_trades_from_dbn_file`; left as 0 in the snapshot-diff + /// path (`extract_trades_from_snapshots`) where the source records + /// don't carry instrument IDs (mostly used by tests). + pub instrument_id: u32, } /// Extract trade ticks from MBP-10 snapshots using consecutive snapshot diffs. @@ -384,6 +392,7 @@ pub fn extract_trades_from_snapshots(snapshots: &[Mbp10Snapshot]) -> Vec Result, volume, timestamp: dt, is_buy, + instrument_id: mbp10.hd.instrument_id, }); } } @@ -485,6 +495,31 @@ pub fn extract_trades_from_dbn_file(file_path: &Path) -> Result, Ok(trades) } +/// Filter `Mbp10Trade`s to keep only the front-month (highest cumulative +/// volume) contract. Mirrors `trades_loader::filter_front_month` for +/// `DbnTrade` — same logic on `Mbp10Trade.instrument_id`. Apply per-file +/// because each quarterly file has a different front-month contract. +fn filter_front_month_mbp10(trades: &[Mbp10Trade]) -> Vec { + if trades.is_empty() { + return Vec::new(); + } + let mut volume_by_id: std::collections::HashMap = + std::collections::HashMap::new(); + for trade in trades { + *volume_by_id.entry(trade.instrument_id).or_insert(0) += trade.volume as u64; + } + let front_month_id = volume_by_id + .into_iter() + .max_by_key(|&(_, vol)| vol) + .map(|(id, _)| id) + .unwrap_or(0); + trades + .iter() + .filter(|t| t.instrument_id == front_month_id) + .cloned() + .collect() +} + /// Recursively collect all .dbn and .dbn.zst files from a directory. fn collect_dbn_files(dir: &Path) -> Vec { let mut files = Vec::new(); @@ -628,17 +663,27 @@ pub fn mbp10_to_imbalance_bars( dbn_files.len() ); - // Collect all trades from all files + // Collect trades from all files, filtering to the front-month (highest- + // volume) contract per file. Without per-file filtering the resulting + // bar stream interleaves contract months — when the dominant ES contract + // rolls over (ESZ24 → ESH25 → ESM25), price jumps appear at every + // rollover, failing the `precompute_features.rs:371-376` price-continuity + // gate. Mirror the per-file pattern in `precompute_features.rs:347-365` + // (uses `filter_front_month` on `DbnTrade` slices); same logic, applied + // to `Mbp10Trade.instrument_id`. let mut all_trades: Vec = Vec::new(); for file_path in &dbn_files { match extract_trades_from_dbn_file(file_path) { Ok(trades) => { - tracing::debug!( - " {} -> {} trades", + let raw_count = trades.len(); + let filtered = filter_front_month_mbp10(&trades); + tracing::info!( + " {} -> {} trades, {} front-month", file_path.file_name().unwrap_or_default().to_string_lossy(), - trades.len() + raw_count, + filtered.len() ); - all_trades.extend(trades); + all_trades.extend(filtered); } Err(e) => { tracing::warn!(