diff --git a/crates/ml/src/hyperopt/adapters/dqn.rs b/crates/ml/src/hyperopt/adapters/dqn.rs index 1dbab5b0f..09feb2cc6 100644 --- a/crates/ml/src/hyperopt/adapters/dqn.rs +++ b/crates/ml/src/hyperopt/adapters/dqn.rs @@ -868,7 +868,7 @@ impl DQNTrainer { MLError::ConfigError("BUG: runtime_handle is None".to_owned()) })?; let (train_data, val_data) = preload_handle - .block_on(loader.load_training_data(data_path_str)) + .block_on(loader.load_training_data(data_path_str, &self.symbol)) .map_err(|e| MLError::TrainingError(format!("Failed to preload data: {e}")))?; // Extract OFI if available @@ -2237,7 +2237,7 @@ impl HyperparameterOptimizable for DQNTrainer { })? } else { info!("Training DQN with DBN directory: {}", data_path_str); - handle.block_on(internal_trainer.train(data_path_str, checkpoint_callback)) + handle.block_on(internal_trainer.train(data_path_str, &self.symbol, checkpoint_callback)) .map_err(|e| MLError::TrainingError(format!("DQN training failed: {e}")))? }; diff --git a/crates/ml/src/trainers/dqn/smoke_tests/feature_coverage.rs b/crates/ml/src/trainers/dqn/smoke_tests/feature_coverage.rs index 88cf90fbf..cd840f760 100644 --- a/crates/ml/src/trainers/dqn/smoke_tests/feature_coverage.rs +++ b/crates/ml/src/trainers/dqn/smoke_tests/feature_coverage.rs @@ -20,7 +20,7 @@ fn test_production_config_trains() -> anyhow::Result<()> { let rt = tokio::runtime::Builder::new_current_thread() .enable_all() .build()?; - let metrics = rt.block_on(trainer.train(&data_dir, |_epoch, _bytes, _best| { + let metrics = rt.block_on(trainer.train(&data_dir, "ES.FUT", |_epoch, _bytes, _best| { Ok("skip".to_owned()) }))?; assert_finite(metrics.loss, "production_loss"); diff --git a/crates/ml/src/trainers/dqn/smoke_tests/gpu_residency.rs b/crates/ml/src/trainers/dqn/smoke_tests/gpu_residency.rs index cafcc5aa2..7a00484d2 100644 --- a/crates/ml/src/trainers/dqn/smoke_tests/gpu_residency.rs +++ b/crates/ml/src/trainers/dqn/smoke_tests/gpu_residency.rs @@ -244,7 +244,7 @@ fn test_train_step_produces_finite_metrics() -> anyhow::Result<()> { let rt = tokio::runtime::Builder::new_current_thread() .enable_all() .build()?; - let metrics = rt.block_on(trainer.train(&data_dir, |_epoch, _bytes, _is_best| { + let metrics = rt.block_on(trainer.train(&data_dir, "ES.FUT", |_epoch, _bytes, _is_best| { Ok(String::new()) }))?; assert_finite(metrics.loss, "final_loss"); diff --git a/crates/ml/src/trainers/dqn/smoke_tests/performance.rs b/crates/ml/src/trainers/dqn/smoke_tests/performance.rs index 0d88cc643..cc5aa8467 100644 --- a/crates/ml/src/trainers/dqn/smoke_tests/performance.rs +++ b/crates/ml/src/trainers/dqn/smoke_tests/performance.rs @@ -29,7 +29,7 @@ fn test_training_throughput_measurement() -> anyhow::Result<()> { .build()?; let start = Instant::now(); - let metrics = rt.block_on(trainer.train(&data_dir(), |_epoch, _bytes, _best| { + let metrics = rt.block_on(trainer.train(&data_dir(), "ES.FUT", |_epoch, _bytes, _best| { Ok("skip".to_owned()) }))?; let elapsed = start.elapsed(); @@ -133,7 +133,7 @@ fn test_real_data_single_epoch() -> anyhow::Result<()> { .build()?; let start = Instant::now(); - let metrics = rt.block_on(trainer.train(&data_dir(), |_epoch, _bytes, _best| { + let metrics = rt.block_on(trainer.train(&data_dir(), "ES.FUT", |_epoch, _bytes, _best| { Ok("skip".to_owned()) }))?; let elapsed = start.elapsed(); diff --git a/crates/ml/src/trainers/dqn/smoke_tests/regression.rs b/crates/ml/src/trainers/dqn/smoke_tests/regression.rs index b179dabba..b356ae4d0 100644 --- a/crates/ml/src/trainers/dqn/smoke_tests/regression.rs +++ b/crates/ml/src/trainers/dqn/smoke_tests/regression.rs @@ -30,7 +30,7 @@ fn test_no_hang_single_epoch() -> anyhow::Result<()> { .build()?; // Must complete 1 epoch — a hang here means VRAM oversubscription returned. - let metrics = rt.block_on(trainer.train(&data_dir, |_epoch, _bytes, _best| { + let metrics = rt.block_on(trainer.train(&data_dir, "ES.FUT", |_epoch, _bytes, _best| { Ok("skip".to_owned()) }))?; @@ -82,7 +82,7 @@ fn test_counterfactual_experiences_in_buffer() -> anyhow::Result<()> { let rt = tokio::runtime::Builder::new_current_thread() .enable_all() .build()?; - let _metrics = rt.block_on(trainer.train(&data_dir, |_epoch, _bytes, _best| { + let _metrics = rt.block_on(trainer.train(&data_dir, "ES.FUT", |_epoch, _bytes, _best| { Ok("skip".to_owned()) }))?; @@ -131,7 +131,7 @@ fn test_gpu_n_episodes_config_honored() -> anyhow::Result<()> { let rt = tokio::runtime::Builder::new_current_thread() .enable_all() .build()?; - let _metrics = rt.block_on(trainer.train(&data_dir, |_epoch, _bytes, _best| { + let _metrics = rt.block_on(trainer.train(&data_dir, "ES.FUT", |_epoch, _bytes, _best| { Ok("skip".to_owned()) }))?; diff --git a/crates/ml/src/trainers/dqn/smoke_tests/training_stability.rs b/crates/ml/src/trainers/dqn/smoke_tests/training_stability.rs index 4f99f583f..afda0baff 100644 --- a/crates/ml/src/trainers/dqn/smoke_tests/training_stability.rs +++ b/crates/ml/src/trainers/dqn/smoke_tests/training_stability.rs @@ -101,7 +101,7 @@ fn test_gradient_norm_bounded_across_epochs() -> anyhow::Result<()> { let rt = tokio::runtime::Builder::new_current_thread() .enable_all() .build()?; - let metrics = rt.block_on(trainer.train(&data_dir, |_epoch, _bytes, _best| { + let metrics = rt.block_on(trainer.train(&data_dir, "ES.FUT", |_epoch, _bytes, _best| { Ok("skip".to_owned()) }))?; @@ -232,7 +232,7 @@ fn test_trading_model_behavior() -> anyhow::Result<()> { let rt = tokio::runtime::Builder::new_current_thread() .enable_all() .build()?; - let metrics = rt.block_on(trainer.train(&data_dir, |_epoch, _bytes, _best| { + let metrics = rt.block_on(trainer.train(&data_dir, "ES.FUT", |_epoch, _bytes, _best| { Ok("skip".to_owned()) }))?; @@ -351,7 +351,7 @@ fn test_50_epoch_convergence() -> anyhow::Result<()> { let rt = tokio::runtime::Builder::new_current_thread() .enable_all() .build()?; - let metrics = rt.block_on(trainer.train(&data_dir, |_epoch, _bytes, _best| { + let metrics = rt.block_on(trainer.train(&data_dir, "ES.FUT", |_epoch, _bytes, _best| { Ok("skip".to_owned()) }))?; @@ -448,7 +448,7 @@ fn test_gpu_collector_auto_initializes() -> anyhow::Result<()> { let rt = tokio::runtime::Builder::new_current_thread() .enable_all() .build()?; - let result = rt.block_on(trainer.train(&data_dir, |_epoch, _bytes, _is_best| { + let result = rt.block_on(trainer.train(&data_dir, "ES.FUT", |_epoch, _bytes, _is_best| { Ok(String::new()) })); assert!(result.is_ok(), "Training must succeed with auto-initialized GPU collector: {:?}", result.err()); diff --git a/crates/ml/src/trainers/dqn/smoke_tests/walk_forward.rs b/crates/ml/src/trainers/dqn/smoke_tests/walk_forward.rs index 154c5c130..94236fe11 100644 --- a/crates/ml/src/trainers/dqn/smoke_tests/walk_forward.rs +++ b/crates/ml/src/trainers/dqn/smoke_tests/walk_forward.rs @@ -36,7 +36,7 @@ fn test_walk_forward_oos_metrics() -> anyhow::Result<()> { let rt = tokio::runtime::Builder::new_current_thread() .enable_all() .build()?; - let metrics = rt.block_on(trainer.train(&data_dir, |_epoch, _bytes, _best| { + let metrics = rt.block_on(trainer.train(&data_dir, "ES.FUT", |_epoch, _bytes, _best| { Ok("skip".to_owned()) }))?; @@ -124,7 +124,7 @@ fn test_walk_forward_no_overfitting_50_epochs() -> anyhow::Result<()> { let rt = tokio::runtime::Builder::new_current_thread() .enable_all() .build()?; - let metrics = rt.block_on(trainer.train(&data_dir, |_epoch, _bytes, _best| { + let metrics = rt.block_on(trainer.train(&data_dir, "ES.FUT", |_epoch, _bytes, _best| { Ok("skip".to_owned()) }))?; @@ -184,7 +184,7 @@ fn test_walk_forward_multi_fold() -> anyhow::Result<()> { let rt = tokio::runtime::Builder::new_current_thread() .enable_all() .build()?; - let metrics = rt.block_on(trainer.train(&data_dir, |_epoch, _bytes, _best| { + let metrics = rt.block_on(trainer.train(&data_dir, "ES.FUT", |_epoch, _bytes, _best| { Ok("skip".to_owned()) }))?; diff --git a/crates/ml/tests/dqn_accumulation_convergence_test.rs b/crates/ml/tests/dqn_accumulation_convergence_test.rs index 4b3e3e35e..dc7e2f5f7 100644 --- a/crates/ml/tests/dqn_accumulation_convergence_test.rs +++ b/crates/ml/tests/dqn_accumulation_convergence_test.rs @@ -120,7 +120,7 @@ async fn test_accumulation_convergence_similar_to_direct() -> Result<()> { let checkpoint_dir_1 = tempfile::tempdir()?; let mut trainer_accum = DQNTrainer::new(hp_accum)?; let _metrics_accum = trainer_accum - .train(&data_dir, |_epoch, data, _is_best| { + .train(&data_dir, "ES.FUT", |_epoch, data, _is_best| { let p = checkpoint_dir_1.path().join("accum.safetensors"); std::fs::write(&p, &data)?; Ok(p.to_string_lossy().to_string()) @@ -141,7 +141,7 @@ async fn test_accumulation_convergence_similar_to_direct() -> Result<()> { let checkpoint_dir_2 = tempfile::tempdir()?; let mut trainer_direct = DQNTrainer::new(hp_direct)?; let _metrics_direct = trainer_direct - .train(&data_dir, |_epoch, data, _is_best| { + .train(&data_dir, "ES.FUT", |_epoch, data, _is_best| { let p = checkpoint_dir_2.path().join("direct.safetensors"); std::fs::write(&p, &data)?; Ok(p.to_string_lossy().to_string()) diff --git a/crates/ml/tests/dqn_early_stopping_termination_test.rs b/crates/ml/tests/dqn_early_stopping_termination_test.rs index bded21aba..1ab5e742a 100644 --- a/crates/ml/tests/dqn_early_stopping_termination_test.rs +++ b/crates/ml/tests/dqn_early_stopping_termination_test.rs @@ -170,7 +170,7 @@ async fn test_early_stopping_terminates_with_error() { }; // Run training - should terminate with error - let result = trainer.train(&data_dir, checkpoint_callback).await; + let result = trainer.train(&data_dir, "ES.FUT", checkpoint_callback).await; // Validate early stopping behavior assert!( @@ -235,7 +235,7 @@ async fn test_gradient_collapse_propagates_error() { }; // Run training - let result = trainer.train(&data_dir, checkpoint_callback).await; + let result = trainer.train(&data_dir, "ES.FUT", checkpoint_callback).await; // Should terminate with error assert!( @@ -297,7 +297,7 @@ async fn test_healthy_training_completes_successfully() { }; // Run training - let result = trainer.train(&data_dir, checkpoint_callback).await; + let result = trainer.train(&data_dir, "ES.FUT", checkpoint_callback).await; // Should complete successfully (no early stopping) assert!( diff --git a/crates/ml/tests/dqn_gradient_accumulation_test.rs b/crates/ml/tests/dqn_gradient_accumulation_test.rs index 20ad5506c..0f19efb92 100644 --- a/crates/ml/tests/dqn_gradient_accumulation_test.rs +++ b/crates/ml/tests/dqn_gradient_accumulation_test.rs @@ -121,7 +121,7 @@ async fn test_accumulation_single_optimizer_step() -> Result<()> { let mut trainer = DQNTrainer::new(hyperparams)?; let metrics = trainer - .train(&data_dir, |_epoch, checkpoint_data, _is_best| { + .train(&data_dir, "ES.FUT", |_epoch, checkpoint_data, _is_best| { let path = checkpoint_dir.path().join("accum_test.safetensors"); std::fs::write(&path, &checkpoint_data)?; Ok(path.to_string_lossy().to_string()) diff --git a/crates/ml/tests/dqn_inference_test.rs b/crates/ml/tests/dqn_inference_test.rs index 69a0b9e69..6bef7ce74 100644 --- a/crates/ml/tests/dqn_inference_test.rs +++ b/crates/ml/tests/dqn_inference_test.rs @@ -166,7 +166,7 @@ async fn test_checkpoint_to_inference() -> Result<()> { let mut best_checkpoint_path: Option = None; let _metrics = trainer - .train(&data_dir, |epoch, checkpoint_data, is_best| { + .train(&data_dir, "ES.FUT", |epoch, checkpoint_data, is_best| { let name = if is_best { "inference_best.safetensors".to_string() } else { diff --git a/crates/ml/tests/dqn_long_training_test.rs b/crates/ml/tests/dqn_long_training_test.rs index a84568f4a..7fd8f9974 100644 --- a/crates/ml/tests/dqn_long_training_test.rs +++ b/crates/ml/tests/dqn_long_training_test.rs @@ -138,7 +138,7 @@ async fn test_dqn_50_epoch_convergence() -> Result<()> { let mut trainer = DQNTrainer::new(hyperparams)?; let _metrics = trainer - .train(&data_dir, |epoch, checkpoint_data, is_best| { + .train(&data_dir, "ES.FUT", |epoch, checkpoint_data, is_best| { let name = if is_best { "long_best.safetensors".to_string() } else { diff --git a/crates/ml/tests/dqn_training_pipeline_test.rs b/crates/ml/tests/dqn_training_pipeline_test.rs index 516c058d3..1d38da0f8 100644 --- a/crates/ml/tests/dqn_training_pipeline_test.rs +++ b/crates/ml/tests/dqn_training_pipeline_test.rs @@ -213,7 +213,7 @@ async fn test_dqn_trains_on_es_fut() -> Result<()> { let mut final_checkpoint_path = PathBuf::new(); let metrics = trainer - .train(&data_dir, |epoch, checkpoint_data, _is_best| { + .train(&data_dir, "ES.FUT", |epoch, checkpoint_data, _is_best| { let path = checkpoint_dir.join(format!("dqn_test_epoch_{}.safetensors", epoch)); std::fs::write(&path, checkpoint_data)?; checkpoint_saved = true; @@ -329,7 +329,7 @@ async fn test_dqn_loss_decreases() -> Result<()> { // Track losses per epoch (would need to modify trainer to expose this) let metrics = trainer - .train(&data_dir, |epoch, checkpoint_data, _is_best| { + .train(&data_dir, "ES.FUT", |epoch, checkpoint_data, _is_best| { let path = checkpoint_dir.join(format!("dqn_loss_test_epoch_{}.safetensors", epoch)); std::fs::write(&path, checkpoint_data)?; Ok(path.to_string_lossy().to_string()) @@ -405,7 +405,7 @@ async fn test_dqn_checkpoint_save_load() -> Result<()> { let mut saved_checkpoint_path = PathBuf::new(); let _metrics = trainer - .train(&data_dir, |epoch, checkpoint_data, _is_best| { + .train(&data_dir, "ES.FUT", |epoch, checkpoint_data, _is_best| { let path = checkpoint_dir.join(format!("dqn_checkpoint_test_epoch_{}.safetensors", epoch)); std::fs::write(&path, checkpoint_data)?; @@ -473,7 +473,7 @@ async fn test_dqn_q_value_predictions() -> Result<()> { let mut trainer = DQNTrainer::new(hyperparams)?; let metrics = trainer - .train(&data_dir, |epoch, checkpoint_data, _is_best| { + .train(&data_dir, "ES.FUT", |epoch, checkpoint_data, _is_best| { let path = checkpoint_dir.join(format!("dqn_qvalue_test_epoch_{}.safetensors", epoch)); std::fs::write(&path, checkpoint_data)?; Ok(path.to_string_lossy().to_string()) @@ -537,7 +537,7 @@ async fn test_dqn_epsilon_greedy() -> Result<()> { let mut trainer = DQNTrainer::new(hyperparams)?; let _metrics = trainer - .train(&data_dir, |epoch, checkpoint_data, _is_best| { + .train(&data_dir, "ES.FUT", |epoch, checkpoint_data, _is_best| { let path = checkpoint_dir.join(format!("dqn_epsilon_test_epoch_{}.safetensors", epoch)); std::fs::write(&path, checkpoint_data)?; Ok(path.to_string_lossy().to_string()) @@ -619,7 +619,7 @@ async fn test_dqn_full_production_training() -> Result<()> { let mut epoch_count = 0; let metrics = trainer - .train(&data_dir, |epoch, checkpoint_data, _is_best| { + .train(&data_dir, "ES.FUT", |epoch, checkpoint_data, _is_best| { epoch_count += 1; let path = if epoch == hyperparams.epochs { production_checkpoint_path.clone() diff --git a/crates/ml/tests/production_training_smoke_test.rs b/crates/ml/tests/production_training_smoke_test.rs index bbcfb5463..d2fca8640 100644 --- a/crates/ml/tests/production_training_smoke_test.rs +++ b/crates/ml/tests/production_training_smoke_test.rs @@ -378,7 +378,7 @@ async fn test_qr_dqn_training_real_data() -> Result<()> { let mut trainer = DQNTrainer::new(hyperparams)?; let metrics = trainer - .train(&data_dir_str, |epoch, checkpoint_data, is_best| { + .train(&data_dir_str, "ES.FUT", |epoch, checkpoint_data, is_best| { let name = if is_best { "qrdqn_best.safetensors".to_string() } else { diff --git a/crates/ml/tests/smoke_test_real_data.rs b/crates/ml/tests/smoke_test_real_data.rs index 1c842aa34..0bd08773e 100644 --- a/crates/ml/tests/smoke_test_real_data.rs +++ b/crates/ml/tests/smoke_test_real_data.rs @@ -816,7 +816,7 @@ async fn smoke_e2e_dqn_training_loop() { info!(data_dir, "Starting E2E DQN training smoke test"); let metrics = trainer - .train(&data_dir, |epoch, checkpoint_data, is_best| { + .train(&data_dir, "ES.FUT", |epoch, checkpoint_data, is_best| { if is_best { let path = checkpoint_dir.path().join("smoke_e2e_best.safetensors"); std::fs::write(&path, &checkpoint_data)?; diff --git a/crates/ml/tests/wave16q_ohlcv_mutation_test.rs b/crates/ml/tests/wave16q_ohlcv_mutation_test.rs index 0015ef467..9ac5272f8 100644 --- a/crates/ml/tests/wave16q_ohlcv_mutation_test.rs +++ b/crates/ml/tests/wave16q_ohlcv_mutation_test.rs @@ -110,7 +110,7 @@ async fn test_ohlcv_raw_prices_preserved_after_preprocessing() -> Result<()> { Ok(String::new()) }; - let result = trainer.train(data_dir, noop_callback).await; + let result = trainer.train(data_dir, "ES.FUT", noop_callback).await; // Training may fail for other reasons, but we need to check OHLCV corruption // even if training fails (the bug manifests during data loading) @@ -186,7 +186,7 @@ async fn test_target_2_is_raw_price_not_zscore() -> Result<()> { let data_dir = "test_data/real/databento/ml_training"; let noop_callback = |_: usize, _: Vec, _: bool| -> Result { Ok(String::new()) }; - let _ = trainer.train(data_dir, noop_callback).await; + let _ = trainer.train(data_dir, "ES.FUT", noop_callback).await; let val_data = trainer.get_val_data(); assert!(!val_data.is_empty(), "Validation data should not be empty"); @@ -220,7 +220,7 @@ async fn test_max_position_realistic_not_100m() -> Result<()> { let data_dir = "test_data/real/databento/ml_training"; let noop_callback = |_: usize, _: Vec, _: bool| -> Result { Ok(String::new()) }; - let _ = trainer.train(data_dir, noop_callback).await; + let _ = trainer.train(data_dir, "ES.FUT", noop_callback).await; let val_data = trainer.get_val_data();