//! PPO Hyperparameter Optimization Adapter //! //! This module provides a production-ready adapter for optimizing PPO //! hyperparameters using the generic optimization framework. It implements: //! //! - Parameter space with log-scale handling for learning rates //! - Training wrapper that integrates with existing PPO pipeline //! - Metrics extraction for episode reward optimization (maximization) //! //! ## Usage Example //! //! ```rust,no_run //! use ml::hyperopt::ArgminOptimizer; //! use ml::hyperopt::adapters::ppo::{PPOTrainer, PPOParams}; //! //! # async fn example() -> anyhow::Result<()> { //! // Create trainer //! let trainer = PPOTrainer::new( //! 1000, // episodes per trial //! )?; //! //! // Run optimization //! let optimizer = ArgminOptimizer::with_trials(30, 5); //! let result = optimizer.optimize(trainer)?; //! //! println!("Best policy LR: {}", result.best_params.policy_learning_rate); //! println!("Best value LR: {}", result.best_params.value_learning_rate); //! println!("Best episode reward: {:.6}", -result.best_objective); // Negated (optimizer minimizes) //! # Ok(()) //! # } //! ``` use ml_core::device::MlDevice; use serde::{Deserialize, Serialize}; use std::fs::OpenOptions; use std::io::Write as IoWrite; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; use tracing::{info, warn}; use crate::common::action::FactoredAction; use crate::cuda_pipeline::signal_adapter::backtest_fitness; use crate::dqn::curiosity::CuriosityModule; use crate::hyperopt::paths::TrainingPaths; use crate::hyperopt::traits::{HardwareBudget, HyperparameterOptimizable, ParameterSpace}; use crate::ppo::composite_reward::CompositeReward; use crate::ppo::gae::GAEConfig; use crate::ppo::ppo::{PPOConfig, PPO}; use crate::ppo::reward_shaping::PPORewardShaper; use crate::ppo::trajectories::TrajectoryBatch; use crate::ppo::trajectory_replay::TrajectoryReplayBuffer; use crate::MLError; use crate::cuda_pipeline::gpu_backtest_evaluator::{GpuBacktestConfig, GpuBacktestEvaluator}; use cudarc::driver::CudaSlice; /// Pure model VRAM in MB (actor + critic + optimizers + gradients). const MODEL_OVERHEAD_MB: f64 = 300.0; /// Total per-trial VRAM in MB during concurrent hyperopt. /// PPO uses less VRAM than DQN (no replay buffer, smaller trajectory batches) /// but CUDA overhead and fragmentation still dominate. /// Conservative estimate: ~3.5 GB/trial on L40S. const TRIAL_VRAM_MB: f64 = 3500.0; /// PPO per-sample memory in MB (45 features x 4B + activations) const MB_PER_SAMPLE: f64 = 0.001; /// PPO hyperparameter space /// /// Defines the hyperparameters to optimize for PPO training: /// - Policy learning rate (log-scale: 1e-6 to 1e-3) /// - Value learning rate (log-scale: 1e-5 to 1e-3) /// - Clip epsilon (linear scale: 0.1 to 0.3) /// - Value loss coefficient (linear scale: 0.5 to 2.0) /// - Entropy coefficient (log-scale: 0.001 to 0.1) /// /// ## Parameter Scaling /// /// - **Log-scale**: Learning rates, entropy_coeff (span multiple orders) /// - **Linear scale**: Clip epsilon, value loss coeff (span single order) /// /// This scaling ensures efficient exploration by argmin's optimization. #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct PPOParams { /// Policy network learning rate (log-scale) pub policy_learning_rate: f64, /// Value network learning rate (log-scale) pub value_learning_rate: f64, /// PPO clip parameter epsilon (linear scale) pub clip_epsilon: f64, /// Value function loss coefficient (linear scale) pub value_loss_coeff: f64, /// Entropy coefficient for exploration (log-scale) pub entropy_coeff: f64, /// Batch size for PPO rollout collection (linear scale) pub batch_size: usize, /// Hidden dimension base for dynamic network sizing. /// Policy network: [base, base/2]. Value network: [base*2, base, base/2]. /// Range: [64, 2048], step=64. Bounded by VRAM via HardwareBudget. pub hidden_dim_base: usize, // --- Ensemble diversity parameters (6D) --- /// GAE discount factor — controls time horizon. /// Low (0.95) = short-term scalper, high (0.999) = trend follower. pub gae_gamma: f64, /// GAE lambda — bias-variance tradeoff in advantage estimation. /// Low (0.8) = more biased but stable, high (1.0) = Monte Carlo-like. pub gae_lambda: f64, /// Mini-batch size for PPO updates within each rollout batch. /// Smaller = noisier gradients (more diverse), larger = more stable. pub mini_batch_size: usize, /// Maximum gradient norm for clipping. Controls learning stability. pub max_grad_norm: f64, /// Maximum absolute position size. Controls risk appetite per trial. pub max_position_absolute: f64, /// Asymmetric clip upper bound. 0.0 = symmetric (None), >0 = clip_epsilon_high. /// Prevents entropy collapse during long training. pub clip_epsilon_high: f64, /// Curiosity-driven exploration weight. 0.0 = disabled, >0 = intrinsic reward scaling. pub curiosity_weight: f64, } impl Default for PPOParams { fn default() -> Self { Self { policy_learning_rate: 3e-5, value_learning_rate: 1e-4, clip_epsilon: 0.2, value_loss_coeff: 1.0, entropy_coeff: 0.05, batch_size: 2048, hidden_dim_base: 128, // Conservative default: policy [128, 64], value [256, 128, 64] gae_gamma: 0.99, gae_lambda: 0.95, mini_batch_size: 512, max_grad_norm: 0.5, max_position_absolute: 2.0, clip_epsilon_high: 0.0, // symmetric by default curiosity_weight: 0.0, // disabled by default } } } impl ParameterSpace for PPOParams { fn continuous_bounds() -> Vec<(f64, f64)> { vec![ (1e-6_f64.ln(), 3e-4_f64.ln()), // 0: policy_learning_rate (log scale, capped at 3e-4) (1e-5_f64.ln(), 1e-4_f64.ln()), // 1: value_learning_rate (log scale, capped at 1e-4 — 1e-3 causes gradient explosion) (0.1, 0.3), // 2: clip_epsilon (linear) (0.5, 2.0), // 3: value_loss_coeff (linear) (0.001_f64.ln(), 0.1_f64.ln()), // 4: entropy_coeff (log scale) (512.0, 8192.0), // 5: batch_size (large for PPO variance reduction) (64.0, 4096.0), // 6: hidden_dim_base (linear, step=64) (0.95, 0.999), // 7: gae_gamma (linear) (0.8, 1.0), // 8: gae_lambda (linear) (128.0, 2048.0), // 9: mini_batch_size (linear) (0.1, 1.0), // 10: max_grad_norm (linear) (0.5, 3.0), // 11: max_position_absolute (linear) (0.0, 0.5), // 12: clip_epsilon_high (linear, 0=symmetric) (0.0, 0.5), // 13: curiosity_weight (linear, 0=off) ] } fn from_continuous(x: &[f64]) -> Result { if x.len() != 14 { return Err(MLError::ConfigError(format!("Expected 14 parameters, got {}", x.len()))); } Ok(Self { policy_learning_rate: x.get(0).copied().ok_or(MLError::ConfigError("Missing policy_learning_rate".to_owned()))?.exp(), value_learning_rate: x.get(1).copied().ok_or(MLError::ConfigError("Missing value_learning_rate".to_owned()))?.exp(), clip_epsilon: x.get(2).copied().ok_or(MLError::ConfigError("Missing clip_epsilon".to_owned()))?.clamp(0.1, 0.3), value_loss_coeff: x.get(3).copied().ok_or(MLError::ConfigError("Missing value_loss_coeff".to_owned()))?.clamp(0.5, 2.0), entropy_coeff: x.get(4).copied().ok_or(MLError::ConfigError("Missing entropy_coeff".to_owned()))?.exp(), batch_size: x.get(5).copied().ok_or(MLError::ConfigError("Missing batch_size".to_owned()))?.round() as usize, hidden_dim_base: ((x.get(6).copied().ok_or(MLError::ConfigError("Missing hidden_dim_base".to_owned()))?.round() / 64.0).round() * 64.0).clamp(64.0, 4096.0) as usize, gae_gamma: x.get(7).copied().ok_or(MLError::ConfigError("Missing gae_gamma".to_owned()))?.clamp(0.95, 0.999), gae_lambda: x.get(8).copied().ok_or(MLError::ConfigError("Missing gae_lambda".to_owned()))?.clamp(0.8, 1.0), mini_batch_size: x.get(9).copied().ok_or(MLError::ConfigError("Missing mini_batch_size".to_owned()))?.round() as usize, max_grad_norm: x.get(10).copied().ok_or(MLError::ConfigError("Missing max_grad_norm".to_owned()))?.clamp(0.1, 1.0), max_position_absolute: x.get(11).copied().ok_or(MLError::ConfigError("Missing max_position_absolute".to_owned()))?.clamp(0.5, 3.0), clip_epsilon_high: x.get(12).copied().ok_or(MLError::ConfigError("Missing clip_epsilon_high".to_owned()))?.clamp(0.0, 0.5), curiosity_weight: x.get(13).copied().ok_or(MLError::ConfigError("Missing curiosity_weight".to_owned()))?.clamp(0.0, 0.5), }) } fn to_continuous(&self) -> Vec { vec![ self.policy_learning_rate.ln(), self.value_learning_rate.ln(), self.clip_epsilon, self.value_loss_coeff, self.entropy_coeff.ln(), self.batch_size as f64, self.hidden_dim_base as f64, self.gae_gamma, self.gae_lambda, self.mini_batch_size as f64, self.max_grad_norm, self.max_position_absolute, self.clip_epsilon_high, self.curiosity_weight, ] } fn param_names() -> Vec<&'static str> { vec![ "policy_learning_rate", "value_learning_rate", "clip_epsilon", "value_loss_coeff", "entropy_coeff", "batch_size", "hidden_dim_base", "gae_gamma", "gae_lambda", "mini_batch_size", "max_grad_norm", "max_position_absolute", "clip_epsilon_high", "curiosity_weight", ] } fn continuous_bounds_for(budget: &HardwareBudget) -> Vec<(f64, f64)> { let mut bounds = Self::continuous_bounds(); // Cap batch_size by VRAM (index 5) if let Some(max_batch) = budget.max_batch_size(MODEL_OVERHEAD_MB, MB_PER_SAMPLE, 512.0, 8192.0) { if let Some(batch_bound) = bounds.get_mut(5) { batch_bound.1 = max_batch; } } // Cap hidden_dim_base by VRAM (index 6) // PPO needs more VRAM than DQN (actor + critic networks) let max_base = budget.max_hidden_dim_base(4, 512, 53, 45); // 4 concurrent, 512 batch, 45 features (42 market + 3 portfolio), 45 factored actions if let Some(dim_bound) = bounds.get_mut(6) { dim_bound.1 = max_base as f64; } // Cap mini_batch_size by VRAM (index 9) — same logic as batch_size if let Some(max_mini) = budget.max_batch_size(MODEL_OVERHEAD_MB, MB_PER_SAMPLE, 128.0, 2048.0) { if let Some(mini_bound) = bounds.get_mut(9) { mini_bound.1 = max_mini; } } bounds } fn model_overhead_mb() -> f64 { TRIAL_VRAM_MB } fn per_sample_mb() -> f64 { MB_PER_SAMPLE } } /// PPO training metrics /// /// Contains all relevant metrics from a PPO training run. /// The primary optimization target is episode reward (avg_episode_reward). #[derive(Debug, Clone, Serialize, Deserialize)] pub struct PPOMetrics { /// Training policy loss pub policy_loss: f64, /// Training value loss pub value_loss: f64, /// Validation policy loss (optimization target) pub val_policy_loss: f64, /// Validation value loss (optimization target) pub val_value_loss: f64, /// Combined loss (weighted sum) pub combined_loss: f64, /// Average episode reward pub avg_episode_reward: f64, /// Number of episodes completed pub episodes_completed: usize, /// GPU walk-forward backtest Sharpe ratio (None when CUDA unavailable or backtest fails) pub backtest_sharpe: Option, /// GPU walk-forward backtest total trade count (None when CUDA unavailable or backtest fails) pub backtest_trades: Option, } /// PPO trainer for hyperparameter optimization /// /// This struct wraps the PPO training pipeline and implements /// `HyperparameterOptimizable` for use with optimization backends. /// /// ## Configuration /// /// - **DBN data dir**: Market data source (OHLCV bars from Databento) /// - **Episodes**: Number of training episodes per trial /// - **Device**: CUDA GPU (falls back to CPU if unavailable) /// - **Features**: 42 market features from extraction pipeline /// /// ## Fixed Architecture /// /// The following parameters are fixed for consistency: /// - `state_dim`: 45 without OFI (42 market + 3 portfolio), or 53 with MBP-10 data (+8 OFI) /// - `num_actions`: 45 (5 exposure x 3 order x 3 urgency) /// - `policy_hidden_dims`: [base, base/2] /// - `value_hidden_dims`: [base*4, base*3, base*2, base, base/2] /// /// ## Optimized Hyperparameters /// /// The following are optimized by `PPOParams`: /// - Policy learning rate /// - Value learning rate /// - Clip epsilon /// - Value loss coefficient /// - Entropy coefficient pub struct PPOTrainer { dbn_data_dir: std::path::PathBuf, episodes: usize, device: MlDevice, /// Pool of GPU devices for multi-GPU trial parallelism. device_pool: Vec, /// Training paths configuration (replaces hardcoded checkpoint directories) training_paths: TrainingPaths, /// Early stopping patience (epochs without improvement before stopping) early_stopping_patience: usize, /// Early stopping minimum epochs (minimum epochs before early stopping can trigger) early_stopping_min_epochs: usize, /// Trial counter for hyperopt — shared across clones for parallel execution trial_counter: Arc, /// Transaction cost in basis points (commission per trade, default 1.0) tx_cost_bps: f64, /// Tick size for spread calculation (default 0.25 for ES futures) tick_size: f64, /// Spread width in ticks (default 1.0) spread_ticks: f64, /// GPU market features [num_bars * 42] — uploaded once, reused across trials features_cuda: Option>, /// GPU market targets [num_bars * 4] — uploaded once, reused across trials targets_cuda: Option>, /// GPU PPO experience collector — initialized on first trial, weights synced per trial gpu_collector: Option, /// Number of bars in the uploaded market data gpu_num_bars: usize, /// Initial capital for portfolio simulation and backtest evaluation. /// Must match the value used in training reward computation. initial_capital: f64, /// Preloaded training data cached across hyperopt trials. /// Loaded once via `preload_data()`, then shared (via Arc) across all trials /// to avoid re-reading `.dbn.zst` files and re-extracting features per trial. preloaded_data: Option>>, } impl std::fmt::Debug for PPOTrainer { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("PPOTrainer") .field("dbn_data_dir", &self.dbn_data_dir) .field("episodes", &self.episodes) .field("device", &self.device) .field("early_stopping_patience", &self.early_stopping_patience) .finish_non_exhaustive() } } impl Clone for PPOTrainer { fn clone(&self) -> Self { Self { dbn_data_dir: self.dbn_data_dir.clone(), episodes: self.episodes, device: self.device.clone(), device_pool: self.device_pool.clone(), training_paths: self.training_paths.clone(), early_stopping_patience: self.early_stopping_patience, early_stopping_min_epochs: self.early_stopping_min_epochs, trial_counter: self.trial_counter.clone(), tx_cost_bps: self.tx_cost_bps, tick_size: self.tick_size, spread_ticks: self.spread_ticks, initial_capital: self.initial_capital, // GPU resources are not cloned — each clone re-initialises on first use features_cuda: None, targets_cuda: None, gpu_collector: None, gpu_num_bars: 0, // Preloaded data is shared via Arc — cheap to clone preloaded_data: self.preloaded_data.clone(), } } } impl PPOTrainer { /// Create a new PPO trainer /// /// # Arguments /// /// * `dbn_data_dir` - Path to directory with DBN market data files /// * `episodes` - Number of training episodes per trial /// /// # Returns /// /// Configured trainer ready for optimization /// /// # Errors /// /// Returns error if: /// - DBN data directory doesn't exist /// - No DBN files found in directory /// - Device initialization fails pub fn new>( dbn_data_dir: P, episodes: usize, ) -> anyhow::Result { let dbn_data_dir = dbn_data_dir.into(); if !dbn_data_dir.exists() { return Err(MLError::ConfigError(format!("DBN data directory not found: {}", dbn_data_dir.display())) .into()); } // Initialize CUDA device — GPU required for RL hyperopt let device = MlDevice::cuda(0) .map_err(|e| anyhow::anyhow!("CUDA GPU required for PPO hyperopt: {}", e))?; info!("PPO Trainer initialized:"); info!(" Data directory: {}", dbn_data_dir.display()); info!(" Device: CUDA GPU"); info!(" Episodes per trial: {}", episodes); // Use temporary default paths - should be replaced with with_training_paths() let training_paths = TrainingPaths::new("/tmp/ml_training", "ppo", "default"); Ok(Self { dbn_data_dir, episodes, device_pool: vec![device.clone()], device, training_paths, early_stopping_patience: 5, // Default: 5 epochs (hyperopt optimized) early_stopping_min_epochs: 5, // Default: 5 epochs (hyperopt optimized) trial_counter: Arc::new(AtomicUsize::new(0)), // Shared across clones for parallel trials tx_cost_bps: 1.0, // 1 bps commission (ES futures typical) tick_size: 0.25, // ES futures tick size spread_ticks: 1.0, // 1 tick spread (ES typical) initial_capital: 35_000.0, // Default: $35K (realistic retail futures) features_cuda: None, targets_cuda: None, gpu_collector: None, gpu_num_bars: 0, preloaded_data: None, // Loaded on first call to preload_data() }) } /// Configure early stopping parameters (overrides defaults) pub fn with_early_stopping(mut self, patience: usize, min_epochs: usize) -> Self { self.early_stopping_patience = patience; self.early_stopping_min_epochs = min_epochs; self } /// Configure transaction costs and spread for realistic reward computation /// /// # Arguments /// /// * `tx_cost_bps` - Commission cost per trade in basis points (e.g. 1.0 for ES) /// * `tick_size` - Minimum price increment (e.g. 0.25 for ES) /// * `spread_ticks` - Typical bid-ask spread in ticks (e.g. 1.0 for ES) pub fn with_costs(mut self, tx_cost_bps: f64, tick_size: f64, spread_ticks: f64) -> Self { self.tx_cost_bps = tx_cost_bps; self.tick_size = tick_size; self.spread_ticks = spread_ticks; info!("PPO costs configured: tx={} bps, spread={} ticks (tick_size={})", tx_cost_bps, spread_ticks, tick_size); self } /// Set initial capital for portfolio simulation and backtest evaluation. pub fn with_initial_capital(mut self, capital: f64) -> Self { self.initial_capital = capital; self } /// Override the compute device (CPU or CUDA). pub fn with_device(mut self, device: MlDevice) -> Self { info!( "Device overridden to: {}", "CUDA" ); self.device_pool = vec![device.clone()]; self.device = device; self } /// Set a pool of GPU devices for multi-GPU trial parallelism. pub fn with_devices(mut self, devices: Vec) -> Self { if devices.is_empty() { return self; } info!("Multi-GPU device pool: {} devices", devices.len()); self.device = devices[0].clone(); self.device_pool = devices; self } /// Set training paths configuration /// /// # Arguments /// /// * `paths` - Training paths configuration /// /// # Returns /// /// Self for method chaining pub fn with_training_paths(mut self, paths: TrainingPaths) -> Self { info!("PPO training paths configured:"); info!(" Run directory: {:?}", paths.run_dir()); info!(" Checkpoints: {:?}", paths.checkpoints_dir()); info!(" Logs: {:?}", paths.logs_dir()); info!(" Hyperopt: {:?}", paths.hyperopt_dir()); self.training_paths = paths; self } /// Preload training data from DBN files once, caching it for reuse across /// all hyperopt trials. This eliminates the per-trial cost of reading /// `.dbn.zst` files, decompressing them, and extracting 42 features. /// /// Call this once before starting the optimization loop. The data is stored /// in `Arc` so cloning the trainer (for parallel trials) is cheap. /// /// # Errors /// /// Returns error if data loading or feature extraction fails pub fn preload_data(&mut self) -> Result<(), MLError> { info!("Preloading PPO training data from: {} ...", self.dbn_data_dir.display()); let preload_start = std::time::Instant::now(); let data = self .load_training_data() .map_err(|e| MLError::TrainingError(format!("Failed to preload data: {}", e)))?; let elapsed = preload_start.elapsed(); info!( "PPO data preloaded: {} samples in {:.1}s (cached for all trials)", data.len(), elapsed.as_secs_f64() ); self.preloaded_data = Some(Arc::new(data)); Ok(()) } /// Returns true if training data has been preloaded. pub fn has_preloaded_data(&self) -> bool { self.preloaded_data.is_some() } /// Upload training data to GPU and initialize the experience collector. /// /// Called once at the start of the first trial. Subsequent trials reuse the /// uploaded data and call `sync_weights()` on the collector. fn ensure_gpu_data( &mut self, training_data: &[([f32; 42], f64)], _ppo_agent: &PPO, ) -> Result<(), MLError> { use tracing::debug; // Data already uploaded — collector weight sync is a no-op until // GpuVarStore integration is wired. The PPO update() path handles // weight synchronization internally via its own CUDA kernels. if self.features_cuda.is_some() && self.targets_cuda.is_some() { debug!("PPO GPU data already uploaded — reusing"); return Ok(()); } let stream = self.device.cuda_stream() .map_err(|_| MLError::ConfigError("CUDA required for PPO training".to_owned()))?; let num_bars = training_data.len(); // Upload features [num_bars * 42] let mut flat_features = Vec::with_capacity(num_bars * 42); for (features, _) in training_data { flat_features.extend_from_slice(features); } match crate::cuda_pipeline::clone_htod_f32(&stream, &flat_features) { Ok(buf) => { info!("PPO CUDA features uploaded: {} bars × 42 ({:.1} MB)", num_bars, (num_bars * 42 * 4) as f64 / 1_048_576.0); self.features_cuda = Some(buf); } Err(e) => { return Err(MLError::TrainingError(format!("PPO CUDA features upload FAILED (no CPU fallback): {e}"))); } } // Upload targets [num_bars * 4]: kernel expects [close, next_close, close_raw, next_close_raw] let mut flat_targets = Vec::with_capacity(num_bars * 4); for (i, (_, price)) in training_data.iter().enumerate() { let close = *price as f32; let next_close = training_data .get(i + 1) .map(|(_, p)| *p as f32) .unwrap_or(close); flat_targets.push(close); flat_targets.push(next_close); flat_targets.push(close); flat_targets.push(next_close); } match crate::cuda_pipeline::clone_htod_f32(&stream, &flat_targets) { Ok(buf) => { info!("PPO CUDA targets uploaded: {} bars × 4 ({:.1} MB)", num_bars, (num_bars * 4 * 4) as f64 / 1_048_576.0); self.targets_cuda = Some(buf); } Err(e) => { self.features_cuda = None; return Err(MLError::TrainingError(format!("PPO CUDA targets upload FAILED (no CPU fallback): {e}"))); } } self.gpu_num_bars = num_bars; // Initialize GPU collector with placeholder VarStores. // PPO actor/critic weights are managed internally by their CUDA networks; // the collector uses these for kernel-side forward passes during experience // collection. Empty VarStores signal the kernel to skip weight-dependent ops. if self.gpu_collector.is_none() { let actor_vars = ml_core::cuda_autograd::GpuVarStore::new(stream.clone()); let critic_vars = ml_core::cuda_autograd::GpuVarStore::new(stream.clone()); let curiosity_vars = ml_core::cuda_autograd::GpuVarStore::new(stream.clone()); match crate::cuda_pipeline::gpu_ppo_collector::GpuPpoExperienceCollector::new( stream.clone(), &actor_vars, &critic_vars, &curiosity_vars, self.initial_capital as f32, self.tick_size as f32 * self.spread_ticks as f32, 0.20, ) { Ok(collector) => { info!("PPO GPU experience collector initialized (zero-roundtrip CUDA kernel)"); self.gpu_collector = Some(collector); } Err(e) => { return Err(MLError::TrainingError(format!("PPO GPU collector init FAILED (no CPU fallback): {e}"))); } } } Ok(()) } /// Collect experiences using GPU kernel, returning a TrajectoryBatch. /// /// Returns Err if GPU collection fails — no CPU fallback. fn gpu_collect_trajectories( &mut self, num_episodes: usize, ) -> Result { use crate::cuda_pipeline::gpu_ppo_collector::PpoCollectorConfig; let collector = self.gpu_collector.as_mut() .ok_or_else(|| MLError::TrainingError("PPO GPU collector not initialized — CPU fallback FORBIDDEN".to_owned()))?; let features_buf = self.features_cuda.as_ref() .ok_or_else(|| MLError::TrainingError("PPO CUDA features not uploaded — CPU fallback FORBIDDEN".to_owned()))?; let targets_buf = self.targets_cuda.as_ref() .ok_or_else(|| MLError::TrainingError("PPO CUDA targets not uploaded — CPU fallback FORBIDDEN".to_owned()))?; let total_bars = self.gpu_num_bars; let n_episodes = (num_episodes as i32).min(128); let timesteps = 100_i32; // Match CPU episode length let usable_bars = (total_bars as i32 - timesteps).max(1); let stride = (usable_bars / n_episodes).max(1); let episode_starts: Vec = (0..n_episodes) .map(|i| (i * stride).rem_euclid(usable_bars)) .collect(); let config = PpoCollectorConfig { n_episodes, timesteps_per_episode: timesteps, total_bars: total_bars as i32, gamma: 0.99, gae_lambda: 0.95, ..Default::default() }; collector.reset_episodes(self.initial_capital as f32, self.tick_size as f32 * self.spread_ticks as f32, 0.20) .map_err(|e| MLError::TrainingError(format!("PPO GPU reset_episodes FAILED (no CPU fallback): {e}")))?; let batch = collector.collect_experiences(features_buf, targets_buf, &episode_starts, &config) .map_err(|e| MLError::TrainingError(format!("PPO GPU experience collection FAILED (no CPU fallback): {e}")))?; info!("GPU PPO collected {} experiences ({} episodes × {} timesteps)", batch.n_episodes * batch.timesteps, batch.n_episodes, batch.timesteps); gpu_ppo_batch_to_trajectory_batch(&batch) } } /// Convert a GPU-resident PPO experience batch to a TrajectoryBatch. /// /// Downloads data from GPU for the hyperopt replay buffer and validation flow. fn gpu_ppo_batch_to_trajectory_batch( batch: &crate::cuda_pipeline::gpu_ppo_collector::PpoExperienceBatch, ) -> Result { let kernel_state_dim = batch.state_dim; let model_state_dim = 45; let total = batch.total(); let states_flat = batch.download_states()?; let actions_raw = batch.download_actions()?; let log_probs_raw = batch.download_log_probs()?; let advantages_raw = batch.download_advantages()?; let returns_raw = batch.download_returns()?; let done_flags_raw = batch.download_done_flags()?; let states: Vec> = states_flat .chunks(kernel_state_dim) .take(total) .map(|c| c.get(..model_state_dim).unwrap_or(c).to_vec()) // cpu-side state truncate .collect(); let hold_action = FactoredAction::new( crate::common::action::ExposureLevel::Flat, crate::common::action::OrderType::Market, crate::common::action::Urgency::Normal, ); let actions: Vec = actions_raw.iter().take(total) .map(|&a| { let idx = (a.clamp(0, 44)) as usize; FactoredAction::from_index(idx).unwrap_or(hold_action) }) .collect(); let log_probs: Vec = log_probs_raw.into_iter().take(total).collect(); let values: Vec = returns_raw.iter().zip(&advantages_raw).take(total) .map(|(r, a)| r - a).collect(); let dones: Vec = done_flags_raw.iter().take(total).map(|&d| d != 0).collect(); let rewards = vec![0.0_f32; states.len()]; Ok(TrajectoryBatch { trajectories: vec![], states, states_flat: vec![], actions, log_probs, values, rewards, dones, advantages: advantages_raw.into_iter().take(total).collect(), returns: returns_raw.into_iter().take(total).collect(), }) } /// Write a log entry to the training log file fn write_training_log_ppo(logs_dir: &std::path::Path, message: &str) -> Result<(), std::io::Error> { let log_file = logs_dir.join("training.log"); let mut file = OpenOptions::new() .create(true) .append(true) .open(log_file)?; let timestamp = chrono::Utc::now().format("%Y-%m-%d %H:%M:%S"); writeln!(file, "[{}] {}", timestamp, message)?; Ok(()) } /// Write trial results to JSON file fn write_trial_result_ppo( hyperopt_dir: &std::path::Path, trial_result: &crate::hyperopt::traits::TrialResult, ) -> Result<(), std::io::Error> { let trials_file = hyperopt_dir.join("trials.json"); // Read existing trials (if any) let mut all_trials = if trials_file.exists() { let content = std::fs::read_to_string(&trials_file)?; serde_json::from_str::>(&content).unwrap_or_default() } else { Vec::new() }; // Append new trial let trial_json = serde_json::to_value(trial_result) .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e))?; all_trials.push(trial_json); // Write back to file (pretty printed) let content = serde_json::to_string_pretty(&all_trials) .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e))?; std::fs::write(&trials_file, content)?; Ok(()) } impl HyperparameterOptimizable for PPOTrainer { type Params = PPOParams; type Metrics = PPOMetrics; fn train_with_params(&mut self, params: Self::Params) -> Result { // START: Add trial timing let trial_start = std::time::Instant::now(); // Get current trial number and increment for next trial (atomic for parallel safety) let current_trial = self.trial_counter.fetch_add(1, Ordering::Relaxed); // Multi-GPU: select device from pool based on trial number (round-robin) let trial_device = self.device_pool.get(current_trial % self.device_pool.len()) .cloned() .unwrap_or_else(|| self.device.clone()); if self.device_pool.len() > 1 { info!("Trial {} → GPU {} of {}", current_trial, current_trial % self.device_pool.len(), self.device_pool.len()); } info!("Training PPO with parameters:"); info!(" Policy LR: {:.6}", params.policy_learning_rate); info!(" Value LR: {:.6}", params.value_learning_rate); info!(" Clip epsilon: {:.3}", params.clip_epsilon); info!(" Value loss coeff: {:.3}", params.value_loss_coeff); info!(" Entropy coeff: {:.6}", params.entropy_coeff); info!(" GAE gamma: {:.4}", params.gae_gamma); info!(" GAE lambda: {:.3}", params.gae_lambda); info!(" Mini-batch size: {}", params.mini_batch_size); info!(" Max grad norm: {:.3}", params.max_grad_norm); info!(" Max position: {:.2}", params.max_position_absolute); info!(" Clip epsilon high: {:.3} ({})", params.clip_epsilon_high, if params.clip_epsilon_high > 0.01 { "asymmetric" } else { "symmetric" }); info!(" Curiosity weight: {:.3} ({})", params.curiosity_weight, if params.curiosity_weight > 0.001 { "enabled" } else { "disabled" }); // Log trial start (ensure directory exists first) std::fs::create_dir_all(self.training_paths.logs_dir()).ok(); write_training_log_ppo( &self.training_paths.logs_dir(), &format!("=== Starting PPO Trial ===\nParams: {:#?}", params), ) .ok(); // Create directories self.training_paths.create_all().map_err(|e| { MLError::ModelError(format!("Failed to create training directories: {}", e)) })?; info!("Training directories created:"); info!(" Logs: {:?}", self.training_paths.logs_dir()); info!(" Hyperopt: {:?}", self.training_paths.hyperopt_dir()); // Create PPO config with trial hyperparameters let ppo_config = PPOConfig { state_dim: 48, // 42 market + 3 portfolio = 45, aligned to 48 for tensor cores num_actions: 63, // 5 exposure × 3 order × 3 urgency (FactoredAction) policy_hidden_dims: { let base = params.hidden_dim_base; vec![base, base / 2] }, value_hidden_dims: { let base = params.hidden_dim_base; vec![base * 4, base * 3, base * 2, base, base / 2] }, policy_learning_rate: params.policy_learning_rate, value_learning_rate: params.value_learning_rate, clip_epsilon: params.clip_epsilon as f32, value_loss_coeff: params.value_loss_coeff as f32, entropy_coeff: params.entropy_coeff as f32, gae_config: GAEConfig { gamma: params.gae_gamma as f32, lambda: params.gae_lambda as f32, normalize_advantages: true, }, batch_size: params.batch_size, mini_batch_size: params.mini_batch_size, num_epochs: 20, max_grad_norm: params.max_grad_norm as f32, early_stopping_enabled: true, early_stopping_patience: self.early_stopping_patience, early_stopping_min_delta: 1e-4, early_stopping_min_epochs: self.early_stopping_min_epochs, max_position_absolute: params.max_position_absolute, transaction_cost_bps: self.tx_cost_bps, // Use configured cost cash_reserve_pct: 0.20, // 20% minimum cash reserve circuit_breaker_threshold: 5, // 5 consecutive failures use_lstm: false, // Standard MLP networks for hyperopt (LSTM not yet integrated) lstm_hidden_dim: 128, lstm_num_layers: 1, lstm_sequence_length: 32, accumulation_steps: 1, clip_epsilon_high: (params.clip_epsilon_high > 0.01) .then_some(params.clip_epsilon_high as f32), use_symlog: true, use_adaptive_entropy: true, use_percentile_scaling: true, }; // Save state_dim before ppo_config is moved into PPO::with_device let state_dim = ppo_config.state_dim; // Create PPO agent let mut ppo_agent = PPO::new(ppo_config) .map_err(|e| MLError::TrainingError(format!("Failed to create PPO agent: {}", e)))?; // Training loop (simplified for hyperopt) let mut total_policy_loss = 0.0; let mut total_value_loss = 0.0; let mut total_reward = 0.0; // Generate all trajectories upfront for train/val split let total_trajectories = self.episodes; // Validate minimum trajectory count for 80/20 split if total_trajectories < 10 { return Err(MLError::ConfigError(format!( "Insufficient trajectories for train/val split: {} < 10 minimum", total_trajectories ))); } // Split trajectories 80/20 for train/val let split_idx = (total_trajectories as f64 * 0.8) as usize; let num_train = split_idx; let num_val = total_trajectories - split_idx; // Validate non-empty validation set if num_val < 1 { return Err(MLError::ConfigError(format!( "Validation set is empty after 80/20 split (train={}, val={})", num_train, num_val ))); } // Warn if validation set is too small if num_val < 5 { warn!( "Validation set is small ({}), may not be representative", num_val ); } // Use preloaded data if available, otherwise load from disk let training_data = if let Some(cached) = &self.preloaded_data { info!( "Using preloaded training data ({} samples, skipping disk I/O)", cached.len() ); cached.as_ref().clone() } else { info!( "Loading training data from: {}", self.dbn_data_dir.display() ); self.load_training_data() .map_err(|e| MLError::TrainingError(format!("Failed to load training data: {}", e)))? }; info!("Loaded {} training samples", training_data.len()); // Split DATA 80/20 (not trajectory count -- that's different) let data_split = (training_data.len() as f64 * 0.8) as usize; let train_data = &training_data[..data_split]; let val_data = &training_data[data_split..]; info!( "Train data: {} samples, Val data: {} samples (train episodes: {}, val episodes: {})", train_data.len(), val_data.len(), num_train, num_val ); // Upload raw market data to GPU and init/sync collector // Uses full dataset (kernel indexes by bar offset). First trial compiles // the CUDA kernel + uploads data; subsequent trials just sync weights. self.ensure_gpu_data(&training_data, &ppo_agent)?; // Create ExO-PPO trajectory replay buffer (M=4 rollouts, 4x sample efficiency) let mut replay_buffer = TrajectoryReplayBuffer::new(4, params.clip_epsilon as f32); info!("ExO-PPO trajectory replay enabled (M=4 rollouts)"); let gae_gamma = params.gae_gamma as f32; let gae_lambda = params.gae_lambda as f32; // Batch episodes: min(64, num_train) to avoid zero batches with small episode counts let batch_episodes = 64.min(num_train); let num_batches = (num_train + batch_episodes - 1) / batch_episodes; // ceil division for _batch_idx in 0..num_batches { let episodes_this_batch = batch_episodes.min(num_train - _batch_idx * batch_episodes); // GPU experience collection let mut trajectory_batch = self.gpu_collect_trajectories(episodes_this_batch)?; // Update PPO with on-policy trajectory batch let (policy_loss, value_loss) = ppo_agent .update(&mut trajectory_batch) .map_err(|e| MLError::TrainingError(format!("PPO update failed: {}", e)))?; total_policy_loss += policy_loss as f64; total_value_loss += value_loss as f64; // Store rollout in ExO-PPO replay buffer for future IS-weighted updates let generation = replay_buffer.generation(); let flattened_states: Vec = trajectory_batch.states.iter().flatten().copied().collect(); let rollout = TrajectoryReplayBuffer::create_rollout( flattened_states, trajectory_batch.actions.iter().map(|a| a.to_index() as u32).collect(), trajectory_batch.log_probs.clone(), trajectory_batch.advantages.clone(), trajectory_batch.returns.clone(), state_dim, generation, ); replay_buffer.store_rollout(rollout); // ExO-PPO: replay past rollouts with IS-weighted advantages if !replay_buffer.is_empty() { let mut replay_loss = 0.0_f64; let replay_count = replay_buffer.len(); for stored_rollout in replay_buffer.rollouts() { // Re-evaluate stored states under current policy to get new log-probs let mut new_log_probs = Vec::with_capacity(stored_rollout.num_steps); for t in 0..stored_rollout.num_steps { if let Some(state_slice) = stored_rollout.get_state(t) { match ppo_agent.act_with_log_prob(state_slice) { Ok((_, lp, _)) => new_log_probs.push(lp), Err(_) => new_log_probs.push(0.0), } } } // Compute IS-weighted surrogate loss for this stored rollout let rollout_loss = replay_buffer.compute_rollout_loss(stored_rollout, &new_log_probs); replay_loss += rollout_loss as f64; } // Weight replay loss at 50% of on-policy (off-policy data is less reliable) if replay_count > 0 { let avg_replay = replay_loss / replay_count as f64; total_reward += avg_replay * 0.5; // Boost reward signal with replay } } // Calculate average reward for this batch let batch_reward: f32 = trajectory_batch.rewards.iter().sum(); total_reward += batch_reward as f64 / episodes_this_batch as f64; } // Compute validation losses on held-out trajectories info!( "Computing validation losses on {} held-out episodes...", num_val ); let mut val_policy_losses = Vec::new(); let mut val_value_losses = Vec::new(); // Generate validation trajectories from held-out data using current policy // Validation uses no curiosity/shaping (evaluate pure extrinsic performance) let mut no_curiosity = None; let mut no_shaper = None; let mut no_composite = None; let mut val_trajectory_batch = self .generate_trajectories_from_data( &ppo_agent, val_data, num_val, &mut no_curiosity, 0.0, &mut no_shaper, &mut no_composite, gae_gamma, gae_lambda, &trial_device, ) .map_err(|e| { MLError::TrainingError(format!("Failed to generate validation trajectories: {}", e)) })?; // Compute validation loss WITHOUT updating policy let (val_policy_loss, val_value_loss) = ppo_agent .compute_losses(&mut val_trajectory_batch) .map_err(|e| { MLError::TrainingError(format!("Failed to compute validation losses: {}", e)) })?; val_policy_losses.push(val_policy_loss as f64); val_value_losses.push(val_value_loss as f64); let avg_val_policy_loss = val_policy_losses.iter().sum::() / val_policy_losses.len() as f64; let avg_val_value_loss = val_value_losses.iter().sum::() / val_value_losses.len() as f64; let avg_policy_loss = total_policy_loss / num_batches as f64; let avg_value_loss = total_value_loss / num_batches as f64; let avg_reward = total_reward / num_batches as f64; // GPU walk-forward backtest (CUDA mandatory) let (bt_sharpe, bt_trades) = match self.run_gpu_backtest(&ppo_agent, &trial_device, ¶ms) { Ok((s, t)) => { info!("PPO GPU backtest: Sharpe={:.4} trades={}", s, t); (Some(s), Some(t)) } Err(e) => { return Err(MLError::TrainingError(format!("PPO GPU backtest FAILED (no CPU fallback): {e}"))); } }; let metrics = PPOMetrics { policy_loss: avg_policy_loss, value_loss: avg_value_loss, val_policy_loss: avg_val_policy_loss, val_value_loss: avg_val_value_loss, combined_loss: avg_policy_loss + params.value_loss_coeff * avg_value_loss, avg_episode_reward: avg_reward, episodes_completed: self.episodes, backtest_sharpe: bt_sharpe, backtest_trades: bt_trades, }; info!("Training completed:"); info!(" Policy loss: {:.6}", metrics.policy_loss); info!(" Value loss: {:.6}", metrics.value_loss); info!(" Val policy loss: {:.6}", metrics.val_policy_loss); info!(" Val value loss: {:.6}", metrics.val_value_loss); info!(" Avg reward: {:.4}", metrics.avg_episode_reward); // END: Add trial completion logging let duration_secs = trial_start.elapsed().as_secs_f64(); write_training_log_ppo( &self.training_paths.logs_dir(), &format!( "Training completed in {:.2}s: val_policy_loss={:.6}, val_value_loss={:.6}", duration_secs, metrics.val_policy_loss, metrics.val_value_loss ), ) .ok(); // CRITICAL FIX: Explicit memory cleanup to prevent OOM between trials // Drop model and data loaders, and sync CUDA to free GPU/RAM info!("Cleaning up resources..."); drop(ppo_agent); drop(val_trajectory_batch); // GPU synchronization via stream sync (forces cudaDeviceSynchronize) if let Ok(stream) = trial_device.cuda_stream() { stream.synchronize().map_err(|e| { MLError::TrainingError(format!("CUDA sync failed: {}", e)) })?; std::thread::sleep(std::time::Duration::from_millis(50)); } info!("Resource cleanup complete"); // Write trial result to JSON (ensure directory exists first) let trial_result = crate::hyperopt::traits::TrialResult { trial_num: current_trial, params, objective: Self::extract_objective(&metrics), duration_secs, metrics: None, }; std::fs::create_dir_all(self.training_paths.hyperopt_dir()).ok(); write_trial_result_ppo(&self.training_paths.hyperopt_dir(), &trial_result).ok(); Ok(metrics) } fn extract_objective(metrics: &Self::Metrics) -> f64 { // Prefer GPU backtest fitness when available (walk-forward Sharpe is more // reliable than episode reward for ranking hyperparameter trials). if let (Some(sharpe), Some(trades)) = (metrics.backtest_sharpe, metrics.backtest_trades) { return backtest_fitness(sharpe, trades, 30, 0.0); } // Fallback: maximize episode rewards (negative because optimizer MINIMIZES) // // We optimize for avg_episode_reward, NOT validation loss, because: // 1. Loss minimization rewards frozen policies (policy_loss=0, KL_div=0) // 2. Low learning rates (e.g., 1e-6) prevent learning but minimize loss // 3. Episode rewards measure actual trading performance // // The optimizer minimizes this objective, so we negate rewards to maximize them. -metrics.avg_episode_reward } } /// Decode OHLCV bars from an already-opened DBN decoder. /// /// Deduplicates by timestamp (keeping highest-volume bar) to handle overlapping /// contracts from `.FUT` parent symbols at roll dates. fn decode_ohlcv_bars( decoder: &mut dbn::decode::dbn::Decoder, ) -> anyhow::Result> { use chrono::DateTime; use dbn::decode::DecodeRecordRef; use dbn::OhlcvMsg; use std::collections::BTreeMap; let mut by_ts: BTreeMap = BTreeMap::new(); let mut total_records = 0_usize; while let Some(record_ref) = decoder .decode_record_ref() .map_err(|e| anyhow::anyhow!("Failed to decode record: {}", e))? { if let Some(ohlcv) = record_ref.get::() { total_records += 1; let ts_nanos = ohlcv.hd.ts_event; let volume = ohlcv.volume as f64; let existing_vol = by_ts.get(&ts_nanos).map(|b| b.volume).unwrap_or(0.0); if volume >= existing_vol { by_ts.insert( ts_nanos, crate::features::extraction::OHLCVBar { timestamp: DateTime::from_timestamp_nanos(ts_nanos as i64), open: ohlcv.open as f64 / 1e9, high: ohlcv.high as f64 / 1e9, low: ohlcv.low as f64 / 1e9, close: ohlcv.close as f64 / 1e9, volume, }, ); } } } let bars: Vec<_> = by_ts.into_values().collect(); let deduped = total_records.saturating_sub(bars.len()); if deduped > 0 { info!(" Deduplicated {}/{} bars by timestamp", deduped, total_records); } Ok(bars) } /// Recursively collect all .dbn and .dbn.zst files from a directory and its subdirectories. fn collect_dbn_files_recursive(dir: &std::path::Path) -> Vec { let mut files = Vec::new(); if let Ok(entries) = std::fs::read_dir(dir) { for entry in entries.flatten() { let path = entry.path(); if path.is_dir() { files.extend(collect_dbn_files_recursive(&path)); } else if path.extension().and_then(|s| s.to_str()) == Some("dbn") || path.to_string_lossy().ends_with(".dbn.zst") { files.push(path); } else { // Skip non-DBN files } } } files.sort(); files } impl PPOTrainer { /// Run GPU walk-forward backtest on validation data using the trained PPO actor. /// /// Uses the last 20% of preloaded bars as validation data. Builds a single /// window, creates a `GpuBacktestEvaluator`, and calls `evaluate()` with /// a forward function that converts PPO 45-action softmax probabilities into /// 5-action exposure scores via `ppo_to_exposure_scores`. /// /// Returns `(sharpe, total_trades)` from the first (and only) window. fn run_gpu_backtest( &self, ppo: &PPO, device: &MlDevice, params: &PPOParams, ) -> Result<(f32, u32), MLError> { let data = self.preloaded_data.as_ref().ok_or_else(|| { MLError::ConfigError("run_gpu_backtest: no preloaded data".to_owned()) })?; // Use last 20% of data for validation let val_start = (data.len() as f64 * 0.8) as usize; let val_data = data.get(val_start..).ok_or_else(|| { MLError::ConfigError("run_gpu_backtest: val slice out of bounds".to_owned()) })?; if val_data.len() < 50 { return Err(MLError::ConfigError(format!( "run_gpu_backtest: insufficient validation data ({} < 50)", val_data.len() ))); } // Build price and feature vectors for a single window let mut prices = Vec::with_capacity(val_data.len()); let mut features = Vec::with_capacity(val_data.len()); for (fv, close_price) in val_data { let close = *close_price as f32; prices.push([close, close, close, close]); // OHLC approximation (same as DQN) features.push(fv.to_vec()); // cpu-side feature copy } let window_prices = vec![prices]; let window_features = vec![features]; // Market features only -- portfolio features (3) added by evaluator's gather_states. let feature_dim = 42; let config = GpuBacktestConfig { max_position: params.max_position_absolute as f32, tx_cost_bps: self.tx_cost_bps as f32, spread_cost: (self.tick_size * self.spread_ticks) as f32, initial_capital: self.initial_capital as f32, ..Default::default() }; let stream = device.cuda_stream() .map_err(|_| MLError::ConfigError("CUDA required for PPO backtest".to_owned()))?; let mut evaluator = GpuBacktestEvaluator::new( &window_prices, &window_features, feature_dim, config, stream, )?; // Forward function: PPO actor -> softmax -> ppo_to_exposure_scores -> [B, 5] // Uses host-side action_probabilities API, then maps to exposure scores on GPU. let metrics = evaluator.evaluate( &|states_gpu: &CudaSlice, batch_size: usize, state_dim: usize| -> Result, MLError> { // Download f32 states to host for PPO forward pass let states_host: Vec = stream.clone_dtoh(states_gpu).map_err(|e| { // gpu-exit: PPO actor forward MLError::ModelError(format!("PPO backtest state download: {e}")) })?; // Run PPO actor forward to get action probabilities per sample let mut action_indices = Vec::with_capacity(batch_size); for b in 0..batch_size { let start = b * state_dim; let end = start + state_dim; let state_slice = states_host.get(start..end).unwrap_or(&[]); match ppo.greedy_action(state_slice) { Ok(action) => action_indices.push(action.to_index() as i32), Err(_) => action_indices.push(0_i32), // Flat fallback } } // Upload action indices to GPU stream.clone_htod(&action_indices).map_err(|e| { MLError::ModelError(format!("PPO backtest action upload: {e}")) }) }, 24, // portfolio_dim: 8 portfolio + 16 multi-timeframe (matches training state layout) )?; let first = metrics.first().ok_or_else(|| { MLError::ModelError("run_gpu_backtest: evaluator returned no windows".to_owned()) })?; Ok((first.sharpe, first.total_trades as u32)) } /// Load training data from Parquet/DBN files /// /// Returns feature vectors (42 dims) with target close prices for trajectory generation fn load_training_data(&self) -> anyhow::Result> { use std::path::Path; let dir_path = Path::new(&self.dbn_data_dir); // Check if directory contains Parquet or DBN files let has_parquet = std::fs::read_dir(dir_path)? .filter_map(|entry| entry.ok()) .any(|entry| entry.path().extension().and_then(|s| s.to_str()) == Some("parquet")); if has_parquet { info!("Found Parquet files, loading from Parquet..."); self.load_from_parquet() } else { info!("No Parquet files found, loading from DBN..."); self.load_from_dbn() } } /// Load training data from Parquet file (optimized path) fn load_from_parquet(&self) -> anyhow::Result> { use crate::features::extraction::OHLCVBar; use arrow::array::{Array, Float64Array, PrimitiveArray, UInt64Array}; use arrow::datatypes::TimestampNanosecondType; use arrow::record_batch::RecordBatch; use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder; use std::fs::File; // Find first Parquet file in directory let parquet_file = std::fs::read_dir(&self.dbn_data_dir)? .filter_map(|entry| entry.ok()) .find(|entry| entry.path().extension().and_then(|s| s.to_str()) == Some("parquet")) .ok_or_else(|| anyhow::anyhow!("No Parquet file found in directory"))? .path(); info!("Loading Parquet file: {}", parquet_file.display()); let file = File::open(&parquet_file)?; let builder = ParquetRecordBatchReaderBuilder::try_new(file)?; let reader = builder.build()?; let mut all_ohlcv_bars = Vec::new(); for batch_result in reader { let batch: RecordBatch = batch_result?; // Extract columns (try both timestamp_ns and ts_event) let timestamp_col = batch .column_by_name("timestamp_ns") .or_else(|| batch.column_by_name("ts_event")) .ok_or_else(|| anyhow::anyhow!("Missing timestamp column"))?; let timestamps = timestamp_col .as_any() .downcast_ref::>() .ok_or_else(|| anyhow::anyhow!("Invalid timestamp column type"))?; let opens = batch .column_by_name("open") .ok_or_else(|| anyhow::anyhow!("Missing 'open' column"))? .as_any() .downcast_ref::() .ok_or_else(|| anyhow::anyhow!("Invalid 'open' column type"))?; let highs = batch .column_by_name("high") .ok_or_else(|| anyhow::anyhow!("Missing 'high' column"))? .as_any() .downcast_ref::() .ok_or_else(|| anyhow::anyhow!("Invalid 'high' column type"))?; let lows = batch .column_by_name("low") .ok_or_else(|| anyhow::anyhow!("Missing 'low' column"))? .as_any() .downcast_ref::() .ok_or_else(|| anyhow::anyhow!("Invalid 'low' column type"))?; let closes = batch .column_by_name("close") .ok_or_else(|| anyhow::anyhow!("Missing 'close' column"))? .as_any() .downcast_ref::() .ok_or_else(|| anyhow::anyhow!("Invalid 'close' column type"))?; let volumes = batch .column_by_name("volume") .ok_or_else(|| anyhow::anyhow!("Missing 'volume' column"))? .as_any() .downcast_ref::() .ok_or_else(|| anyhow::anyhow!("Invalid 'volume' column type"))?; for i in 0..batch.num_rows() { let timestamp_ns = timestamps.value(i); let timestamp = chrono::DateTime::from_timestamp_nanos(timestamp_ns); let bar = OHLCVBar { timestamp, open: opens.value(i), high: highs.value(i), low: lows.value(i), close: closes.value(i), volume: volumes.value(i) as f64, }; all_ohlcv_bars.push(bar); } } info!("Loaded {} OHLCV bars from Parquet", all_ohlcv_bars.len()); // Sort bars chronologically all_ohlcv_bars.sort_by_key(|bar| bar.timestamp); // Extract features and create training data self.extract_features_and_targets(&all_ohlcv_bars) } /// Load training data from DBN files fn load_from_dbn(&self) -> anyhow::Result> { use dbn::decode::DbnDecoder; let dbn_files = collect_dbn_files_recursive(std::path::Path::new(&self.dbn_data_dir)); if dbn_files.is_empty() { return Err(anyhow::anyhow!("No DBN files found in directory")); } info!("Found {} DBN files to load", dbn_files.len()); let mut all_bars = Vec::new(); for dbn_file in &dbn_files { info!("Loading DBN file: {}", dbn_file.display()); let is_zstd = dbn_file.to_string_lossy().ends_with(".dbn.zst"); let bars = if is_zstd { let mut decoder = DbnDecoder::from_zstd_file(dbn_file) .map_err(|e| anyhow::anyhow!("Failed to open zstd DBN file {}: {}", dbn_file.display(), e))?; decode_ohlcv_bars(&mut decoder)? } else { let mut decoder = DbnDecoder::from_file(dbn_file) .map_err(|e| anyhow::anyhow!("Failed to open DBN file {}: {}", dbn_file.display(), e))?; decode_ohlcv_bars(&mut decoder)? }; all_bars.extend(bars); } info!("Loaded {} OHLCV bars from {} DBN files", all_bars.len(), dbn_files.len()); all_bars.sort_by_key(|bar| bar.timestamp); self.extract_features_and_targets(&all_bars) } /// Attempt to load OFI features from MBP10 data. /// Returns None if MBP10 data is not available or on error. fn load_ofi_features(&self) -> Option> { use crate::features::mbp10_loader::load_ofi_features_parallel; let mbp10_dir = self .dbn_data_dir .parent() .map(|p| p.join("mbp10"))?; // Derive trades directory as sibling of the OHLCV data dir let trades_dir = self.dbn_data_dir.parent().map(|p| p.join("trades")); load_ofi_features_parallel(&mbp10_dir, trades_dir.as_deref(), |dir| { collect_dbn_files_recursive(dir) }).map(|ofi8| { ofi8.into_iter().map(|row| { let mut wide = [0.0_f64; 20]; wide[..8].copy_from_slice(&row); wide }).collect() }) } /// Extract 42-feature vectors and targets from OHLCV bars. fn extract_features_and_targets( &self, ohlcv_bars: &[crate::features::extraction::OHLCVBar], ) -> anyhow::Result> { use crate::features::extraction::extract_ml_features; info!( "Extracting 42-feature vectors from {} OHLCV bars...", ohlcv_bars.len() ); // Need at least 50 bars for warmup period if ohlcv_bars.len() < 51 { return Err(anyhow::anyhow!( "Insufficient OHLCV bars for feature extraction: {} < 51", ohlcv_bars.len() )); } // Extract features using production API (returns Vec<[f64; 42]>) let feature_vectors = extract_ml_features(ohlcv_bars) .map_err(|e| anyhow::anyhow!("Feature extraction failed: {}", e))?; info!( "Extracted {} feature vectors (42 dimensions each)", feature_vectors.len() ); // Load OFI features if MBP10 data is available let ofi_features = self.load_ofi_features(); let ofi_count = ofi_features.as_ref().map_or(0, |v| v.len()); if ofi_count > 0 { info!( "Overlaying {} OFI features onto positions 45-52", ofi_count ); } // Create training data pairs (features, target) // Target: Next bar's close price (autoregressive prediction) // Features are 42-dim market features (matches train_baseline) let mut training_data = Vec::new(); for i in 0..feature_vectors.len().saturating_sub(1) { // Target is next bar's close (skip warmup period) let target_idx = i + 51; // +50 warmup + 1 next bar if target_idx < ohlcv_bars.len() { let next_close = ohlcv_bars[target_idx].close; // Convert [f64; 42] to [f32; 42] let mut feature_array = [0.0_f32; 42]; for (j, &val) in feature_vectors[i].iter().enumerate() { feature_array[j] = val as f32; } training_data.push((feature_array, next_close)); } } // Note: the last feature vector is excluded because there is no "next bar" // to compute a meaningful return against. Including it with target=self // would create a ~0 return sample that biases the model toward HOLD. info!( "Created {} training samples with 42-feature market vectors", training_data.len() ); Ok(training_data) } /// Generate trajectories from real market data using the PPO agent's policy. /// /// Creates episodes by simulating trading on market data, using the PPO /// agent to select actions with real log-probabilities and value estimates. /// When a curiosity module is provided, intrinsic novelty rewards are added /// to the extrinsic reward signal, scaled by `curiosity_weight`. fn generate_trajectories_from_data( &self, agent: &PPO, data: &[([f32; 42], f64)], num_episodes: usize, curiosity: &mut Option, curiosity_weight: f64, reward_shaper: &mut Option, composite_reward: &mut Option, gae_gamma: f32, gae_lambda: f32, _device: &MlDevice, ) -> anyhow::Result { use crate::ppo::trajectories::{Trajectory, TrajectoryStep}; use rand::Rng; let mut rng = rand::thread_rng(); let mut trajectories = Vec::new(); // Episode length: use up to 100 steps or available data let max_episode_length = 100.min(data.len() / num_episodes.max(1)); for _episode_idx in 0..num_episodes { let mut trajectory = Trajectory::new(); // Reset per-episode state for reward components if let Some(ref mut shaper) = reward_shaper { shaper.reset(); } if let Some(ref mut comp) = composite_reward { comp.reset(); } // Start at random position in data let start_idx = if data.len() > max_episode_length { rng.gen_range(0..data.len() - max_episode_length) } else { 0 }; for step_idx in 0..max_episode_length { let data_idx = start_idx + step_idx; if data_idx >= data.len() { break; } let (features, target_price) = &data[data_idx]; // Convert features to state vector let state = features.to_vec(); // cpu-side feature copy // Use PPO agent for action selection with real log_prob and value let (action, log_prob, value) = match agent.act_with_log_prob(&state) { Ok((a, lp, v)) => (a, lp, v), Err(_) => { // Fallback: random factored action with uniform log_prob let random_idx = rng.gen_range(0..45_usize); let a = FactoredAction::from_index(random_idx).unwrap_or_else(|_| { FactoredAction::new( crate::common::action::ExposureLevel::Flat, crate::common::action::OrderType::Market, crate::common::action::Urgency::Normal, ) }); (a, -(45.0_f32).ln(), 0.0_f32) } }; // Compute reward based on price movement and action, minus transaction costs let mut reward = if step_idx + 1 < max_episode_length && data_idx + 1 < data.len() { let current_price = target_price; let next_price = data.get(data_idx + 1).map(|(_, p)| *p).unwrap_or(*current_price); let price_change = (next_price - current_price) / current_price; // Transaction cost: commission + spread (deducted on Buy/Sell) let spread_cost = if current_price.abs() > 1e-10 { self.tick_size * self.spread_ticks / current_price } else { 0.0 }; let total_cost = (self.tx_cost_bps * 0.0001 + spread_cost) as f32; // Use target_exposure for PnL: +1.0 (long), -1.0 (short), 0.0 (flat) let exposure = action.target_exposure() as f32; if exposure.abs() > f32::EPSILON { exposure * price_change as f32 - total_cost } else { 0.0 // Flat position — no PnL } } else { 0.0 }; // Add intrinsic curiosity reward if enabled if let Some(ref mut curiosity_mod) = curiosity { if let Some((next_feat, _)) = data.get(data_idx + 1) { // Build state tensors for curiosity module (expects [batch, 35]) // Use first 35 features from the 42-dim state vector let state_slice: Vec = features.iter().take(35).copied().collect(); let next_slice: Vec = next_feat.iter().take(35).copied().collect(); if let Ok(intrinsic) = curiosity_mod.calculate_curiosity_reward( &state_slice, 1, 35, action, &next_slice, ) { reward += (curiosity_weight * intrinsic) as f32; } } } // Apply composite risk-adjusted reward (downside dev + differential return) if let Some(ref mut comp) = composite_reward { reward = comp.compute(reward); } // Apply reward shaping (hold penalty + rolling Sharpe + diversity bonus) if let Some(ref mut shaper) = reward_shaper { let is_flat = action.target_exposure().abs() < f64::EPSILON; // Signal: price change exceeds 0.01% threshold let has_signal = if step_idx + 1 < max_episode_length { data.get(data_idx + 1) .map(|(_, next_p)| { let current_p = target_price; (next_p - current_p).abs() / current_p.abs().max(1e-10) > 0.0001 }) .unwrap_or(false) } else { false }; reward = shaper.shape_reward( reward, is_flat, has_signal, action.to_index(), ); } let done = step_idx + 1 >= max_episode_length; let step = TrajectoryStep::new(state, action, log_prob, value, reward, done); trajectory.add_step(step); if done { break; } } // Mark last step as done if let Some(last_step) = trajectory.steps.last_mut() { last_step.done = true; } trajectories.push(trajectory); } // Compute advantages and returns using GAE (params from hyperopt search space) let gamma = gae_gamma; let lambda = gae_lambda; let mut advantages = Vec::new(); let mut returns = Vec::new(); for trajectory in &trajectories { let rewards = trajectory.get_rewards(); let values = trajectory.get_values(); let dones = trajectory.get_dones(); // Compute GAE advantages let mut traj_advantages = Vec::new(); let mut last_gae = 0.0; for t in (0..rewards.len()).rev() { let reward = rewards[t]; let value = values[t]; let next_value = if t + 1 < values.len() { values[t + 1] } else { 0.0 }; let done = dones[t]; let mask = if done { 0.0 } else { 1.0 }; let delta = reward + gamma * next_value * mask - value; let gae = delta + gamma * lambda * mask * last_gae; traj_advantages.push(gae); last_gae = gae; } traj_advantages.reverse(); advantages.extend(traj_advantages.clone()); // Compute returns as advantage + value let traj_returns: Vec = traj_advantages .iter() .zip(values.iter()) .map(|(adv, val)| adv + val) .collect(); returns.extend(traj_returns); } // Create trajectory batch Ok(TrajectoryBatch::from_trajectories( trajectories, advantages, returns, )) } } #[cfg(test)] mod tests { use super::*; #[test] fn test_ppo_params_roundtrip() -> Result<(), MLError> { let params = PPOParams { policy_learning_rate: 3e-5, value_learning_rate: 1e-4, clip_epsilon: 0.2, value_loss_coeff: 1.0, entropy_coeff: 0.05, batch_size: 2048, hidden_dim_base: 128, gae_gamma: 0.99, gae_lambda: 0.95, mini_batch_size: 512, max_grad_norm: 0.5, max_position_absolute: 2.0, clip_epsilon_high: 0.0, curiosity_weight: 0.0, }; let continuous = params.to_continuous(); assert_eq!(continuous.len(), 14, "to_continuous should produce 14D vector"); let recovered = PPOParams::from_continuous(&continuous)?; assert!((recovered.policy_learning_rate - params.policy_learning_rate).abs() < 1e-10); assert!((recovered.value_learning_rate - params.value_learning_rate).abs() < 1e-10); assert!((recovered.clip_epsilon - params.clip_epsilon).abs() < 1e-10); assert!((recovered.value_loss_coeff - params.value_loss_coeff).abs() < 1e-10); assert!((recovered.entropy_coeff - params.entropy_coeff).abs() < 1e-10); assert!((recovered.gae_gamma - params.gae_gamma).abs() < 1e-10); assert!((recovered.gae_lambda - params.gae_lambda).abs() < 1e-10); assert_eq!(recovered.mini_batch_size, params.mini_batch_size); assert!((recovered.max_grad_norm - params.max_grad_norm).abs() < 1e-10); assert!((recovered.max_position_absolute - params.max_position_absolute).abs() < 1e-10); assert!((recovered.clip_epsilon_high - params.clip_epsilon_high).abs() < 1e-10); assert!((recovered.curiosity_weight - params.curiosity_weight).abs() < 1e-10); Ok(()) } #[test] fn test_ppo_params_bounds() { let bounds = PPOParams::continuous_bounds(); assert_eq!(bounds.len(), 14); // 14D: 7 original + 6 ensemble diversity + 1 curiosity // Check log-scale bounds are reasonable assert!(bounds.get(0).is_some_and(|b| b.0 < b.1)); // policy_learning_rate assert!(bounds.get(1).is_some_and(|b| b.0 < b.1)); // value_learning_rate assert!(bounds.get(4).is_some_and(|b| b.0 < b.1)); // entropy_coeff // Check linear bounds (original) assert_eq!(bounds.get(2).copied(), Some((0.1, 0.3))); // clip_epsilon assert_eq!(bounds.get(3).copied(), Some((0.5, 2.0))); // value_loss_coeff assert_eq!(bounds.get(5).copied(), Some((512.0, 8192.0))); // batch_size assert_eq!(bounds.get(6).copied(), Some((64.0, 4096.0))); // hidden_dim_base // Check ensemble diversity bounds assert_eq!(bounds.get(7).copied(), Some((0.95, 0.999))); // gae_gamma assert_eq!(bounds.get(8).copied(), Some((0.8, 1.0))); // gae_lambda assert_eq!(bounds.get(9).copied(), Some((128.0, 2048.0))); // mini_batch_size assert_eq!(bounds.get(10).copied(), Some((0.1, 1.0))); // max_grad_norm assert_eq!(bounds.get(11).copied(), Some((0.5, 3.0))); // max_position_absolute assert_eq!(bounds.get(12).copied(), Some((0.0, 0.5))); // clip_epsilon_high // Check curiosity bounds assert_eq!(bounds.get(13).copied(), Some((0.0, 0.5))); // curiosity_weight } #[test] fn test_param_names() { let names = PPOParams::param_names(); assert_eq!(names.len(), 14); // 14 tunable hyperparameters (7 original + 6 ensemble diversity + 1 curiosity) assert_eq!(names.get(0).copied(), Some("policy_learning_rate")); assert_eq!(names.get(1).copied(), Some("value_learning_rate")); assert_eq!(names.get(2).copied(), Some("clip_epsilon")); assert_eq!(names.get(3).copied(), Some("value_loss_coeff")); assert_eq!(names.get(4).copied(), Some("entropy_coeff")); assert_eq!(names.get(5).copied(), Some("batch_size")); assert_eq!(names.get(6).copied(), Some("hidden_dim_base")); assert_eq!(names.get(7).copied(), Some("gae_gamma")); assert_eq!(names.get(8).copied(), Some("gae_lambda")); assert_eq!(names.get(9).copied(), Some("mini_batch_size")); assert_eq!(names.get(10).copied(), Some("max_grad_norm")); assert_eq!(names.get(11).copied(), Some("max_position_absolute")); assert_eq!(names.get(12).copied(), Some("clip_epsilon_high")); assert_eq!(names.get(13).copied(), Some("curiosity_weight")); } #[test] fn test_objective_function_maximizes_reward() { // Test that objective function returns NEGATIVE rewards (for maximization) // Scenario 1: Positive reward → Negative objective (optimizer will minimize to -∞) let metrics_positive = PPOMetrics { policy_loss: 0.5, value_loss: 0.3, val_policy_loss: 0.4, val_value_loss: 0.2, combined_loss: 0.8, avg_episode_reward: 100.0, // Good performance episodes_completed: 1000, backtest_sharpe: None, backtest_trades: None, }; let objective_positive = PPOTrainer::extract_objective(&metrics_positive); assert_eq!( objective_positive, -100.0, "Objective should be negative of reward" ); // Scenario 2: Negative reward → Positive objective (optimizer will avoid) let metrics_negative = PPOMetrics { policy_loss: 0.5, value_loss: 0.3, val_policy_loss: 0.4, val_value_loss: 0.2, combined_loss: 0.8, avg_episode_reward: -50.0, // Poor performance episodes_completed: 1000, backtest_sharpe: None, backtest_trades: None, }; let objective_negative = PPOTrainer::extract_objective(&metrics_negative); assert_eq!( objective_negative, 50.0, "Negative reward should give positive objective" ); // Scenario 3: Zero reward let metrics_zero = PPOMetrics { policy_loss: 0.0, value_loss: 0.0, val_policy_loss: 0.0, val_value_loss: 0.0, combined_loss: 0.0, avg_episode_reward: 0.0, // Neutral performance episodes_completed: 1000, backtest_sharpe: None, backtest_trades: None, }; let objective_zero = PPOTrainer::extract_objective(&metrics_zero); assert_eq!( objective_zero, 0.0, "Zero reward should give zero objective" ); // Verify that higher rewards result in LOWER objectives (better for minimization) assert!( objective_positive < objective_zero, "Higher reward should have lower objective" ); assert!( objective_zero < objective_negative, "Negative reward should have higher objective" ); } #[test] fn test_objective_ignores_loss_metrics() { // Verify that objective function does NOT use loss metrics // This ensures we're not accidentally rewarding frozen policies let metrics_high_reward_high_loss = PPOMetrics { policy_loss: 10.0, // High loss value_loss: 10.0, // High loss val_policy_loss: 10.0, // High validation loss val_value_loss: 10.0, // High validation loss combined_loss: 40.0, // High combined loss avg_episode_reward: 200.0, // But high reward (good trading) episodes_completed: 1000, backtest_sharpe: None, backtest_trades: None, }; let metrics_low_reward_low_loss = PPOMetrics { policy_loss: 0.0, // Zero loss (frozen policy) value_loss: 0.0, // Zero loss val_policy_loss: 0.0, // Zero validation loss val_value_loss: 0.0, // Zero validation loss combined_loss: 0.0, // Zero combined loss avg_episode_reward: 10.0, // But low reward (poor trading) episodes_completed: 1000, backtest_sharpe: None, backtest_trades: None, }; let obj_high_reward = PPOTrainer::extract_objective(&metrics_high_reward_high_loss); let obj_low_reward = PPOTrainer::extract_objective(&metrics_low_reward_low_loss); // High reward should be preferred (lower objective) despite high loss assert!(obj_high_reward < obj_low_reward, "High reward should be preferred over low loss. Got: high_reward_obj={}, low_loss_obj={}", obj_high_reward, obj_low_reward); } #[test] fn test_objective_uses_backtest_fitness_when_available() { // When GPU backtest results are available, extract_objective should use // backtest_fitness instead of -avg_episode_reward. let metrics_with_backtest = PPOMetrics { policy_loss: 0.5, value_loss: 0.3, val_policy_loss: 0.4, val_value_loss: 0.2, combined_loss: 0.8, avg_episode_reward: 100.0, episodes_completed: 1000, backtest_sharpe: Some(1.5), backtest_trades: Some(50), }; let obj = PPOTrainer::extract_objective(&metrics_with_backtest); // backtest_fitness(1.5, 50, 30, 0.0) = -1.5 (trades >= min_trades) assert!((obj - (-1.5)).abs() < 1e-9, "Expected backtest_fitness(-1.5), got {obj}"); // Verify fallback: without backtest, should be -avg_episode_reward let metrics_no_backtest = PPOMetrics { backtest_sharpe: None, backtest_trades: None, ..metrics_with_backtest.clone() }; let obj_fallback = PPOTrainer::extract_objective(&metrics_no_backtest); assert!((obj_fallback - (-100.0)).abs() < 1e-9, "Expected -avg_episode_reward (-100.0), got {obj_fallback}"); } #[test] fn test_objective_backtest_few_trades_penalty() { // When trades < min_trades, backtest_fitness applies a linear penalty let metrics = PPOMetrics { policy_loss: 0.0, value_loss: 0.0, val_policy_loss: 0.0, val_value_loss: 0.0, combined_loss: 0.0, avg_episode_reward: 0.0, episodes_completed: 100, backtest_sharpe: Some(2.0), backtest_trades: Some(15), // half of min_trades=30 }; let obj = PPOTrainer::extract_objective(&metrics); // backtest_fitness(2.0, 15, 30, 0.0) = -2.0 * (15/30) = -1.0 assert!((obj - (-1.0)).abs() < 1e-9, "Expected penalized fitness (-1.0), got {obj}"); } #[test] fn test_curiosity_weight_in_params() -> Result<(), MLError> { let params = PPOParams { curiosity_weight: 0.25, ..Default::default() }; let continuous = params.to_continuous(); assert_eq!(continuous.len(), 14); let recovered = PPOParams::from_continuous(&continuous)?; assert!((recovered.curiosity_weight - 0.25).abs() < 1e-10); Ok(()) } }