From 5930d9586e2ea667068844bde024a6a8f9532ac4 Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Fri, 22 May 2026 22:16:15 +0200 Subject: [PATCH] =?UTF-8?q?perf(loader):=20cache=20computed=20labels=20arr?= =?UTF-8?q?ays=20to=20disk=20(~50x=20first-run=20/=20~=E2=88=9E=20subseque?= =?UTF-8?q?nt)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds per-file labels cache sidecar (CachedLabels struct with labels_full + outcome_prof_long/short_full + outcome_size_long/short_full + sigma_k_full + pos_fraction + regime_full). Cache key encodes (horizons, outcome_label_cost, instrument_filter) so any of those changing invalidates the cache. The load path also rejects caches whose regime_full length doesn't match the freshly-loaded snapshot count, as a defense against re-downloaded quarters whose underlying MBP-10 sidecar was refreshed but whose labels sidecar wasn't. Stacks with SPEED-C (~400x on Welford) and SPEED-A (~4-8x parallel): - First run on a file: pay the existing label-generation cost, write the cache. - Subsequent runs: load arrays directly from bincode sidecar — ~1s/file vs ~3 min/file previously. Inference-only runs skip the cache write to avoid polluting it with all-default vectors that would mis-serve a later non-inference run. Cache key suffix format: 'labels_h_

_

_cost_' to keep filenames portable (no embedded '.' from f32 formatting) and preserve exact-float identity via raw-bit encoding of the cost. --- crates/ml-alpha/src/data/loader.rs | 318 +++++++++++++++++++++++++---- 1 file changed, 280 insertions(+), 38 deletions(-) diff --git a/crates/ml-alpha/src/data/loader.rs b/crates/ml-alpha/src/data/loader.rs index 86992845b..5e2052c2e 100644 --- a/crates/ml-alpha/src/data/loader.rs +++ b/crates/ml-alpha/src/data/loader.rs @@ -28,6 +28,104 @@ use crate::cfc::snap_features::{Mbp10RawInput, ES_TICK_SIZE, REGIME_DIM}; use crate::heads::N_HORIZONS; use crate::multi_horizon_labels::{generate_labels, generate_outcome_labels_ab}; +/// Cached on-disk shape of per-file label + regime arrays. Sidecar lives +/// alongside the predecoded MBP-10 snapshot sidecar in `cfg.predecoded_dir`. +/// +/// Cache key (encoded in the sidecar filename) includes (horizons, +/// outcome_label_cost, instrument_filter) because labels depend on these. +/// The existing predecoded-MBP-10 sidecar's `instrument_filter` part of +/// its filename plus the horizons/cost suffix here give a complete cache +/// key. +/// +/// Refilled on schema-mismatch errors (deserialize failure) — the bincode +/// payload's struct shape is itself the schema check. The load path also +/// checks `regime_full.len() == snapshots.len()` as a defensive length +/// guard against silently-stale caches (the underlying MBP-10 sidecar is +/// invalidated by mtime+size mismatch in `ml_features::predecoded`, but +/// that doesn't propagate to a label-length check here without this +/// guard). +#[derive(serde::Serialize, serde::Deserialize)] +struct CachedLabels { + labels_full: [Vec; N_HORIZONS], + outcome_prof_long_full: [Vec; N_HORIZONS], + outcome_prof_short_full: [Vec; N_HORIZONS], + outcome_size_long_full: [Vec; N_HORIZONS], + outcome_size_short_full: [Vec; N_HORIZONS], + sigma_k_full: [Vec; N_HORIZONS], + pos_fraction: Vec, + regime_full: Vec<[f32; REGIME_DIM]>, +} + +/// Cache-key suffix for the labels sidecar. Encodes (horizons, cost, +/// instrument_filter) so changes to any of these properties invalidate +/// the cache. Format: `labels_h_

_

