diff --git a/crates/ml-alpha/src/data/loader.rs b/crates/ml-alpha/src/data/loader.rs index 80c6de515..972d0f628 100644 --- a/crates/ml-alpha/src/data/loader.rs +++ b/crates/ml-alpha/src/data/loader.rs @@ -2,8 +2,9 @@ //! //! 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 N_HORIZONS binary -//! labels computed via `multi_horizon_labels::generate_labels`. +//! yields multi-resolution windows of `Mbp10RawInput` (length = +//! `multi_resolution.total_positions()`) plus N_HORIZONS 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 @@ -117,8 +118,11 @@ pub struct MultiHorizonLoaderConfig { /// Directory where predecoded sidecars live (`load_or_predecode_mbp10` /// arg — typically the same dir as `files` or a sibling). pub predecoded_dir: PathBuf, - /// Number of consecutive snapshots per training sequence. - pub seq_len: usize, + /// Multi-resolution input window. `total_positions()` defines the + /// sequence length the trainer sees; `required_lookback_ticks()` is + /// the per-anchor pre-edge snapshot count the loader needs from each + /// source file. + pub multi_resolution: crate::data::aggregation::MultiResolutionConfig, /// Forward-horizon offsets in snapshots. Length tracks /// `ml_alpha::heads::N_HORIZONS` so the loader's label layout matches /// the per-horizon heads. @@ -181,26 +185,27 @@ pub fn discover_mbp10_files_sorted(root: &std::path::Path) -> Result, - /// `[N_HORIZONS][seq_len]` BCE labels. NaN at positions where the + /// `[N_HORIZONS][total_positions]` BCE 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; N_HORIZONS], - /// `[N_HORIZONS][seq_len]` Candidate-B prof-binary labels for the long - /// direction (1.0 if `ΔP_K - 2×cost > 0`, 0.0 otherwise, NaN at right + /// `[N_HORIZONS][total_positions]` Candidate-B prof-binary labels for the + /// long direction (1.0 if `ΔP_K - 2×cost > 0`, 0.0 otherwise, NaN at right /// edge or non-finite price). See [`crate::multi_horizon_labels::generate_outcome_labels_ab`]. pub outcome_prof_long: [Vec; N_HORIZONS], - /// `[N_HORIZONS][seq_len]` Candidate-B prof-binary labels for the short - /// direction. + /// `[N_HORIZONS][total_positions]` Candidate-B prof-binary labels for the + /// short direction. pub outcome_prof_short: [Vec; N_HORIZONS], - /// `[N_HORIZONS][seq_len]` Candidate-B σ-normalized signed-return targets - /// for the long direction. NaN where prof_long==0 or σ_K not yet warmed. + /// `[N_HORIZONS][total_positions]` Candidate-B σ-normalized signed-return + /// targets for the long direction. NaN where prof_long==0 or σ_K not yet warmed. pub outcome_size_long: [Vec; N_HORIZONS], - /// `[N_HORIZONS][seq_len]` Candidate-B σ-normalized signed-return targets - /// for the short direction. Same NaN semantics as `outcome_size_long`. + /// `[N_HORIZONS][total_positions]` Candidate-B σ-normalized signed-return + /// targets for the short direction. Same NaN semantics as `outcome_size_long`. pub outcome_size_short: [Vec; N_HORIZONS], - /// `[N_HORIZONS][seq_len]` rolling 1000-step Welford std of K-step + /// `[N_HORIZONS][total_positions]` rolling 1000-step Welford std of K-step /// log-returns at each snapshot. NaN during the warmup window. Used /// downstream for σ-normalized size targets and any in-trainer scaling. pub sigma_k: [Vec; N_HORIZONS], @@ -282,12 +287,13 @@ impl MultiHorizonLoader { let rng = ChaCha8Rng::seed_from_u64(cfg.seed); let max_horizon = *cfg.horizons.iter().max().expect("non-empty horizons"); - // Training requires `seq_len + max_horizon + 1` snapshots/file to compute - // forward labels; inference only needs `seq_len` (no forward window). + // Multi-resolution lookback: each anchor needs `required_lookback_ticks` + // raw snapshots BEFORE the forward edge to fill all aggregation scales. + let lookback = cfg.multi_resolution.required_lookback_ticks(); let min_size = if cfg.inference_only { - cfg.seq_len.max(1) + lookback.max(1) } else { - cfg.seq_len + max_horizon + 1 + lookback + max_horizon + 1 }; let mut files_loaded: Vec = Vec::with_capacity(files.len()); for path in &files { @@ -300,7 +306,7 @@ impl MultiHorizonLoader { path = %path.display(), snapshots = snapshots.len(), min_size, - "skipping file (too few snapshots for seq_len + max_horizon)", + "skipping file (too few snapshots for required_lookback_ticks + max_horizon)", ); continue; } @@ -465,11 +471,11 @@ impl MultiHorizonLoader { let n_files = self.files_loaded.len(); let file_idx = self.rng.gen_range(0..n_files); let lf = &self.files_loaded[file_idx]; - let needed = self.cfg.seq_len + max_horizon; + let needed = self.cfg.multi_resolution.total_positions() + max_horizon; anyhow::ensure!( lf.snapshots.len() > needed, - "file too short for seq_len={} max_horizon={}: {} <= {}", - self.cfg.seq_len, max_horizon, lf.snapshots.len(), needed + "file too short for total_positions={} max_horizon={}: {} <= {}", + self.cfg.multi_resolution.total_positions(), max_horizon, lf.snapshots.len(), needed ); let max_anchor = lf.snapshots.len() - needed; let anchor: usize = self.rng.gen_range(0..max_anchor); @@ -481,13 +487,13 @@ impl MultiHorizonLoader { let mut outcome_size_short: [Vec; N_HORIZONS] = Default::default(); let mut sigma_k: [Vec; N_HORIZONS] = Default::default(); for h in 0..N_HORIZONS { - let mut row = Vec::with_capacity(self.cfg.seq_len); - let mut row_prof_long = Vec::with_capacity(self.cfg.seq_len); - let mut row_prof_short = Vec::with_capacity(self.cfg.seq_len); - let mut row_size_long = Vec::with_capacity(self.cfg.seq_len); - let mut row_size_short = Vec::with_capacity(self.cfg.seq_len); - let mut row_sigma = Vec::with_capacity(self.cfg.seq_len); - for k in 0..self.cfg.seq_len { + let mut row = Vec::with_capacity(self.cfg.multi_resolution.total_positions()); + let mut row_prof_long = Vec::with_capacity(self.cfg.multi_resolution.total_positions()); + let mut row_prof_short = Vec::with_capacity(self.cfg.multi_resolution.total_positions()); + let mut row_size_long = Vec::with_capacity(self.cfg.multi_resolution.total_positions()); + let mut row_size_short = Vec::with_capacity(self.cfg.multi_resolution.total_positions()); + let mut row_sigma = Vec::with_capacity(self.cfg.multi_resolution.total_positions()); + for k in 0..self.cfg.multi_resolution.total_positions() { row.push(lf.labels_full[h][anchor + k]); row_prof_long.push(lf.outcome_prof_long_full[h][anchor + k]); row_prof_short.push(lf.outcome_prof_short_full[h][anchor + k]); @@ -502,14 +508,42 @@ impl MultiHorizonLoader { outcome_size_short[h] = row_size_short; sigma_k[h] = row_sigma; } - let mut sequence = Vec::with_capacity(self.cfg.seq_len); - for k in 0..self.cfg.seq_len { - let idx = anchor + k; - let cur = &lf.snapshots[idx]; - let prev_idx = if idx == 0 { 0 } else { idx - 1 }; - let prev = &lf.snapshots[prev_idx]; - sequence.push(convert(cur, prev, lf.regime_full[idx])); + // Multi-resolution window construction. The forward edge is + // `anchor + total_positions` exclusive (the snapshot AT + // `anchor + total_positions - 1` is the newest position in the + // emitted sequence). We walk fine→coarse from the edge backward + // so each scale's positions cover earlier-in-time windows than + // the previous scale, then reverse so the emitted sequence is + // chronological (oldest-first). + let scales = self.cfg.multi_resolution.scales(); + let total = self.cfg.multi_resolution.total_positions(); + let edge = anchor + total; // exclusive upper bound + + let mut sequence_rev: Vec = Vec::with_capacity(total); + let mut cursor = edge; + for &(scale, count) in scales { + for _ in 0..count { + let window_end = cursor; // exclusive + let window_start = window_end - scale; // inclusive + let mut window_raws: Vec = Vec::with_capacity(scale); + for idx in window_start..window_end { + let cur = &lf.snapshots[idx]; + let prev_idx = if idx == 0 { 0 } else { idx - 1 }; + let prev = &lf.snapshots[prev_idx]; + window_raws.push(convert(cur, prev, lf.regime_full[idx])); + } + if window_raws.len() == 1 { + sequence_rev.push(window_raws.into_iter().next().unwrap()); + } else { + sequence_rev.push( + crate::data::aggregation::aggregate_window(window_raws.as_slice()), + ); + } + cursor = window_start; + } } + let sequence: Vec = sequence_rev.into_iter().rev().collect(); + debug_assert_eq!(sequence.len(), total); self.yielded += 1; Ok(Some(LabeledSequence { @@ -695,7 +729,7 @@ mod inference_mode_tests { Some(MultiHorizonLoaderConfig { files, predecoded_dir: mbp10.clone(), - seq_len: 1, + multi_resolution: crate::data::aggregation::MultiResolutionConfig::from_scales(vec![(1, 1)]).unwrap(), horizons: crate::heads::HORIZONS, n_max_sequences: 1, seed: 0xCAFEF00D, @@ -777,7 +811,7 @@ mod inference_mode_tests { let cfg = MultiHorizonLoaderConfig { files: vec![std::path::PathBuf::from("/tmp/__nonexistent_foxhunt.dbn.zst")], predecoded_dir: std::path::PathBuf::from("/tmp"), - seq_len: 1, + multi_resolution: crate::data::aggregation::MultiResolutionConfig::from_scales(vec![(1, 1)]).unwrap(), horizons: crate::heads::HORIZONS, n_max_sequences: 1, seed: 0,