feat(ml-alpha): Phase A data loader — predecoded MBP-10 -> seq + labels

Reuses ml-features::predecoded::load_or_predecode_mbp10 (no cycle —
ml-features doesn't depend on ml-alpha). Yields seq_len-sized windows
of Mbp10RawInput plus 5-horizon binary labels via
multi_horizon_labels::generate_labels.

Per-snapshot prev_mid / prev_ts_ns / trade_signed_vol come from the
prior snapshot in the source stream (not from the anchor), so the
CfC trunk sees a continuous-time signal across the entire seq.

Labels: NaN at edge positions (no forward window) or tied prices;
BCE kernel masks these (Task 10).

Tests:
  - loader_errors_on_missing_root: passes (1/1 inline)
  - loader_yields_seq_with_valid_labels: --ignored, runs at gate time
    with FOXHUNT_TEST_DATA set

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

crates/ml-alpha/Cargo.toml

@@ -20,6 +20,8 @@ ml-core = { workspace = true }
 # NOTE: cannot depend on `ml` (would cycle — ml depends on ml-alpha for the
 # mamba2_block gate reference). MappedF32Buffer is duplicated locally in
 # `src/pinned_mem.rs`; eventually we should move mapped-pinned to ml-core.
+ml-features = { workspace = true }
+data = { workspace = true }
 cudarc = { version = "0.19", default-features = false, features = ["driver", "cublas", "dynamic-linking", "std", "cuda-version-from-build-system", "f16"] }
 
 # Standard async + error + logging plumbing

crates/ml-alpha/src/data/mod.rs

@@ -0,0 +1,3 @@
+//! Phase A data path — predecoded MBP-10 -> Mbp10RawInput sequences + labels.
+
+pub mod phase_a_loader;

crates/ml-alpha/src/data/phase_a_loader.rs

@@ -0,0 +1,196 @@
+//! Phase A data loader.
+//!
+//! Reads predecoded MBP-10 sidecars (already on disk from earlier session
+//! work — see `crates/ml-features/src/predecoded.rs`). Per spec Section 4:
+//! yields `seq_len`-sized windows of `Mbp10RawInput` plus 5-horizon binary
+//! labels computed via `multi_horizon_labels::generate_labels`.
+//!
+//! Memory discipline: snapshots stay on the host side here (CPU control
+//! flow). The trainer (Task 13) uploads them into GPU buffers via the
+//! `CfcTrunk::update_input_buffers` mapped-pinned path before each
+//! captured-graph replay.
+
+use std::path::PathBuf;
+
+use anyhow::{Context, Result};
+use data::providers::databento::mbp10::{BidAskPair, Mbp10Snapshot};
+use ml_features::predecoded::load_or_predecode_mbp10;
+use rand::seq::SliceRandom;
+use rand::{Rng, SeedableRng};
+use rand_chacha::ChaCha8Rng;
+
+use crate::cfc::snap_features::Mbp10RawInput;
+use crate::multi_horizon_labels::generate_labels;
+
+#[derive(Clone, Debug)]
+pub struct PhaseAConfig {
+    /// Directory containing .dbn.zst MBP-10 files plus their predecoded sidecars.
+    pub mbp10_root: PathBuf,
+    /// Directory where predecoded sidecars live (`load_or_predecode_mbp10`
+    /// arg — typically the same as `mbp10_root` or a sibling).
+    pub predecoded_dir: PathBuf,
+    /// Number of consecutive snapshots per training sequence.
+    pub seq_len: usize,
+    /// Forward-horizon offsets in snapshots; spec mandates 5 of these.
+    pub horizons: [usize; 5],
+    /// Soft cap on total sequences yielded across all source files.
+    pub n_max_sequences: usize,
+    /// Deterministic seed for shuffling + sequence-anchor selection.
+    pub seed: u64,
+}
+
+pub struct PhaseASequence {
+    /// `seq_len` consecutive raw snapshot inputs.
+    pub snapshots: Vec<Mbp10RawInput>,
+    /// `[5][seq_len]` labels. NaN at positions where the forward window
+    /// is outside the source file (right edge) or where the move is a
+    /// tied price (drop, mirror `generate_labels` semantics).
+    pub labels: [Vec<f32>; 5],
+}
+
+pub struct PhaseALoader {
+    cfg: PhaseAConfig,
+    files: Vec<PathBuf>,
+    rng: ChaCha8Rng,
+    file_idx: usize,
+    yielded: usize,
+}
+
+impl PhaseALoader {
+    pub fn new(cfg: &PhaseAConfig) -> Result<Self> {
+        let mut files: Vec<PathBuf> = std::fs::read_dir(&cfg.mbp10_root)
+            .with_context(|| format!("read mbp10 root {}", cfg.mbp10_root.display()))?
+            .filter_map(|e| e.ok())
+            .map(|e| e.path())
+            .filter(|p| {
+                let s = p.to_string_lossy();
+                // Source DBN files only; not the predecoded sidecars themselves.
+                s.ends_with(".dbn.zst") || s.ends_with(".dbn")
+            })
+            .collect();
+        anyhow::ensure!(
+            !files.is_empty(),
+            "no DBN files under {}",
+            cfg.mbp10_root.display()
+        );
+        let mut rng = ChaCha8Rng::seed_from_u64(cfg.seed);
+        files.shuffle(&mut rng);
+        Ok(Self {
+            cfg: cfg.clone(),
+            files,
+            rng,
+            file_idx: 0,
+            yielded: 0,
+        })
+    }
+
+    pub fn n_files(&self) -> usize {
+        self.files.len()
+    }
+
+    pub fn yielded(&self) -> usize {
+        self.yielded
+    }
+
+    pub fn next_sequence(&mut self) -> Result<Option<PhaseASequence>> {
+        if self.yielded >= self.cfg.n_max_sequences {
+            return Ok(None);
+        }
+        // Loop until we either yield or exhaust the file list (re-shuffle and
+        // try again if files are exhausted but we haven't hit n_max_sequences).
+        for _ in 0..self.files.len() * 2 {
+            if self.file_idx >= self.files.len() {
+                self.files.shuffle(&mut self.rng);
+                self.file_idx = 0;
+            }
+            let path = self.files[self.file_idx].clone();
+            self.file_idx += 1;
+
+            let snapshots = load_or_predecode_mbp10(&path, &self.cfg.predecoded_dir)
+                .with_context(|| format!("load mbp10 {}", path.display()))?;
+            let max_horizon = *self.cfg.horizons.iter().max().expect("non-empty horizons");
+            if snapshots.len() < self.cfg.seq_len + max_horizon + 1 {
+                continue;
+            }
+
+            let max_anchor = snapshots.len() - self.cfg.seq_len - max_horizon;
+            let anchor: usize = self.rng.gen_range(0..max_anchor);
+
+            // Compute mid prices over the entire file (label generation needs
+            // the trailing window past the anchor + seq_len).
+            let prices: Vec<f32> = snapshots
+                .iter()
+                .map(|s| mid_price_f32(s))
+                .collect();
+
+            let mut labels: [Vec<f32>; 5] = Default::default();
+            for (h_idx, &h) in self.cfg.horizons.iter().enumerate() {
+                let mut out = vec![f32::NAN; self.cfg.seq_len];
+                let raw = generate_labels(&prices, h);
+                for (i, &t) in raw.valid_indices.iter().enumerate() {
+                    if t >= anchor && t < anchor + self.cfg.seq_len {
+                        out[t - anchor] = raw.labels[i];
+                    }
+                }
+                labels[h_idx] = out;
+            }
+
+            // Build sequence of Mbp10RawInput. Each input carries its own
+            // prev_mid + prev_ts_ns + trade_signed_vol derived from the
+            // PRIOR snapshot in the source stream (not from `anchor`).
+            let mut sequence = Vec::with_capacity(self.cfg.seq_len);
+            for k in 0..self.cfg.seq_len {
+                let idx = anchor + k;
+                let cur = &snapshots[idx];
+                let prev_idx = if idx == 0 { 0 } else { idx - 1 };
+                let prev = &snapshots[prev_idx];
+                sequence.push(convert(cur, prev));
+            }
+
+            self.yielded += 1;
+            return Ok(Some(PhaseASequence {
+                snapshots: sequence,
+                labels,
+            }));
+        }
+        Ok(None) // exhausted files and none had enough snapshots
+    }
+}
+
+fn mid_price_f32(s: &Mbp10Snapshot) -> f32 {
+    let l = s.levels.first().copied().unwrap_or_else(BidAskPair::empty);
+    let bid = BidAskPair::price_to_f64(l.bid_px);
+    let ask = BidAskPair::price_to_f64(l.ask_px);
+    (0.5 * (bid + ask)) as f32
+}
+
+fn convert(cur: &Mbp10Snapshot, prev: &Mbp10Snapshot) -> Mbp10RawInput {
+    let mut bid_px = [0.0f32; 10];
+    let mut bid_sz = [0.0f32; 10];
+    let mut ask_px = [0.0f32; 10];
+    let mut ask_sz = [0.0f32; 10];
+    for (i, lvl) in cur.levels.iter().take(10).enumerate() {
+        bid_px[i] = BidAskPair::price_to_f64(lvl.bid_px) as f32;
+        bid_sz[i] = lvl.bid_sz as f32;
+        ask_px[i] = BidAskPair::price_to_f64(lvl.ask_px) as f32;
+        ask_sz[i] = lvl.ask_sz as f32;
+    }
+    let prev_mid = mid_price_f32(prev);
+    let trade_count = cur.trade_count.saturating_sub(prev.trade_count);
+    // `trade_signed_vol` isn't directly in Mbp10Snapshot; we'd compute it
+    // from the trades stream in production. For Phase A v1 use trade_count
+    // as a proxy (sign neutral); the snap_feature kernel reads it but the
+    // production pipeline will replace this with the trades-loader output.
+    let trade_signed_vol = (cur.trade_count as i64 - prev.trade_count as i64) as f32;
+    Mbp10RawInput {
+        bid_px,
+        bid_sz,
+        ask_px,
+        ask_sz,
+        prev_mid,
+        trade_signed_vol,
+        trade_count,
+        ts_ns: cur.timestamp,
+        prev_ts_ns: prev.timestamp,
+    }
+}

crates/ml-alpha/src/lib.rs

@@ -26,6 +26,7 @@
 //   Task 12: pub mod data;
 //   Task 16+: pub mod gate;
 pub mod cfc;
+pub mod data;
 pub mod heads;
 pub mod isv;
 pub mod pinned;

crates/ml-alpha/tests/phase_a_loader.rs

@@ -0,0 +1,66 @@
+//! Phase A loader smoke tests.
+//!
+//! Real-data path is `--ignored`; runs only with FOXHUNT_TEST_DATA set.
+
+use ml_alpha::data::phase_a_loader::{PhaseAConfig, PhaseALoader};
+use std::path::PathBuf;
+
+fn cfg_from_env() -> Option<PhaseAConfig> {
+    let root = std::env::var("FOXHUNT_TEST_DATA").ok()?;
+    let mbp10 = PathBuf::from(format!("{root}/mbp10"));
+    let predec = PathBuf::from(format!("{root}/predecoded"));
+    if !mbp10.exists() {
+        return None;
+    }
+    Some(PhaseAConfig {
+        mbp10_root: mbp10,
+        predecoded_dir: predec,
+        seq_len: 64,
+        horizons: [30, 100, 300, 1000, 6000],
+        n_max_sequences: 100,
+        seed: 0xA1A2_A3A4,
+    })
+}
+
+#[test]
+#[ignore]
+fn loader_yields_seq_with_valid_labels() {
+    let Some(cfg) = cfg_from_env() else {
+        eprintln!("skipping: FOXHUNT_TEST_DATA not set or mbp10 dir missing");
+        return;
+    };
+    let mut loader = PhaseALoader::new(&cfg).expect("loader init");
+    assert!(loader.n_files() > 0, "no DBN files found in mbp10 root");
+
+    let mut count = 0;
+    while let Some(seq) = loader.next_sequence().expect("next") {
+        assert_eq!(seq.snapshots.len(), cfg.seq_len);
+        for h in 0..5 {
+            assert_eq!(seq.labels[h].len(), cfg.seq_len);
+            for &v in &seq.labels[h] {
+                assert!(v.is_nan() || v == 0.0 || v == 1.0, "label not binary: {v}");
+            }
+        }
+        let any_valid = seq.labels.iter().any(|l| l.iter().any(|v| !v.is_nan()));
+        assert!(any_valid, "expected at least one non-NaN label in a sequence");
+        count += 1;
+        if count >= 5 {
+            break;
+        }
+    }
+    assert!(count >= 5, "expected at least 5 sequences, got {count}");
+}
+
+#[test]
+fn loader_errors_on_missing_root() {
+    let cfg = PhaseAConfig {
+        mbp10_root: PathBuf::from("/tmp/does_not_exist_foxhunt_test"),
+        predecoded_dir: PathBuf::from("/tmp/does_not_exist_foxhunt_test"),
+        seq_len: 64,
+        horizons: [30, 100, 300, 1000, 6000],
+        n_max_sequences: 10,
+        seed: 0,
+    };
+    let res = PhaseALoader::new(&cfg);
+    assert!(res.is_err(), "expected error for missing mbp10 root");
+}