diff --git a/crates/config/clippy.toml b/crates/config/clippy.toml index f4370e10c..a8fc27159 100644 --- a/crates/config/clippy.toml +++ b/crates/config/clippy.toml @@ -67,7 +67,7 @@ array-size-threshold = 256000 allow-unwrap-in-tests = true # Minimum Supported Rust Version -msrv = "1.75" +msrv = "1.85" # Standard library items to avoid (empty = allow all) disallowed-names = [] diff --git a/crates/ml/examples/evaluate_baseline.rs b/crates/ml/examples/evaluate_baseline.rs index e91844cab..4cba08a9f 100644 --- a/crates/ml/examples/evaluate_baseline.rs +++ b/crates/ml/examples/evaluate_baseline.rs @@ -280,7 +280,7 @@ fn compute_metrics(returns: &[f64], bars_per_year: f64) -> ComputedMetrics { let mut max_drawdown = 0.0_f64; for &ret in returns { - equity += ret; + equity *= 1.0 + ret; if equity > peak { peak = equity; } @@ -314,8 +314,8 @@ fn compute_metrics(returns: &[f64], bars_per_year: f64) -> ComputedMetrics { 0.0 }; - // Total return - let total_return_pct = sum * 100.0; + // Total return (from compounded equity curve) + let total_return_pct = (equity - 1.0) * 100.0; ComputedMetrics { sharpe_ratio, @@ -356,11 +356,7 @@ fn evaluate_dqn_fold( num_actions: args.num_actions, hidden_dims: { let base = hp_usize(hp, "hidden_dim_base").unwrap_or(128); - if base > 128 { - vec![base, base / 2, base / 4] // GPU-scaled: 3-layer network - } else { - vec![128, 64] // Default small network - } + vec![base, base / 2, base / 4] // Always 3-layer to match trainer.rs }, learning_rate: 1e-4, gamma: 0.95, // must match train_baseline (shorter horizon, 20 bars) @@ -532,7 +528,7 @@ fn evaluate_ppo_fold( }, value_hidden_dims: { let base = hp_usize(hp, "hidden_dim_base").unwrap_or(128); - vec![base * 2, base, base / 2] + vec![base * 4, base * 3, base * 2, base, base / 2] // 5-layer to match trainer }, policy_learning_rate: 3e-4, value_learning_rate: 1e-3, diff --git a/crates/ml/examples/hyperopt_baseline_rl.rs b/crates/ml/examples/hyperopt_baseline_rl.rs index 6541d0ce6..651dcd5d3 100644 --- a/crates/ml/examples/hyperopt_baseline_rl.rs +++ b/crates/ml/examples/hyperopt_baseline_rl.rs @@ -200,9 +200,13 @@ fn run_dqn_hyperopt(args: &Args, parallel: usize, device: &candle_core::Device) info!(" Total trials: {}", result.all_trials.len()); info!(" Elapsed: {:.1}s", elapsed); - let best_params_json = serde_json::to_value(&result.best_params) - .ok() - .unwrap_or(Value::Null); + let best_params_json = match serde_json::to_value(&result.best_params) { + Ok(v) => v, + Err(e) => { + warn!("Failed to serialize best hyperopt params: {}", e); + Value::Null + } + }; Ok(build_model_result( result.best_objective, @@ -276,9 +280,13 @@ fn run_ppo_hyperopt(args: &Args, parallel: usize, device: &candle_core::Device) info!(" Total trials: {}", result.all_trials.len()); info!(" Elapsed: {:.1}s", elapsed); - let best_params_json = serde_json::to_value(&result.best_params) - .ok() - .unwrap_or(Value::Null); + let best_params_json = match serde_json::to_value(&result.best_params) { + Ok(v) => v, + Err(e) => { + warn!("Failed to serialize best hyperopt params: {}", e); + Value::Null + } + }; Ok(build_model_result( result.best_objective, diff --git a/crates/ml/examples/train_baseline_rl.rs b/crates/ml/examples/train_baseline_rl.rs index 04a820401..770065deb 100644 --- a/crates/ml/examples/train_baseline_rl.rs +++ b/crates/ml/examples/train_baseline_rl.rs @@ -75,16 +75,16 @@ struct Args { #[arg(long)] hyperopt_params: Option, - /// Feature dimension (must match `extract_ml_features` output) - #[arg(long, default_value_t = 51)] + /// Feature dimension (51 market + 3 portfolio = 54, must match trainer `state_dim`) + #[arg(long, default_value_t = 54)] feature_dim: usize, /// Early stopping patience (epochs without improvement) #[arg(long, default_value_t = 10)] patience: usize, - /// Number of actions for the DQN/PPO action space - #[arg(long, default_value_t = 3)] + /// Number of actions (45 = 5 exposure x 3 order x 3 urgency) + #[arg(long, default_value_t = 45)] num_actions: usize, /// Walk-forward: initial training window in months diff --git a/crates/ml/src/dqn/agent.rs b/crates/ml/src/dqn/agent.rs index 559f48c83..519257169 100644 --- a/crates/ml/src/dqn/agent.rs +++ b/crates/ml/src/dqn/agent.rs @@ -555,28 +555,25 @@ impl DQNAgent { } fn update_target_network_weights(&mut self) -> Result<(), MLError> { - // Implement soft update of target network using Polyak averaging - let tau = self.config.tau; // BUG #4 FIX: Use configurable tau from config + let tau = self.config.tau; let main_vars = self.q_network.vars(); let target_vars = self.target_network.vars(); - // Soft update: θ_target = τ * θ_main + (1 - τ) * θ_target - if let (Ok(main_data), Ok(target_data)) = - (main_vars.data().lock(), target_vars.data().lock()) - { - for (main_var_name, main_var) in main_data.iter() { - if let Some(target_var) = target_data.get(main_var_name) { - // Get current values - let main_value = main_var.as_tensor(); - let target_value = target_var.as_tensor(); + // Soft update: theta_target = tau * theta_main + (1 - tau) * theta_target + let main_data = main_vars.data().lock().map_err(|e| { + MLError::LockError(format!("Failed to lock main network vars: {}", e)) + })?; + let target_data = target_vars.data().lock().map_err(|e| { + MLError::LockError(format!("Failed to lock target network vars: {}", e)) + })?; - // Compute soft update - let new_target_value = ((main_value * tau)? + (target_value * (1.0 - tau))?)?; - - // Update target variable - target_var.set(&new_target_value)?; - } + for (main_var_name, main_var) in main_data.iter() { + if let Some(target_var) = target_data.get(main_var_name) { + let main_value = main_var.as_tensor(); + let target_value = target_var.as_tensor(); + let new_target_value = ((main_value * tau)? + (target_value * (1.0 - tau))?)?; + target_var.set(&new_target_value)?; } } @@ -661,12 +658,20 @@ impl DQNAgent { }) .map_err(|e| MLError::TrainingError(format!("Failed to serialize checkpoint: {}", e)))?; - let mut file = File::create(path).map_err(|e| { - MLError::TrainingError(format!("Failed to create checkpoint file: {}", e)) + // Atomic write: write to .tmp sibling, then rename (POSIX atomic) + let tmp_path = path.with_extension("json.tmp"); + let mut file = File::create(&tmp_path).map_err(|e| { + MLError::TrainingError(format!("Failed to create tmp checkpoint file: {}", e)) })?; file.write_all(checkpoint_data.as_bytes()) - .map_err(|e| MLError::TrainingError(format!("Failed to write checkpoint: {}", e)))?; + .map_err(|e| MLError::TrainingError(format!("Failed to write tmp checkpoint: {}", e)))?; + + std::fs::rename(&tmp_path, path).map_err(|e| { + // Best-effort cleanup of tmp file on rename failure + drop(std::fs::remove_file(&tmp_path)); + MLError::TrainingError(format!("Failed to rename checkpoint: {}", e)) + })?; Ok(()) } diff --git a/crates/ml/src/dqn/dqn.rs b/crates/ml/src/dqn/dqn.rs index 62e6f9ec6..6c0cb2823 100644 --- a/crates/ml/src/dqn/dqn.rs +++ b/crates/ml/src/dqn/dqn.rs @@ -1909,7 +1909,9 @@ impl DQN { .unsqueeze(1)? .broadcast_as((batch_size, num_quantiles))?; - let target_quantiles = (rewards_broadcast + (next_quantiles * not_done_broadcast)? * gamma as f64)? + let gamma_t = Tensor::full(gamma, &[batch_size, num_quantiles], &device) + .map_err(|e| MLError::TrainingError(format!("Failed to create gamma tensor: {}", e)))?; + let target_quantiles = (rewards_broadcast + (next_quantiles * not_done_broadcast)?.broadcast_mul(&gamma_t)?)? .detach(); // Per-sample quantile Huber loss (Dabney et al. 2018b, Eq. 10) @@ -2099,7 +2101,11 @@ impl DQN { let abs_diff = weighted_diff.abs()?; // Element-wise Huber loss (use weighted_diff for loss computation) - let squared_loss = ((&weighted_diff * &weighted_diff)? * 0.5)?; // 0.5 * x^2 + let half = Tensor::from_vec(vec![0.5_f32; batch_size], batch_size, device) + .map_err(|e| { + MLError::TrainingError(format!("Failed to create half tensor: {}", e)) + })?; + let squared_loss = (&weighted_diff * &weighted_diff)?.broadcast_mul(&half)?; // 0.5 * x^2 // Create delta tensor for operations let delta_tensor = Tensor::from_vec(vec![delta; batch_size], batch_size, device) @@ -2361,6 +2367,47 @@ impl DQN { } } + /// Update the optimizer's learning rate by recreating it with `new_lr = current_lr * decay_factor`. + /// + /// Called by the LR scheduler at epoch boundaries. If the optimizer has not + /// been initialised yet (no `train_step` calls), this is a no-op. + pub fn update_learning_rate(&mut self, decay_factor: f64) -> Result<(), MLError> { + if let Some(ref optimizer) = self.optimizer { + let current_lr = optimizer.learning_rate(); + let new_lr = current_lr * decay_factor; + + let adam_params = ParamsAdam { + lr: new_lr, + beta_1: 0.9, + beta_2: 0.999, + eps: 1.5e-4, // Rainbow DQN standard (matches train_step init) + weight_decay: (self.config.weight_decay > 0.0) + .then_some(Decay::DecoupledWeightDecay(self.config.weight_decay)), + amsgrad: false, + }; + + // Collect vars with the same priority as train_step: + // IQN+base > hybrid > dueling > standard + let mut vars = if let Some(ref dist_dueling_net) = self.dist_dueling_q_network { + dist_dueling_net.vars().all_vars() + } else if let Some(ref dueling_net) = self.dueling_q_network { + dueling_net.vars().all_vars() + } else { + self.q_network.vars().all_vars() + }; + if let Some(ref iqn_net) = self.iqn_network { + vars.extend(iqn_net.vars().all_vars()); + } + + self.optimizer = Some( + Adam::new(vars, adam_params).map_err(|e| { + MLError::TrainingError(format!("Failed to update learning rate: {}", e)) + })?, + ); + } + Ok(()) + } + /// Update target networks using cosine-annealed EMA (soft) or hard copy. /// /// This is the shared implementation used by both [`train_step`] and diff --git a/crates/ml/src/dqn/ensemble.rs b/crates/ml/src/dqn/ensemble.rs index 01528cabd..68b79bc34 100644 --- a/crates/ml/src/dqn/ensemble.rs +++ b/crates/ml/src/dqn/ensemble.rs @@ -24,7 +24,7 @@ //! use ml::dqn::{DQNEnsemble, EnsembleConfig, VotingStrategy, DQNConfig}; //! //! // Create ensemble with 3 agents -//! let config = EnsembleConfig::new(3, 128, VotingStrategy::QValueWeighted); +//! let config = EnsembleConfig::new(3, 128, VotingStrategy::QValueWeighted)?; //! let mut ensemble = DQNEnsemble::new(config)?; //! //! // Select action via ensemble voting @@ -92,14 +92,14 @@ impl EnsembleConfig { /// * `num_agents` - Number of agents (3-5) /// * `state_dim` - State dimension /// * `voting_strategy` - Voting strategy - pub fn new(num_agents: usize, state_dim: usize, voting_strategy: VotingStrategy) -> Self { - assert!( - (3..=5).contains(&num_agents), - "num_agents must be 3-5, got {}", - num_agents - ); + pub fn new(num_agents: usize, state_dim: usize, voting_strategy: VotingStrategy) -> Result { + if !(3..=5).contains(&num_agents) { + return Err(MLError::ConfigError { + reason: format!("num_agents must be 3-5, got {}", num_agents), + }); + } - Self { + Ok(Self { num_agents, state_dim, voting_strategy, @@ -108,16 +108,16 @@ impl EnsembleConfig { thompson_decay: 0.99, enable_diversity_penalty: true, diversity_penalty_weight: 0.1, - } + }) } /// Create config for production use (conservative, QValueWeighted) - pub fn production(state_dim: usize) -> Self { + pub fn production(state_dim: usize) -> Result { Self::new(3, state_dim, VotingStrategy::QValueWeighted) } /// Create config for exploration (aggressive, MaxVariance) - pub fn exploration(state_dim: usize) -> Self { + pub fn exploration(state_dim: usize) -> Result { Self::new(5, state_dim, VotingStrategy::MaxVariance) } } @@ -802,7 +802,7 @@ mod tests { #[test] fn test_ensemble_creation() -> anyhow::Result<()> { - let config = EnsembleConfig::new(3, 128, VotingStrategy::Majority); + let config = EnsembleConfig::new(3, 128, VotingStrategy::Majority)?; let ensemble = DQNEnsemble::new(config)?; assert_eq!(ensemble.num_agents(), 3); @@ -812,20 +812,18 @@ mod tests { #[test] fn test_ensemble_config_validation() { - let result = std::panic::catch_unwind(|| { - EnsembleConfig::new(2, 128, VotingStrategy::Majority) // Too few agents - }); + // Too few agents + let result = EnsembleConfig::new(2, 128, VotingStrategy::Majority); assert!(result.is_err()); - let result = std::panic::catch_unwind(|| { - EnsembleConfig::new(6, 128, VotingStrategy::Majority) // Too many agents - }); + // Too many agents + let result = EnsembleConfig::new(6, 128, VotingStrategy::Majority); assert!(result.is_err()); } #[test] fn test_majority_voting() -> anyhow::Result<()> { - let config = EnsembleConfig::new(3, 128, VotingStrategy::Majority); + let config = EnsembleConfig::new(3, 128, VotingStrategy::Majority)?; let ensemble = DQNEnsemble::new(config)?; // 2 Buy, 1 Sell -> Buy wins @@ -843,7 +841,7 @@ mod tests { #[test] fn test_q_value_weighted_voting() -> anyhow::Result<()> { - let config = EnsembleConfig::new(3, 128, VotingStrategy::QValueWeighted); + let config = EnsembleConfig::new(3, 128, VotingStrategy::QValueWeighted)?; let ensemble = DQNEnsemble::new(config)?; let actions = vec![TradingAction::Buy, TradingAction::Sell, TradingAction::Hold]; @@ -862,7 +860,7 @@ mod tests { #[test] fn test_min_variance_voting() -> anyhow::Result<()> { - let config = EnsembleConfig::new(3, 128, VotingStrategy::MinVariance); + let config = EnsembleConfig::new(3, 128, VotingStrategy::MinVariance)?; let ensemble = DQNEnsemble::new(config)?; let q_values = vec![ @@ -881,7 +879,7 @@ mod tests { #[test] fn test_max_variance_voting() -> anyhow::Result<()> { - let config = EnsembleConfig::new(3, 128, VotingStrategy::MaxVariance); + let config = EnsembleConfig::new(3, 128, VotingStrategy::MaxVariance)?; let ensemble = DQNEnsemble::new(config)?; let q_values = vec![ @@ -899,7 +897,7 @@ mod tests { #[test] fn test_action_selection_consistency() -> anyhow::Result<()> { - let config = EnsembleConfig::new(3, 128, VotingStrategy::Majority); + let config = EnsembleConfig::new(3, 128, VotingStrategy::Majority)?; let mut ensemble = DQNEnsemble::new(config)?; // Select action 10 times (should not crash) @@ -917,7 +915,7 @@ mod tests { #[test] fn test_training_with_shared_buffer() -> anyhow::Result<()> { - let mut config = EnsembleConfig::new(3, 128, VotingStrategy::Majority); + let mut config = EnsembleConfig::new(3, 128, VotingStrategy::Majority)?; config.shared_replay_buffer = true; let mut ensemble = DQNEnsemble::new(config)?; @@ -940,7 +938,7 @@ mod tests { #[test] fn test_training_with_separate_buffers() -> anyhow::Result<()> { - let config = EnsembleConfig::new(3, 128, VotingStrategy::Majority); + let config = EnsembleConfig::new(3, 128, VotingStrategy::Majority)?; let mut ensemble = DQNEnsemble::new(config)?; let experience = Experience::new( @@ -959,7 +957,7 @@ mod tests { #[test] fn test_epsilon_update() -> anyhow::Result<()> { - let config = EnsembleConfig::new(3, 128, VotingStrategy::Majority); + let config = EnsembleConfig::new(3, 128, VotingStrategy::Majority)?; let mut ensemble = DQNEnsemble::new(config)?; let initial_epsilons = ensemble.get_epsilons(); @@ -976,7 +974,7 @@ mod tests { #[test] fn test_temperature_update() -> anyhow::Result<()> { - let config = EnsembleConfig::new(3, 128, VotingStrategy::Majority); + let config = EnsembleConfig::new(3, 128, VotingStrategy::Majority)?; let mut ensemble = DQNEnsemble::new(config)?; let initial_temps = ensemble.get_temperatures(); @@ -992,18 +990,20 @@ mod tests { } #[test] - fn test_production_config() { - let config = EnsembleConfig::production(128); + fn test_production_config() -> anyhow::Result<()> { + let config = EnsembleConfig::production(128)?; assert_eq!(config.num_agents, 3); assert_eq!(config.voting_strategy, VotingStrategy::QValueWeighted); assert!(!config.shared_replay_buffer); + Ok(()) } #[test] - fn test_exploration_config() { - let config = EnsembleConfig::exploration(128); + fn test_exploration_config() -> anyhow::Result<()> { + let config = EnsembleConfig::exploration(128)?; assert_eq!(config.num_agents, 5); assert_eq!(config.voting_strategy, VotingStrategy::MaxVariance); + Ok(()) } #[test] @@ -1041,7 +1041,7 @@ mod tests { #[test] fn test_architectural_diversity() -> anyhow::Result<()> { - let config = EnsembleConfig::new(5, 128, VotingStrategy::Majority); + let config = EnsembleConfig::new(5, 128, VotingStrategy::Majority)?; let ensemble = DQNEnsemble::new(config)?; // Verify agents have different architectures diff --git a/crates/ml/src/dqn/portfolio_tracker.rs b/crates/ml/src/dqn/portfolio_tracker.rs index 857ddc2d9..a8db4c7ea 100644 --- a/crates/ml/src/dqn/portfolio_tracker.rs +++ b/crates/ml/src/dqn/portfolio_tracker.rs @@ -312,8 +312,9 @@ impl PortfolioTracker { self.position_size * price // position_size is negative, so this is negative }; - // Apply Phase 1 transaction cost - let phase1_tx_cost = phase1_cost * tx_cost_rate; + // Apply Phase 1 transaction cost (scaled by urgency) + let urgency_mult = action.urgency_weight() as f32; + let phase1_tx_cost = phase1_cost * tx_cost_rate * urgency_mult; self.cumulative_transaction_costs += phase1_tx_cost; self.cash += phase1_cash_change - phase1_tx_cost; @@ -367,7 +368,7 @@ impl PortfolioTracker { // Apply Phase 2 transaction cost let phase2_cost = actual_phase2_position.abs() * price; - let phase2_tx_cost = phase2_cost * tx_cost_rate; + let phase2_tx_cost = phase2_cost * tx_cost_rate * urgency_mult; self.cumulative_transaction_costs += phase2_tx_cost; self.cash -= actual_phase2_position * price + phase2_tx_cost; @@ -383,14 +384,15 @@ impl PortfolioTracker { } // Calculate position change - // Non-reversal path: apply transaction costs + // Non-reversal path: apply transaction costs (scaled by urgency) let tx_cost_rate = action.transaction_cost() as f32; + let urgency_mult = action.urgency_weight() as f32; let position_delta = target_position - self.position_size; - + // Calculate transaction cost for this trade if position_delta.abs() > 0.0 { let trade_value = position_delta.abs() * price; - let tx_cost = trade_value * tx_cost_rate; + let tx_cost = trade_value * tx_cost_rate * urgency_mult; self.cumulative_transaction_costs += tx_cost; // Update cash accounting for transaction costs diff --git a/crates/ml/src/dqn/quantile_regression.rs b/crates/ml/src/dqn/quantile_regression.rs index 03caf7d3b..180426533 100644 --- a/crates/ml/src/dqn/quantile_regression.rs +++ b/crates/ml/src/dqn/quantile_regression.rs @@ -335,22 +335,23 @@ pub fn quantile_huber_loss( let abs_errors = td_errors.abs()?; let kappa_tensor = Tensor::full(kappa, abs_errors.shape(), device)?; - // L_κ(u) = {0.5*u² if |u|≤κ, κ(|u|-0.5κ) if |u|>κ} - let quadratic = ((&td_errors * &td_errors)? * 0.5)?; // 0.5*u² + // L_kappa(u) = {0.5*u^2 if |u|<=kappa, kappa(|u|-0.5*kappa) if |u|>kappa} + let half = Tensor::full(0.5_f32, abs_errors.shape(), device)?; + let quadratic = (&td_errors * &td_errors)?.broadcast_mul(&half)?; // 0.5*u^2 let half_kappa = Tensor::full(kappa / 2.0, abs_errors.shape(), device)?; let kappa_scalar = Tensor::full(kappa, abs_errors.shape(), device)?; - let linear = ((&abs_errors - &half_kappa)? * &kappa_scalar)?; // κ(|u| - 0.5κ) + let linear = ((&abs_errors - &half_kappa)? * &kappa_scalar)?; // kappa(|u| - 0.5*kappa) - // Mask for |u| ≤ κ + // Mask for |u| <= kappa let mask = abs_errors.le(&kappa_tensor)?; let huber_loss = mask.where_cond(&quadratic, &linear)?; - // Quantile asymmetry: ρ_τ(u) = |τ - 𝟙{u < 0}| * L_κ(u) + // Quantile asymmetry: rho_tau(u) = |tau - 1{u < 0}| * L_kappa(u) let zero_tensor = Tensor::zeros(td_errors.shape(), DType::F32, device)?; - let indicator = td_errors.lt(&zero_tensor)?; // 𝟙{u < 0} + let indicator = td_errors.lt(&zero_tensor)?; // 1{u < 0} let indicator_f32 = indicator.to_dtype(DType::F32)?; - // |τ - 𝟙{u < 0}| + // |tau - 1{u < 0}| let asymmetric_weight = (taus - indicator_f32)?.abs()?; // ρ_τ(u) = asymmetric_weight * huber_loss @@ -389,22 +390,23 @@ pub fn quantile_huber_loss_per_sample( let abs_errors = td_errors.abs()?; let kappa_tensor = Tensor::full(kappa, abs_errors.shape(), device)?; - // L_κ(u) = {0.5*u² if |u|≤κ, κ(|u|-0.5κ) if |u|>κ} - let quadratic = ((&td_errors * &td_errors)? * 0.5)?; // 0.5*u² + // L_kappa(u) = {0.5*u^2 if |u|<=kappa, kappa(|u|-0.5*kappa) if |u|>kappa} + let half = Tensor::full(0.5_f32, abs_errors.shape(), device)?; + let quadratic = (&td_errors * &td_errors)?.broadcast_mul(&half)?; // 0.5*u^2 let half_kappa = Tensor::full(kappa / 2.0, abs_errors.shape(), device)?; let kappa_scalar = Tensor::full(kappa, abs_errors.shape(), device)?; - let linear = ((&abs_errors - &half_kappa)? * &kappa_scalar)?; // κ(|u| - 0.5κ) + let linear = ((&abs_errors - &half_kappa)? * &kappa_scalar)?; // kappa(|u| - 0.5*kappa) - // Mask for |u| ≤ κ + // Mask for |u| <= kappa let mask = abs_errors.le(&kappa_tensor)?; let huber_loss = mask.where_cond(&quadratic, &linear)?; - // Quantile asymmetry: ρ_τ(u) = |τ - 𝟙{u < 0}| * L_κ(u) + // Quantile asymmetry: rho_tau(u) = |tau - 1{u < 0}| * L_kappa(u) let zero_tensor = Tensor::zeros(td_errors.shape(), DType::F32, device)?; - let indicator = td_errors.lt(&zero_tensor)?; // 𝟙{u < 0} + let indicator = td_errors.lt(&zero_tensor)?; // 1{u < 0} let indicator_f32 = indicator.to_dtype(DType::F32)?; - // |τ - 𝟙{u < 0}| + // |tau - 1{u < 0}| let asymmetric_weight = (taus - indicator_f32)?.abs()?; // ρ_τ(u) = asymmetric_weight * huber_loss → [batch, num_quantiles] diff --git a/crates/ml/src/dqn/regime_conditional.rs b/crates/ml/src/dqn/regime_conditional.rs index f14021d39..7e66cb9cc 100644 --- a/crates/ml/src/dqn/regime_conditional.rs +++ b/crates/ml/src/dqn/regime_conditional.rs @@ -831,6 +831,14 @@ impl RegimeConditionalDQN { self.volatile_head.reset_target_network()?; Ok(()) } + + /// Update learning rate for all regime heads + pub fn update_learning_rate(&mut self, decay_factor: f64) -> Result<(), crate::MLError> { + self.trending_head.update_learning_rate(decay_factor)?; + self.ranging_head.update_learning_rate(decay_factor)?; + self.volatile_head.update_learning_rate(decay_factor)?; + Ok(()) + } } #[cfg(test)] diff --git a/crates/ml/src/dqn/reward.rs b/crates/ml/src/dqn/reward.rs index 1f6bded5f..8eae9f5cc 100644 --- a/crates/ml/src/dqn/reward.rs +++ b/crates/ml/src/dqn/reward.rs @@ -860,15 +860,21 @@ impl RewardFunction { Decimal::try_from(*current_state.portfolio_features.get(0).unwrap_or(&100000.0) as f64) .unwrap_or(Decimal::try_from(100000.0).unwrap_or(Decimal::ZERO)); - // For percentage-based penalty: cost_rate × position_change - // This gives a penalty proportional to the trade size - // Example: 1.0 position change × 0.0015 = 0.0015 penalty (0.15% of portfolio) - let cost_penalty = position_change * tx_cost_rate; - + // Apply urgency multiplier: Patient=0.5x, Normal=1.0x, Aggressive=1.5x + // This ensures the urgency dimension affects the learned value function during training + let urgency_mult = Decimal::try_from(action.urgency_weight()) + .unwrap_or(Decimal::ONE); + + // For percentage-based penalty: cost_rate × position_change × urgency + // This gives a penalty proportional to the trade size and urgency + // Example: 1.0 position change × 0.0015 × 1.5 (aggressive) = 0.00225 penalty + let cost_penalty = position_change * tx_cost_rate * urgency_mult; + tracing::trace!( - "Transaction cost: position_change={:.4}, tx_rate={:.4}, penalty={:.6}", + "Transaction cost: position_change={:.4}, tx_rate={:.4}, urgency={:.2}, penalty={:.6}", position_change, tx_cost_rate, + urgency_mult, cost_penalty ); @@ -1138,15 +1144,15 @@ mod tests { ); let market_cost = reward_fn.calculate_cost_penalty(market_action, ¤t_state, &next_state); - // Expected: 1.0 position change × 0.0015 = 0.0015 (0.15%) + // Expected: 1.0 position change × 0.0015 × 1.0 (Normal urgency) = 0.0015 (0.15%) let expected_market = Decimal::try_from(0.0015).unwrap(); assert!( (market_cost - expected_market).abs() < Decimal::try_from(0.0001).unwrap(), - "Market order cost should be 0.0015, got {}", + "Market+Normal order cost should be 0.0015, got {}", market_cost ); - // Test LimitMaker order (0.05% fee) + // Test LimitMaker order (0.05% fee) with Patient urgency (0.5x) let limit_action = FactoredAction::new( ExposureLevel::Long100, OrderType::LimitMaker, @@ -1154,15 +1160,15 @@ mod tests { ); let limit_cost = reward_fn.calculate_cost_penalty(limit_action, ¤t_state, &next_state); - // Expected: 1.0 position change × 0.0005 = 0.0005 (0.05%) - let expected_limit = Decimal::try_from(0.0005).unwrap(); + // Expected: 1.0 position change × 0.0005 × 0.5 (Patient urgency) = 0.00025 + let expected_limit = Decimal::try_from(0.00025).unwrap(); assert!( (limit_cost - expected_limit).abs() < Decimal::try_from(0.0001).unwrap(), - "LimitMaker order cost should be 0.0005, got {}", + "LimitMaker+Patient order cost should be 0.00025, got {}", limit_cost ); - // Test IoC order (0.10% fee) + // Test IoC order (0.10% fee) with Aggressive urgency (1.5x) let ioc_action = FactoredAction::new( ExposureLevel::Long100, OrderType::IoC, @@ -1170,18 +1176,19 @@ mod tests { ); let ioc_cost = reward_fn.calculate_cost_penalty(ioc_action, ¤t_state, &next_state); - // Expected: 1.0 position change × 0.0010 = 0.0010 (0.10%) - let expected_ioc = Decimal::try_from(0.0010).unwrap(); + // Expected: 1.0 position change × 0.0010 × 1.5 (Aggressive urgency) = 0.0015 + let expected_ioc = Decimal::try_from(0.0015).unwrap(); assert!( (ioc_cost - expected_ioc).abs() < Decimal::try_from(0.0001).unwrap(), - "IoC order cost should be 0.0010, got {}", + "IoC+Aggressive order cost should be 0.0015, got {}", ioc_cost ); - // Verify market order is 3x more expensive than limit maker + // Verify market+normal is 6x more expensive than limit+patient + // (0.0015 / 0.00025 = 6.0) assert!( - market_cost > limit_cost * Decimal::try_from(2.5).unwrap(), - "Market order ({}) should be ~3x more expensive than LimitMaker ({})", + market_cost > limit_cost * Decimal::try_from(5.0).unwrap(), + "Market+Normal ({}) should be ~6x more expensive than LimitMaker+Patient ({})", market_cost, limit_cost ); diff --git a/crates/ml/src/trainers/dqn/config.rs b/crates/ml/src/trainers/dqn/config.rs index 4fe0e6645..e43cbbc96 100644 --- a/crates/ml/src/trainers/dqn/config.rs +++ b/crates/ml/src/trainers/dqn/config.rs @@ -395,6 +395,17 @@ impl DQNAgentType { pub fn step_replay_buffer(&self) { self.memory().step(); } + + /// Update learning rate for the optimizer(s) by applying a decay factor. + /// + /// For Standard agents, updates the single optimizer. + /// For RegimeConditional agents, updates all three regime head optimizers. + pub fn update_learning_rate(&mut self, decay_factor: f64) -> Result<(), crate::MLError> { + match self { + Self::Standard(agent) => agent.update_learning_rate(decay_factor), + Self::RegimeConditional(agent) => agent.update_learning_rate(decay_factor), + } + } } /// DQN training hyperparameters from gRPC request diff --git a/crates/ml/src/trainers/dqn/trainer.rs b/crates/ml/src/trainers/dqn/trainer.rs index 502f585e6..5e2b2518b 100644 --- a/crates/ml/src/trainers/dqn/trainer.rs +++ b/crates/ml/src/trainers/dqn/trainer.rs @@ -412,7 +412,7 @@ impl DQNTrainer { per_alpha: hyperparams.per_alpha, per_beta_start: hyperparams.per_beta_start, per_beta_max: 1.0, - per_beta_annealing_steps: hyperparams.epochs * 70, // ~70 steps/epoch estimate + per_beta_annealing_steps: hyperparams.epochs * 1000, // ~1000 steps/epoch (130k bars / 128 batch_size) // Wave 2.1: Dueling Networks (ENABLED BY DEFAULT - Wave 6.4) use_dueling: hyperparams.use_dueling, @@ -1200,7 +1200,11 @@ impl DQNTrainer { let reward = reward_decimal.to_f32().unwrap_or(0.0); let max_q = max_q_values[i] as f64; - // Loss = (predicted_q - reward)^2 + // Reward prediction error: (max_Q - reward)^2 + // NOTE: This is NOT the Bellman residual (which would be (r + gamma*Q_next - Q_pred)^2). + // It measures how well the Q-network's greedy Q-value predicts immediate rewards, + // serving as a proxy for overfitting detection — rising values indicate the + // Q-function is drifting from grounded reward signals. let loss = (max_q - reward as f64).powi(2); total_loss += loss; } @@ -2677,8 +2681,13 @@ impl DQNTrainer { self.lr_scheduler.get_initial_lr() ); } - // Note: Actual optimizer LR update would happen here in a real implementation - // For now, we log the scheduled LR for monitoring purposes + // Apply scheduled LR to the optimizer + let initial_lr = self.lr_scheduler.get_initial_lr(); + if initial_lr > 0.0 { + let decay_factor = current_lr / initial_lr; + self.agent.write().await.update_learning_rate(decay_factor)?; + } + training_metrics::set_learning_rate("dqn", "current", current_lr); // WAVE 9-11: Log Q-value range for production monitoring if train_step_count > 0 { @@ -2731,7 +2740,6 @@ impl DQNTrainer { training_metrics::set_q_value_stats("dqn", "current", q_mean, q_max); training_metrics::set_gradient_norm("dqn", "current", avg_grad_norm); training_metrics::set_epoch_duration("dqn", "current", epoch_duration.as_secs_f64()); - training_metrics::set_learning_rate("dqn", "current", current_lr); { let agent = self.agent.read().await; if let Ok(buf_size) = agent.get_replay_buffer_size() { diff --git a/services/trading_service/src/services/enhanced_ml.rs b/services/trading_service/src/services/enhanced_ml.rs index 4fff39d14..e83de7599 100644 --- a/services/trading_service/src/services/enhanced_ml.rs +++ b/services/trading_service/src/services/enhanced_ml.rs @@ -1444,11 +1444,13 @@ impl RealDQNModel { use ml::dqn::agent::DQNAgent; use ml::dqn::DQNConfig; - // DQN configuration matching paper trading config + // TODO: Load architecture config from checkpoint metadata instead of hardcoding + // DQN configuration matching production training defaults + // (51 market + 3 portfolio features, 5 exposure x 3 order x 3 urgency factored actions) let config = DQNConfig { - state_dim: 16, - num_actions: 3, - hidden_dims: vec![256, 128], + state_dim: 54, + num_actions: 45, + hidden_dims: vec![128, 64, 32], learning_rate: 0.0001, gamma: 0.99, epsilon_start: 0.1, @@ -1638,7 +1640,9 @@ impl RealPPOModel { use ml::ppo::gae::GAEConfig; use ml::ppo::{PPOConfig, WorkingPPO}; - // PPO configuration matching paper trading config + // TODO: Load architecture config from checkpoint metadata instead of hardcoding + // PPO configuration matching production training defaults + // (51 market + 3 portfolio features, 5 exposure x 3 order x 3 urgency factored actions) let gae_config = GAEConfig { gamma: 0.99, lambda: 0.95, @@ -1646,10 +1650,10 @@ impl RealPPOModel { }; let config = PPOConfig { - state_dim: 16, - num_actions: 3, - policy_hidden_dims: vec![256, 128], - value_hidden_dims: vec![256, 128], + state_dim: 54, + num_actions: 45, + policy_hidden_dims: vec![128, 64, 32], + value_hidden_dims: vec![512, 384, 256, 128, 64], policy_learning_rate: 0.0003, value_learning_rate: 0.001, clip_epsilon: 0.2,