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

perf(precompute): parallel trades load + predecoded sidecar cache

Browse Source 
Flamegraph of precompute_features on 1Q ES showed 62% of CPU time in
zstd decompression, 6% in DBN FSM parsing, and only 2% in the actual
feature math — single-threaded zstd was the bottleneck, not compute.

Two fixes:

1. Per-quarter parallelism on the volume-bar trades loop (was sequential
   `for file in &trade_files`); brings it in line with the OFI path that
   already used par_iter.

2. Predecoded sidecar cache in `crates/ml-features/src/predecoded.rs`:
   first call to a `.dbn.zst` writes a bincode'd Vec<Mbp10Snapshot> or
   Vec<DbnTrade> under `<output_dir>/predecoded/`. Subsequent calls
   deserialize the sidecar and skip zstd entirely. An mtime+size header
   self-invalidates the sidecar when the source changes — no manual
   flush needed when a quarter is re-downloaded.

   Local 1Q ES results:
   - cold (writes sidecar): 40.7s (was 39.3s; +1.4s for write)
   - warm (HIT):             4.7s  (8.7× faster)
   - zstd in flat perf:      62% → 0% of CPU samples
   - sidecar disk per Q:     ~150MB

The sidecar layer also auto-dedupes within a single run: the OFI section
re-loads trades, but the second call hits the sidecar that the
volume-bar section wrote moments earlier.

CLI: `--rebuild-predecoded` purges sidecars for cold-path testing or
after a wire-format change to Mbp10Snapshot / DbnTrade. Sidecars also
self-invalidate on format-version mismatch so old caches are skipped
silently rather than mis-deserializing.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
jgrusewski
2026-05-16 13:55:32 +02:00
parent 86de265fc8
commit e7ce4395e8
6 changed files with 363 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

112
crates/ml/examples/precompute_features.rs
View File

