gitadmin/foxhunt
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(ml-alpha): walk-forward CV via file-list-driven loader

Browse Source 
mhzs7 reported val mean_auc=0.726 on 3a196382f — but the trainer
constructed both train and val MultiHorizonLoader with the SAME
`mbp10_root: cli.mbp10_data_dir`. The two loaders only differed by
seed. So val sequences were held-out-by-anchor from the same files
train sampled from; not temporally OOS. Per
`pearl_single_window_oos_is_not_oos.md` a single-window result that
doesn't enforce time-ordered separation can collapse across true
walk-forward folds.

Refactor: drop `mbp10_root` from `MultiHorizonLoaderConfig` (which
forced caller to share the dir between train and val). New API takes
an explicit `files: Vec<PathBuf>` — the loader preserves the order
given and does no internal shuffle, so callers control temporal
ordering. Added `discover_mbp10_files_sorted(root)` helper that
enumerates a dir and sorts by filename (chronological under the
`ES.FUT_<YEAR>-Q<n>.dbn.zst` convention).

alpha_train.rs splits the discovered files by 3 new CLI flags:
  --cv-fold <k>            (default 0)
  --cv-n-folds <N>         (default 1 — single fold)
  --cv-train-window <W>    (default 0 — auto)

Single-fold default (cv_n_folds=1): train on all files except the
last, val on the last file. This replaces the old "same files for
both" bug; even runs that don't think about CV now get a temporal
split by default.

Sliding-window CV (cv_n_folds > 1): fold k trains on files
[k..k+W] and validates on file [k+W]. With 9 quarterly files
(2024-Q1..2026-Q1) and `--cv-n-folds 3`, the natural layout is:

  fold 0: train 2024-Q1..2024-Q4 (W=4) → val 2025-Q1
  fold 1: train 2024-Q2..2025-Q1       → val 2025-Q2
  fold 2: train 2024-Q3..2025-Q2       → val 2025-Q3
  blind holdout: 2025-Q4, 2026-Q1

Threaded the flags through scripts/argo-alpha-perception.sh and
infra/k8s/argo/alpha-perception-template.yaml so each fold submits
as an independent workflow.

Updated tests/multi_horizon_loader.rs to the new API:
  loader_yields_seq_with_valid_labels — exercises discover + load.
  loader_errors_on_empty_files       — replaces missing-root test.
  discover_errors_on_missing_root    — pinpoints the discover step.

Honors:
  - feedback_no_partial_refactor.md — every consumer migrated atomically.
  - feedback_no_legacy_aliases.md — no `mbp10_root` shim left behind.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
jgrusewski
2026-05-17 16:36:48 +02:00
parent 3a196382f0
commit eb51c0f9cd
5 changed files with 167 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

92
crates/ml-alpha/examples/alpha_train.rs
View File

@@ -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_<YEAR>-Q<n>.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<PathBuf>, Vec<PathBuf>) = 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::<Vec<_>>(),
val_files = ?val_files.iter().map(|p| p.file_name().unwrap().to_string_lossy().to_string()).collect::<Vec<_>>(),
"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,

64
crates/ml-alpha/src/data/loader.rs
View File

@@ -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<PathBuf>,
/// 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_<YEAR>-Q<n>.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<Vec<PathBuf>> {
let mut files: Vec<PathBuf> = 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<Mbp10RawInput>,
@@ -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<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();
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<PathBuf> = 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(),

21
crates/ml-alpha/tests/multi_horizon_loader.rs
View File

@@ -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<MultiHorizonLoaderConfig> {
@@ -12,8 +14,9 @@ fn cfg_from_env() -> Option<MultiHorizonLoaderConfig> {
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");
}

9
infra/k8s/argo/alpha-perception-template.yaml
View File

@@ -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 ==="

12
scripts/argo-alpha-perception.sh
View File

@@ -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 <n> Train sequences per epoch (default: $N_TRAIN_SEQS)
--n-val-seqs <n> Val sequences per epoch (default: $N_VAL_SEQS)
--seed <n> Random seed (default: $SEED)
--cv-fold <k> CV fold index (default: $CV_FOLD)
--cv-n-folds <N> Total CV folds (default: $CV_N_FOLDS)
--cv-train-window <W> 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