diff --git a/ml/src/trainers/dqn.rs b/ml/src/trainers/dqn.rs index 6f60ddc28..119a0babf 100644 --- a/ml/src/trainers/dqn.rs +++ b/ml/src/trainers/dqn.rs @@ -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 = 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, // Used for portfolio feature population ) -> Result { - // 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 = 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::>(), - &[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 = 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 = 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)], + ) -> Result { + let mut stats = FeatureStatistics::new(225); + + for (feature_vec, _) in samples { + let features: Vec = 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)], + ) -> 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 = 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)]