From 7dd953aa2b47f0460c4847a561b038291583bc6f Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Thu, 21 May 2026 11:30:50 +0200 Subject: [PATCH] feat(intervention-b): add LoaderMode::Sequential + attn_pool gate Loader can now serve sequences in temporal order (Sequential mode); when active, the trainer skips attn_pool reset at non-file-boundary sequence boundaries (next commit will use this for stateful CfC + MTER). Random mode preserved as default; ZERO behavioral change for existing runs. Phased commit 1/8 of intervention B per docs/superpowers/specs/2026-05-21-crt-train-intervention-b-multi-timescale-readout.md --- crates/ml-alpha/examples/alpha_train.rs | 14 ++- crates/ml-alpha/src/data/loader.rs | 112 ++++++++++++++++++ crates/ml-alpha/src/trainer/perception.rs | 33 +++++- crates/ml-alpha/tests/multi_horizon_loader.rs | 4 +- crates/ml-alpha/tests/perception_overfit.rs | 9 ++ crates/ml-backtesting/src/harness.rs | 3 +- crates/ml-backtesting/tests/ring3_replay.rs | 3 +- crates/ml-backtesting/tests/trainer_parity.rs | 3 +- infra/k8s/argo/alpha-perception-template.yaml | 3 + 9 files changed, 178 insertions(+), 6 deletions(-) diff --git a/crates/ml-alpha/examples/alpha_train.rs b/crates/ml-alpha/examples/alpha_train.rs index 231f696a3..fdac88f3e 100644 --- a/crates/ml-alpha/examples/alpha_train.rs +++ b/crates/ml-alpha/examples/alpha_train.rs @@ -24,7 +24,7 @@ use anyhow::{Context, Result}; use clap::Parser; use ml_alpha::cfc::snap_features::Mbp10RawInput; -use ml_alpha::data::loader::{MultiHorizonLoader, MultiHorizonLoaderConfig}; +use ml_alpha::data::loader::{LoaderMode, MultiHorizonLoader, MultiHorizonLoaderConfig}; use ml_alpha::eval::auc::{compute_auc, AucInput}; use ml_alpha::heads::N_HORIZONS; use ml_alpha::trainer::perception::{auto_horizon_weights, PerceptionTrainer, PerceptionTrainerConfig}; @@ -153,6 +153,11 @@ struct Cli { #[arg(long, default_value_t = 0)] cv_train_window: usize, + /// Sequence-sampling mode. `random` = legacy; `sequential` = required for MTER (CRT.train intervention B). + /// See docs/superpowers/specs/2026-05-21-crt-train-intervention-b-multi-timescale-readout.md. + #[arg(long, default_value = "random")] + loader_mode: String, + /// CRT.train intervention A — output-smoothness regularizer. /// /// λ[h] = base × (HORIZONS[h] / HORIZONS[0]). Default 0.0 disables @@ -277,6 +282,8 @@ fn main() -> Result<()> { ); anyhow::ensure!(cli.batch_size >= 1, "batch_size must be >= 1"); + let loader_mode: LoaderMode = cli.loader_mode.parse() + .map_err(|e: String| anyhow::anyhow!("invalid --loader-mode={}: {e}", cli.loader_mode))?; let trainer_cfg = PerceptionTrainerConfig { seq_len: cli.seq_len, mamba2_state_dim: cli.mamba2_state_dim, @@ -287,6 +294,7 @@ fn main() -> Result<()> { n_batch: cli.batch_size, smoothness_base_lambda: cli.smoothness_base_lambda, kernel_step_trace_path: cli.kernel_step_trace.clone(), + loader_mode, }; let mut trainer = PerceptionTrainer::new(&dev, &trainer_cfg).context("trainer init")?; @@ -417,6 +425,7 @@ fn main() -> Result<()> { n_max_sequences: cli.n_train_seqs, seed: cli.seed, inference_only: false, + mode: loader_mode, }) .context("train loader")?; let mut val_loader = MultiHorizonLoader::new(&MultiHorizonLoaderConfig { @@ -427,6 +436,7 @@ fn main() -> Result<()> { n_max_sequences: cli.n_val_seqs, seed: cli.seed.wrapping_add(0xC0FFEE), inference_only: false, + mode: loader_mode, }) .context("val loader")?; tracing::info!( @@ -470,6 +480,8 @@ fn main() -> Result<()> { let snap_refs: Vec<&[Mbp10RawInput]> = snap_batch.iter().map(|v| v.as_slice()).collect(); let label_refs: Vec<&[[f32; N_HORIZONS]]> = label_batch.iter().map(|v| v.as_slice()).collect(); + let is_boundary = train_loader.is_file_boundary(); + trainer.notify_file_boundary(is_boundary); let loss = trainer.step_batched(&snap_refs, &label_refs).context("train step_batched")?; epoch_train_loss += loss; epoch_train_steps += 1; diff --git a/crates/ml-alpha/src/data/loader.rs b/crates/ml-alpha/src/data/loader.rs index e058947a5..3f9604361 100644 --- a/crates/ml-alpha/src/data/loader.rs +++ b/crates/ml-alpha/src/data/loader.rs @@ -21,6 +21,35 @@ use rand_chacha::ChaCha8Rng; use crate::cfc::snap_features::{Mbp10RawInput, ES_TICK_SIZE, REGIME_DIM}; use crate::multi_horizon_labels::generate_labels; +/// Sequence-sampling mode for the loader. +/// +/// - `Random`: original behaviour. `next_sequence()` picks a random file and random anchor. +/// - `Sequential`: each file is consumed as consecutive K-position chunks in temporal order. +/// Required for MTER (CRT.train intervention B) so training-time h_view distribution +/// matches inference-time geometry. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum LoaderMode { + Random, + Sequential, +} + +impl Default for LoaderMode { + fn default() -> Self { + Self::Random + } +} + +impl std::str::FromStr for LoaderMode { + type Err = String; + fn from_str(s: &str) -> Result { + match s { + "random" => Ok(Self::Random), + "sequential" => Ok(Self::Sequential), + other => Err(format!("unknown loader mode: {other} (expected 'random' or 'sequential')")), + } + } +} + /// EMA smoothing factors. Half-life ≈ ln(2)/α snapshots. const ALPHA_MED: f32 = 0.02; // half-life ≈ 35 snapshots ≈ 9s @ 250ms const ALPHA_SLOW: f32 = 0.0005; // half-life ≈ 1400 snapshots ≈ 6 min @@ -113,6 +142,8 @@ pub struct MultiHorizonLoaderConfig { /// LOB backtest harness; trainer always passes `false`. See /// `docs/superpowers/specs/2026-05-18-real-lob-integration-design.md` §1. pub inference_only: bool, + /// Sequence-sampling mode. Default `Random` for backward compat. MTER training requires `Sequential`. + pub mode: LoaderMode, } /// Discover MBP-10 files under `root` and return them sorted by filename. @@ -175,6 +206,13 @@ pub struct MultiHorizonLoader { /// when `cfg.inference_only == true`. inference_file_idx: usize, inference_snap_idx: usize, + /// Sequential-mode cursor: current file index in `files_loaded`. + sequential_file_idx: usize, + /// Sequential-mode cursor: current anchor position WITHIN the current file. + sequential_anchor_idx: usize, + /// `true` if the most recent `next_sequence()` call crossed a file boundary. + /// Cleared on each `next_sequence()` call; consulted by trainer via `is_file_boundary()`. + last_call_was_file_boundary: bool, } impl MultiHorizonLoader { @@ -244,6 +282,9 @@ impl MultiHorizonLoader { files_loaded, inference_file_idx: 0, inference_snap_idx: 0, + sequential_file_idx: 0, + sequential_anchor_idx: 0, + last_call_was_file_boundary: false, }) } @@ -319,6 +360,20 @@ impl MultiHorizonLoader { } pub fn next_sequence(&mut self) -> Result> { + match self.cfg.mode { + LoaderMode::Random => self.next_sequence_random(), + LoaderMode::Sequential => self.next_sequence_sequential(), + } + } + + /// `true` if the most recent `next_sequence()` call crossed a file boundary in sequential mode. + /// Trainer consults this to decide whether to re-bootstrap stateful buffers. + /// Always returns `false` in `Random` mode. + pub fn is_file_boundary(&self) -> bool { + self.last_call_was_file_boundary + } + + fn next_sequence_random(&mut self) -> Result> { if self.yielded >= self.cfg.n_max_sequences { return Ok(None); } @@ -361,6 +416,61 @@ impl MultiHorizonLoader { self.yielded += 1; Ok(Some(LabeledSequence { snapshots: sequence, labels })) } + + fn next_sequence_sequential(&mut self) -> Result> { + if self.yielded >= self.cfg.n_max_sequences { + return Ok(None); + } + let max_horizon = *self.cfg.horizons.iter().max().expect("non-empty horizons"); + let needed = self.cfg.seq_len + max_horizon; + + // Advance cursor: find the next file with enough remaining snapshots from current anchor. + let mut file_boundary = false; + loop { + if self.sequential_file_idx >= self.files_loaded.len() { + // Wrap around to the start of the file list. Treat the wrap as a file boundary + // so the trainer re-bootstraps state. + self.sequential_file_idx = 0; + self.sequential_anchor_idx = 0; + file_boundary = true; + } + let lf = &self.files_loaded[self.sequential_file_idx]; + if self.sequential_anchor_idx + needed <= lf.snapshots.len() { + break; + } + // This file has no more room; advance. + self.sequential_file_idx += 1; + self.sequential_anchor_idx = 0; + file_boundary = true; + } + self.last_call_was_file_boundary = file_boundary; + + let lf = &self.files_loaded[self.sequential_file_idx]; + let anchor = self.sequential_anchor_idx; + + // Build the labelled sequence (mirror logic from `next_sequence_random`). + let mut labels: [Vec; 5] = Default::default(); + for h in 0..5 { + let mut row = Vec::with_capacity(self.cfg.seq_len); + for k in 0..self.cfg.seq_len { + row.push(lf.labels_full[h][anchor + k]); + } + labels[h] = row; + } + 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])); + } + + // Advance cursor by seq_len (no overlap between sequences). + self.sequential_anchor_idx += self.cfg.seq_len; + self.yielded += 1; + Ok(Some(LabeledSequence { snapshots: sequence, labels })) + } } fn mid_price_f32(s: &Mbp10Snapshot) -> f32 { @@ -538,6 +648,7 @@ mod inference_mode_tests { n_max_sequences: 1, seed: 0xCAFEF00D, inference_only, + mode: LoaderMode::default(), }) } @@ -618,6 +729,7 @@ mod inference_mode_tests { n_max_sequences: 1, seed: 0, inference_only: false, + mode: LoaderMode::default(), }; if let Ok(mut loader) = MultiHorizonLoader::new(&cfg) { let err = loader.next_inference_input(); diff --git a/crates/ml-alpha/src/trainer/perception.rs b/crates/ml-alpha/src/trainer/perception.rs index a43440f45..57b35c2c1 100644 --- a/crates/ml-alpha/src/trainer/perception.rs +++ b/crates/ml-alpha/src/trainer/perception.rs @@ -46,6 +46,7 @@ use rand::{Rng, SeedableRng}; use rand_chacha::ChaCha8Rng; use crate::cfc::snap_features::{Mbp10RawInput, ES_TICK_SIZE, FEATURE_DIM, REGIME_DIM}; +use crate::data::loader::LoaderMode; use crate::heads::{HEAD_MID_DIM, HIDDEN_DIM, N_HORIZONS}; use crate::mamba2_block::{ Mamba2AdamW, Mamba2AdamWConfig, Mamba2BackwardGradsBuffers, Mamba2BackwardScratch, @@ -101,6 +102,10 @@ pub struct PerceptionTrainerConfig { /// Per `feedback_no_feature_flags`: gated by the compile-time /// `kernel-step-trace` feature; specific name justifies the gate. pub kernel_step_trace_path: Option, + + /// Sequence-sampling mode. Affects whether attn_pool resets at every sequence + /// (Random) or only at file boundaries (Sequential). + pub loader_mode: LoaderMode, } impl Default for PerceptionTrainerConfig { @@ -115,6 +120,7 @@ impl Default for PerceptionTrainerConfig { n_batch: 1, smoothness_base_lambda: 0.0, kernel_step_trace_path: None, + loader_mode: LoaderMode::Random, } } } @@ -146,6 +152,10 @@ pub fn auto_horizon_weights(_seq_len: usize, _horizons: &[usize; N_HORIZONS]) -> pub struct PerceptionTrainer { cfg: PerceptionTrainerConfig, stream: Arc, + /// Set by `notify_file_boundary()` BEFORE each `step_batched` call in sequential mode. + /// Trainer uses this to gate attn_pool reset (and later, stateful CfC + MTER bootstrap). + /// Always treated as `true` in Random mode (every sequence is a boundary). + last_seen_file_boundary: bool, // Modules + cached function handles _snap_module: Arc, @@ -1203,6 +1213,7 @@ impl PerceptionTrainer { let k = cfg.seq_len; Ok(Self { cfg: cfg.clone(), + last_seen_file_boundary: true, h_new_per_k_d: stream.alloc_zeros::(k * cfg.n_batch * n_hid)?, probs_per_k_d: stream.alloc_zeros::(k * cfg.n_batch * N_HORIZONS)?, labels_per_k_d: stream.alloc_zeros::(k * cfg.n_batch * N_HORIZONS)?, @@ -1893,7 +1904,20 @@ impl PerceptionTrainer { // zero-initialised h_old at k=0 with this context vector. // Shared mem: K floats (scores) + BLOCK floats (reduce) + // HIDDEN_DIM floats (context) = (K + 128 + 128) * 4 bytes. - { + // + // MTER (CRT.train intervention B) commit 1: gate the bootstrap + // launch on `last_seen_file_boundary`. In Random mode the host + // bool is unconditionally true so the launch is always recorded + // into the captured graph — bit-identical to pre-gate behavior. + // In Sequential mode mid-file, the gate skips re-bootstrap so + // the previous step's pooled context carries forward. (Future + // commits in this intervention introduce stateful CfC + MTER + // bootstrap that consume the same boundary signal.) + let need_attn_pool_bootstrap = match self.cfg.loader_mode { + LoaderMode::Random => true, + LoaderMode::Sequential => self.last_seen_file_boundary, + }; + if need_attn_pool_bootstrap { let k_i32 = k_seq as i32; let n_batch_attn = b_sz as i32; let shared = (k_seq + 128 + HIDDEN_DIM) * std::mem::size_of::(); @@ -2832,6 +2856,13 @@ impl PerceptionTrainer { &self.cfg } + /// Trainer-orchestrator API. Call this BEFORE `step()`/`step_batched()` to signal whether + /// the sequence we're about to consume crosses a file boundary in sequential mode. + /// Loaders provide this via `MultiHorizonLoader::is_file_boundary()`. + pub fn notify_file_boundary(&mut self, is_boundary: bool) { + self.last_seen_file_boundary = is_boundary; + } + /// X11 inference entry point: forward-only pass over `snapshots` /// (length `cfg.n_batch * cfg.seq_len` per the trainer's batching). /// Returns per-horizon probabilities flattened as `[K, B, N_HORIZONS]` diff --git a/crates/ml-alpha/tests/multi_horizon_loader.rs b/crates/ml-alpha/tests/multi_horizon_loader.rs index 02c83d5ac..d6f8200a7 100644 --- a/crates/ml-alpha/tests/multi_horizon_loader.rs +++ b/crates/ml-alpha/tests/multi_horizon_loader.rs @@ -3,7 +3,7 @@ //! Real-data path is `--ignored`; runs only with FOXHUNT_TEST_DATA set. use ml_alpha::data::loader::{ - discover_mbp10_files_sorted, MultiHorizonLoader, MultiHorizonLoaderConfig, + discover_mbp10_files_sorted, LoaderMode, MultiHorizonLoader, MultiHorizonLoaderConfig, }; use std::path::PathBuf; @@ -23,6 +23,7 @@ fn cfg_from_env() -> Option { n_max_sequences: 100, seed: 0xA1A2_A3A4, inference_only: false, + mode: LoaderMode::default(), }) } @@ -65,6 +66,7 @@ fn loader_errors_on_empty_files() { n_max_sequences: 10, seed: 0, inference_only: false, + mode: LoaderMode::default(), }; let res = MultiHorizonLoader::new(&cfg); assert!(res.is_err(), "expected error for empty file list"); diff --git a/crates/ml-alpha/tests/perception_overfit.rs b/crates/ml-alpha/tests/perception_overfit.rs index f72cae6dc..68063e7a7 100644 --- a/crates/ml-alpha/tests/perception_overfit.rs +++ b/crates/ml-alpha/tests/perception_overfit.rs @@ -7,6 +7,7 @@ //! end-to-end and that all 6 AdamW optimizers actually move weights. use ml_alpha::cfc::snap_features::Mbp10RawInput; +use ml_alpha::data::loader::LoaderMode; use ml_alpha::trainer::perception::{PerceptionTrainer, PerceptionTrainerConfig}; use ml_core::device::MlDevice; @@ -74,6 +75,7 @@ fn stacked_trainer_loss_shrinks_on_constant_signal() { n_batch: 1, smoothness_base_lambda: 0.0, kernel_step_trace_path: None, + loader_mode: LoaderMode::Random, }; let mut trainer = PerceptionTrainer::new(&dev, &cfg).expect("init"); @@ -149,6 +151,7 @@ fn stacked_trainer_loss_shrinks_with_stride_4() { n_batch: 1, smoothness_base_lambda: 0.0, kernel_step_trace_path: None, + loader_mode: LoaderMode::Random, }; let mut trainer = PerceptionTrainer::new(&dev, &cfg).expect("init"); @@ -204,6 +207,7 @@ fn stacked_trainer_loss_shrinks_at_batch_32() { n_batch: 32, smoothness_base_lambda: 0.0, kernel_step_trace_path: None, + loader_mode: LoaderMode::Random, }; let mut trainer = PerceptionTrainer::new(&dev, &cfg).expect("init"); @@ -303,6 +307,7 @@ fn evaluate_alone_succeeds() { n_batch: 1, smoothness_base_lambda: 0.0, kernel_step_trace_path: None, + loader_mode: LoaderMode::Random, }; let mut trainer = PerceptionTrainer::new(&dev, &cfg).expect("init"); let ts = 1_000_000u64; @@ -332,6 +337,7 @@ fn evaluate_works_after_captured_training_step() { n_batch: 1, smoothness_base_lambda: 0.0, kernel_step_trace_path: None, + loader_mode: LoaderMode::Random, }; let mut trainer = PerceptionTrainer::new(&dev, &cfg).expect("init"); @@ -368,6 +374,7 @@ fn evaluate_works_after_capture_no_replay() { n_batch: 1, smoothness_base_lambda: 0.0, kernel_step_trace_path: None, + loader_mode: LoaderMode::Random, }; let mut trainer = PerceptionTrainer::new(&dev, &cfg).expect("init"); let ts = 1_000_000u64; @@ -399,6 +406,7 @@ fn horizon_ema_and_lambda_track_after_training() { n_batch: 1, smoothness_base_lambda: 0.0, kernel_step_trace_path: None, + loader_mode: LoaderMode::Random, }; let mut trainer = PerceptionTrainer::new(&dev, &cfg).expect("init"); @@ -456,6 +464,7 @@ fn evaluate_works_after_warmup_only() { n_batch: 1, smoothness_base_lambda: 0.0, kernel_step_trace_path: None, + loader_mode: LoaderMode::Random, }; let mut trainer = PerceptionTrainer::new(&dev, &cfg).expect("init"); let ts = 1_000_000u64; diff --git a/crates/ml-backtesting/src/harness.rs b/crates/ml-backtesting/src/harness.rs index c31c72d36..bef6af9f6 100644 --- a/crates/ml-backtesting/src/harness.rs +++ b/crates/ml-backtesting/src/harness.rs @@ -13,7 +13,7 @@ use anyhow::{Context, Result}; use ml_alpha::cfc::snap_features::Mbp10RawInput; use ml_alpha::data::loader::{ - discover_mbp10_files_sorted, MultiHorizonLoader, MultiHorizonLoaderConfig, + discover_mbp10_files_sorted, LoaderMode, MultiHorizonLoader, MultiHorizonLoaderConfig, }; use ml_alpha::trainer::perception::PerceptionTrainer; use ml_core::device::MlDevice; @@ -149,6 +149,7 @@ impl BacktestHarness { seed: 0, // A1: decision_stride removed; inference mode walks every event at stride=1. inference_only: true, + mode: LoaderMode::default(), }; let loader = MultiHorizonLoader::new(&loader_cfg)?; diff --git a/crates/ml-backtesting/tests/ring3_replay.rs b/crates/ml-backtesting/tests/ring3_replay.rs index 1a329e487..ed5e26e2d 100644 --- a/crates/ml-backtesting/tests/ring3_replay.rs +++ b/crates/ml-backtesting/tests/ring3_replay.rs @@ -25,7 +25,7 @@ use anyhow::Result; use ml_alpha::data::loader::{ - discover_mbp10_files_sorted, MultiHorizonLoader, MultiHorizonLoaderConfig, + discover_mbp10_files_sorted, LoaderMode, MultiHorizonLoader, MultiHorizonLoaderConfig, }; use ml_backtesting::sim::LobSimCuda; use ml_core::device::MlDevice; @@ -48,6 +48,7 @@ fn try_loader() -> Option { n_max_sequences: 0, seed: 0xCAFE_F00D, inference_only: true, + mode: LoaderMode::default(), }; match MultiHorizonLoader::new(&cfg) { Ok(l) => Some(l), diff --git a/crates/ml-backtesting/tests/trainer_parity.rs b/crates/ml-backtesting/tests/trainer_parity.rs index 074686bd3..ed81c3c97 100644 --- a/crates/ml-backtesting/tests/trainer_parity.rs +++ b/crates/ml-backtesting/tests/trainer_parity.rs @@ -16,7 +16,7 @@ use anyhow::Result; use ml_alpha::data::loader::{ - discover_mbp10_files_sorted, MultiHorizonLoader, MultiHorizonLoaderConfig, + discover_mbp10_files_sorted, LoaderMode, MultiHorizonLoader, MultiHorizonLoaderConfig, }; use std::path::PathBuf; @@ -35,6 +35,7 @@ fn try_loader(inference_only: bool) -> Option { n_max_sequences: 1, seed: 0xCAFEF00D, inference_only, + mode: LoaderMode::default(), }; match MultiHorizonLoader::new(&cfg) { Ok(l) => Some(l), diff --git a/infra/k8s/argo/alpha-perception-template.yaml b/infra/k8s/argo/alpha-perception-template.yaml index 95ed8384b..87004865d 100644 --- a/infra/k8s/argo/alpha-perception-template.yaml +++ b/infra/k8s/argo/alpha-perception-template.yaml @@ -76,6 +76,8 @@ spec: value: "1" - name: cv-train-window value: "0" + - name: loader-mode + value: "random" - name: smoothness-base-lambda value: "0.0" - name: kernel-step-trace-enable @@ -469,6 +471,7 @@ spec: --cv-fold {{workflow.parameters.cv-fold}} \ --cv-n-folds {{workflow.parameters.cv-n-folds}} \ --cv-train-window {{workflow.parameters.cv-train-window}} \ + --loader-mode "{{workflow.parameters.loader-mode}}" \ --smoothness-base-lambda {{workflow.parameters.smoothness-base-lambda}} \ ${TRACE_FLAG} \ $EXTRA_FLAGS