fix(mbp10): filter front-month per file in mbp10_to_imbalance_bars
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) <noreply@anthropic.com>
This commit is contained in:
@@ -343,6 +343,14 @@ pub struct Mbp10Trade {
|
||||
pub timestamp: DateTime<Utc>,
|
||||
/// 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<Mbp10Tr
|
||||
volume,
|
||||
timestamp: dt,
|
||||
is_buy,
|
||||
instrument_id: 0, // snapshot-diff path has no per-record id
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -466,6 +475,7 @@ pub fn extract_trades_from_dbn_file(file_path: &Path) -> Result<Vec<Mbp10Trade>,
|
||||
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<Vec<Mbp10Trade>,
|
||||
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<Mbp10Trade> {
|
||||
if trades.is_empty() {
|
||||
return Vec::new();
|
||||
}
|
||||
let mut volume_by_id: std::collections::HashMap<u32, u64> =
|
||||
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<std::path::PathBuf> {
|
||||
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<Mbp10Trade> = 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!(
|
||||
|
||||
Reference in New Issue
Block a user