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

refactor: all train() and load_training_data() callers pass explicit symbol

Browse Source 
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
jgrusewski
2026-04-04 10:02:37 +02:00
parent d99abd3844
commit d2d80b7392
16 changed files with 35 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
crates/ml/src/hyperopt/adapters/dqn.rs
View File

@@ -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}")))?
};

2
crates/ml/src/trainers/dqn/smoke_tests/feature_coverage.rs
View File

@@ -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");

2
crates/ml/src/trainers/dqn/smoke_tests/gpu_residency.rs
View File

@@ -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");

4
crates/ml/src/trainers/dqn/smoke_tests/performance.rs
View File

@@ -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();

6
crates/ml/src/trainers/dqn/smoke_tests/regression.rs
View File

@@ -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())
}))?;

8
crates/ml/src/trainers/dqn/smoke_tests/training_stability.rs
View File

@@ -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());

6
crates/ml/src/trainers/dqn/smoke_tests/walk_forward.rs
View File

@@ -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())
}))?;

4
crates/ml/tests/dqn_accumulation_convergence_test.rs
View File

@@ -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())

6
crates/ml/tests/dqn_early_stopping_termination_test.rs
View File

@@ -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!(

2
crates/ml/tests/dqn_gradient_accumulation_test.rs
View File

@@ -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())

2
crates/ml/tests/dqn_inference_test.rs
View File

@@ -166,7 +166,7 @@ async fn test_checkpoint_to_inference() -> Result<()> {
let mut best_checkpoint_path: Option<PathBuf> = 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 {

2
crates/ml/tests/dqn_long_training_test.rs
View File

@@ -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 {

12
crates/ml/tests/dqn_training_pipeline_test.rs
View File

@@ -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()

2
crates/ml/tests/production_training_smoke_test.rs
View File

@@ -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 {

2
crates/ml/tests/smoke_test_real_data.rs
View File

@@ -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)?;

6
crates/ml/tests/wave16q_ohlcv_mutation_test.rs
View File

@@ -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<u8>, _: bool| -> Result<String> { 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<u8>, _: bool| -> Result<String> { 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();