gitadmin/foxhunt
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(sp20): parallelise per-file MBP-10 trade extraction in mbp10_to_imbalance_bars

Browse Source 
Bottleneck A of two: the loop in `mbp10_to_imbalance_bars` that calls
`extract_trades_from_dbn_file` + `filter_front_month_mbp10` per file
ran serially across the 9 quarterly DBN files. Each file is independent
(different contract universe per quarter), the front-month filter is
purely intra-file, and the final `all_trades.sort_by(|a, b| timestamp)`
re-sequences across files — so reduce order across files is irrelevant
to correctness.

Switched the loop to `dbn_files.par_iter().filter_map(...).collect()`,
mirroring the trades_loader-side pattern in
`precompute_features.rs:551-565`. Per-file logging (raw count → filtered
front-month count) preserved verbatim. On the `ci-compile-cpu` 28-core
node decoding 9 .dbn.zst files, the file-decoder/zstd-decompress phase
should drop from ~9× single-file latency to ~1× — bounded by the slowest
single file.

This is the trivial half of the parallelisation. Bottleneck B (the
sequential `ImbalanceBarSampler` pass over the concatenated trade list)
follows in the next commit with time-bucket sharding + warmup overlap +
bit-equivalence test, mirroring the SP20 OFI pattern at
`crates/ml-features/src/ofi_calculator.rs::compute_ofi_per_bar_parallel`.

Build: `cargo check -p ml-features` clean.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
jgrusewski
2026-05-10 12:37:45 +02:00
parent 032026dc07
commit 8f5c64e108
1 changed files with 40 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

62
crates/ml-features/src/mbp10_loader.rs
View File

@@ -681,28 +681,46 @@ pub fn mbp10_to_imbalance_bars(
// gate. Mirror the per-file pattern in `precompute_features.rs:347-365` // gate. Mirror the per-file pattern in `precompute_features.rs:347-365`
// (uses `filter_front_month` on `DbnTrade` slices); same logic, applied // (uses `filter_front_month` on `DbnTrade` slices); same logic, applied
// to `Mbp10Trade.instrument_id`. // to `Mbp10Trade.instrument_id`.
let mut all_trades: Vec<Mbp10Trade> = Vec::new(); //
for file_path in &dbn_files { // SP20 parallelism: `extract_trades_from_dbn_file` is a pure
match extract_trades_from_dbn_file(file_path) { // file → Vec<Mbp10Trade> function and the 9 quarterly files are fully
Ok(trades) => { // independent (each carries its own contract universe). Mirror the
let raw_count = trades.len(); // par_iter pattern in `precompute_features.rs:551-565` so a 28-core
let filtered = filter_front_month_mbp10(&trades); // CPU node can decode the files concurrently. Front-month filter is
tracing::info!( // applied per-file as before — semantics unchanged. Final ordering
" {} -> {} trades, {} front-month", // is enforced by the `all_trades.sort_by(|a, b| ...)` step below, so
file_path.file_name().unwrap_or_default().to_string_lossy(), // file-level reduce order is irrelevant for correctness.
raw_count, let per_file_trades: Vec<Vec<Mbp10Trade>> = {
filtered.len() use rayon::prelude::*;
); dbn_files
all_trades.extend(filtered); .par_iter()
} .filter_map(|file_path| match extract_trades_from_dbn_file(file_path) {
Err(e) => { Ok(trades) => {
tracing::warn!( let raw_count = trades.len();
"Failed to extract trades from {}: {}", let filtered = filter_front_month_mbp10(&trades);
file_path.display(), tracing::info!(
e " {} -> {} trades, {} front-month",
); file_path.file_name().unwrap_or_default().to_string_lossy(),
} raw_count,
} filtered.len()
);
Some(filtered)
}
Err(e) => {
tracing::warn!(
"Failed to extract trades from {}: {}",
file_path.display(),
e
);
None
}
})
.collect()
};
let mut all_trades: Vec<Mbp10Trade> =
Vec::with_capacity(per_file_trades.iter().map(Vec::len).sum());
for chunk in per_file_trades {
all_trades.extend(chunk);
} }
if all_trades.is_empty() { if all_trades.is_empty() {