From 8f5c64e1088896e2e6ed8c1e746b7cf42981e133 Mon Sep 17 00:00:00 2001
From: jgrusewski <jgrusewski@localhost>
Date: Sun, 10 May 2026 12:37:45 +0200
Subject: [PATCH] feat(sp20): parallelise per-file MBP-10 trade extraction in
 mbp10_to_imbalance_bars
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

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>
---
 crates/ml-features/src/mbp10_loader.rs | 62 +++++++++++++++++---------
 1 file changed, 40 insertions(+), 22 deletions(-)

diff --git a/crates/ml-features/src/mbp10_loader.rs b/crates/ml-features/src/mbp10_loader.rs
index 50046d2aa..3dfb32748 100644
--- a/crates/ml-features/src/mbp10_loader.rs
+++ b/crates/ml-features/src/mbp10_loader.rs
@@ -681,28 +681,46 @@ pub fn mbp10_to_imbalance_bars(
     // 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) => {
-                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(),
-                    raw_count,
-                    filtered.len()
-                );
-                all_trades.extend(filtered);
-            }
-            Err(e) => {
-                tracing::warn!(
-                    "Failed to extract trades from {}: {}",
-                    file_path.display(),
-                    e
-                );
-            }
-        }
+    //
+    // SP20 parallelism: `extract_trades_from_dbn_file` is a pure
+    // file → Vec<Mbp10Trade> function and the 9 quarterly files are fully
+    // independent (each carries its own contract universe). Mirror the
+    // par_iter pattern in `precompute_features.rs:551-565` so a 28-core
+    // CPU node can decode the files concurrently. Front-month filter is
+    // applied per-file as before — semantics unchanged. Final ordering
+    // is enforced by the `all_trades.sort_by(|a, b| ...)` step below, so
+    // file-level reduce order is irrelevant for correctness.
+    let per_file_trades: Vec<Vec<Mbp10Trade>> = {
+        use rayon::prelude::*;
+        dbn_files
+            .par_iter()
+            .filter_map(|file_path| match extract_trades_from_dbn_file(file_path) {
+                Ok(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(),
+                        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() {