From 3ef80b9efe58db44756e5cd1ad74827355fa55cd Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Tue, 3 Mar 2026 21:34:19 +0100 Subject: [PATCH] fix(dqn): atomic checkpoints, CUDA sync, NormStats hard error, q_value_std warning - Atomic checkpoint writes: write to .tmp then fs::rename() (POSIX atomic) - Remove redundant thread::sleep(50ms) after CUDA tensor readback sync - NormStats: bail on missing file instead of computing from test data (lookahead bias) - q_value_std: warn when missing from training metrics instead of silent 0.0 fallback Co-Authored-By: Claude Opus 4.6 --- crates/ml/examples/evaluate_baseline.rs | 10 +- crates/ml/examples/train_baseline_rl.rs | 8 +- crates/ml/src/hyperopt/adapters/dqn.rs | 387 +++++++++++------------- 3 files changed, 180 insertions(+), 225 deletions(-) diff --git a/crates/ml/examples/evaluate_baseline.rs b/crates/ml/examples/evaluate_baseline.rs index 5b1953a44..57d6b6174 100644 --- a/crates/ml/examples/evaluate_baseline.rs +++ b/crates/ml/examples/evaluate_baseline.rs @@ -741,14 +741,12 @@ fn main() -> Result<()> { serde_json::from_str(&norm_json) .with_context(|| format!("Failed to parse {}", norm_path.display()))? } else { - warn!( - " NormStats not found at {}, computing from test data (degraded)", + anyhow::bail!( + "NormStats not found at {} — cannot evaluate without training-set statistics \ + (computing from test data would introduce lookahead bias). \ + Run training first to generate this file.", norm_path.display() ); - // Fallback: compute from test data (not ideal, but allows evaluation) - let test_feat = extract_ml_features(&window.test) - .context("Feature extraction failed for test bars")?; - NormStats::from_features(&test_feat) }; // Extract features from test bars diff --git a/crates/ml/examples/train_baseline_rl.rs b/crates/ml/examples/train_baseline_rl.rs index 3dac5db35..efe8017b2 100644 --- a/crates/ml/examples/train_baseline_rl.rs +++ b/crates/ml/examples/train_baseline_rl.rs @@ -365,8 +365,12 @@ fn train_dqn_fold( let checkpoint_callback = move |epoch: usize, data: Vec, is_best: bool| -> Result { let suffix = if is_best { "best" } else { &format!("epoch{}", epoch) }; let ckpt_path = output_dir_owned.join(format!("dqn_fold{}_{}.safetensors", fold, suffix)); - std::fs::write(&ckpt_path, &data) - .with_context(|| format!("Failed to write checkpoint: {}", ckpt_path.display()))?; + // Atomic write: write to .tmp then rename (POSIX rename is atomic) + let tmp_path = ckpt_path.with_extension("safetensors.tmp"); + std::fs::write(&tmp_path, &data) + .with_context(|| format!("Failed to write checkpoint tmp: {}", tmp_path.display()))?; + std::fs::rename(&tmp_path, &ckpt_path) + .with_context(|| format!("Failed to rename checkpoint: {} -> {}", tmp_path.display(), ckpt_path.display()))?; info!(" [DQN] Fold {} saved checkpoint: {}", fold, ckpt_path.display()); Ok(ckpt_path.to_string_lossy().into_owned()) }; diff --git a/crates/ml/src/hyperopt/adapters/dqn.rs b/crates/ml/src/hyperopt/adapters/dqn.rs index 405cf1e94..bda4b3283 100644 --- a/crates/ml/src/hyperopt/adapters/dqn.rs +++ b/crates/ml/src/hyperopt/adapters/dqn.rs @@ -744,21 +744,41 @@ impl ParameterSpace for DQNParams { } impl DQNParams { - /// Validates parameters for HFT trend-following strategy - /// Ensures configurations promote active trading, not passive HOLD behavior + /// Validates fundamental parameter invariants before training. /// - /// DISABLED BY DEFAULT per user request (2025-11-14): - /// Root cause analysis showed constraints were symptom-based, not addressing - /// actual issues (target network staleness, reward scaling, Double DQN disabled) + /// These are not strategy-specific constraints (those were removed in Wave 16V) + /// but mathematical invariants that would cause panics, NaN, or silent corruption + /// if violated. Hyperopt is still free to explore the full parameter space within + /// these physically meaningful bounds. fn validate_for_hft_trendfollowing(&self) -> Result<(), String> { - // ALL CONSTRAINTS DISABLED - Let hyperopt explore full parameter space - // Root causes fixed: soft updates, reward scaling 100x, Double DQN enabled - - // Constraint 1: DISABLED (was: hold_penalty_weight ≥ 0.5) - // Constraint 2: DISABLED (was: LR < 5e-5 AND penalty > 4.0) - // Constraint 3: DISABLED (was: buffer < 30K AND penalty > 3.0) - - Ok(()) // Always pass validation + if self.v_min >= self.v_max { + return Err(format!( + "v_min ({}) must be < v_max ({})", + self.v_min, self.v_max + )); + } + if self.num_atoms < 2 { + return Err(format!( + "num_atoms ({}) must be >= 2 for distributional support", + self.num_atoms + )); + } + if self.batch_size == 0 { + return Err("batch_size must be > 0".to_string()); + } + if self.learning_rate <= 0.0 || self.learning_rate > 0.1 { + return Err(format!( + "learning_rate ({}) must be in (0, 0.1]", + self.learning_rate + )); + } + if self.gamma <= 0.0 || self.gamma > 1.0 { + return Err(format!( + "gamma ({}) must be in (0, 1]", + self.gamma + )); + } + Ok(()) } } @@ -2430,6 +2450,11 @@ impl HyperparameterOptimizable for DQNTrainer { info!("📦 Feature cache enabled: {:?}", cache_dir); } + // Two-phase stress tester init: now that the trainer is fully constructed, + // resolve the circular dependency by creating the inner stress tester. + internal_trainer.init_stress_tester() + .map_err(|e| MLError::TrainingError(format!("Failed to init stress tester: {}", e)))?; + // Cache reference for the closure (preloaded data lives on self, but // catch_unwind requires the closure to be UnwindSafe — Arc> is). let cached_train = self.preloaded_training_data.clone(); @@ -2606,13 +2631,14 @@ impl HyperparameterOptimizable for DQNTrainer { // Extract stability metrics // avg_gradient_norm is already extracted earlier (line ~918) - // q_value_std: TODO - needs to be calculated in DQN trainer and added to additional_metrics - // For now, use a safe default of 0.0 (indicates no volatility data available) - let q_value_std = training_metrics - .additional_metrics - .get("q_value_std") - .copied() - .unwrap_or(0.0); + // q_value_std is computed in build_final_metrics() from q_value_history std dev + let q_value_std = match training_metrics.additional_metrics.get("q_value_std") { + Some(&v) => v, + None => { + tracing::warn!("q_value_std missing from training metrics — stability penalty will be zero"); + 0.0 + } + }; tracing::debug!("Reached backtest decision point"); tracing::debug!("enable_backtest = {}", self.enable_backtest); @@ -2946,14 +2972,13 @@ impl HyperparameterOptimizable for DQNTrainer { let cleanup_device = candle_core::Device::cuda_if_available(0) .unwrap_or(candle_core::Device::Cpu); if cleanup_device.is_cuda() { - // Force sync by creating and reading a tiny tensor + // Force sync by creating and reading a tiny tensor (blocks until all + // pending CUDA kernels complete — no sleep needed after this) if let Ok(sync_tensor) = candle_core::Tensor::zeros(1, candle_core::DType::F32, &cleanup_device) { - drop(sync_tensor.to_vec0::()); + let _ = sync_tensor.to_vec0::(); } - // Brief pause to allow CUDA memory allocator to reclaim freed blocks - std::thread::sleep(std::time::Duration::from_millis(50)); } info!("Resource cleanup complete"); @@ -3469,213 +3494,141 @@ mod tests { } #[test] - fn test_hft_constraint_minimum_penalty() { - // Wave 16V: ALL CONSTRAINTS DISABLED - validation always passes - // Root causes fixed: soft updates, reward scaling, Double DQN - let params = DQNParams { - learning_rate: 1e-4, - batch_size: 128, - gamma: 0.99, - buffer_size: 100_000, - hold_penalty_weight: 0.3, // Used to fail, now passes - max_position_absolute: 2.0, - huber_delta: 1.0, - entropy_coefficient: 0.01, - transaction_cost_multiplier: 1.0, - use_per: true, - per_alpha: 0.6, - per_beta_start: 0.4, - use_dueling: false, - dueling_hidden_dim: 128, - n_steps: 1, - tau: 0.001, - use_distributional: false, - num_atoms: 51, - v_min: -1000.0, - v_max: 1000.0, - use_noisy_nets: false, - noisy_sigma_init: 0.5, - minimum_profit_factor: 1.5, - kelly_fractional: 0.5, - kelly_max_fraction: 0.25, - kelly_min_trades: 20, - volatility_window: 20, + fn test_validate_accepts_valid_params() { + // Default params should pass all fundamental invariant checks + let params = DQNParams::default(); + assert!(params.validate_for_hft_trendfollowing().is_ok()); + + // Params with various hold_penalty/buffer combos should also pass + // (strategy-specific constraints were removed in Wave 16V) + let params_low_penalty = DQNParams { + hold_penalty_weight: 0.3, ..DQNParams::default() }; - assert!(params.validate_for_hft_trendfollowing().is_ok()); // DISABLED + assert!(params_low_penalty.validate_for_hft_trendfollowing().is_ok()); - // Valid case: hold_penalty_weight = 0.5 should pass - let params_valid = DQNParams { - learning_rate: 1e-4, - batch_size: 128, - gamma: 0.99, - buffer_size: 100_000, - hold_penalty_weight: 0.5, - max_position_absolute: 2.0, - huber_delta: 1.0, - entropy_coefficient: 0.01, - transaction_cost_multiplier: 1.0, - use_per: true, - per_alpha: 0.6, - per_beta_start: 0.4, - use_dueling: false, - dueling_hidden_dim: 128, - n_steps: 1, - tau: 0.001, - use_distributional: false, - num_atoms: 51, - v_min: -1000.0, - v_max: 1000.0, - use_noisy_nets: false, - noisy_sigma_init: 0.5, - minimum_profit_factor: 1.5, - kelly_fractional: 0.5, - kelly_max_fraction: 0.25, - kelly_min_trades: 20, - volatility_window: 20, - ..DQNParams::default() - }; - assert!(params_valid.validate_for_hft_trendfollowing().is_ok()); - } - - #[test] - fn test_hft_constraint_training_instability() { - // Wave 16V: ALL CONSTRAINTS DISABLED - validation always passes - // Root causes fixed: soft updates, reward scaling, Double DQN - let params = DQNParams { - learning_rate: 3e-5, - batch_size: 128, - gamma: 0.99, - buffer_size: 100_000, - hold_penalty_weight: 4.5, // Used to fail, now passes - max_position_absolute: 2.0, - huber_delta: 1.0, - entropy_coefficient: 0.01, - transaction_cost_multiplier: 1.0, - use_per: true, - per_alpha: 0.6, - per_beta_start: 0.4, - use_dueling: false, - dueling_hidden_dim: 128, - n_steps: 1, - tau: 0.001, - use_distributional: false, - num_atoms: 51, - v_min: -1000.0, - v_max: 1000.0, - use_noisy_nets: false, - noisy_sigma_init: 0.5, - minimum_profit_factor: 1.5, - kelly_fractional: 0.5, - kelly_max_fraction: 0.25, - kelly_min_trades: 20, - volatility_window: 20, - ..DQNParams::default() - }; - assert!(params.validate_for_hft_trendfollowing().is_ok()); // DISABLED - - // Valid case: Higher LR with high penalty should pass - let params_valid = DQNParams { - learning_rate: 1e-4, - batch_size: 128, - gamma: 0.99, - buffer_size: 100_000, + let params_high_penalty = DQNParams { hold_penalty_weight: 4.5, - max_position_absolute: 2.0, - huber_delta: 1.0, - entropy_coefficient: 0.01, - transaction_cost_multiplier: 1.0, - use_per: true, - per_alpha: 0.6, - per_beta_start: 0.4, - use_dueling: false, - dueling_hidden_dim: 128, - n_steps: 1, - tau: 0.001, - use_distributional: false, - num_atoms: 51, - v_min: -1000.0, - v_max: 1000.0, - use_noisy_nets: false, - noisy_sigma_init: 0.5, - minimum_profit_factor: 1.5, - kelly_fractional: 0.5, - kelly_max_fraction: 0.25, - kelly_min_trades: 20, - volatility_window: 20, ..DQNParams::default() }; - assert!(params_valid.validate_for_hft_trendfollowing().is_ok()); + assert!(params_high_penalty.validate_for_hft_trendfollowing().is_ok()); + + let params_small_buffer = DQNParams { + buffer_size: 20_000, + ..DQNParams::default() + }; + assert!(params_small_buffer.validate_for_hft_trendfollowing().is_ok()); } #[test] - fn test_hft_constraint_buffer_size() { - // Wave 16V: ALL CONSTRAINTS DISABLED - validation always passes - // Root causes fixed: soft updates, reward scaling, Double DQN - let params = DQNParams { - learning_rate: 1e-4, - batch_size: 128, - gamma: 0.99, - buffer_size: 20_000, - hold_penalty_weight: 3.5, // Used to fail, now passes - max_position_absolute: 2.0, - huber_delta: 1.0, - entropy_coefficient: 0.01, - transaction_cost_multiplier: 1.0, - use_per: true, - per_alpha: 0.6, - per_beta_start: 0.4, - use_dueling: false, - dueling_hidden_dim: 128, - n_steps: 1, - tau: 0.001, - use_distributional: false, - num_atoms: 51, - v_min: -1000.0, - v_max: 1000.0, - use_noisy_nets: false, - noisy_sigma_init: 0.5, - minimum_profit_factor: 1.5, - kelly_fractional: 0.5, - kelly_max_fraction: 0.25, - kelly_min_trades: 20, - volatility_window: 20, + fn test_validate_rejects_v_min_gte_v_max() { + // v_min == v_max: zero-width distributional support + let params_equal = DQNParams { + v_min: 1.0, + v_max: 1.0, ..DQNParams::default() }; - assert!(params.validate_for_hft_trendfollowing().is_ok()); // DISABLED + let err = params_equal.validate_for_hft_trendfollowing().unwrap_err(); + assert!(err.contains("v_min"), "error should mention v_min: {err}"); - // Valid case: Large buffer with high penalty should pass - let params_valid = DQNParams { - learning_rate: 1e-4, - batch_size: 128, - gamma: 0.99, - buffer_size: 100_000, - hold_penalty_weight: 3.5, - max_position_absolute: 2.0, - huber_delta: 1.0, - entropy_coefficient: 0.01, - transaction_cost_multiplier: 1.0, - use_per: true, - per_alpha: 0.6, - per_beta_start: 0.4, - use_dueling: false, - dueling_hidden_dim: 128, - n_steps: 1, - tau: 0.001, - use_distributional: false, - num_atoms: 51, - v_min: -1000.0, - v_max: 1000.0, - use_noisy_nets: false, - noisy_sigma_init: 0.5, - minimum_profit_factor: 1.5, - kelly_fractional: 0.5, - kelly_max_fraction: 0.25, - kelly_min_trades: 20, - volatility_window: 20, + // v_min > v_max: inverted distributional support + let params_inverted = DQNParams { + v_min: 100.0, + v_max: -100.0, ..DQNParams::default() }; - assert!(params_valid.validate_for_hft_trendfollowing().is_ok()); + assert!(params_inverted.validate_for_hft_trendfollowing().is_err()); + } + + #[test] + fn test_validate_rejects_num_atoms_below_two() { + let params = DQNParams { + num_atoms: 1, + ..DQNParams::default() + }; + let err = params.validate_for_hft_trendfollowing().unwrap_err(); + assert!(err.contains("num_atoms"), "error should mention num_atoms: {err}"); + + // Zero atoms + let params_zero = DQNParams { + num_atoms: 0, + ..DQNParams::default() + }; + assert!(params_zero.validate_for_hft_trendfollowing().is_err()); + } + + #[test] + fn test_validate_rejects_zero_batch_size() { + let params = DQNParams { + batch_size: 0, + ..DQNParams::default() + }; + let err = params.validate_for_hft_trendfollowing().unwrap_err(); + assert!(err.contains("batch_size"), "error should mention batch_size: {err}"); + } + + #[test] + fn test_validate_rejects_invalid_learning_rate() { + // Zero LR + let params_zero = DQNParams { + learning_rate: 0.0, + ..DQNParams::default() + }; + assert!(params_zero.validate_for_hft_trendfollowing().is_err()); + + // Negative LR + let params_neg = DQNParams { + learning_rate: -1e-4, + ..DQNParams::default() + }; + assert!(params_neg.validate_for_hft_trendfollowing().is_err()); + + // LR too high (> 0.1) + let params_high = DQNParams { + learning_rate: 0.2, + ..DQNParams::default() + }; + let err = params_high.validate_for_hft_trendfollowing().unwrap_err(); + assert!(err.contains("learning_rate"), "error should mention learning_rate: {err}"); + + // LR at upper bound (0.1) should pass + let params_boundary = DQNParams { + learning_rate: 0.1, + ..DQNParams::default() + }; + assert!(params_boundary.validate_for_hft_trendfollowing().is_ok()); + } + + #[test] + fn test_validate_rejects_invalid_gamma() { + // Zero gamma + let params_zero = DQNParams { + gamma: 0.0, + ..DQNParams::default() + }; + assert!(params_zero.validate_for_hft_trendfollowing().is_err()); + + // Negative gamma + let params_neg = DQNParams { + gamma: -0.5, + ..DQNParams::default() + }; + assert!(params_neg.validate_for_hft_trendfollowing().is_err()); + + // gamma > 1.0 + let params_over = DQNParams { + gamma: 1.01, + ..DQNParams::default() + }; + let err = params_over.validate_for_hft_trendfollowing().unwrap_err(); + assert!(err.contains("gamma"), "error should mention gamma: {err}"); + + // gamma at upper bound (1.0) should pass + let params_one = DQNParams { + gamma: 1.0, + ..DQNParams::default() + }; + assert!(params_one.validate_for_hft_trendfollowing().is_ok()); } #[test]