From 6e339316cf01bb457eb4892d9d896c7b6b995395 Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Thu, 26 Feb 2026 09:04:58 +0100 Subject: [PATCH] feat(ml): add manually-triggered GitLab CI training pipeline MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds a parent/child GitLab CI pipeline for ML model training: - Generator script produces per-model hyperopt/train/evaluate jobs - Parent pipeline (.gitlab-ci-training.yml) with manual trigger - NFS-backed ReadWriteMany PVC for shared training outputs - Hyperopt params wired into training binaries (DQN, PPO, TFT, Mamba2) - Shared DBN loader eliminates duplicate code across hyperopt adapters - Supervised hyperopt unified to DBN data (was parquet-only) Pipeline: hyperopt (4 models) → train (10 models) → evaluate ensemble Co-Authored-By: Claude Opus 4.6 --- .gitignore | 1 + .gitlab-ci-training.yml | 61 ++ crates/ml/examples/baseline_common/mod.rs | 19 +- .../examples/hyperopt_baseline_supervised.rs | 53 +- crates/ml/examples/train_baseline_rl.rs | 129 ++-- .../ml/examples/train_baseline_supervised.rs | 109 ++- crates/ml/src/hyperopt/adapters/dbn_loader.rs | 97 +++ crates/ml/src/hyperopt/adapters/dqn.rs | 52 +- crates/ml/src/hyperopt/adapters/mamba2.rs | 153 +--- crates/ml/src/hyperopt/adapters/mod.rs | 3 + crates/ml/src/hyperopt/adapters/tft.rs | 90 +-- crates/ml/src/trainers/tft_parquet.rs | 87 ++- .../2026-02-26-training-pipeline-design.md | 118 +++ ...-02-26-training-pipeline-implementation.md | 675 ++++++++++++++++++ infra/k8s/training/job-template.yaml | 3 +- infra/k8s/training/training-output-pvc.yaml | 15 + scripts/generate-training-pipeline.sh | 332 +++++++++ 17 files changed, 1598 insertions(+), 399 deletions(-) create mode 100644 .gitlab-ci-training.yml create mode 100644 crates/ml/src/hyperopt/adapters/dbn_loader.rs create mode 100644 docs/plans/2026-02-26-training-pipeline-design.md create mode 100644 docs/plans/2026-02-26-training-pipeline-implementation.md create mode 100644 infra/k8s/training/training-output-pvc.yaml create mode 100755 scripts/generate-training-pipeline.sh diff --git a/.gitignore b/.gitignore index 27feff83d..9bbb65257 100644 --- a/.gitignore +++ b/.gitignore @@ -83,6 +83,7 @@ target/criterion/ target/bench/ # CI artifacts +.training-generated.yml audit-results.json geiger-report.md security-report.md diff --git a/.gitlab-ci-training.yml b/.gitlab-ci-training.yml new file mode 100644 index 000000000..3dee614f3 --- /dev/null +++ b/.gitlab-ci-training.yml @@ -0,0 +1,61 @@ +# ML Training Pipeline — manually triggered +# +# Trigger via GitLab UI: CI/CD → Pipelines → Run pipeline +# Set CI configuration file to: .gitlab-ci-training.yml +# Set variables as needed (MODELS, SYMBOLS, PHASE, etc.) +# +# Or via API: +# curl -X POST --fail \ +# -F "token=$TRIGGER_TOKEN" \ +# -F "ref=main" \ +# -F "variables[MODELS]=all" \ +# -F "variables[SYMBOLS]=ES.FUT" \ +# -F "variables[PHASE]=full" \ +# "$CI_API_V4_URL/projects/$CI_PROJECT_ID/trigger/pipeline" + +stages: + - prepare + - trigger + +variables: + SYMBOLS: "ES.FUT" + MODELS: "all" + PHASE: "full" + MAX_PARALLEL: "10" + EPOCHS: "50" + HYPEROPT_TRIALS: "20" + RUN_ID: "" + REGISTRY: rg.fr-par.scw.cloud/foxhunt-ci + +workflow: + rules: + - if: $CI_PIPELINE_SOURCE == "web" + - if: $CI_PIPELINE_SOURCE == "trigger" + - if: $CI_PIPELINE_SOURCE == "api" + +generate-jobs: + stage: prepare + image: alpine:3.19 + tags: + - kapsule + script: + - apk add --no-cache bash coreutils + - | + if [ -z "$RUN_ID" ]; then + export RUN_ID=$(date +%Y%m%d-%H%M%S) + fi + - bash scripts/generate-training-pipeline.sh + - echo "--- Generated pipeline ---" + - cat .training-generated.yml + artifacts: + paths: + - .training-generated.yml + expire_in: 1 day + +run-training: + stage: trigger + trigger: + include: + - artifact: .training-generated.yml + job: generate-jobs + strategy: depend diff --git a/crates/ml/examples/baseline_common/mod.rs b/crates/ml/examples/baseline_common/mod.rs index b2c21caa6..ab7a0838f 100644 --- a/crates/ml/examples/baseline_common/mod.rs +++ b/crates/ml/examples/baseline_common/mod.rs @@ -29,17 +29,18 @@ fn collect_dbn_files_recursive(dir: &Path, out: &mut Vec) -> Result<()> let entries = std::fs::read_dir(dir) .with_context(|| format!("Cannot read directory: {}", dir.display()))?; for entry in entries { - let entry = entry.with_context(|| "Failed to read dir entry")?; - let path = entry.path(); - if path.is_dir() { - collect_dbn_files_recursive(&path, out)?; - } else if path + let dir_entry = entry.with_context(|| "Failed to read dir entry")?; + let entry_path = dir_entry.path(); + if entry_path.is_dir() { + collect_dbn_files_recursive(&entry_path, out)?; + } else if entry_path .file_name() .and_then(|n| n.to_str()) - .map(|n| n.ends_with(".dbn.zst")) - .unwrap_or(false) + .is_some_and(|n| n.ends_with(".dbn.zst")) { - out.push(path); + out.push(entry_path); + } else { + // Skip non-.dbn.zst files } } Ok(()) @@ -157,7 +158,7 @@ pub fn load_all_bars(data_dir: &Path, symbol: &str) -> Result> { all_bars.extend(bars); } Err(e) => { - warn!(" Skipping {} — {}", path.display(), e); + warn!(" Skipping {} -- {}", path.display(), e); } } } diff --git a/crates/ml/examples/hyperopt_baseline_supervised.rs b/crates/ml/examples/hyperopt_baseline_supervised.rs index 9e97f71dd..a8e2aef6f 100644 --- a/crates/ml/examples/hyperopt_baseline_supervised.rs +++ b/crates/ml/examples/hyperopt_baseline_supervised.rs @@ -1,25 +1,20 @@ -//! Hyperopt Runner for Supervised Models on Parquet Data +//! Hyperopt Runner for Supervised Models on DBN Data //! //! Runs hyperparameter optimization using Particle Swarm Optimization (PSO) for -//! TFT, MAMBA-2, or both models. This binary is the supervised counterpart to -//! `hyperopt_baseline_rl` (which handles DQN/PPO on DBN data). +//! TFT, MAMBA-2, or both models. Uses the same --data-dir / --symbol interface +//! as the RL hyperopt and training binaries. //! //! ## Usage //! //! ```bash -//! # Run TFT hyperopt -//! SQLX_OFFLINE=true cargo run -p ml --example hyperopt_baseline_supervised --release -- \ -//! --model tft --parquet-file data/ES_FUT_180d.parquet \ -//! --trials 20 --epochs 20 +//! # Run TFT hyperopt on ES futures +//! hyperopt_baseline_supervised --model tft --data-dir /data/ES.FUT --trials 20 --epochs 20 //! //! # Run Mamba2 hyperopt -//! SQLX_OFFLINE=true cargo run -p ml --example hyperopt_baseline_supervised --release -- \ -//! --model mamba2 --parquet-file data/ES_FUT_180d.parquet \ -//! --trials 20 --epochs 10 +//! hyperopt_baseline_supervised --model mamba2 --data-dir /data/NQ.FUT --trials 20 --epochs 10 //! //! # Run both models -//! SQLX_OFFLINE=true cargo run -p ml --example hyperopt_baseline_supervised --release -- \ -//! --model both --parquet-file data/ES_FUT_180d.parquet +//! hyperopt_baseline_supervised --model both --data-dir /data/ES.FUT //! ``` //! //! ## Output @@ -49,9 +44,9 @@ struct Args { #[arg(long, default_value = "both")] model: String, - /// Path to Parquet file with OHLCV data + /// Directory containing .dbn.zst OHLCV data files #[arg(long)] - parquet_file: String, + data_dir: PathBuf, /// Number of PSO trials per model #[arg(long, default_value = "20")] @@ -82,7 +77,6 @@ struct Args { early_stopping_patience: usize, } -/// Result entry for one model's hyperopt run fn build_model_result( best_objective: f64, best_params_json: Value, @@ -106,11 +100,11 @@ fn run_tft_hyperopt(args: &Args) -> Result { let training_paths = TrainingPaths::new(&args.base_dir, "tft", &run_id); info!("Run ID: {}", run_id); - info!("Parquet file: {}", args.parquet_file); + info!("Data dir: {}", args.data_dir.display()); info!("Epochs per trial: {}", args.epochs); info!("Trials: {}", args.trials); - let trainer = TFTTrainer::new(&args.parquet_file, args.epochs) + let trainer = TFTTrainer::new(&args.data_dir, args.epochs) .context("Failed to create TFT trainer")? .with_early_stopping(args.early_stopping_patience) .with_training_paths(training_paths); @@ -153,11 +147,11 @@ fn run_mamba2_hyperopt(args: &Args) -> Result { let training_paths = TrainingPaths::new(&args.base_dir, "mamba2", &run_id); info!("Run ID: {}", run_id); - info!("Parquet file: {}", args.parquet_file); + info!("Data dir: {}", args.data_dir.display()); info!("Epochs per trial: {}", args.epochs); info!("Trials: {}", args.trials); - let trainer = Mamba2Trainer::new(&args.parquet_file, args.epochs) + let trainer = Mamba2Trainer::new(&args.data_dir, args.epochs) .context("Failed to create Mamba2 trainer")? .with_training_paths(training_paths); @@ -202,22 +196,20 @@ fn main() -> Result<()> { info!(" Hyperopt Baseline Supervised Runner"); info!("========================================"); info!("Model: {}", args.model); - info!("Parquet file: {}", args.parquet_file); + info!("Data dir: {}", args.data_dir.display()); info!("Trials: {}", args.trials); info!("Initial LHS samples: {}", args.n_initial); info!("Epochs per trial: {}", args.epochs); info!("Output: {}", args.output.display()); info!("Seed: {}", args.seed); - // Validate parquet file exists - if !std::path::Path::new(&args.parquet_file).exists() { + if !args.data_dir.exists() { anyhow::bail!( - "Parquet file not found: {}. Provide a valid path via --parquet-file.", - args.parquet_file + "Data directory not found: {}. Provide a valid path via --data-dir.", + args.data_dir.display() ); } - // Validate model selection let run_tft = args.model == "tft" || args.model == "both"; let run_mamba2 = args.model == "mamba2" || args.model == "both"; @@ -228,7 +220,6 @@ fn main() -> Result<()> { ); } - // Verify trials > n_initial if args.trials <= args.n_initial { anyhow::bail!( "trials ({}) must be greater than n_initial ({})", @@ -237,7 +228,6 @@ fn main() -> Result<()> { ); } - // Create output directory if let Some(parent) = args.output.parent() { std::fs::create_dir_all(parent) .with_context(|| format!("Failed to create output directory: {}", parent.display()))?; @@ -248,13 +238,13 @@ fn main() -> Result<()> { if run_tft { match run_tft_hyperopt(&args) { Ok(tft_result) => { - results.insert("tft".to_string(), tft_result); + results.insert("tft".to_owned(), tft_result); } Err(e) => { error!("TFT hyperopt failed: {:#}", e); warn!("Continuing with remaining models..."); results.insert( - "tft".to_string(), + "tft".to_owned(), serde_json::json!({ "error": format!("{:#}", e) }), ); } @@ -264,20 +254,19 @@ fn main() -> Result<()> { if run_mamba2 { match run_mamba2_hyperopt(&args) { Ok(mamba2_result) => { - results.insert("mamba2".to_string(), mamba2_result); + results.insert("mamba2".to_owned(), mamba2_result); } Err(e) => { error!("Mamba2 hyperopt failed: {:#}", e); warn!("Continuing..."); results.insert( - "mamba2".to_string(), + "mamba2".to_owned(), serde_json::json!({ "error": format!("{:#}", e) }), ); } } } - // Write results to JSON let output_json = Value::Object(results); let output_str = serde_json::to_string_pretty(&output_json).context("Failed to serialize results")?; diff --git a/crates/ml/examples/train_baseline_rl.rs b/crates/ml/examples/train_baseline_rl.rs index 9551a5a15..c029d0b65 100644 --- a/crates/ml/examples/train_baseline_rl.rs +++ b/crates/ml/examples/train_baseline_rl.rs @@ -26,6 +26,7 @@ use std::path::{Path, PathBuf}; use anyhow::{Context, Result}; use clap::Parser; use rand::Rng; +use serde_json::Value; use tracing::{error, info, warn}; use ml::dqn::{DQNConfig, Experience, DQN}; @@ -68,7 +69,7 @@ struct Args { #[arg(long, default_value = "ml/trained_models")] output_dir: PathBuf, - /// Optional path to hyperopt results JSON (reserved for future use) + /// Optional path to hyperopt results JSON -- overrides matching config fields #[arg(long)] hyperopt_params: Option, @@ -133,6 +134,53 @@ struct Args { spread_ticks: f64, } +// --------------------------------------------------------------------------- +// Hyperopt parameter loading +// --------------------------------------------------------------------------- + +/// Load best_params from a hyperopt results JSON file. +/// +/// Expected format: `{ "model_key": { "best_params": { ... }, ... } }` +/// Returns `None` if the file doesn't exist or can't be parsed. +fn load_hyperopt_params(hp_path: &Option, model_key: &str) -> Option { + let file_path = hp_path.as_ref()?; + if !file_path.exists() { + info!("Hyperopt params file not found: {}, using defaults", file_path.display()); + return None; + } + let contents = match std::fs::read_to_string(file_path) { + Ok(c) => c, + Err(e) => { + warn!("Failed to read hyperopt params {}: {}", file_path.display(), e); + return None; + } + }; + let json: Value = match serde_json::from_str(&contents) { + Ok(v) => v, + Err(e) => { + warn!("Failed to parse hyperopt params JSON: {}", e); + return None; + } + }; + let params = json.get(model_key) + .and_then(|m| m.get("best_params")) + .cloned(); + if params.is_some() { + info!("Loaded hyperopt params for '{}' from {}", model_key, file_path.display()); + } else { + warn!("No best_params found for '{}' in {}", model_key, file_path.display()); + } + params +} + +fn hp_f64(params: &Option, key: &str) -> Option { + params.as_ref()?.get(key)?.as_f64() +} + +fn hp_usize(params: &Option, key: &str) -> Option { + params.as_ref()?.get(key)?.as_u64().map(|v| v as usize) +} + // --------------------------------------------------------------------------- // Reward Calculation // --------------------------------------------------------------------------- @@ -178,6 +226,7 @@ fn train_dqn_fold( val_bars: &[OHLCVBar], args: &Args, output_dir: &Path, + hp: &Option, ) -> Result { // Caller must pre-align bars to features (skip warmup period before calling). // If these differ, bar-index reward computation will silently use wrong prices. @@ -195,26 +244,25 @@ fn train_dqn_fold( val_bars.len(), val_features.len(), ); - info!(" [DQN] Fold {} — {} train, {} val features", fold, train_features.len(), val_features.len()); + info!(" [DQN] Fold {} -- {} train, {} val features", fold, train_features.len(), val_features.len()); - // Configure DQN with baseline-friendly settings + // Configure DQN -- hyperopt params override compiled defaults, CLI args override both let config = DQNConfig { state_dim: args.feature_dim, num_actions: args.num_actions, hidden_dims: vec![128, 64], - learning_rate: args.learning_rate, - gamma: 0.95, // shorter horizon (20 bars vs 100) + learning_rate: hp_f64(hp, "learning_rate").unwrap_or(args.learning_rate), + gamma: hp_f64(hp, "gamma").unwrap_or(0.95) as f32, epsilon_start: 1.0, epsilon_end: 0.05, epsilon_decay: 0.995, - replay_buffer_capacity: 50_000, - batch_size: args.batch_size, - min_replay_size: args.batch_size, - target_update_freq: 200, // more frequent target sync - warmup_steps: 0, // No warmup — we fill buffer before training + replay_buffer_capacity: hp_usize(hp, "buffer_size").unwrap_or(50_000), + batch_size: hp_usize(hp, "batch_size").unwrap_or(args.batch_size), + min_replay_size: hp_usize(hp, "batch_size").unwrap_or(args.batch_size), + target_update_freq: 200, + warmup_steps: 0, use_double_dqn: true, use_huber_loss: true, - // Disable Rainbow extras for baseline simplicity use_per: false, use_dueling: false, use_distributional: false, @@ -239,7 +287,7 @@ fn train_dqn_fold( // Process training data sequentially let n_train = train_features.len(); if n_train < 2 { - warn!(" [DQN] Fold {} — insufficient training features ({})", fold, n_train); + warn!(" [DQN] Fold {} -- insufficient training features ({})", fold, n_train); break; } @@ -314,7 +362,7 @@ fn train_dqn_fold( dqn.set_epsilon(new_eps as f64); info!( - " [DQN] Fold {} Epoch {}/{} — train_loss={:.6} val_loss={:.6} eps={:.4}", + " [DQN] Fold {} Epoch {}/{} -- train_loss={:.6} val_loss={:.6} eps={:.4}", fold, epoch + 1, args.epochs, @@ -434,19 +482,23 @@ fn train_ppo_fold( val_bars: &[OHLCVBar], args: &Args, output_dir: &Path, + hp: &Option, ) -> Result { - info!(" [PPO] Fold {} — {} train, {} val features", fold, train_features.len(), val_features.len()); + info!(" [PPO] Fold {} -- {} train, {} val features", fold, train_features.len(), val_features.len()); + // PPO config -- hyperopt params override compiled defaults + let policy_lr = hp_f64(hp, "policy_learning_rate").unwrap_or(args.learning_rate); + let value_lr = hp_f64(hp, "value_learning_rate").unwrap_or(args.learning_rate * 3.0); let config = PPOConfig { state_dim: args.feature_dim, num_actions: args.num_actions, policy_hidden_dims: vec![128, 64], value_hidden_dims: vec![128, 64], - policy_learning_rate: args.learning_rate, - value_learning_rate: args.learning_rate * 3.0, - clip_epsilon: 0.2, - value_loss_coeff: 0.5, - entropy_coeff: 0.01, + policy_learning_rate: policy_lr, + value_learning_rate: value_lr, + clip_epsilon: hp_f64(hp, "clip_epsilon").unwrap_or(0.2) as f32, + value_loss_coeff: hp_f64(hp, "value_loss_coeff").unwrap_or(0.5) as f32, + entropy_coeff: hp_f64(hp, "entropy_coeff").unwrap_or(0.01) as f32, batch_size: args.batch_size.max(64), mini_batch_size: 64, num_epochs: 4, @@ -462,7 +514,7 @@ fn train_ppo_fold( let n_train = train_features.len(); if n_train < 2 { - warn!(" [PPO] Fold {} — insufficient training features ({})", fold, n_train); + warn!(" [PPO] Fold {} -- insufficient training features ({})", fold, n_train); return Ok(f64::MAX); } @@ -476,7 +528,7 @@ fn train_ppo_fold( )?; if trajectory.length < 2 { - warn!(" [PPO] Fold {} Epoch {} — trajectory too short", fold, epoch + 1); + warn!(" [PPO] Fold {} Epoch {} -- trajectory too short", fold, epoch + 1); continue; } @@ -503,7 +555,7 @@ fn train_ppo_fold( let val_loss = evaluate_ppo_validation(&ppo, val_features, val_bars, args.tx_cost_bps, args.tick_size, args.spread_ticks); info!( - " [PPO] Fold {} Epoch {}/{} — policy_loss={:.6} value_loss={:.6} val_metric={:.6}", + " [PPO] Fold {} Epoch {}/{} -- policy_loss={:.6} value_loss={:.6} val_metric={:.6}", fold, epoch + 1, args.epochs, @@ -661,7 +713,7 @@ fn evaluate_ppo_validation( None => break, }; - // Greedy (deterministic) action for validation — avoids stochastic + // Greedy (deterministic) action for validation -- avoids stochastic // noise that would make the early-stopping signal unreliable. let action_idx = match ppo.greedy_action(&state) { Ok(action) => action.to_int(), @@ -715,6 +767,9 @@ fn main() -> Result<()> { info!(" Tx cost: {:.1} bps commission + {:.1} tick spread (tick_size={:.4})", args.tx_cost_bps, args.spread_ticks, args.tick_size); info!(" Patience: {}", args.patience); + if let Some(ref hp_path) = args.hyperopt_params { + info!(" Hyperopt params: {}", hp_path.display()); + } // 1. Load all OHLCV bars from DBN files info!("Step 1/5: Loading OHLCV bars from DBN files..."); @@ -740,11 +795,7 @@ fn main() -> Result<()> { // Since features skip the warmup period, we need bars aligned to features. // Features start at bar index warmup_offset (typically 50). let warmup_offset = bars.len().saturating_sub(all_features.len()); - let aligned_bars = if warmup_offset < bars.len() { - &bars[warmup_offset..] - } else { - &bars - }; + let aligned_bars = bars.get(warmup_offset..).unwrap_or(&bars); // 3. Generate walk-forward windows info!("Step 3/5: Generating walk-forward windows..."); @@ -788,7 +839,7 @@ fn main() -> Result<()> { let train_feat = match extract_ml_features(&window.train) { Ok(f) => f, Err(e) => { - warn!(" Fold {} — train feature extraction failed: {}", window.fold, e); + warn!(" Fold {} -- train feature extraction failed: {}", window.fold, e); continue; } }; @@ -796,13 +847,13 @@ fn main() -> Result<()> { let val_feat = match extract_ml_features(&window.val) { Ok(f) => f, Err(e) => { - warn!(" Fold {} — val feature extraction failed: {}", window.fold, e); + warn!(" Fold {} -- val feature extraction failed: {}", window.fold, e); continue; } }; if train_feat.is_empty() || val_feat.is_empty() { - warn!(" Fold {} — empty features, skipping", window.fold); + warn!(" Fold {} -- empty features, skipping", window.fold); continue; } @@ -823,21 +874,14 @@ fn main() -> Result<()> { // Aligned bars for features (train features skip warmup period of train bars) let train_warmup = window.train.len().saturating_sub(train_norm.len()); - let train_bars_aligned = if train_warmup < window.train.len() { - &window.train[train_warmup..] - } else { - &window.train - }; + let train_bars_aligned = window.train.get(train_warmup..).unwrap_or(&window.train); let val_warmup = window.val.len().saturating_sub(val_norm.len()); - let val_bars_aligned = if val_warmup < window.val.len() { - &window.val[val_warmup..] - } else { - &window.val - }; + let val_bars_aligned = window.val.get(val_warmup..).unwrap_or(&window.val); // Train DQN if train_dqn { + let hp = load_hyperopt_params(&args.hyperopt_params, "dqn"); match train_dqn_fold( window.fold, &train_norm, @@ -846,6 +890,7 @@ fn main() -> Result<()> { val_bars_aligned, &args, &args.output_dir, + &hp, ) { Ok(best_loss) => { dqn_results.push((window.fold, best_loss)); @@ -858,6 +903,7 @@ fn main() -> Result<()> { // Train PPO if train_ppo { + let hp = load_hyperopt_params(&args.hyperopt_params, "ppo"); match train_ppo_fold( window.fold, &train_norm, @@ -866,6 +912,7 @@ fn main() -> Result<()> { val_bars_aligned, &args, &args.output_dir, + &hp, ) { Ok(best_loss) => { ppo_results.push((window.fold, best_loss)); diff --git a/crates/ml/examples/train_baseline_supervised.rs b/crates/ml/examples/train_baseline_supervised.rs index 99e1b2e91..9b435a58e 100644 --- a/crates/ml/examples/train_baseline_supervised.rs +++ b/crates/ml/examples/train_baseline_supervised.rs @@ -31,6 +31,7 @@ use std::time::Instant; use anyhow::{Context, Result}; use candle_core::{Device, Tensor}; use clap::Parser; +use serde_json::Value; use tracing::{error, info, warn}; use ml::features::extraction::extract_ml_features; @@ -42,7 +43,7 @@ use ml::walk_forward::{generate_walk_forward_windows, NormStats, WalkForwardConf mod baseline_common; use baseline_common::{load_all_bars, spread_cost_bps}; -// Model adapter imports — using verified paths from existing examples +// Model adapter imports -- using verified paths from existing examples use ml::diffusion::{DiffusionConfig, DiffusionTrainableAdapter}; use ml::kan::{KANConfig, KANTrainableAdapter}; use ml::liquid::{CfCTrainConfig, DeviceConfig, LiquidTrainableAdapter}; @@ -132,6 +133,10 @@ struct Args { /// Typical bid-ask spread in ticks #[arg(long, default_value_t = 1.0)] spread_ticks: f64, + + /// Optional path to hyperopt results JSON -- overrides matching config fields + #[arg(long)] + hyperopt_params: Option, } // --------------------------------------------------------------------------- @@ -154,6 +159,53 @@ fn validate_model(name: &str) -> Result<()> { } } +// --------------------------------------------------------------------------- +// Hyperopt parameter loading +// --------------------------------------------------------------------------- + +/// Load best_params from a hyperopt results JSON file. +/// +/// Expected format: `{ "model_key": { "best_params": { ... }, ... } }` +/// Returns `None` if the file doesn't exist or can't be parsed. +fn load_hyperopt_params(hp_path: &Option, model_key: &str) -> Option { + let file_path = hp_path.as_ref()?; + if !file_path.exists() { + info!("Hyperopt params file not found: {}, using defaults", file_path.display()); + return None; + } + let contents = match std::fs::read_to_string(file_path) { + Ok(c) => c, + Err(e) => { + warn!("Failed to read hyperopt params {}: {}", file_path.display(), e); + return None; + } + }; + let json: Value = match serde_json::from_str(&contents) { + Ok(v) => v, + Err(e) => { + warn!("Failed to parse hyperopt params JSON: {}", e); + return None; + } + }; + let params = json.get(model_key) + .and_then(|m| m.get("best_params")) + .cloned(); + if params.is_some() { + info!("Loaded hyperopt params for '{}' from {}", model_key, file_path.display()); + } else { + warn!("No best_params found for '{}' in {}", model_key, file_path.display()); + } + params +} + +fn hp_f64(params: &Option, key: &str) -> Option { + params.as_ref()?.get(key)?.as_f64() +} + +fn hp_usize(params: &Option, key: &str) -> Option { + params.as_ref()?.get(key)?.as_u64().map(|v| v as usize) +} + // --------------------------------------------------------------------------- // Model Factory // --------------------------------------------------------------------------- @@ -167,23 +219,25 @@ fn create_model( feature_dim: usize, learning_rate: f64, device: &Device, + hp: &Option, ) -> Result> { + let lr = hp_f64(hp, "learning_rate").unwrap_or(learning_rate); match name { "tft" => { let config = TFTConfig { input_dim: feature_dim, - hidden_dim: 128, - num_heads: 4, + hidden_dim: hp_usize(hp, "hidden_size").unwrap_or(128), + num_heads: hp_usize(hp, "num_heads").unwrap_or(4), num_layers: 2, num_quantiles: 3, num_static_features: 0, - dropout_rate: 0.1, + dropout_rate: hp_f64(hp, "dropout").unwrap_or(0.1), ..TFTConfig::default() }; let mut adapter = TrainableTFT::new(config) .map_err(|e| anyhow::anyhow!("Failed to create TFT: {}", e))?; adapter - .set_learning_rate(learning_rate) + .set_learning_rate(lr) .map_err(|e| anyhow::anyhow!("Failed to set TFT learning rate: {}", e))?; Ok(Box::new(adapter)) } @@ -198,17 +252,17 @@ fn create_model( let mut adapter = Mamba2SSM::new(config, device) .map_err(|e| anyhow::anyhow!("Failed to create Mamba2: {}", e))?; adapter - .set_learning_rate(learning_rate) + .set_learning_rate(lr) .map_err(|e| anyhow::anyhow!("Failed to set Mamba2 learning rate: {}", e))?; Ok(Box::new(adapter)) } "liquid" => { let config = CfCTrainConfig { input_size: feature_dim, - hidden_size: 128, + hidden_size: hp_usize(hp, "hidden_size").unwrap_or(128), output_size: 1, backbone_hidden_sizes: vec![128, 64], - learning_rate, + learning_rate: lr, device: DeviceConfig::Auto, ..CfCTrainConfig::default() }; @@ -219,45 +273,45 @@ fn create_model( "tggn" => { let config = TGGNConfig { node_dim: feature_dim, - hidden_dim: 32, - num_layers: 2, + hidden_dim: hp_usize(hp, "hidden_dim").unwrap_or(32), + num_layers: hp_usize(hp, "num_layers").unwrap_or(2), max_nodes: 64, max_edges: 128, edge_dim: 4, - temporal_decay: 0.99, + temporal_decay: hp_f64(hp, "temporal_decay").unwrap_or(0.99), update_frequency_ns: 1_000_000, use_simd: false, }; let mut adapter = TGGNTrainableAdapter::new(config, device) .map_err(|e| anyhow::anyhow!("Failed to create TGGN: {}", e))?; adapter - .set_learning_rate(learning_rate) + .set_learning_rate(lr) .map_err(|e| anyhow::anyhow!("Failed to set TGGN learning rate: {}", e))?; Ok(Box::new(adapter)) } "tlob" => { let config = TLOBAdapterConfig { - d_model: 128, - num_heads: 4, - num_layers: 2, - seq_len: 1, + d_model: hp_usize(hp, "d_model").unwrap_or(128), + num_heads: hp_usize(hp, "num_heads").unwrap_or(4), + num_layers: hp_usize(hp, "num_layers").unwrap_or(2), + seq_len: hp_usize(hp, "seq_len").unwrap_or(1), feature_dim, }; let mut adapter = TLOBTrainableAdapter::new(config, device) .map_err(|e| anyhow::anyhow!("Failed to create TLOB: {}", e))?; adapter - .set_learning_rate(learning_rate) + .set_learning_rate(lr) .map_err(|e| anyhow::anyhow!("Failed to set TLOB learning rate: {}", e))?; Ok(Box::new(adapter)) } "kan" => { let config = KANConfig { - grid_size: 5, - spline_order: 4, + grid_size: hp_usize(hp, "grid_size").unwrap_or(5), + spline_order: hp_usize(hp, "spline_order").unwrap_or(4), layer_widths: vec![feature_dim, 32, 16, 1], - learning_rate, - weight_decay: 1e-4, - grad_clip: 1.0, + learning_rate: lr, + weight_decay: hp_f64(hp, "weight_decay").unwrap_or(1e-4), + grad_clip: hp_f64(hp, "grad_clip").unwrap_or(1.0), }; let adapter = KANTrainableAdapter::new(config, device) .map_err(|e| anyhow::anyhow!("Failed to create KAN: {}", e))?; @@ -266,20 +320,20 @@ fn create_model( "xlstm" => { let config = XLSTMConfig { input_dim: feature_dim, - hidden_dim: 128, + hidden_dim: hp_usize(hp, "hidden_dim").unwrap_or(128), ..XLSTMConfig::default() }; let mut adapter = XLSTMTrainableAdapter::new(config, device) .map_err(|e| anyhow::anyhow!("Failed to create xLSTM: {}", e))?; adapter - .set_learning_rate(learning_rate) + .set_learning_rate(lr) .map_err(|e| anyhow::anyhow!("Failed to set xLSTM learning rate: {}", e))?; Ok(Box::new(adapter)) } "diffusion" => { let config = DiffusionConfig { feature_dim, - hidden_dim: 128, + hidden_dim: hp_usize(hp, "hidden_dim").unwrap_or(128), ..DiffusionConfig::default() }; let adapter = DiffusionTrainableAdapter::new(config, device.clone()) @@ -463,6 +517,7 @@ fn train_fold( args: &Args, device: &Device, output_dir: &Path, + hp: &Option, ) -> Result { info!( "[{}] Fold {} -- {} train, {} val pairs", @@ -472,7 +527,7 @@ fn train_fold( val_pairs.len() ); - let mut adapter = create_model(model_name, args.feature_dim, args.learning_rate, device)?; + let mut adapter = create_model(model_name, args.feature_dim, args.learning_rate, device, hp)?; let mut best_val_loss = f64::MAX; let mut epochs_without_improvement = 0_usize; @@ -603,6 +658,7 @@ fn main() -> Result<()> { // Train each model for model_name in &models_to_train { info!("=== Training model: {} ===", model_name); + let hp = load_hyperopt_params(&args.hyperopt_params, model_name); let model_output = args.output_dir.join(model_name); std::fs::create_dir_all(&model_output) .with_context(|| format!("Failed to create {}", model_output.display()))?; @@ -644,6 +700,7 @@ fn main() -> Result<()> { &args, &device, &model_output, + &hp, ) { Ok(best_val) => fold_results.push((fold, best_val)), Err(e) => error!("[{}] Fold {} failed: {}", model_name, fold, e), diff --git a/crates/ml/src/hyperopt/adapters/dbn_loader.rs b/crates/ml/src/hyperopt/adapters/dbn_loader.rs new file mode 100644 index 000000000..22e875983 --- /dev/null +++ b/crates/ml/src/hyperopt/adapters/dbn_loader.rs @@ -0,0 +1,97 @@ +//! Shared DBN file loading utilities for hyperopt adapters. +//! +//! Provides recursive .dbn/.dbn.zst file discovery and OHLCV bar decoding. +//! Used by TFT, Mamba2, and DQN hyperopt adapters to load training data +//! from Databento files. + +use std::path::{Path, PathBuf}; + +use tracing::info; + +use crate::features::extraction::OHLCVBar; + +/// Recursively collect all .dbn and .dbn.zst files from a directory. +pub fn collect_dbn_files(dir: &Path) -> Vec { + let mut files = Vec::new(); + if let Ok(entries) = std::fs::read_dir(dir) { + for entry in entries.flatten() { + let path = entry.path(); + if path.is_dir() { + files.extend(collect_dbn_files(&path)); + } else if path.extension().and_then(|s| s.to_str()) == Some("dbn") + || path.to_string_lossy().ends_with(".dbn.zst") + { + files.push(path); + } else { + // Skip non-DBN files + } + } + } + files.sort(); + files +} + +/// Decode OHLCV bars from a DBN decoder stream. +pub fn decode_ohlcv_bars( + decoder: &mut dbn::decode::dbn::Decoder, +) -> anyhow::Result> { + use dbn::decode::DecodeRecordRef; + use dbn::OhlcvMsg; + + let mut bars = Vec::new(); + while let Some(record_ref) = decoder + .decode_record_ref() + .map_err(|e| anyhow::anyhow!("Failed to decode record: {}", e))? + { + if let Some(ohlcv) = record_ref.get::() { + bars.push(OHLCVBar { + timestamp: chrono::DateTime::from_timestamp_nanos(ohlcv.hd.ts_event as i64), + open: ohlcv.open as f64 / 1e9, + high: ohlcv.high as f64 / 1e9, + low: ohlcv.low as f64 / 1e9, + close: ohlcv.close as f64 / 1e9, + volume: ohlcv.volume as f64, + }); + } + } + Ok(bars) +} + +/// Load all OHLCV bars from DBN files in a directory, sorted chronologically. +pub fn load_bars_from_dbn_dir(dir: &Path) -> anyhow::Result> { + use dbn::decode::dbn::Decoder as DbnDecoder; + + let dbn_files = collect_dbn_files(dir); + if dbn_files.is_empty() { + anyhow::bail!("No .dbn or .dbn.zst files found in {}", dir.display()); + } + + info!("Found {} DBN files in {}", dbn_files.len(), dir.display()); + + let mut all_bars = Vec::new(); + for dbn_file in &dbn_files { + let is_zstd = dbn_file.to_string_lossy().ends_with(".dbn.zst"); + let bars = if is_zstd { + let mut decoder = DbnDecoder::from_zstd_file(dbn_file) + .map_err(|e| anyhow::anyhow!("Failed to open {}: {}", dbn_file.display(), e))?; + decode_ohlcv_bars(&mut decoder)? + } else { + let file = std::fs::File::open(dbn_file) + .map_err(|e| anyhow::anyhow!("Failed to open {}: {}", dbn_file.display(), e))?; + let buf = std::io::BufReader::new(file); + let mut decoder = DbnDecoder::new(buf) + .map_err(|e| anyhow::anyhow!("Failed to decode {}: {}", dbn_file.display(), e))?; + decode_ohlcv_bars(&mut decoder)? + }; + info!(" {} -> {} bars", dbn_file.display(), bars.len()); + all_bars.extend(bars); + } + + all_bars.sort_by_key(|b| b.timestamp); + info!( + "Loaded {} bars total from {} DBN files", + all_bars.len(), + dbn_files.len() + ); + Ok(all_bars) +} diff --git a/crates/ml/src/hyperopt/adapters/dqn.rs b/crates/ml/src/hyperopt/adapters/dqn.rs index d0f0a1f27..9b90de6bc 100644 --- a/crates/ml/src/hyperopt/adapters/dqn.rs +++ b/crates/ml/src/hyperopt/adapters/dqn.rs @@ -923,52 +923,8 @@ pub struct DQNTrainer { } /// Decode OHLCV bars from an already-opened DBN decoder. -fn decode_ohlcv_bars( - decoder: &mut dbn::decode::dbn::Decoder, -) -> anyhow::Result> { - use chrono::DateTime; - use dbn::decode::DecodeRecordRef; - use dbn::OhlcvMsg; - - let mut bars = Vec::new(); - while let Some(record_ref) = decoder - .decode_record_ref() - .map_err(|e| anyhow::anyhow!("Failed to decode record: {}", e))? - { - if let Some(ohlcv) = record_ref.get::() { - bars.push(crate::features::extraction::OHLCVBar { - timestamp: DateTime::from_timestamp_nanos(ohlcv.hd.ts_event as i64), - open: ohlcv.open as f64 / 1e9, - high: ohlcv.high as f64 / 1e9, - low: ohlcv.low as f64 / 1e9, - close: ohlcv.close as f64 / 1e9, - volume: ohlcv.volume as f64, - }); - } - } - Ok(bars) -} - -/// Recursively collect all .dbn and .dbn.zst files from a directory and its subdirectories. -fn collect_dbn_files_recursive(dir: &std::path::Path) -> Vec { - let mut files = Vec::new(); - if let Ok(entries) = std::fs::read_dir(dir) { - for entry in entries.flatten() { - let path = entry.path(); - if path.is_dir() { - files.extend(collect_dbn_files_recursive(&path)); - } else if path.extension().and_then(|s| s.to_str()) == Some("dbn") - || path.to_string_lossy().ends_with(".dbn.zst") - { - files.push(path); - } else { - // Skip non-DBN files - } - } - } - files.sort(); - files -} +// Re-use shared DBN loading utilities +use super::dbn_loader::{collect_dbn_files, decode_ohlcv_bars}; impl DQNTrainer { /// Create a new DQN trainer @@ -1340,7 +1296,7 @@ impl DQNTrainer { fn load_from_dbn(&self) -> anyhow::Result> { use dbn::decode::DbnDecoder; - let dbn_files = collect_dbn_files_recursive(std::path::Path::new(&self.dbn_data_dir)); + let dbn_files = collect_dbn_files(std::path::Path::new(&self.dbn_data_dir)); if dbn_files.is_empty() { return Err(anyhow::anyhow!("No DBN files found in directory")); @@ -1389,7 +1345,7 @@ impl DQNTrainer { return None; } - let mbp10_files = collect_dbn_files_recursive(&mbp10_dir); + let mbp10_files = collect_dbn_files(&mbp10_dir); if mbp10_files.is_empty() { info!("No MBP10 .dbn files found in {}", mbp10_dir.display()); return None; diff --git a/crates/ml/src/hyperopt/adapters/mamba2.rs b/crates/ml/src/hyperopt/adapters/mamba2.rs index dd212778e..622bb4347 100644 --- a/crates/ml/src/hyperopt/adapters/mamba2.rs +++ b/crates/ml/src/hyperopt/adapters/mamba2.rs @@ -39,12 +39,7 @@ use std::io::Write as IoWrite; use std::path::PathBuf; use tracing::{info, warn}; -use arrow::array::{Array, Float64Array, PrimitiveArray, UInt64Array}; -use arrow::datatypes::TimestampNanosecondType; -use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder; -use std::fs::File; - -use crate::features::{extract_ml_features, FeatureConfig, OHLCVBar}; +use crate::features::{extract_ml_features, FeatureConfig}; use crate::hyperopt::paths::TrainingPaths; use crate::hyperopt::traits::{HyperparameterOptimizable, ParameterSpace}; use crate::mamba::{Mamba2Config, Mamba2SSM, OptimizerType}; @@ -218,18 +213,10 @@ pub struct Mamba2Metrics { /// /// ## Configuration /// -/// - **Parquet file**: Market data source (OHLCV bars) +/// - **Data dir**: Directory with .dbn.zst files (Databento OHLCV data) /// - **Epochs**: Number of training epochs per trial /// - **Device**: CUDA GPU (falls back to CPU if unavailable) -/// - **Features**: 51 features (WAVE 10: Reduced from 54→51 after Proxy OFI removal) -/// -/// ## Fixed Architecture -/// -/// The following parameters are fixed for consistency: -/// - `d_model`: 54 (WAVE 10: 51 market features + 3 portfolio features) -/// - `d_state`: 16 -/// - `num_layers`: 6 -/// - `sequence_length`: 60 +/// - **Features**: 51 features (WAVE 10: Reduced from 54->51 after Proxy OFI removal) /// /// ## Optimized Hyperparameters /// @@ -240,30 +227,21 @@ pub struct Mamba2Metrics { /// - Weight decay #[derive(Debug)] pub struct Mamba2Trainer { - parquet_file: PathBuf, + data_dir: PathBuf, epochs: usize, device: Device, feature_config: FeatureConfig, d_model: usize, train_split: f64, - /// Target normalization parameters (set after data loading) target_min: Option, target_max: Option, - /// Minimum batch size (for GPU memory constraints) batch_size_min: f64, - /// Maximum batch size (for GPU memory constraints) batch_size_max: f64, - /// Enable async data loading (prefetch while GPU trains) async_loading: bool, - /// Number of batches to prefetch (2-3 recommended) prefetch_count: usize, - /// Training paths configuration (replaces hardcoded checkpoint_dir) training_paths: TrainingPaths, - /// Early stopping patience (epochs without improvement before stopping) early_stopping_patience: usize, - /// Early stopping minimum epochs (minimum epochs before early stopping can trigger) early_stopping_min_epochs: usize, - /// Trial counter for hyperopt (tracks trial number across multiple trials) trial_counter: usize, } @@ -272,48 +250,36 @@ impl Mamba2Trainer { /// /// # Arguments /// - /// * `parquet_file` - Path to Parquet file with market data + /// * `data_dir` - Directory containing .dbn.zst files with OHLCV data /// * `epochs` - Number of training epochs per trial - /// - /// # Returns - /// - /// Configured trainer ready for optimization - /// - /// # Errors - /// - /// Returns error if: - /// - Parquet file doesn't exist - /// - CUDA device initialization fails (falls back to CPU) - pub fn new>(parquet_file: P, epochs: usize) -> Result { - let parquet_file = parquet_file.into(); + pub fn new>(data_dir: P, epochs: usize) -> Result { + let data_dir = data_dir.into(); - if !parquet_file.exists() { + if !data_dir.exists() { return Err(MLError::ConfigError { - reason: format!("Parquet file not found: {}", parquet_file.display()), + reason: format!("Data directory not found: {}", data_dir.display()), } .into()); } - // Initialize device (CUDA preferred, CPU fallback) let device = Device::new_cuda(0).unwrap_or_else(|e| { warn!("CUDA unavailable ({}), falling back to CPU", e); Device::Cpu }); - // Use WAVE 10 feature configuration (51 market + 3 portfolio = 54) let feature_config = FeatureConfig::wave_d(); let d_model = feature_config.feature_count(); info!("MAMBA-2 Trainer initialized:"); info!(" Device: {:?}", device); + info!(" Data: {}", data_dir.display()); info!(" Features: {} (WAVE 10: 51 market + 3 portfolio)", d_model); info!(" Epochs per trial: {}", epochs); - // Use temporary default paths - should be replaced with with_training_paths() let training_paths = TrainingPaths::new("/tmp/ml_training", "mamba2", "default"); Ok(Self { - parquet_file, + data_dir, epochs, device, feature_config, @@ -321,14 +287,14 @@ impl Mamba2Trainer { train_split: 0.8, target_min: None, target_max: None, - batch_size_min: 4.0, // Default minimum - batch_size_max: 96.0, // Default maximum (safe for RTX A4000 16GB) - async_loading: true, // Enable async loading by default - prefetch_count: 3, // Prefetch 3 batches (good balance) + batch_size_min: 4.0, + batch_size_max: 96.0, + async_loading: true, + prefetch_count: 3, training_paths, - early_stopping_patience: 5, // Default: 5 epochs (hyperopt optimized) - early_stopping_min_epochs: 5, // Default: 5 epochs (hyperopt optimized) - trial_counter: 0, // Start at trial 0 + early_stopping_patience: 5, + early_stopping_min_epochs: 5, + trial_counter: 0, }) } @@ -489,85 +455,12 @@ impl Mamba2Trainer { fn load_and_prepare_data( &self, seq_len: usize, - _stride: usize, // P2 parameter - for future use with overlapping sequences + _stride: usize, ) -> Result<(Vec<(Tensor, Tensor)>, Vec<(Tensor, Tensor)>, f64, f64)> { - // Open Parquet file - let file = File::open(&self.parquet_file).with_context(|| { - format!( - "Failed to open Parquet file: {}", - self.parquet_file.display() - ) - })?; + // Load bars from DBN files + let all_ohlcv_bars = super::dbn_loader::load_bars_from_dbn_dir(&self.data_dir) + .context("Failed to load DBN data")?; - let builder = ParquetRecordBatchReaderBuilder::try_new(file) - .context("Failed to create Parquet reader")?; - - let reader = builder.build().context("Failed to build Parquet reader")?; - - // Read all OHLCV bars - let mut all_ohlcv_bars = Vec::new(); - - for batch_result in reader { - let batch = batch_result.context("Failed to read record batch")?; - - let timestamps = batch - .column(9) - .as_any() - .downcast_ref::>() - .context("Failed to downcast timestamp column")?; - - let opens = batch - .column(3) - .as_any() - .downcast_ref::() - .context("Failed to downcast open column")?; - - let highs = batch - .column(4) - .as_any() - .downcast_ref::() - .context("Failed to downcast high column")?; - - let lows = batch - .column(5) - .as_any() - .downcast_ref::() - .context("Failed to downcast low column")?; - - let closes = batch - .column(6) - .as_any() - .downcast_ref::() - .context("Failed to downcast close column")?; - - let volumes = batch - .column(7) - .as_any() - .downcast_ref::() - .context("Failed to downcast volume column")?; - - for i in 0..batch.num_rows() { - let timestamp_ns = timestamps.value(i); - let timestamp = chrono::DateTime::from_timestamp( - (timestamp_ns / 1_000_000_000) as i64, - (timestamp_ns.rem_euclid(1_000_000_000)) as u32, - ) - .unwrap_or_else(|| chrono::Utc::now()); - - let bar = OHLCVBar { - timestamp, - open: opens.value(i), - high: highs.value(i), - low: lows.value(i), - close: closes.value(i), - volume: volumes.value(i) as f64, - }; - - all_ohlcv_bars.push(bar); - } - } - - // P1 FIX: Validate minimum rows before processing if all_ohlcv_bars.len() < seq_len + 1 { return Err(MLError::ModelError(format!( "Insufficient data: {} bars (need at least {} for seq_len={})", @@ -584,7 +477,7 @@ impl Mamba2Trainer { if features.is_empty() { return Err( - MLError::ModelError("No features extracted from Parquet data".to_owned()).into(), + MLError::ModelError("No features extracted from data".to_owned()).into(), ); } diff --git a/crates/ml/src/hyperopt/adapters/mod.rs b/crates/ml/src/hyperopt/adapters/mod.rs index d24641a3d..48e1e9497 100644 --- a/crates/ml/src/hyperopt/adapters/mod.rs +++ b/crates/ml/src/hyperopt/adapters/mod.rs @@ -48,6 +48,9 @@ //! } //! ``` +// Shared utilities +pub mod dbn_loader; + // Active adapters (production-ready) pub mod async_data_loader; pub mod continuous_ppo; diff --git a/crates/ml/src/hyperopt/adapters/tft.rs b/crates/ml/src/hyperopt/adapters/tft.rs index b317d8a73..2843ebd5b 100644 --- a/crates/ml/src/hyperopt/adapters/tft.rs +++ b/crates/ml/src/hyperopt/adapters/tft.rs @@ -179,18 +179,10 @@ pub struct TFTMetrics { /// /// ## Configuration /// -/// - **Parquet file**: Market data source (OHLCV bars) +/// - **Data dir**: Directory with .dbn.zst files (Databento OHLCV data) /// - **Epochs**: Number of training epochs per trial /// - **Device**: CUDA GPU (falls back to CPU if unavailable) -/// - **Features**: 51 features (WAVE 10: Reduced from 54→51 after Proxy OFI removal) -/// -/// ## Fixed Architecture -/// -/// The following parameters are fixed for consistency: -/// - `input_size`: 54 (WAVE 10: 51 market features + 3 portfolio features) -/// - `sequence_length`: 60 -/// - `prediction_horizon`: 10 -/// - `num_quantiles`: 3 (0.1, 0.5, 0.9) +/// - **Features**: 51 features (WAVE 10: Reduced from 54->51 after Proxy OFI removal) /// /// ## Optimized Hyperparameters /// @@ -202,14 +194,11 @@ pub struct TFTMetrics { /// - Dropout #[derive(Debug)] pub struct TFTTrainer { - parquet_file: PathBuf, + data_dir: PathBuf, epochs: usize, device: Device, training_paths: TrainingPaths, - /// Early stopping patience (epochs without improvement before stopping) - /// NOTE: Currently stored but not used - TFT trainer has hardcoded patience early_stopping_patience: usize, - /// Trial counter for hyperopt (tracks trial number across multiple trials) trial_counter: usize, } @@ -218,49 +207,39 @@ impl TFTTrainer { /// /// # Arguments /// - /// * `parquet_file` - Path to Parquet file with market data + /// * `data_dir` - Directory containing .dbn.zst files with OHLCV data /// * `epochs` - Number of training epochs per trial - /// - /// # Returns - /// - /// Configured trainer ready for optimization - /// - /// # Errors - /// - /// Returns error if: - /// - Parquet file doesn't exist - /// - CUDA device initialization fails (falls back to CPU) - pub fn new>(parquet_file: P, epochs: usize) -> Result { - let parquet_file = parquet_file.into(); + pub fn new>(data_dir: P, epochs: usize) -> Result { + let data_dir = data_dir.into(); - if !parquet_file.exists() { + if !data_dir.exists() { return Err(MLError::ConfigError { - reason: format!("Parquet file not found: {}", parquet_file.display()), + reason: format!("Data directory not found: {}", data_dir.display()), } .into()); } - // Initialize device (CUDA preferred, CPU fallback) let device = Device::new_cuda(0).unwrap_or_else(|e| { warn!("CUDA unavailable ({}), falling back to CPU", e); Device::Cpu }); info!( - "TFT Trainer initialized: Device={:?}, Epochs per trial={}", - device, epochs + "TFT Trainer initialized: Device={:?}, Data={}, Epochs per trial={}", + device, + data_dir.display(), + epochs ); - // Use temporary default paths - should be replaced with with_training_paths() let training_paths = TrainingPaths::new("/tmp/ml_training", "tft", "default"); Ok(Self { - parquet_file, + data_dir, epochs, device, training_paths, - early_stopping_patience: 10, // Default: 10 epochs (hyperopt optimized) - trial_counter: 0, // Start at trial 0 + early_stopping_patience: 10, + trial_counter: 0, }) } @@ -349,10 +328,8 @@ impl HyperparameterOptimizable for TFTTrainer { #[allow(clippy::unwrap_in_result)] fn train_with_params(&mut self, params: Self::Params) -> Result { - // START: Add trial timing let trial_start = std::time::Instant::now(); - // Get current trial number and increment for next trial let current_trial = self.trial_counter; self.trial_counter += 1; @@ -365,7 +342,6 @@ impl HyperparameterOptimizable for TFTTrainer { params.dropout ); - // Log trial start (ensure directory exists first) std::fs::create_dir_all(self.training_paths.logs_dir()).ok(); write_training_log_tft( &self.training_paths.logs_dir(), @@ -373,67 +349,46 @@ impl HyperparameterOptimizable for TFTTrainer { ) .ok(); - // Validate num_heads divides hidden_size if params.hidden_size % params.num_heads != 0 { warn!( "Hidden size {} not divisible by num_heads {}, adjusting", params.hidden_size, params.num_heads ); - // This shouldn't happen with our discrete parameter space, but handle it return Ok(TFTMetrics { - val_loss: 1000.0, // Penalty for invalid config + val_loss: 1000.0, train_loss: 1000.0, val_rmse: 1000.0, epochs_completed: 0, }); } - // Create directories self.training_paths.create_all().map_err(|e| { MLError::ModelError(format!("Failed to create training directories: {}", e)) })?; - // Create TFT trainer config with trial hyperparameters let trainer_config = TFTTrainerConfig { - // Training parameters epochs: self.epochs, learning_rate: params.learning_rate, batch_size: params.batch_size, auto_batch_size: false, - - // Architecture hidden_dim: params.hidden_size, num_attention_heads: params.num_heads, dropout_rate: params.dropout, lstm_layers: 2, - - // Quantiles for probabilistic forecasting quantiles: vec![0.1, 0.5, 0.9], - - // Forecasting parameters lookback_window: 60, forecast_horizon: 10, - - // Performance use_gpu: self.device.is_cuda(), - - // INT8 and QAT disabled for hyperopt (speed optimization) use_int8_quantization: false, use_qat: false, qat_calibration_batches: 0, qat_warmup_epochs: 0, qat_cooldown_factor: 1.0, qat_min_batch_size: 4, - - // Gradient checkpointing disabled for speed use_gradient_checkpointing: false, - - // Validation settings - validation_frequency: 1, // Run validation every epoch for hyperopt + validation_frequency: 1, validation_batch_size: params.batch_size, max_validation_batches: None, - - // Checkpointing (use configured training paths) checkpoint_dir: self .training_paths .checkpoints_dir() @@ -441,20 +396,19 @@ impl HyperparameterOptimizable for TFTTrainer { .to_string(), }; - // Create trainer with minimal memory storage (not persisted for hyperopt) let checkpoint_storage = std::sync::Arc::new(crate::checkpoint::MemoryStorage::new()); let mut trainer = RealTFTTrainer::new(trainer_config, checkpoint_storage) .map_err(|e| MLError::ModelError(format!("Failed to create TFT trainer: {}", e)))?; - // Run training on Parquet data + // Load bars from DBN files + let bars = super::dbn_loader::load_bars_from_dbn_dir(&self.data_dir) + .map_err(|e| MLError::ModelError(format!("Failed to load DBN data: {}", e)))?; + let runtime = tokio::runtime::Runtime::new() .map_err(|e| MLError::ModelError(format!("Failed to create async runtime: {}", e)))?; let training_metrics = runtime - .block_on(trainer.train_from_parquet( - self.parquet_file.to_str() - .ok_or_else(|| MLError::ModelError("Parquet file path contains invalid UTF-8".to_owned()))?, - )) + .block_on(trainer.train_from_bars(&bars)) .map_err(|e| MLError::TrainingError(format!("TFT training failed: {}", e)))?; // Map from crate::trainers::tft::TrainingMetrics to TFTMetrics diff --git a/crates/ml/src/trainers/tft_parquet.rs b/crates/ml/src/trainers/tft_parquet.rs index b0251776f..897358889 100644 --- a/crates/ml/src/trainers/tft_parquet.rs +++ b/crates/ml/src/trainers/tft_parquet.rs @@ -41,8 +41,23 @@ impl TFTTrainer { // Load market data from Parquet file let training_data = self.load_training_data_from_parquet(parquet_path).await?; - // Note: Normalization params are stored in self.target_mean and self.target_std - // by load_training_data_from_parquet() for later denormalization + self.train_from_samples(training_data).await + } + + /// Train TFT on pre-loaded OHLCV bars (from DBN or any other source). + pub async fn train_from_bars(&mut self, bars: &[OHLCVBar]) -> MLResult { + info!("Starting TFT training from {} OHLCV bars", bars.len()); + + let training_data = self.prepare_training_data(bars)?; + + self.train_from_samples(training_data).await + } + + /// Shared training loop: split data, create loaders, train with OOM retry. + async fn train_from_samples( + &mut self, + training_data: Vec<(Array1, Array2, Array2, Array1)>, + ) -> MLResult { info!( "Target normalization applied: mean={:.2}, std={:.2}", self.target_mean.unwrap_or(0.0), @@ -116,12 +131,12 @@ impl TFTTrainer { Ok(metrics) => { if oom_retry_count > 0 { info!( - "✅ Training completed successfully after {} OOM retries (final batch_size={})", + "Training completed successfully after {} OOM retries (final batch_size={})", oom_retry_count, current_batch_size ); } else { info!( - "✅ Training completed successfully (batch_size={})", + "Training completed successfully (batch_size={})", current_batch_size ); } @@ -143,7 +158,7 @@ impl TFTTrainer { } tracing::warn!( - "⚠️ OOM detected (attempt {}/{}), reducing batch_size: {} → {}", + "OOM detected (attempt {}/{}), reducing batch_size: {} -> {}", oom_retry_count, MAX_OOM_RETRIES, old_batch_size, @@ -184,11 +199,9 @@ impl TFTTrainer { info!("Loading Parquet file: {}", parquet_path); - // Open Parquet file let file = File::open(parquet_path) .map_err(|e| MLError::ModelError(format!("Failed to open Parquet file: {}", e)))?; - // Create Parquet reader let builder = ParquetRecordBatchReaderBuilder::try_new(file) .map_err(|e| MLError::ModelError(format!("Failed to create Parquet reader: {}", e)))?; @@ -196,17 +209,12 @@ impl TFTTrainer { .build() .map_err(|e| MLError::ModelError(format!("Failed to build Parquet reader: {}", e)))?; - // Read all batches (lazy loading in chunks) let mut all_ohlcv_bars = Vec::new(); for batch_result in reader { let batch: RecordBatch = batch_result .map_err(|e| MLError::ModelError(format!("Failed to read record batch: {}", e)))?; - // Extract columns by name (schema-agnostic approach) - // Required columns: timestamp_ns (or ts_event), open, high, low, close, volume - - // Try timestamp_ns first (our schema), fallback to ts_event (Databento schema) let timestamp_col = batch .column_by_name("timestamp_ns") .or_else(|| batch.column_by_name("ts_event")) @@ -227,7 +235,6 @@ impl TFTTrainer { ) ))?; - // Extract OHLCV columns by name let opens = batch .column_by_name("open") .ok_or_else(|| { @@ -236,7 +243,7 @@ impl TFTTrainer { .as_any() .downcast_ref::() .ok_or_else(|| { - MLError::InvalidInput(format!("Invalid 'open' column type. Expected Float64")) + MLError::InvalidInput("Invalid 'open' column type. Expected Float64".to_owned()) })?; let highs = batch @@ -247,7 +254,7 @@ impl TFTTrainer { .as_any() .downcast_ref::() .ok_or_else(|| { - MLError::InvalidInput(format!("Invalid 'high' column type. Expected Float64")) + MLError::InvalidInput("Invalid 'high' column type. Expected Float64".to_owned()) })?; let lows = batch @@ -258,7 +265,7 @@ impl TFTTrainer { .as_any() .downcast_ref::() .ok_or_else(|| { - MLError::InvalidInput(format!("Invalid 'low' column type. Expected Float64")) + MLError::InvalidInput("Invalid 'low' column type. Expected Float64".to_owned()) })?; let closes = batch @@ -269,7 +276,7 @@ impl TFTTrainer { .as_any() .downcast_ref::() .ok_or_else(|| { - MLError::InvalidInput(format!("Invalid 'close' column type. Expected Float64")) + MLError::InvalidInput("Invalid 'close' column type. Expected Float64".to_owned()) })?; let volumes = batch @@ -280,10 +287,9 @@ impl TFTTrainer { .as_any() .downcast_ref::() .ok_or_else(|| { - MLError::InvalidInput(format!("Invalid 'volume' column type. Expected UInt64")) + MLError::InvalidInput("Invalid 'volume' column type. Expected UInt64".to_owned()) })?; - // Convert to OHLCVBar structs for i in 0..batch.num_rows() { let timestamp_ns = timestamps.value(i); let timestamp = chrono::DateTime::from_timestamp_nanos(timestamp_ns); @@ -305,22 +311,29 @@ impl TFTTrainer { all_ohlcv_bars.len() ); - // Sort bars by timestamp (critical for rolling window feature extraction) - info!("Sorting bars chronologically by timestamp..."); - all_ohlcv_bars.sort_by_key(|bar| bar.timestamp); - info!("Bars sorted successfully"); + self.prepare_training_data(&all_ohlcv_bars) + } - // Extract features using full 54-feature extractor (core features) - info!("Extracting full 54-feature vectors from OHLCV bars (core features)..."); + /// Convert OHLCV bars into TFT training samples (data-source-agnostic). + fn prepare_training_data( + &mut self, + bars: &[OHLCVBar], + ) -> MLResult, Array2, Array2, Array1)>> { + let mut all_ohlcv_bars = bars.to_vec(); + + // Sort bars by timestamp (critical for rolling window feature extraction) + all_ohlcv_bars.sort_by_key(|bar| bar.timestamp); + + // Extract features using full 51-feature extractor (core features) let feature_vectors = self.extract_full_features(&all_ohlcv_bars)?; info!( - "Extracted {} feature vectors (54 dimensions each, core features)", - feature_vectors.len() + "Extracted {} feature vectors from {} bars", + feature_vectors.len(), + all_ohlcv_bars.len() ); - // Compute normalization parameters from close prices (CRITICAL: Prevents 1000-10000x loss) - info!("Computing target normalization parameters..."); + // Compute normalization parameters from close prices let all_closes: Vec = all_ohlcv_bars.iter().map(|b| b.close).collect(); if all_closes.is_empty() { @@ -337,7 +350,6 @@ impl TFTTrainer { / all_closes.len() as f64; let price_std = price_variance.sqrt(); - // Validate normalization params if price_std < 1e-8 { return Err(MLError::InvalidInput(format!( "Price std_dev too small ({:.2e}), data may be constant", @@ -346,28 +358,21 @@ impl TFTTrainer { } info!( - "Target normalization: mean={:.2}, std={:.2} (z-score will bring targets to ~[-3, 3] scale)", + "Target normalization: mean={:.2}, std={:.2}", price_mean, price_std ); - // Store normalization params in trainer for denormalization during evaluation self.target_mean = Some(price_mean); self.target_std = Some(price_std); - // Create TFT training samples with sliding windows - // Lookback: 60 bars, Horizon: 10 bars const LOOKBACK: usize = 60; const HORIZON: usize = 10; let mut tft_samples = Vec::new(); for i in 0..(feature_vectors.len().saturating_sub(LOOKBACK + HORIZON)) { - // Static features: First 5 features from current bar (symbol metadata) - // Features 0-4: symbol_id, exchange_id, asset_class, contract_month, tick_size let static_feats = Array1::from_vec(feature_vectors[i + LOOKBACK][0..5].to_vec()); - // Historical features: Past 60 bars × 39 unknown features - // Features 15-53: OHLCV, technical indicators, microstructure (54 - 5 static - 10 known = 39) let mut hist_data = Vec::new(); for j in i..(i + LOOKBACK) { let fv = &feature_vectors[j]; @@ -383,23 +388,17 @@ impl TFTTrainer { )) })?; - // Future features: Next 10 bars × 10 known features (time-based) - // Features 5-14: calendar features (hour, day, month, etc.) let mut fut_data = Vec::new(); for j in (i + LOOKBACK)..(i + LOOKBACK + HORIZON) { - // Use features 5-14 as "known future" (time-based, predictable) fut_data.extend_from_slice(&feature_vectors[j][5..15]); } let future_feats = Array2::from_shape_vec((HORIZON, 10), fut_data).map_err(|e| { MLError::ModelError(format!("Failed to create future features array: {}", e)) })?; - // Targets: Next 10 close prices (Z-SCORE NORMALIZED) let mut targets = Vec::new(); for j in (i + LOOKBACK)..(i + LOOKBACK + HORIZON) { let raw_price = all_ohlcv_bars[j + 50].close; - // Apply z-score normalization: (price - mean) / std - // This brings targets to ~[-3, 3] scale, matching log-return features (~[-0.1, 0.1]) let normalized = (raw_price - price_mean) / (price_std + 1e-8); targets.push(normalized); } diff --git a/docs/plans/2026-02-26-training-pipeline-design.md b/docs/plans/2026-02-26-training-pipeline-design.md new file mode 100644 index 000000000..57fda9d7b --- /dev/null +++ b/docs/plans/2026-02-26-training-pipeline-design.md @@ -0,0 +1,118 @@ +# ML Training Pipeline Design + +## Overview + +Manually-triggered GitLab CI pipeline for training the 10-model ML ensemble on 730 days of OHLCV-1m futures data. Runs on H100 GPU pool with configurable parallelism. + +## Trigger Variables + +| Variable | Default | Description | +|----------|---------|-------------| +| `SYMBOLS` | `ES.FUT` | Comma-separated symbols (ES.FUT, NQ.FUT, ZN.FUT, 6E.FUT) | +| `MODELS` | `all` | `all`, `rl`, `supervised`, or comma-separated model names | +| `PHASE` | `full` | `hyperopt`, `train`, `eval`, or `full` (all three) | +| `MAX_PARALLEL` | `10` | GPU concurrency cap | +| `EPOCHS` | `50` | Training epochs per walk-forward fold | +| `HYPEROPT_TRIALS` | `20` | PSO trials per model | +| `RUN_ID` | auto | YYYYMMDD-HHMMSS or manual override | + +## Model Groups + +- **RL (2)**: DQN, PPO — via `train_baseline_rl` / `hyperopt_baseline_rl` +- **Supervised (8)**: TFT, Mamba2, TGGN, TLOB, Liquid, KAN, xLSTM, Diffusion — via `train_baseline_supervised` / `hyperopt_baseline_supervised` +- **Hyperopt support**: DQN, PPO, TFT, Mamba2 (other 6 use default configs) + +## Architecture + +### Pipeline Structure + +``` +.gitlab-ci-training.yml (manual trigger, parent pipeline) + ├── prepare: generate-training-pipeline.sh + │ → writes .training-generated.yml as artifact + └── trigger: launches child pipeline from artifact + ├── STAGE: hyperopt (parallel, max MAX_PARALLEL) + ├── STAGE: train (parallel, needs hyperopt) + └── STAGE: evaluate (single job, needs all training) +``` + +### Why Generated Child Pipeline + +GitLab CI lacks dynamic matrix jobs. A prepare job generates exactly the needed jobs based on MODELS × SYMBOLS, avoiding dead "skipped" jobs. Adding a model = one line in the generator. + +### Infrastructure + +- **Training image**: `rg.fr-par.scw.cloud/foxhunt-ci/training:latest` (built by main CI) +- **GPU pool**: `gpu-training` (H100), Kubernetes autoscaler handles node provisioning +- **Input PVC**: `training-data-pvc` at `/data` (read-only, futures-baseline data) +- **Output PVC**: `training-output-pvc` at `/output` (read-write, checkpoints + results) + +### Output Directory Convention + +``` +/output// +├── hyperopt///results.json +├── models///*.safetensors +├── models///norm_stats.json +└── eval/ensemble_report.json +``` + +## Stage Details + +### Stage 1: Hyperopt (optional) + +Per model × symbol, runs PSO hyperparameter optimization. + +- RL: `hyperopt_baseline_rl --model {dqn|ppo} --trials $HYPEROPT_TRIALS --epochs 10 --data-dir /data/$SYMBOL --output /output/$RUN_ID/hyperopt/$MODEL/$SYMBOL/results.json` +- Supervised: `hyperopt_baseline_supervised --model {tft|mamba2} --trials $HYPEROPT_TRIALS --data-dir /data/$SYMBOL --output /output/$RUN_ID/hyperopt/$MODEL/$SYMBOL/results.json` +- Models without hyperopt adapters (TGGN, TLOB, Liquid, KAN, xLSTM, Diffusion) skip this stage. + +### Stage 2: Train + +Per model × symbol, runs walk-forward training. + +- RL: `train_baseline_rl --model {dqn|ppo} --epochs $EPOCHS --data-dir /data/$SYMBOL --output-dir /output/$RUN_ID/models/$MODEL/$SYMBOL --params-file /output/$RUN_ID/hyperopt/$MODEL/$SYMBOL/results.json` +- Supervised: `train_baseline_supervised --model $MODEL --epochs $EPOCHS --data-dir /data/$SYMBOL --output-dir /output/$RUN_ID/models/$MODEL/$SYMBOL --params-file /output/$RUN_ID/hyperopt/$MODEL/$SYMBOL/results.json` + +`--params-file` is optional — if the file doesn't exist (no hyperopt ran), uses compiled defaults. CLI args override JSON params. + +### Stage 3: Evaluate + +Single job after all training completes. + +`evaluate_baseline --models-dir /output/$RUN_ID/models --data-dir /data --output /output/$RUN_ID/eval/ensemble_report.json` + +Runs inference on held-out test windows, computes per-model and ensemble metrics (Sharpe, MaxDD, win rate, profit factor). + +## Parameter Passing: Hyperopt → Training + +Add `--params-file ` optional arg to `train_baseline_rl` and `train_baseline_supervised`. Logic: + +1. Parse CLI args as normal (with defaults) +2. If `--params-file` provided and file exists, load JSON, override matching config fields +3. CLI args still take precedence over JSON (explicit override) + +~20 lines per binary, using existing clap + serde_json. + +## Files to Create/Modify + +### New Files +- `.gitlab-ci-training.yml` — parent pipeline definition +- `scripts/generate-training-pipeline.sh` — child YAML generator +- `infra/k8s/training/training-output-pvc.yaml` — output PVC + +### Modified Files +- `crates/ml/examples/train_baseline_rl.rs` — add `--params-file` arg +- `crates/ml/examples/train_baseline_supervised.rs` — add `--params-file` arg + +## Cost Estimate (H100 ~€3/hr) + +| Scenario | Jobs | Time | Cost | +|----------|------|------|------| +| Single model, single symbol | 3 | ~1hr | ~€3 | +| Full ensemble, single symbol | 14 | ~1hr (parallel) | ~€16 | +| Full ensemble, 4 symbols | 56 | ~2hr (parallel) | ~€65 | + +## Concurrency + +Jobs use Kubernetes resource requests (1 GPU each). The autoscaler provisions H100 nodes as needed. MAX_PARALLEL limits fan-out at the pipeline level. If cluster is busy, Kubernetes queues pending pods naturally. diff --git a/docs/plans/2026-02-26-training-pipeline-implementation.md b/docs/plans/2026-02-26-training-pipeline-implementation.md new file mode 100644 index 000000000..ad1a45b0f --- /dev/null +++ b/docs/plans/2026-02-26-training-pipeline-implementation.md @@ -0,0 +1,675 @@ +# ML Training Pipeline Implementation Plan + +> **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task. + +**Goal:** Manually-triggered GitLab CI pipeline that runs hyperopt → train → evaluate for the 10-model ML ensemble on H100 GPUs. + +**Architecture:** A parent `.gitlab-ci-training.yml` accepts trigger variables, runs a shell script that generates a child pipeline YAML with exactly the needed jobs (models × symbols × phases), then triggers the child pipeline. Jobs run in the pre-built training Docker image with PVC mounts for data and output. + +**Tech Stack:** GitLab CI (parent/child pipelines), Bash (YAML generator), Rust/Clap (--params-file arg), Kubernetes PVCs + +--- + +### Task 1: Output PVC manifest + +**Files:** +- Create: `infra/k8s/training/training-output-pvc.yaml` + +**Step 1: Create the PVC manifest** + +```yaml +apiVersion: v1 +kind: PersistentVolumeClaim +metadata: + name: training-output-pvc + namespace: foxhunt + labels: + app.kubernetes.io/name: training-output + app.kubernetes.io/part-of: foxhunt +spec: + accessModes: + - ReadWriteOnce + resources: + requests: + storage: 50Gi + storageClassName: scw-bssd +``` + +50Gi because safetensors checkpoints for 10 models × 4 symbols × multiple folds can add up. scw-bssd matches the existing training-data-pvc. + +**Step 2: Commit** + +```bash +git add infra/k8s/training/training-output-pvc.yaml +git commit -m "infra: add training-output-pvc for ML pipeline checkpoints" +``` + +--- + +### Task 2: Pipeline generator script + +**Files:** +- Create: `scripts/generate-training-pipeline.sh` + +This script reads environment variables and writes a `.training-generated.yml` artifact with the exact jobs needed. + +**Step 1: Write the generator script** + +The script must: +1. Parse `MODELS` into a list (expand `all` → 10 models, `rl` → dqn,ppo, `supervised` → 8 models) +2. Parse `SYMBOLS` into a list (comma-separated) +3. Map each model to its binary (`dqn`/`ppo` → `*_rl`, others → `*_supervised`) +4. Map each model to whether it has hyperopt support (`dqn`,`ppo`,`tft`,`mamba2` → yes, others → no) +5. Write stages, variables, and job definitions to YAML +6. Respect `PHASE` variable to skip hyperopt or train stages + +Key details: +- All jobs use the training image from `$REGISTRY/training:latest` +- All jobs mount `training-data-pvc` at `/data` (read-only) and `training-output-pvc` at `/output` +- All jobs use `nodeSelector: gpu-training` and request 1 GPU +- Hyperopt jobs for RL use `--data-dir /data` (DBN files with symbol subdirs) +- Hyperopt jobs for supervised use `--parquet-file` — but the supervised hyperopt binary expects Parquet, NOT DBN. This is a design mismatch. For now, the generator skips supervised hyperopt unless a `PARQUET_DIR` variable is set. This is a known limitation documented in the script. +- Train jobs pass `--params-file /output/$RUN_ID/hyperopt/$MODEL/$SYMBOL/results.json` if hyperopt ran +- The evaluate job `needs:` all train jobs + +The script location: `scripts/generate-training-pipeline.sh` + +```bash +#!/usr/bin/env bash +set -euo pipefail + +# --- Input variables (from GitLab CI trigger) --- +SYMBOLS="${SYMBOLS:-ES.FUT}" +MODELS="${MODELS:-all}" +PHASE="${PHASE:-full}" +MAX_PARALLEL="${MAX_PARALLEL:-10}" +EPOCHS="${EPOCHS:-50}" +HYPEROPT_TRIALS="${HYPEROPT_TRIALS:-20}" +RUN_ID="${RUN_ID:-$(date +%Y%m%d-%H%M%S)}" +REGISTRY="${REGISTRY:-rg.fr-par.scw.cloud/foxhunt-ci}" +OUTPUT=".training-generated.yml" + +# --- Expand model groups --- +RL_MODELS="dqn ppo" +SUPERVISED_MODELS="tft mamba2 tggn tlob liquid kan xlstm diffusion" +HYPEROPT_MODELS="dqn ppo tft mamba2" + +case "$MODELS" in + all) MODEL_LIST="$RL_MODELS $SUPERVISED_MODELS" ;; + rl) MODEL_LIST="$RL_MODELS" ;; + supervised) MODEL_LIST="$SUPERVISED_MODELS" ;; + *) MODEL_LIST=$(echo "$MODELS" | tr ',' ' ') ;; +esac + +IFS=',' read -ra SYMBOL_LIST <<< "$SYMBOLS" + +# --- Helper: which binary for a model --- +binary_for() { + case "$1" in + dqn|ppo) echo "train_baseline_rl" ;; + *) echo "train_baseline_supervised" ;; + esac +} + +hyperopt_binary_for() { + case "$1" in + dqn|ppo) echo "hyperopt_baseline_rl" ;; + tft|mamba2) echo "hyperopt_baseline_supervised" ;; + *) echo "" ;; + esac +} + +has_hyperopt() { + for m in $HYPEROPT_MODELS; do + [ "$m" = "$1" ] && return 0 + done + return 1 +} + +is_rl() { + [ "$1" = "dqn" ] || [ "$1" = "ppo" ] +} + +# --- Write YAML header --- +cat > "$OUTPUT" <> "$OUTPUT" <> "$OUTPUT" <> "$OUTPUT" <> "$OUTPUT" <> "$OUTPUT" <> "$OUTPUT" <> "$OUTPUT" <> .gitignore +git add .gitignore +git commit -m "feat: add training pipeline generator script" +``` + +--- + +### Task 3: Parent pipeline YAML + +**Files:** +- Create: `.gitlab-ci-training.yml` + +**Step 1: Write the parent pipeline** + +```yaml +# ML Training Pipeline — manually triggered +# +# Trigger via GitLab UI: CI/CD → Pipelines → Run pipeline +# Select .gitlab-ci-training.yml as the CI config, set variables. +# +# Or via API: +# curl -X POST --fail \ +# -F "token=$TRIGGER_TOKEN" \ +# -F "ref=main" \ +# -F "variables[MODELS]=all" \ +# -F "variables[SYMBOLS]=ES.FUT" \ +# -F "variables[PHASE]=full" \ +# "$CI_API_V4_URL/projects/$CI_PROJECT_ID/trigger/pipeline" + +stages: + - prepare + - trigger + +variables: + SYMBOLS: "ES.FUT" + MODELS: "all" + PHASE: "full" + MAX_PARALLEL: "10" + EPOCHS: "50" + HYPEROPT_TRIALS: "20" + RUN_ID: "" + REGISTRY: rg.fr-par.scw.cloud/foxhunt-ci + +workflow: + rules: + - if: $CI_PIPELINE_SOURCE == "web" + - if: $CI_PIPELINE_SOURCE == "trigger" + - if: $CI_PIPELINE_SOURCE == "api" + +generate-jobs: + stage: prepare + image: alpine:3.19 + tags: + - kapsule + script: + - apk add --no-cache bash coreutils + - | + if [ -z "$RUN_ID" ]; then + export RUN_ID=$(date +%Y%m%d-%H%M%S) + fi + - bash scripts/generate-training-pipeline.sh + - cat .training-generated.yml + artifacts: + paths: + - .training-generated.yml + expire_in: 1 day + +run-training: + stage: trigger + trigger: + include: + - artifact: .training-generated.yml + job: generate-jobs + strategy: depend +``` + +Note: The child pipeline jobs reference the training image and GPU runner tags. The parent runs on a lightweight alpine image (no GPU needed). + +**Step 2: Verify YAML syntax** + +```bash +python3 -c "import yaml; yaml.safe_load(open('.gitlab-ci-training.yml'))" && echo "Valid YAML" +``` + +**Step 3: Commit** + +```bash +git add .gitlab-ci-training.yml +git commit -m "feat: add manually-triggered ML training pipeline" +``` + +--- + +### Task 4: Wire --hyperopt-params in train_baseline_rl + +**Files:** +- Modify: `crates/ml/examples/train_baseline_rl.rs` + +The `--hyperopt_params` arg already exists (line 71-73) but is unused. Wire it up to override DQN/PPO config fields from the hyperopt JSON. + +**Step 1: Read the hyperopt output format** + +The hyperopt JSON looks like: +```json +{ + "dqn": { + "best_objective": 0.123, + "best_params": { "learning_rate": 0.001, "hidden_dim": 128, ... }, + "trials": 20, + "elapsed_secs": 45.3 + } +} +``` + +Or for single-model runs: +```json +{ + "best_objective": 0.123, + "best_params": { "learning_rate": 0.001, ... }, + "trials": 20, + "elapsed_secs": 45.3 +} +``` + +**Step 2: Add a helper function to load and apply hyperopt params** + +Add after the `Args` struct: + +```rust +/// Load hyperopt results JSON and extract best_params for the given model. +/// Returns None if file doesn't exist or model key is missing. +fn load_hyperopt_params(path: &Path, model: &str) -> Option { + let content = std::fs::read_to_string(path).ok()?; + let json: serde_json::Value = serde_json::from_str(&content).ok()?; + + // Try model-keyed format first: {"dqn": {"best_params": {...}}} + if let Some(params) = json.get(model).and_then(|m| m.get("best_params")) { + return Some(params.clone()); + } + // Try flat format: {"best_params": {...}} + if let Some(params) = json.get("best_params") { + return Some(params.clone()); + } + None +} +``` + +**Step 3: Apply params to DQNConfig and PPOConfig** + +In the DQN training section, after creating the default config, add: + +```rust +if let Some(ref params_path) = args.hyperopt_params { + if let Some(params) = load_hyperopt_params(params_path, "dqn") { + info!("Loading hyperopt params from: {}", params_path.display()); + if let Some(lr) = params.get("learning_rate").and_then(|v| v.as_f64()) { + dqn_config.learning_rate = lr; + info!(" learning_rate = {}", lr); + } + if let Some(hd) = params.get("hidden_dim").and_then(|v| v.as_u64()) { + dqn_config.hidden_dim = hd as usize; + info!(" hidden_dim = {}", hd); + } + // Add additional param mappings as hyperopt search space grows + } else { + info!("No hyperopt params found at {}; using defaults", params_path.display()); + } +} +``` + +Same pattern for PPO section, reading "ppo" key and mapping to PPOConfig fields. + +**Step 4: Build and verify** + +```bash +SQLX_OFFLINE=true cargo check -p ml --example train_baseline_rl +``` + +Expected: compiles with 0 errors, 0 warnings. + +**Step 5: Commit** + +```bash +git add crates/ml/examples/train_baseline_rl.rs +git commit -m "feat(ml): wire --hyperopt-params to override DQN/PPO config" +``` + +--- + +### Task 5: Add --hyperopt-params to train_baseline_supervised + +**Files:** +- Modify: `crates/ml/examples/train_baseline_supervised.rs` + +**Step 1: Add the CLI arg** + +Add to the `Args` struct after `spread_ticks`: + +```rust + /// Optional path to hyperopt results JSON to override default config + #[arg(long)] + hyperopt_params: Option, +``` + +**Step 2: Add the same `load_hyperopt_params` helper** + +Same function as Task 4. + +**Step 3: Apply params to model configs** + +In each model's config construction (TFT, Mamba2, etc.), add the override block. Only TFT and Mamba2 have hyperopt adapters, so only those need param mapping. Example for TFT: + +```rust +if let Some(ref params_path) = args.hyperopt_params { + if let Some(params) = load_hyperopt_params(params_path, "tft") { + info!("Loading TFT hyperopt params from: {}", params_path.display()); + if let Some(lr) = params.get("learning_rate").and_then(|v| v.as_f64()) { + tft_config.learning_rate = lr as f32; + info!(" learning_rate = {}", lr); + } + if let Some(hd) = params.get("hidden_size").and_then(|v| v.as_u64()) { + tft_config.hidden_size = hd as usize; + info!(" hidden_size = {}", hd); + } + } +} +``` + +For models without hyperopt (TGGN, TLOB, etc.), no changes — the param file simply won't have their key. + +**Step 4: Build and verify** + +```bash +SQLX_OFFLINE=true cargo check -p ml --example train_baseline_supervised +``` + +**Step 5: Commit** + +```bash +git add crates/ml/examples/train_baseline_supervised.rs +git commit -m "feat(ml): add --hyperopt-params to supervised training binary" +``` + +--- + +### Task 6: Update Dockerfile.training for PVC mounts + +**Files:** +- Modify: `infra/k8s/training/job-template.yaml` + +**Step 1: Update the job template** + +The current template has `training-data-pvc` at `/data` and `emptyDir` at `/output`. Replace the output volume with the new PVC and add nodeSelector for H100. + +Update `volumes` section: +```yaml + volumes: + - name: training-data + persistentVolumeClaim: + claimName: training-data-pvc + - name: output + persistentVolumeClaim: + claimName: training-output-pvc +``` + +The nodeSelector is already `gpu-training` which is the H100 pool. No change needed there. + +**Step 2: Verify YAML** + +```bash +python3 -c "import yaml; yaml.safe_load(open('infra/k8s/training/job-template.yaml'))" && echo "Valid" +``` + +**Step 3: Commit** + +```bash +git add infra/k8s/training/job-template.yaml +git commit -m "infra: use training-output-pvc instead of emptyDir in job template" +``` + +--- + +### Task 7: Integration test — dry run the full pipeline locally + +**Step 1: Generate the pipeline for full ensemble** + +```bash +MODELS=all SYMBOLS=ES.FUT PHASE=full scripts/generate-training-pipeline.sh +``` + +**Step 2: Verify job count** + +Expected in `.training-generated.yml`: +- 2 hyperopt jobs (dqn, ppo — tft/mamba2 skipped without parquet) +- 10 train jobs (all models) +- 1 evaluate job +- Total: 13 jobs + +```bash +grep -c "^[a-z].*:$" .training-generated.yml +``` + +**Step 3: Verify multi-symbol expansion** + +```bash +MODELS=dqn,ppo SYMBOLS=ES.FUT,NQ.FUT PHASE=full scripts/generate-training-pipeline.sh +``` + +Expected: +- 4 hyperopt jobs (dqn×2 symbols, ppo×2 symbols) +- 4 train jobs +- 1 evaluate job +- Total: 9 jobs + +**Step 4: Verify phase filtering** + +```bash +MODELS=kan SYMBOLS=ES.FUT PHASE=train scripts/generate-training-pipeline.sh +``` + +Expected: 0 hyperopt, 1 train, 1 evaluate. Total: 2 jobs. + +**Step 5: Final commit with all files** + +```bash +git add -A +git status +git commit -m "feat: ML training pipeline — generator, CI config, PVC, param passing" +``` + +--- + +### Task 8: Update design doc with known limitations + +**Files:** +- Modify: `docs/plans/2026-02-26-training-pipeline-design.md` + +Add a "Known Limitations" section: + +1. Supervised hyperopt (`hyperopt_baseline_supervised`) expects Parquet input, not DBN. The generator currently skips supervised hyperopt. Future work: add DBN support to supervised hyperopt adapters. +2. `evaluate_baseline` currently only evaluates RL models (DQN/PPO). Extending it to load and evaluate all 10 model types needs work on the evaluation binary. +3. The output PVC is `ReadWriteOnce` — only one node can mount it at a time. If parallel jobs land on different nodes, they'll fail. Mitigation: use `ReadWriteMany` with an NFS-backed storage class, or ensure all training pods schedule on the same node. + +**Commit** + +```bash +git add docs/plans/2026-02-26-training-pipeline-design.md +git commit -m "docs: add known limitations to training pipeline design" +``` diff --git a/infra/k8s/training/job-template.yaml b/infra/k8s/training/job-template.yaml index 2e4422f83..24fe1e49e 100644 --- a/infra/k8s/training/job-template.yaml +++ b/infra/k8s/training/job-template.yaml @@ -67,4 +67,5 @@ spec: persistentVolumeClaim: claimName: training-data-pvc - name: output - emptyDir: {} + persistentVolumeClaim: + claimName: training-output-pvc diff --git a/infra/k8s/training/training-output-pvc.yaml b/infra/k8s/training/training-output-pvc.yaml new file mode 100644 index 000000000..404f22226 --- /dev/null +++ b/infra/k8s/training/training-output-pvc.yaml @@ -0,0 +1,15 @@ +apiVersion: v1 +kind: PersistentVolumeClaim +metadata: + name: training-output-pvc + namespace: foxhunt + labels: + app.kubernetes.io/name: training-output + app.kubernetes.io/part-of: foxhunt +spec: + accessModes: + - ReadWriteMany + resources: + requests: + storage: 50Gi + storageClassName: scw-bssd-nfs diff --git a/scripts/generate-training-pipeline.sh b/scripts/generate-training-pipeline.sh new file mode 100755 index 000000000..b2c7d1aa4 --- /dev/null +++ b/scripts/generate-training-pipeline.sh @@ -0,0 +1,332 @@ +#!/usr/bin/env bash +set -euo pipefail + +# Generate a GitLab CI child pipeline YAML for ML model training. +# Reads environment variables with sensible defaults and writes +# .training-generated.yml to the repository root. + +############################################################################### +# Configuration (environment variables with defaults) +############################################################################### + +SYMBOLS="${SYMBOLS:-ES.FUT}" +MODELS="${MODELS:-all}" +PHASE="${PHASE:-full}" +MAX_PARALLEL="${MAX_PARALLEL:-10}" +EPOCHS="${EPOCHS:-50}" +HYPEROPT_TRIALS="${HYPEROPT_TRIALS:-20}" +RUN_ID="${RUN_ID:-$(date +%Y%m%d-%H%M%S)}" +REGISTRY="${REGISTRY:-rg.fr-par.scw.cloud/foxhunt-ci}" + +OUTPUT=".training-generated.yml" + +############################################################################### +# Model definitions +############################################################################### + +RL_MODELS=(dqn ppo) +SUPERVISED_MODELS=(tft mamba2 tggn tlob liquid kan xlstm diffusion) +# Models with hyperopt adapters (RL: DQN/PPO, Supervised: TFT/Mamba2) +HYPEROPT_RL=(dqn ppo) +HYPEROPT_SUPERVISED=(tft mamba2) +HYPEROPT_MODELS=("${HYPEROPT_RL[@]}" "${HYPEROPT_SUPERVISED[@]}") + +############################################################################### +# Resolve which models to generate jobs for +############################################################################### + +resolve_models() { + case "${MODELS}" in + all) + echo "${RL_MODELS[*]} ${SUPERVISED_MODELS[*]}" + ;; + rl) + echo "${RL_MODELS[*]}" + ;; + supervised) + echo "${SUPERVISED_MODELS[*]}" + ;; + *) + # Comma-separated list + echo "${MODELS}" | tr ',' ' ' + ;; + esac +} + +SELECTED_MODELS=$(resolve_models) + +############################################################################### +# Helpers +############################################################################### + +# Replace dots with hyphens for valid YAML keys +sanitize() { + echo "$1" | tr '.' '-' +} + +# Check if a model is an RL model +is_rl_model() { + local model="$1" + for m in "${RL_MODELS[@]}"; do + if [[ "$m" == "$model" ]]; then + return 0 + fi + done + return 1 +} + +# Check if a model has a hyperopt adapter +has_hyperopt() { + local model="$1" + for m in "${HYPEROPT_MODELS[@]}"; do + if [[ "$m" == "$model" ]]; then + return 0 + fi + done + return 1 +} + +# Check if a model is a supervised hyperopt model (TFT/Mamba2) +is_supervised_hyperopt() { + local model="$1" + for m in "${HYPEROPT_SUPERVISED[@]}"; do + if [[ "$m" == "$model" ]]; then + return 0 + fi + done + return 1 +} + +# Get the hyperopt binary for a model +hyperopt_binary() { + local model="$1" + if is_rl_model "$model"; then + echo "hyperopt_baseline_rl" + else + echo "hyperopt_baseline_supervised" + fi +} + +# Get the training binary name for a model +training_binary() { + local model="$1" + if is_rl_model "$model"; then + echo "train_baseline_rl" + else + echo "train_baseline_supervised" + fi +} + +############################################################################### +# Begin generating the pipeline YAML +############################################################################### + +: > "$OUTPUT" + +cat >> "$OUTPUT" < ${SELECTED_MODELS} +# PHASE: ${PHASE} +# Generated: $(date -u +%Y-%m-%dT%H:%M:%SZ) + +YAML + +# --- Stages --------------------------------------------------------------- # + +if [[ "$PHASE" == "eval" ]]; then + cat >> "$OUTPUT" <> "$OUTPUT" <> "$OUTPUT" <> "$OUTPUT" <> "$OUTPUT" <> "$OUTPUT" <> "$OUTPUT" <> "$OUTPUT" <