diff --git a/crates/ml-alpha/Cargo.toml b/crates/ml-alpha/Cargo.toml index ae65cef2c..b2289f7c9 100644 --- a/crates/ml-alpha/Cargo.toml +++ b/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 diff --git a/crates/ml-alpha/src/data/mod.rs b/crates/ml-alpha/src/data/mod.rs new file mode 100644 index 000000000..965768f19 --- /dev/null +++ b/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; diff --git a/crates/ml-alpha/src/data/phase_a_loader.rs b/crates/ml-alpha/src/data/phase_a_loader.rs new file mode 100644 index 000000000..c5d3d9135 --- /dev/null +++ b/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, + /// `[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; 5], +} + +pub struct PhaseALoader { + cfg: PhaseAConfig, + files: Vec, + rng: ChaCha8Rng, + file_idx: usize, + yielded: usize, +} + +impl PhaseALoader { + pub fn new(cfg: &PhaseAConfig) -> Result { + let mut files: Vec = 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> { + 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 = snapshots + .iter() + .map(|s| mid_price_f32(s)) + .collect(); + + let mut labels: [Vec; 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, + } +} diff --git a/crates/ml-alpha/src/lib.rs b/crates/ml-alpha/src/lib.rs index e671a2aeb..dd2c27012 100644 --- a/crates/ml-alpha/src/lib.rs +++ b/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; diff --git a/crates/ml-alpha/tests/phase_a_loader.rs b/crates/ml-alpha/tests/phase_a_loader.rs new file mode 100644 index 000000000..d9086838f --- /dev/null +++ b/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 { + 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"); +}