perf(loader): cache computed labels arrays to disk (~50x first-run / ~∞ subsequent)

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<H0>_<H1>_<H2>_cost<HEX>_<FILTER>'
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

@@ -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<f32>; N_HORIZONS],
+    outcome_prof_long_full: [Vec<f32>; N_HORIZONS],
+    outcome_prof_short_full: [Vec<f32>; N_HORIZONS],
+    outcome_size_long_full: [Vec<f32>; N_HORIZONS],
+    outcome_size_short_full: [Vec<f32>; N_HORIZONS],
+    sigma_k_full: [Vec<f32>; N_HORIZONS],
+    pos_fraction: Vec<f32>,
+    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<H0>_<H1>_<H2>_cost<HEX>_<FILTER>`.
+/// 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_dir>/<source_stem>.<suffix>.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<Option<CachedLabels>> {
+    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<f32>; N_HORIZONS] = Default::default();
-            let mut outcome_prof_long_full: [Vec<f32>; N_HORIZONS] = Default::default();
-            let mut outcome_prof_short_full: [Vec<f32>; N_HORIZONS] = Default::default();
-            let mut outcome_size_long_full: [Vec<f32>; N_HORIZONS] = Default::default();
-            let mut outcome_size_short_full: [Vec<f32>; N_HORIZONS] = Default::default();
-            let mut sigma_k_full: [Vec<f32>; 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<f32> = vec![0.0_f32; 2 * N_HORIZONS];
-            if !cfg.inference_only {
-                let prices: Vec<f32> = 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<f32>; N_HORIZONS] = Default::default();
+                let mut outcome_prof_long_full: [Vec<f32>; N_HORIZONS] = Default::default();
+                let mut outcome_prof_short_full: [Vec<f32>; N_HORIZONS] = Default::default();
+                let mut outcome_size_long_full: [Vec<f32>; N_HORIZONS] = Default::default();
+                let mut outcome_size_short_full: [Vec<f32>; N_HORIZONS] = Default::default();
+                let mut sigma_k_full: [Vec<f32>; 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<f32> = vec![0.0_f32; 2 * N_HORIZONS];
+                if !cfg.inference_only {
+                    let prices: Vec<f32> = 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)
+        );
+    }
+}