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

fix(dqn): production hardening — LR scheduler, urgency costs, F32 traps, atomic IO

Browse Source 
Wire LR scheduler to actually update the Adam optimizer (was logged but
never applied). Add update_learning_rate to DQN, RegimeConditionalDQN,
and DQNAgentType so decay_factor propagates through all agent variants.

Fix train/eval parity: CLI defaults 51→54 features, 3→45 actions;
DQN eval always uses 3-layer hidden_dims; PPO eval uses 5-layer value
network matching trainer. enhanced_ml.rs hardcoded config updated from
state_dim=16/num_actions=3 to 54/45.

Fix Candle F32/BF16 traps: replace `Tensor * 0.5` (f64 literal) with
broadcast_mul(Tensor::full(0.5_f32)) in quantile_regression.rs (2x),
dqn.rs Huber loss, and IQN gamma multiplication. Prevents panics on
Ampere+ BF16 GPUs.

Add urgency_weight() multiplier to training cost model in reward.rs
and portfolio_tracker.rs — urgency dimension (Patient/Normal/Aggressive)
now affects learned value function, matching evaluate_baseline.rs.

Fix equity tracking: additive (equity += ret) → multiplicative
(equity *= 1.0 + ret) in compute_metrics. Fix total return calc.

Fix PER beta annealing: epochs*70 → epochs*1000 (~1015 actual steps
per epoch for 130k bars / batch 128).

Fix silent target network freeze: mutex lock failure now propagates
error instead of silently skipping weight update.

Make save_checkpoint atomic (write .tmp then rename). Make NormStats
write atomic with error logging instead of silent discard.

Convert EnsembleConfig::new assert! → Result<Self, MLError> with
14 call site updates. Fix hyperopt result serialization to warn
instead of silently dropping to Value::Null.

Fix clippy MSRV mismatch: clippy.toml 1.75 → 1.85 matching Cargo.toml.

2732 tests pass, 0 clippy warnings across workspace.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
jgrusewski
2026-03-03 23:04:53 +01:00
parent d289b2f264
commit fc0754c63f
14 changed files with 225 additions and 127 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
crates/config/clippy.toml
View File

@@ -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 = []

14
crates/ml/examples/evaluate_baseline.rs
View File

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

20
crates/ml/examples/hyperopt_baseline_rl.rs
View File

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

8
crates/ml/examples/train_baseline_rl.rs
View File

@@ -75,16 +75,16 @@ struct Args {
#[arg(long)]
hyperopt_params: Option<PathBuf>,
/// 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

45
crates/ml/src/dqn/agent.rs
View File

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

51
crates/ml/src/dqn/dqn.rs
View File

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

64
crates/ml/src/dqn/ensemble.rs
View File

@@ -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<Self, MLError> {
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, MLError> {
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, MLError> {
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

14
crates/ml/src/dqn/portfolio_tracker.rs
View File

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

30
crates/ml/src/dqn/quantile_regression.rs
View File

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

8
crates/ml/src/dqn/regime_conditional.rs
View File

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

45
crates/ml/src/dqn/reward.rs
View File

@@ -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, &current_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, &current_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, &current_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
);

11
crates/ml/src/trainers/dqn/config.rs
View File

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

18
crates/ml/src/trainers/dqn/trainer.rs
View File

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

22
services/trading_service/src/services/enhanced_ml.rs
View File

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