@@ -184,6 +184,17 @@ struct Opts {
/// Only used when `--row-unit snapshot`.
#[arg(long, default_value_t = 100)]
snapshot_warmup: usize,
/// Purge the predecoded-sidecar cache before loading.
///
/// Predecoded sidecars (`<basename>.<kind>.predecoded.bin` under
/// `<output_dir>/predecoded/`) skip zstd-DBN decompression on subsequent
/// precompute runs. They self-invalidate when the source file's mtime or
/// size changes, so this flag is normally unnecessary — pass it only after
/// a format change to `Mbp10Snapshot` / `DbnTrade` (bumps fail to
/// deserialize) or when explicitly testing the cold path.
#[arg(long)]
rebuild_predecoded: bool,
}
// ---------------------------------------------------------------------------
@@ -217,6 +228,20 @@ async fn main() -> Result<()> {
let mbp10_dir = opts.mbp10_data_dir.as_ref().map(PathBuf::from);
let trades_dir = opts.trades_data_dir.as_ref().map(PathBuf::from);
// Predecoded-sidecar cache lives under output_dir/predecoded/ so it sits
// next to the .fxcache outputs (same writability assumptions). First call
// for any source `.dbn.zst` pays zstd; subsequent calls deserialize the
// sidecar and skip decompression entirely (zstd was 62% of wall time on
// the 2026-05-16 1Q ES profile).
let predecoded_dir = output_dir.join("predecoded");
std::fs::create_dir_all(&predecoded_dir)
.with_context(|| format!("Failed to create predecoded dir {}", predecoded_dir.display()))?;
if opts.rebuild_predecoded {
let removed = ml::features::predecoded::purge_predecoded_dir(&predecoded_dir)
.with_context(|| format!("Failed to purge {}", predecoded_dir.display()))?;
info!("--rebuild-predecoded: purged {} sidecar(s) from {}", removed, predecoded_dir.display());
}
// ── Validate inputs ──────────────────────────────────────────────────────
if !data_dir.exists() {
anyhow::bail!("Data directory not found: {}", data_dir.display());
@@ -358,23 +383,40 @@ async fn main() -> Result<()> {
// Load trades PER FILE and filter front-month within each file.
// Each quarterly DBN file has a different front-month contract (e.g. ESH24, ESM24).
// Filtering globally would select only ONE quarter's contract.
let mut all_front_month_trades: Vec<ml::features::DbnTrade> = Vec::new();
let mut total_raw = 0_usize;
for file in &trade_files {
match ml::features::load_trades_sync(file) {
Ok(trades) => {
let raw_count = trades.len();
total_raw += raw_count;
let filtered = ml::features::filter_front_month(&trades);
info!(" {} -> {} trades, {} front-month",
file.file_name().unwrap_or_default().to_string_lossy(),
raw_count, filtered.len());
all_front_month_trades.extend(filtered);
}
Err(e) => {
tracing::warn!(" Failed {:?}: {e}", file.file_name());
}
}
//
// Parallel across quarters via rayon; each file goes through the
// predecoded-sidecar cache so a re-run of `precompute_features` skips
// zstd decompression entirely.
let per_file: Vec<(usize, Vec<ml::features::DbnTrade>)> = {
use rayon::prelude::*;
trade_files
.par_iter()
.filter_map(|file| {
match ml::features::predecoded::load_or_predecode_trades(file, &predecoded_dir) {
Ok(trades) => {
let raw_count = trades.len();
let filtered = ml::features::filter_front_month(&trades);
info!(
" {} -> {} trades, {} front-month",
file.file_name().unwrap_or_default().to_string_lossy(),
raw_count,
filtered.len()
);
Some((raw_count, filtered))
}
Err(e) => {
tracing::warn!(" Failed {:?}: {e}", file.file_name());
None
}
}
})
.collect()
};
let total_raw: usize = per_file.iter().map(|(r, _)| r).sum();
let mut all_front_month_trades: Vec<ml::features::DbnTrade> =
Vec::with_capacity(per_file.iter().map(|(_, t)| t.len()).sum());
for (_, t) in per_file {
all_front_month_trades.extend(t);
}
all_front_month_trades.sort_by_key(|t| t.timestamp);
info!("Total trades: {} raw, {} front-month", total_raw, all_front_month_trades.len());
@@ -521,8 +563,7 @@ async fn main() -> Result<()> {
const OFI_DIM: usize = ml_core::state_layout::OFI_DIM;
let t2 = Instant::now();
let ofi: Vec<[f64; OFI_DIM]> = if let Some(ref mbp10_path) = mbp10_dir {
use ml::features::mbp10_loader::load_ofi_features_parallel;
use ml::features::trades_loader::load_trades_sync;
use ml::features::ofi_calculator::OFICalculator;
info!("Loading MBP-10 snapshots from {}...", mbp10_path.display());
let mut mbp10_files = collect_dbn_files_recursive(mbp10_path);
@@ -532,21 +573,17 @@ async fn main() -> Result<()> {
info!("No MBP-10 files found, OFI will be zeros");
vec![[0.0; OFI_DIM]; n]
} else {
// Load MBP-10 snapshots (parallel)
use data::providers::databento::dbn_parser::DbnParser;
// Load MBP-10 snapshots (parallel, with predecoded sidecar)
let per_file: Vec<Vec<_>> = {
use rayon::prelude::*;
mbp10_files.par_iter().filter_map(|file| {
let parser = match DbnParser::new() {
Ok(p) => p,
Err(e) => { tracing::warn!("Parser init failed: {e}"); return None; }
};
let mut snapshots = Vec::new();
match parser.parse_mbp10_streaming(file, 100, |snap| {
snapshots.push(snap.clone());
}) {
Ok(_) => {
info!(" {} -> {} snapshots", file.file_name().unwrap_or_default().to_string_lossy(), snapshots.len());
match ml::features::predecoded::load_or_predecode_mbp10(file, &predecoded_dir) {
Ok(snapshots) => {
info!(
" {} -> {} snapshots",
file.file_name().unwrap_or_default().to_string_lossy(),
snapshots.len()
);
Some(snapshots)
}
Err(e) => {
@@ -562,24 +599,29 @@ async fn main() -> Result<()> {
all_snapshots.sort_by_key(|s| s.timestamp);
info!("Loaded {} MBP-10 snapshots in {:.1}s", all_snapshots.len(), t2.elapsed().as_secs_f64());
// Load trades (parallel) and filter front-month per-file.
// Load trades (parallel, with predecoded sidecar) and filter
// front-month per-file.
//
// Pre-2026-05-09: this branch loaded trades unfiltered, leaking
// off-contract trade activity into the per-bar OFI/VPIN/Kyle's
// Lambda computations during contract rollover windows. Volume
// bar formation already filters front-month at line 354 of this
// file; the OFI side did not. Audit confirmed the latent bug
// bar formation already filters front-month at the trades-for-bars
// load above; the OFI side did not. Audit confirmed the latent bug
// affected every prior MBP-10+trades production run.
//
// Fix: mirror the per-file `filter_front_month` call from the
// volume bar path so OFI windows only see in-month trades.
//
// Trade loads go through `load_or_predecode_trades` so the sidecar
// cache is shared with the bar-formation path above (one decode of
// each `.dbn.zst` per process — the OFI re-load is a sidecar HIT).
let all_trades = if let Some(ref tdir) = trades_dir {
let mut trade_files = collect_dbn_files_recursive(tdir);
trade_files.sort();
let per_file_trades: Vec<Vec<_>> = {
use rayon::prelude::*;
trade_files.par_iter().filter_map(|path| {
match load_trades_sync(path) {
match ml::features::predecoded::load_or_predecode_trades(path, &predecoded_dir) {
Ok(t) => {
let raw_count = t.len();
let filtered = ml::features::filter_front_month(&t);