diff --git a/crates/ml-alpha/examples/alpha_train.rs b/crates/ml-alpha/examples/alpha_train.rs index 215070b7a..e7b093398 100644 --- a/crates/ml-alpha/examples/alpha_train.rs +++ b/crates/ml-alpha/examples/alpha_train.rs @@ -112,6 +112,28 @@ struct Cli { /// previous unbatched behavior. Validation always runs at B=1. #[arg(long, default_value_t = 1)] batch_size: usize, + + /// Sliding-window cross-validation: index of THIS fold (0-indexed). + /// Combined with `--cv-n-folds` and `--cv-train-window` to split + /// the discovered MBP-10 files chronologically. Default 0 means + /// "no CV — use the first fold of a 1-fold split" (legacy behavior: + /// train = all-but-last file, val = last file). + #[arg(long, default_value_t = 0)] + cv_fold: usize, + + /// Total number of sliding-window CV folds. 1 disables CV (single + /// train/val split). When >1, fold k trains on the W files starting + /// at `cv_fold_offset(k)` and validates on the file immediately + /// after that window. Folds are submitted as independent runs. + #[arg(long, default_value_t = 1)] + cv_n_folds: usize, + + /// Size of each fold's TRAIN window in files. 0 means "use all + /// files except the val file" (expanding-window mode at 1 fold). + /// >0 enables sliding-window mode where the training data has a + /// fixed temporal extent. + #[arg(long, default_value_t = 0)] + cv_train_window: usize, } #[derive(Serialize)] @@ -247,13 +269,77 @@ fn main() -> Result<()> { // sequence yields one optimizer step). Used by cosine decay. let total_steps_budget = cli.epochs * cli.n_train_seqs; - // PRELOAD: construct loaders ONCE (each preloads all files into RAM). + // ── Walk-forward CV file split ──────────────────────────────────── + // Files are discovered in chronological order (filenames follow + // ES.FUT_-Q.dbn.zst so sort = chronological). Then we + // slice train_files / val_files based on the CV flags. Train and + // val NEVER share a file — every val sample is strictly OOS in + // time vs every train sample. + let all_files = ml_alpha::data::loader::discover_mbp10_files_sorted(&cli.mbp10_data_dir) + .context("discover MBP-10 files")?; + anyhow::ensure!( + cli.cv_n_folds >= 1, + "--cv-n-folds must be >= 1 (got {})", cli.cv_n_folds + ); + anyhow::ensure!( + cli.cv_fold < cli.cv_n_folds, + "--cv-fold {} must be < --cv-n-folds {}", cli.cv_fold, cli.cv_n_folds + ); + let n_files = all_files.len(); + let (train_files, val_files): (Vec, Vec) = if cli.cv_n_folds == 1 { + // Single fold: train on all files except the LAST, val on the LAST file. + // Legacy callers that didn't think about CV at all get a temporal split + // by default instead of the old "same files for train and val" bug. + anyhow::ensure!(n_files >= 2, + "need >= 2 files for single-fold temporal split, found {} under {}", + n_files, cli.mbp10_data_dir.display()); + let val_idx = n_files - 1; + ( + all_files[..val_idx].to_vec(), + vec![all_files[val_idx].clone()], + ) + } else { + // Sliding-window CV. Train window has fixed extent `W`; val is the + // single file immediately after. Fold k uses files [k .. k+W] for + // train and file [k+W] for val. + let w = if cli.cv_train_window > 0 { + cli.cv_train_window + } else { + // Default: leave room for n_folds val files at the tail. + n_files.saturating_sub(cli.cv_n_folds) + }; + anyhow::ensure!(w >= 1, "cv_train_window resolves to 0 (n_files={n_files}, n_folds={})", cli.cv_n_folds); + anyhow::ensure!( + cli.cv_fold + w + 1 <= n_files, + "CV layout overflows: cv_fold={} + window={} + 1 val file > {} files available", + cli.cv_fold, w, n_files + ); + let train_start = cli.cv_fold; + let train_end = train_start + w; // exclusive + let val_idx = train_end; + ( + all_files[train_start..train_end].to_vec(), + vec![all_files[val_idx].clone()], + ) + }; + tracing::info!( + cv_fold = cli.cv_fold, + cv_n_folds = cli.cv_n_folds, + cv_train_window = cli.cv_train_window, + n_train_files = train_files.len(), + n_val_files = val_files.len(), + train_files = ?train_files.iter().map(|p| p.file_name().unwrap().to_string_lossy().to_string()).collect::>(), + val_files = ?val_files.iter().map(|p| p.file_name().unwrap().to_string_lossy().to_string()).collect::>(), + "walk-forward CV file split", + ); + + // PRELOAD: construct loaders ONCE (each preloads its files into RAM). // Per-epoch we call reset(seed) to re-shuffle anchor sampling — no // disk IO. Cuts wall time ~5× on a 6-epoch run, ~7× on 15 epochs. tracing::info!("preloading train + val MBP-10 files into RAM (slow startup, fast per-epoch)…"); let preload_start = std::time::Instant::now(); let mut train_loader = MultiHorizonLoader::new(&MultiHorizonLoaderConfig { - mbp10_root: cli.mbp10_data_dir.clone(), + files: train_files, predecoded_dir: cli.predecoded_dir.clone(), seq_len: cli.seq_len, horizons, @@ -262,7 +348,7 @@ fn main() -> Result<()> { }) .context("train loader")?; let mut val_loader = MultiHorizonLoader::new(&MultiHorizonLoaderConfig { - mbp10_root: cli.mbp10_data_dir.clone(), + files: val_files, predecoded_dir: cli.predecoded_dir.clone(), seq_len: cli.seq_len, horizons, diff --git a/crates/ml-alpha/src/data/loader.rs b/crates/ml-alpha/src/data/loader.rs index 498041fdb..9b4e398a7 100644 --- a/crates/ml-alpha/src/data/loader.rs +++ b/crates/ml-alpha/src/data/loader.rs @@ -15,7 +15,6 @@ 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; @@ -88,10 +87,14 @@ fn compute_regime_features(snapshots: &[Mbp10Snapshot]) -> Vec<[f32; REGIME_DIM] #[derive(Clone, Debug)] pub struct MultiHorizonLoaderConfig { - /// Directory containing .dbn.zst MBP-10 files plus their predecoded sidecars. - pub mbp10_root: PathBuf, + /// Explicit list of .dbn / .dbn.zst MBP-10 files to load. Order + /// is preserved (no internal shuffle) so callers can use file + /// ordering to control temporal train / val splits. Use + /// [`discover_mbp10_files_sorted`] to enumerate a directory in + /// chronological-filename order. + pub files: Vec, /// Directory where predecoded sidecars live (`load_or_predecode_mbp10` - /// arg — typically the same as `mbp10_root` or a sibling). + /// 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, @@ -99,10 +102,32 @@ pub struct MultiHorizonLoaderConfig { 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. + /// Deterministic seed for sequence-anchor selection within files. + /// (Files themselves are taken in the order given in `files`; this + /// seed only randomises which snapshot inside each file becomes + /// the start of each yielded sequence.) pub seed: u64, } +/// Discover MBP-10 files under `root` and return them sorted by filename. +/// Filenames follow the `ES.FUT_-Q.dbn.zst` convention, so +/// lexicographic sort = chronological. Caller can then slice for +/// walk-forward CV. +pub fn discover_mbp10_files_sorted(root: &std::path::Path) -> Result> { + let mut files: Vec = std::fs::read_dir(root) + .with_context(|| format!("read mbp10 root {}", root.display()))? + .filter_map(|e| e.ok()) + .map(|e| e.path()) + .filter(|p| { + let s = p.to_string_lossy(); + s.ends_with(".dbn.zst") || s.ends_with(".dbn") + }) + .collect(); + anyhow::ensure!(!files.is_empty(), "no DBN files under {}", root.display()); + files.sort(); + Ok(files) +} + pub struct LabeledSequence { /// `seq_len` consecutive raw snapshot inputs. pub snapshots: Vec, @@ -143,25 +168,18 @@ pub struct MultiHorizonLoader { } impl MultiHorizonLoader { - /// Construct + preload all MBP-10 files. Slow (~50s × N_files), - /// done once. Subsequent `next_sequence` calls are pure memory. + /// Construct + preload the MBP-10 files listed in `cfg.files`, + /// in the order given (no shuffle — file order is the caller's + /// responsibility, see [`discover_mbp10_files_sorted`]). Slow + /// (~50s × N_files), done once at construction. Subsequent + /// `next_sequence` calls are pure memory. pub fn new(cfg: &MultiHorizonLoaderConfig) -> 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(); - s.ends_with(".dbn.zst") || s.ends_with(".dbn") - }) - .collect(); anyhow::ensure!( - !files.is_empty(), - "no DBN files under {}", - cfg.mbp10_root.display() + !cfg.files.is_empty(), + "MultiHorizonLoaderConfig.files is empty — pass at least one file" ); - let mut rng = ChaCha8Rng::seed_from_u64(cfg.seed); - files.shuffle(&mut rng); + let files: Vec = cfg.files.clone(); + let rng = ChaCha8Rng::seed_from_u64(cfg.seed); let max_horizon = *cfg.horizons.iter().max().expect("non-empty horizons"); let min_size = cfg.seq_len + max_horizon + 1; @@ -193,8 +211,8 @@ impl MultiHorizonLoader { } anyhow::ensure!( !files_loaded.is_empty(), - "no files had enough snapshots ({}+ required) under {}", - min_size, cfg.mbp10_root.display() + "no files had enough snapshots ({}+ required) across {} input files", + min_size, cfg.files.len() ); tracing::info!( n_files = files_loaded.len(), diff --git a/crates/ml-alpha/tests/multi_horizon_loader.rs b/crates/ml-alpha/tests/multi_horizon_loader.rs index ae49eb1a0..ec6c1fbc3 100644 --- a/crates/ml-alpha/tests/multi_horizon_loader.rs +++ b/crates/ml-alpha/tests/multi_horizon_loader.rs @@ -2,7 +2,9 @@ //! //! Real-data path is `--ignored`; runs only with FOXHUNT_TEST_DATA set. -use ml_alpha::data::loader::{MultiHorizonLoaderConfig, MultiHorizonLoader}; +use ml_alpha::data::loader::{ + discover_mbp10_files_sorted, MultiHorizonLoader, MultiHorizonLoaderConfig, +}; use std::path::PathBuf; fn cfg_from_env() -> Option { @@ -12,8 +14,9 @@ fn cfg_from_env() -> Option { if !mbp10.exists() { return None; } + let files = discover_mbp10_files_sorted(&mbp10).ok()?; Some(MultiHorizonLoaderConfig { - mbp10_root: mbp10, + files, predecoded_dir: predec, seq_len: 64, horizons: [30, 100, 300, 1000, 6000], @@ -52,9 +55,9 @@ fn loader_yields_seq_with_valid_labels() { } #[test] -fn loader_errors_on_missing_root() { +fn loader_errors_on_empty_files() { let cfg = MultiHorizonLoaderConfig { - mbp10_root: PathBuf::from("/tmp/does_not_exist_foxhunt_test"), + files: Vec::new(), predecoded_dir: PathBuf::from("/tmp/does_not_exist_foxhunt_test"), seq_len: 64, horizons: [30, 100, 300, 1000, 6000], @@ -62,5 +65,13 @@ fn loader_errors_on_missing_root() { seed: 0, }; let res = MultiHorizonLoader::new(&cfg); - assert!(res.is_err(), "expected error for missing mbp10 root"); + assert!(res.is_err(), "expected error for empty file list"); +} + +#[test] +fn discover_errors_on_missing_root() { + let res = discover_mbp10_files_sorted(&PathBuf::from( + "/tmp/does_not_exist_foxhunt_test", + )); + assert!(res.is_err(), "expected error for missing root dir"); } diff --git a/infra/k8s/argo/alpha-perception-template.yaml b/infra/k8s/argo/alpha-perception-template.yaml index 4fb499b44..f7d501d46 100644 --- a/infra/k8s/argo/alpha-perception-template.yaml +++ b/infra/k8s/argo/alpha-perception-template.yaml @@ -70,6 +70,12 @@ spec: value: "mean_auc" - name: early-stop-patience value: "5" + - name: cv-fold + value: "0" + - name: cv-n-folds + value: "1" + - name: cv-train-window + value: "0" volumes: - name: git-ssh-key @@ -437,6 +443,9 @@ spec: --batch-size {{workflow.parameters.batch-size}} \ --early-stop-metric {{workflow.parameters.early-stop-metric}} \ --early-stop-patience {{workflow.parameters.early-stop-patience}} \ + --cv-fold {{workflow.parameters.cv-fold}} \ + --cv-n-folds {{workflow.parameters.cv-n-folds}} \ + --cv-train-window {{workflow.parameters.cv-train-window}} \ $EXTRA_FLAGS echo "=== Training complete ===" diff --git a/scripts/argo-alpha-perception.sh b/scripts/argo-alpha-perception.sh index 8f8cad0de..d830ea220 100755 --- a/scripts/argo-alpha-perception.sh +++ b/scripts/argo-alpha-perception.sh @@ -31,6 +31,9 @@ BATCH_SIZE=1 AUTO_HORIZON_WEIGHTS=false EARLY_STOP_METRIC=mean_auc EARLY_STOP_PATIENCE=5 +CV_FOLD=0 +CV_N_FOLDS=1 +CV_TRAIN_WINDOW=0 WATCH=false usage() { @@ -47,6 +50,9 @@ Usage: $0 [OPTIONS] --n-train-seqs Train sequences per epoch (default: $N_TRAIN_SEQS) --n-val-seqs Val sequences per epoch (default: $N_VAL_SEQS) --seed Random seed (default: $SEED) + --cv-fold CV fold index (default: $CV_FOLD) + --cv-n-folds Total CV folds (default: $CV_N_FOLDS) + --cv-train-window Files per train window (default: $CV_TRAIN_WINDOW = auto) --watch Follow logs via argo watch EOF } @@ -68,6 +74,9 @@ while [[ $# -gt 0 ]]; do --auto-horizon-weights) AUTO_HORIZON_WEIGHTS=true; shift ;; --early-stop-metric) EARLY_STOP_METRIC="$2"; shift 2 ;; --early-stop-patience) EARLY_STOP_PATIENCE="$2"; shift 2 ;; + --cv-fold) CV_FOLD="$2"; shift 2 ;; + --cv-n-folds) CV_N_FOLDS="$2"; shift 2 ;; + --cv-train-window) CV_TRAIN_WINDOW="$2"; shift 2 ;; --watch) WATCH=true; shift ;; -h|--help) usage; exit 0 ;; *) echo "Unknown option: $1"; usage; exit 1 ;; @@ -135,4 +144,7 @@ argo submit -n foxhunt --from=wftmpl/alpha-perception \ -p auto-horizon-weights="$AUTO_HORIZON_WEIGHTS" \ -p early-stop-metric="$EARLY_STOP_METRIC" \ -p early-stop-patience="$EARLY_STOP_PATIENCE" \ + -p cv-fold="$CV_FOLD" \ + -p cv-n-folds="$CV_N_FOLDS" \ + -p cv-train-window="$CV_TRAIN_WINDOW" \ $WATCH_FLAG