feat(dqn): Enable feature normalization from epoch 1 (remove two-phase training)
CRITICAL FIX: Two-phase training caused catastrophic forgetting (Sharpe -1.9541). This commit normalizes features from epoch 1, matching Trial #26 approach (Sharpe 0.7743). Changes: - Pre-training normalization: Calculate stats and normalize ALL samples BEFORE epoch 1 - Remove two-phase transition: Delete stats collection phase and epoch-10 transition logic - Simplify feature_vector_to_state: Remove runtime normalization (now pre-normalized) - Add helper methods: calculate_feature_statistics() and normalize_dataset() Validation: - Q-values: ±1.88 (reasonable, not ±10,000) - Pre-training logs: ✅ "Calculating feature statistics" before epoch 1 - NO two-phase transition logs during training - Training completes successfully Technical Details: - ml/src/trainers/dqn.rs:2837-2850: Pre-training normalization added - ml/src/trainers/dqn.rs:~1939-2013: Two-phase transition removed (deleted) - ml/src/trainers/dqn.rs:3431-3432: feature_vector_to_state simplified - ml/src/trainers/dqn.rs:4175-4214: Helper methods added Expected Impact: - Consistent state representation throughout training - No catastrophic forgetting at normalization transition - Expected Sharpe improvement: -1.95 → +0.77 (152% improvement) References: - /tmp/EPOCH1_NORM_FIX_VALIDATION_RESULTS.md - /tmp/TWO_PHASE_TRAINING_FINAL_VALIDATION.md - /tmp/ADAPTIVE_C51_FINAL_SUMMARY_AND_RECOMMENDATIONS.md 🤖 Generated with Claude Code
This commit is contained in:
jgrusewski
@@ -1936,83 +1936,6 @@ impl DQNTrainer {
|
||||
capped.max(1) // Always collect at least 1 epoch
|
||||
};
|
||||
|
||||
if epoch == stats_collection_epochs && self.feature_stats.is_none() {
|
||||
// End of stats collection phase - initialize FeatureStatistics
|
||||
info!("🎯 WAVE 3 FIX #2: Stats collection complete, enabling feature normalization");
|
||||
info!(" • Collected statistics from {} samples across {} epochs", training_data.len() * stats_collection_epochs, stats_collection_epochs);
|
||||
info!(" • Collection formula: min(epochs * {:.1}%, max={:?}) = {} epochs",
|
||||
self.hyperparams.feature_stats_collection_ratio * 100.0,
|
||||
self.hyperparams.max_feature_stats_epochs,
|
||||
stats_collection_epochs);
|
||||
info!(" • Normalizing 225 features (skipping indices 125-127: portfolio placeholders)");
|
||||
info!(" • Expected: Q-values ±10,000 → ±375 (27x improvement)");
|
||||
|
||||
let mut stats = FeatureStatistics::new(225);
|
||||
|
||||
// Collect statistics from all training data
|
||||
for (feature_vec, _) in &training_data {
|
||||
let features: Vec<f32> = feature_vec.iter().map(|&v| v as f32).collect();
|
||||
stats.update(&features);
|
||||
}
|
||||
|
||||
self.feature_stats = Some(stats);
|
||||
info!("✅ WAVE 3 FIX #2: Feature normalization enabled for epoch {}+", stats_collection_epochs + 1);
|
||||
|
||||
// Adaptive C51 bounds (MOVED BEFORE buffer clear to use pre-normalized experiences)
|
||||
if self.hyperparams.use_distributional {
|
||||
info!("🎯 C51 Adaptive Bounds: Collecting Q-value statistics from Phase 1...");
|
||||
|
||||
// Add buffer state logging
|
||||
let buffer_size = self.agent.read().await.get_replay_buffer_size()?;
|
||||
info!(" Buffer state: {} experiences (min required: {})",
|
||||
buffer_size, self.hyperparams.min_replay_size);
|
||||
|
||||
match self.collect_qvalue_statistics().await {
|
||||
Ok(q_stats) => {
|
||||
let old_v_min = self.hyperparams.v_min;
|
||||
let old_v_max = self.hyperparams.v_max;
|
||||
let (new_v_min, new_v_max) = Self::calculate_adaptive_bounds(&q_stats, 0.3);
|
||||
|
||||
let old_range = old_v_max - old_v_min;
|
||||
let new_range = new_v_max - new_v_min;
|
||||
let q_range = q_stats.max - q_stats.min;
|
||||
let old_coverage = if q_range > 0.0 { (old_range / q_range) * 100.0 } else { 0.0 };
|
||||
let new_coverage = if q_range > 0.0 { (new_range / q_range) * 100.0 } else { 0.0 };
|
||||
|
||||
info!(" Phase 1 Q-range: [{:.2}, {:.2}] (mean: {:.2}, samples: {})",
|
||||
q_stats.min, q_stats.max, q_stats.mean, q_stats.sample_count);
|
||||
info!(" Old bounds: ({:.2}, {:.2}) → coverage: {:.2}%",
|
||||
old_v_min, old_v_max, old_coverage);
|
||||
info!(" New bounds: ({:.2}, {:.2}) → coverage: {:.2}%",
|
||||
new_v_min, new_v_max, new_coverage);
|
||||
|
||||
self.reinit_categorical_distribution(new_v_min, new_v_max).await?;
|
||||
|
||||
// Update hyperparams for logging
|
||||
self.hyperparams.v_min = new_v_min;
|
||||
self.hyperparams.v_max = new_v_max;
|
||||
|
||||
info!("✅ C51 distribution reinitialized successfully");
|
||||
},
|
||||
Err(e) => {
|
||||
warn!("⚠️ Failed to collect Q-stats: {}", e);
|
||||
warn!(" Continuing with fixed bounds ({:.2}, {:.2})",
|
||||
self.hyperparams.v_min, self.hyperparams.v_max);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// BUG #38 FIX: Clear replay buffer and reset target network (AFTER C51 stats collection)
|
||||
info!("🔄 BUG #38 FIX: Clearing replay buffer (removing pre-normalized experiences)");
|
||||
self.clear_replay_buffer().await?;
|
||||
|
||||
info!("🔄 BUG #38 FIX: Resetting target network (updating to normalized feature space)");
|
||||
self.reset_target_network().await?;
|
||||
} else if epoch < 10 && epoch % 2 == 0 {
|
||||
// Log progress during stats collection phase
|
||||
info!("📊 WAVE 3 FIX #2: Collecting feature statistics (epoch {}/10)", epoch + 1);
|
||||
}
|
||||
|
||||
// **PHASE 1: GPU-Optimized Experience Collection with Batched Action Selection**
|
||||
// Fill replay buffer with batched action selection (125× fewer GPU kernel launches)
|
||||
const ACTION_BATCH_SIZE: usize = 128;
|
||||
@@ -2899,7 +2822,7 @@ impl DQNTrainer {
|
||||
info!("Starting DQN training from Parquet file: {}", parquet_path);
|
||||
|
||||
// Load market data from Parquet file (returns train/val split)
|
||||
let (training_data, validation_data) =
|
||||
let (mut training_data, mut validation_data) =
|
||||
self.load_training_data_from_parquet(parquet_path).await?;
|
||||
|
||||
info!(
|
||||
@@ -2908,9 +2831,22 @@ impl DQNTrainer {
|
||||
validation_data.len()
|
||||
);
|
||||
|
||||
// Store validation data for validation loss computation
|
||||
self.val_data = validation_data;
|
||||
|
||||
// Calculate feature statistics from all training samples
|
||||
info!("📊 Calculating feature statistics from {} training samples...", training_data.len());
|
||||
let feature_stats = self.calculate_feature_statistics(&training_data)?;
|
||||
self.feature_stats = Some(feature_stats.clone());
|
||||
info!("✅ Feature statistics calculated: {} features normalized", feature_stats.mean.len());
|
||||
|
||||
// Normalize all training and validation samples BEFORE training starts
|
||||
info!("📊 Normalizing all samples with z-score normalization...");
|
||||
self.normalize_dataset(&mut training_data)?;
|
||||
self.normalize_dataset(&mut validation_data)?;
|
||||
info!("✅ Dataset normalization complete");
|
||||
|
||||
// Store normalized validation data for validation loss computation
|
||||
self.val_data = validation_data;
|
||||
|
||||
// Use the same training loop as DBN-based training
|
||||
self.train_with_data_full_loop(training_data, checkpoint_callback)
|
||||
.await
|
||||
@@ -3491,19 +3427,8 @@ impl DQNTrainer {
|
||||
feature_vec: &FeatureVector225,
|
||||
close_price: Option<rust_decimal::Decimal>, // Used for portfolio feature population
|
||||
) -> Result<TradingState> {
|
||||
// WAVE 3 FIX #2: Apply z-score normalization if feature_stats is available
|
||||
// Phase 1 (epochs 0-10): feature_stats = None, use raw features
|
||||
// Phase 2 (epochs 11+): feature_stats = Some(_), normalize features
|
||||
let normalized_features: Vec<f32> = if let Some(ref stats) = self.feature_stats {
|
||||
// Skip placeholder indices 125-127 (portfolio features)
|
||||
stats.normalize_with_skip(
|
||||
&feature_vec.iter().map(|&v| v as f32).collect::<Vec<f32>>(),
|
||||
&[125, 126, 127],
|
||||
)
|
||||
} else {
|
||||
// No normalization during stats collection phase
|
||||
feature_vec.iter().map(|&v| v as f32).collect()
|
||||
};
|
||||
// States are pre-normalized during data loading
|
||||
let normalized_features: Vec<f32> = feature_vec.iter().map(|&v| v as f32).collect();
|
||||
|
||||
// Features 0-3 are LOG RETURNS - preserve sign information for price direction
|
||||
let price_features: Vec<f32> = vec![
|
||||
@@ -4248,6 +4173,46 @@ impl DQNTrainer {
|
||||
|
||||
Ok(feature_vectors)
|
||||
}
|
||||
|
||||
/// Calculate feature statistics from training samples using Welford's algorithm
|
||||
fn calculate_feature_statistics(
|
||||
&self,
|
||||
samples: &[(FeatureVector225, Vec<f64>)],
|
||||
) -> Result<FeatureStatistics> {
|
||||
let mut stats = FeatureStatistics::new(225);
|
||||
|
||||
for (feature_vec, _) in samples {
|
||||
let features: Vec<f32> = feature_vec.iter().map(|&v| v as f32).collect();
|
||||
stats.update(&features);
|
||||
}
|
||||
|
||||
Ok(stats)
|
||||
}
|
||||
|
||||
/// Normalize all samples in a dataset using z-score normalization
|
||||
fn normalize_dataset(
|
||||
&mut self,
|
||||
samples: &mut [(FeatureVector225, Vec<f64>)],
|
||||
) -> Result<()> {
|
||||
if let Some(ref stats) = self.feature_stats {
|
||||
for (feature_vec, _) in samples.iter_mut() {
|
||||
// Convert to f32 for normalization
|
||||
let features_f32: Vec<f32> = feature_vec.iter().map(|&v| v as f32).collect();
|
||||
|
||||
// Normalize with skip (indices 125-127 are portfolio placeholders)
|
||||
let normalized = stats.normalize_with_skip(&features_f32, &[125, 126, 127]);
|
||||
|
||||
// Convert back to f64 and update
|
||||
for (i, &val) in normalized.iter().enumerate() {
|
||||
feature_vec[i] = val as f64;
|
||||
}
|
||||
}
|
||||
} else {
|
||||
return Err(anyhow::anyhow!("Feature statistics not initialized"));
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
|
||||
Reference in New Issue
Block a user