_cost_`. +/// The cost is encoded as the f32's raw bit pattern in lowercase hex to +/// keep '.' out of the filename (portable across filesystems) and to +/// preserve exact-float identity (0.5 != 0.50000001 → different bits → +/// different cache key). +fn labels_cache_suffix( + horizons: &[usize; N_HORIZONS], + cost_price_units: f32, + filter: InstrumentFilter, +) -> String { + let cost_bits = cost_price_units.to_bits(); + let filter_tag = match filter { + InstrumentFilter::All => String::from("all"), + InstrumentFilter::Id(id) => format!("instr{id}"), + InstrumentFilter::FrontMonth => String::from("front_month"), + }; + format!( + "labels_h{}_{}_{}_cost{:08x}_{}", + horizons[0], horizons[1], horizons[2], cost_bits, filter_tag + ) +} + +/// Path to the labels-cache sidecar for `source` under `predecoded_dir`. +/// Layout: `/..predecoded.bin`. +fn labels_cache_path( + source: &std::path::Path, + predecoded_dir: &std::path::Path, + horizons: &[usize; N_HORIZONS], + cost_price_units: f32, + filter: InstrumentFilter, +) -> PathBuf { + let suffix = labels_cache_suffix(horizons, cost_price_units, filter); + let stem = source + .file_name() + .and_then(|s| s.to_str()) + .unwrap_or("unknown"); + predecoded_dir.join(format!("{stem}.{suffix}.predecoded.bin")) +} + +/// Try to read cached labels. Returns `Ok(Some(...))` on hit, +/// `Ok(None)` on miss (file does not exist), and `Err` only on IO / +/// decode errors. Callers treat `Err` the same as a miss (recompute) +/// but log a warning. +fn try_read_labels_cache(path: &std::path::Path) -> Result> { + use std::fs::File; + use std::io::BufReader; + if !path.exists() { + return Ok(None); + } + let f = File::open(path).with_context(|| format!("open labels cache {}", path.display()))?; + let reader = BufReader::new(f); + let cached: CachedLabels = bincode::deserialize_from(reader) + .with_context(|| format!("decode labels cache {}", path.display()))?; + Ok(Some(cached)) +} + +/// Write labels cache to disk. Best-effort: failures are logged but don't +/// halt the training run. +fn write_labels_cache(path: &std::path::Path, cached: &CachedLabels) -> Result<()> { + use std::fs::File; + use std::io::BufWriter; + let f = File::create(path).with_context(|| format!("create labels cache {}", path.display()))?; + let writer = BufWriter::new(f); + bincode::serialize_into(writer, cached) + .with_context(|| format!("encode labels cache {}", path.display()))?; + Ok(()) +} + /// EMA smoothing factors. Half-life ≈ ln(2)/α snapshots. /// /// Empirical ES MBP-10 median inter-event Δt = 74µs (NOT 250ms as the @@ -316,46 +414,154 @@ impl MultiHorizonLoader { ); return Ok(None); } - let mut labels_full: [Vec; N_HORIZONS] = Default::default(); - let mut outcome_prof_long_full: [Vec; N_HORIZONS] = Default::default(); - let mut outcome_prof_short_full: [Vec; N_HORIZONS] = Default::default(); - let mut outcome_size_long_full: [Vec; N_HORIZONS] = Default::default(); - let mut outcome_size_short_full: [Vec; N_HORIZONS] = Default::default(); - let mut sigma_k_full: [Vec; N_HORIZONS] = Default::default(); - // Default pos_fraction is zero per direction/horizon; populated below in - // training mode. Length is fixed so downstream broadcast is uniform. - let mut pos_fraction: Vec = vec![0.0_f32; 2 * N_HORIZONS]; - if !cfg.inference_only { - let prices: Vec = snapshots.iter().map(mid_price_f32).collect(); - for (h_idx, &h) in cfg.horizons.iter().enumerate() { - let mut full = vec![f32::NAN; snapshots.len()]; - let raw = generate_labels(&prices, h); - for (i, &t) in raw.valid_indices.iter().enumerate() { - full[t] = raw.labels[i]; - } - labels_full[h_idx] = full; + // SPEED-B (2026-05-22): on-disk cache for the per-file label + + // regime arrays so subsequent training runs skip the multi-second + // label-generation pass. Cache key encodes (horizons, cost, + // instrument_filter). Inference-only runs skip the cache WRITE + // (they don't compute labels) but DO honor a hit if present — + // the cached arrays would just go unused in that mode. + let labels_cache = labels_cache_path( + path, + &cfg.predecoded_dir, + &cfg.horizons, + cfg.outcome_label_cost, + cfg.instrument_filter, + ); + let cached = match try_read_labels_cache(&labels_cache) { + Ok(Some(c)) if c.regime_full.len() == snapshots.len() => { + tracing::info!( + path = %labels_cache.display(), + n_snapshots = snapshots.len(), + "labels cache HIT", + ); + Some(c) } - // Candidate-B aux supervision: paired prof-binary + σ-normalized - // size-regression targets per (direction, horizon). Replaces the - // old D-style asymmetric loss-aversion labels. - let outcome = generate_outcome_labels_ab( - &prices, - &cfg.horizons, - cfg.outcome_label_cost, + Ok(Some(c)) => { + // Stale length — the underlying MBP-10 sidecar was + // refreshed (re-downloaded quarter) but the labels cache + // wasn't. Treat as miss, recompute, overwrite. + tracing::warn!( + path = %labels_cache.display(), + cached_n = c.regime_full.len(), + actual_n = snapshots.len(), + "labels cache length mismatch — recomputing", + ); + None + } + Ok(None) => None, + Err(e) => { + tracing::warn!( + path = %labels_cache.display(), + err = %e, + "labels cache read failed — recomputing", + ); + None + } + }; + + let ( + labels_full, + outcome_prof_long_full, + outcome_prof_short_full, + outcome_size_long_full, + outcome_size_short_full, + sigma_k_full, + pos_fraction, + regime_full, + ) = if let Some(c) = cached { + ( + c.labels_full, + c.outcome_prof_long_full, + c.outcome_prof_short_full, + c.outcome_size_long_full, + c.outcome_size_short_full, + c.sigma_k_full, + c.pos_fraction, + c.regime_full, ) - .with_context(|| format!( - "generate_outcome_labels_ab for {} (cost={})", - path.display(), - cfg.outcome_label_cost, - ))?; - outcome_prof_long_full = outcome.y_prof_long; - outcome_prof_short_full = outcome.y_prof_short; - outcome_size_long_full = outcome.y_size_long; - outcome_size_short_full = outcome.y_size_short; - sigma_k_full = outcome.sigma_k; - pos_fraction = outcome.pos_fraction; - } - let regime_full = compute_regime_features(&snapshots); + } else { + let mut labels_full: [Vec; N_HORIZONS] = Default::default(); + let mut outcome_prof_long_full: [Vec; N_HORIZONS] = Default::default(); + let mut outcome_prof_short_full: [Vec; N_HORIZONS] = Default::default(); + let mut outcome_size_long_full: [Vec; N_HORIZONS] = Default::default(); + let mut outcome_size_short_full: [Vec; N_HORIZONS] = Default::default(); + let mut sigma_k_full: [Vec; N_HORIZONS] = Default::default(); + // Default pos_fraction is zero per direction/horizon; populated below + // in training mode. Length is fixed so downstream broadcast is uniform. + let mut pos_fraction: Vec = vec![0.0_f32; 2 * N_HORIZONS]; + if !cfg.inference_only { + let prices: Vec = snapshots.iter().map(mid_price_f32).collect(); + for (h_idx, &h) in cfg.horizons.iter().enumerate() { + let mut full = vec![f32::NAN; snapshots.len()]; + let raw = generate_labels(&prices, h); + for (i, &t) in raw.valid_indices.iter().enumerate() { + full[t] = raw.labels[i]; + } + labels_full[h_idx] = full; + } + // Candidate-B aux supervision: paired prof-binary + σ-normalized + // size-regression targets per (direction, horizon). Replaces the + // old D-style asymmetric loss-aversion labels. + let outcome = generate_outcome_labels_ab( + &prices, + &cfg.horizons, + cfg.outcome_label_cost, + ) + .with_context(|| format!( + "generate_outcome_labels_ab for {} (cost={})", + path.display(), + cfg.outcome_label_cost, + ))?; + outcome_prof_long_full = outcome.y_prof_long; + outcome_prof_short_full = outcome.y_prof_short; + outcome_size_long_full = outcome.y_size_long; + outcome_size_short_full = outcome.y_size_short; + sigma_k_full = outcome.sigma_k; + pos_fraction = outcome.pos_fraction; + } + let regime_full = compute_regime_features(&snapshots); + + // Best-effort write to cache for next run. Skipped when + // inference_only since labels_full/outcome_* are all-default + // and a stored cache there would mis-serve a later + // non-inference run that hit it. + if !cfg.inference_only { + let payload = CachedLabels { + labels_full: labels_full.clone(), + outcome_prof_long_full: outcome_prof_long_full.clone(), + outcome_prof_short_full: outcome_prof_short_full.clone(), + outcome_size_long_full: outcome_size_long_full.clone(), + outcome_size_short_full: outcome_size_short_full.clone(), + sigma_k_full: sigma_k_full.clone(), + pos_fraction: pos_fraction.clone(), + regime_full: regime_full.clone(), + }; + if let Err(e) = write_labels_cache(&labels_cache, &payload) { + tracing::warn!( + path = %labels_cache.display(), + err = %e, + "labels cache write failed (training continues)", + ); + } else { + tracing::info!( + path = %labels_cache.display(), + n_snapshots = snapshots.len(), + "labels cache WRITE", + ); + } + } + + ( + labels_full, + outcome_prof_long_full, + outcome_prof_short_full, + outcome_size_long_full, + outcome_size_short_full, + sigma_k_full, + pos_fraction, + regime_full, + ) + }; Ok(Some(LoadedFile { snapshots, labels_full, @@ -852,3 +1058,39 @@ mod inference_mode_tests { // If new() failed (no file), the assertion is moot — covered elsewhere. } } + +#[cfg(test)] +mod labels_cache_tests { + use super::*; + + #[test] + fn labels_cache_suffix_is_deterministic_and_keys_on_inputs() { + let h1 = [100, 300, 1000]; + let h2 = [100, 300, 1001]; // different last horizon + let c1 = 0.5_f32; + let c2 = 0.25_f32; + let f1 = InstrumentFilter::All; + let f2 = InstrumentFilter::FrontMonth; + + // Same inputs → same suffix + assert_eq!( + labels_cache_suffix(&h1, c1, f1), + labels_cache_suffix(&h1, c1, f1) + ); + // Different horizon → different suffix + assert_ne!( + labels_cache_suffix(&h1, c1, f1), + labels_cache_suffix(&h2, c1, f1) + ); + // Different cost → different suffix + assert_ne!( + labels_cache_suffix(&h1, c1, f1), + labels_cache_suffix(&h1, c2, f1) + ); + // Different filter → different suffix + assert_ne!( + labels_cache_suffix(&h1, c1, f1), + labels_cache_suffix(&h1, c1, f2) + ); + } +}