From cd8a45d8a8343b5140a083b52e3e3139fbc07514 Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Sat, 28 Feb 2026 18:28:15 +0100 Subject: [PATCH] fix(gpu): wire PPO GPU experience collection, fix debug log levels, increase hyperopt deadline MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Wire GPU experience collection into PPO hyperopt adapter with CPU fallback (ensure_gpu_data, gpu_collect_trajectories, batch-to-trajectory converter) - Downgrade 6 ERROR-level debug logs to debug!() in DQN hyperopt adapter - Increase hyperopt activeDeadlineSeconds from 1h to 6h (job-template + train.sh) - Mark 3 data-dependent real_data_loader tests as #[ignore] - Fix 4 clippy map_or → is_none_or warnings in trading_service Co-Authored-By: Claude Opus 4.6 --- crates/ml/src/hyperopt/adapters/dqn.rs | 12 +- crates/ml/src/hyperopt/adapters/ppo.rs | 315 +++++++++++++++++- crates/ml/src/real_data_loader.rs | 3 + infra/k8s/training/job-template.yaml | 2 +- infra/scripts/train.sh | 6 +- .../trading_service/src/core/order_manager.rs | 4 +- services/trading_service/src/services/risk.rs | 4 +- 7 files changed, 327 insertions(+), 19 deletions(-) diff --git a/crates/ml/src/hyperopt/adapters/dqn.rs b/crates/ml/src/hyperopt/adapters/dqn.rs index 8fa71ca16..b8a441843 100644 --- a/crates/ml/src/hyperopt/adapters/dqn.rs +++ b/crates/ml/src/hyperopt/adapters/dqn.rs @@ -2083,7 +2083,7 @@ impl HyperparameterOptimizable for DQNTrainer { type Metrics = DQNMetrics; fn train_with_params(&mut self, params: Self::Params) -> Result { - tracing::error!("DEBUG: train_with_params called"); + tracing::debug!("train_with_params called"); // START: Add trial timing let trial_start = std::time::Instant::now(); @@ -2490,8 +2490,8 @@ impl HyperparameterOptimizable for DQNTrainer { }, }; - tracing::error!("DEBUG: Training completed successfully, extracting metrics"); - tracing::error!("DEBUG: Checking validation data - internal_trainer.get_val_data().len() = {}", internal_trainer.get_val_data().len()); + tracing::debug!("Training completed successfully, extracting metrics"); + tracing::debug!("Checking validation data - internal_trainer.get_val_data().len() = {}", internal_trainer.get_val_data().len()); // Extract metrics from TrainingMetrics struct // Note: TrainingMetrics.loss is a single f64, not a Vec @@ -2576,12 +2576,12 @@ impl HyperparameterOptimizable for DQNTrainer { .copied() .unwrap_or(0.0); - tracing::error!("DEBUG: Reached backtest decision point"); - tracing::error!("DEBUG: self.enable_backtest = {}", self.enable_backtest); + tracing::debug!("Reached backtest decision point"); + tracing::debug!("enable_backtest = {}", self.enable_backtest); // Run backtest if enabled let backtest_metrics = if self.enable_backtest { - tracing::error!("DEBUG: Backtest is enabled, starting backtest..."); + tracing::debug!("Backtest is enabled, starting backtest..."); tracing::info!("Running backtest on validation data..."); // Get validation data from trainer diff --git a/crates/ml/src/hyperopt/adapters/ppo.rs b/crates/ml/src/hyperopt/adapters/ppo.rs index 3fe011118..3ee237dd3 100644 --- a/crates/ml/src/hyperopt/adapters/ppo.rs +++ b/crates/ml/src/hyperopt/adapters/ppo.rs @@ -207,7 +207,6 @@ pub struct PPOMetrics { /// - Clip epsilon /// - Value loss coefficient /// - Entropy coefficient -#[derive(Debug, Clone)] pub struct PPOTrainer { dbn_data_dir: std::path::PathBuf, episodes: usize, @@ -226,6 +225,55 @@ pub struct PPOTrainer { tick_size: f64, /// Spread width in ticks (default 1.0) spread_ticks: f64, + /// GPU market features [num_bars * 51] — uploaded once, reused across trials + #[cfg(feature = "cuda")] + features_cuda: Option>, + /// GPU market targets [num_bars * 4] — uploaded once, reused across trials + #[cfg(feature = "cuda")] + targets_cuda: Option>, + /// GPU PPO experience collector — initialized on first trial, weights synced per trial + #[cfg(feature = "cuda")] + gpu_collector: Option, + /// Number of bars in the uploaded market data + #[cfg(feature = "cuda")] + gpu_num_bars: usize, +} + +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(), + 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, + // GPU resources are not cloned — each clone re-initialises on first use + #[cfg(feature = "cuda")] + features_cuda: None, + #[cfg(feature = "cuda")] + targets_cuda: None, + #[cfg(feature = "cuda")] + gpu_collector: None, + #[cfg(feature = "cuda")] + gpu_num_bars: 0, + } + } } impl PPOTrainer { @@ -282,6 +330,14 @@ impl PPOTrainer { 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) + #[cfg(feature = "cuda")] + features_cuda: None, + #[cfg(feature = "cuda")] + targets_cuda: None, + #[cfg(feature = "cuda")] + gpu_collector: None, + #[cfg(feature = "cuda")] + gpu_num_bars: 0, }) } @@ -336,6 +392,235 @@ impl PPOTrainer { self.training_paths = paths; self } + + /// 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. + #[cfg(feature = "cuda")] + fn ensure_gpu_data( + &mut self, + training_data: &[([f32; 51], f64)], + ppo_agent: &PPO, + ) { + use tracing::debug; + + // Data already uploaded — just sync weights for the new trial's model + if self.features_cuda.is_some() && self.targets_cuda.is_some() { + if let Some(ref mut collector) = self.gpu_collector { + if let Err(e) = collector.sync_weights( + ppo_agent.actor.vars(), + ppo_agent.critic.vars(), + ) { + warn!("PPO GPU weight sync failed: {e}"); + } else { + debug!("PPO GPU weights synced for new trial"); + } + } + return; + } + + let cuda_dev = match &self.device { + candle_core::Device::Cuda(d) => d, + candle_core::Device::Cpu | candle_core::Device::Metal(_) => return, + }; + let stream = cuda_dev.cuda_stream(); + let num_bars = training_data.len(); + + // Upload features [num_bars * 51] + let mut flat_features = Vec::with_capacity(num_bars * 51); + for (features, _) in training_data { + flat_features.extend_from_slice(features); + } + match stream.memcpy_stod(&flat_features) { + Ok(buf) => { + info!("PPO CUDA features uploaded: {} bars × 51 ({:.1} MB)", + num_bars, (num_bars * 51 * 4) as f64 / 1_048_576.0); + self.features_cuda = Some(buf); + } + Err(e) => { + warn!("PPO CUDA features upload failed: {e}"); + return; + } + } + + // 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 stream.memcpy_stod(&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) => { + warn!("PPO CUDA targets upload failed: {e}"); + self.features_cuda = None; // Both must succeed + return; + } + } + + self.gpu_num_bars = num_bars; + + // Initialize GPU collector with current model's weights + if self.gpu_collector.is_none() { + let actor_vars = ppo_agent.actor.vars(); + let critic_vars = ppo_agent.critic.vars(); + match crate::cuda_pipeline::gpu_ppo_collector::GpuPpoExperienceCollector::new( + stream, + actor_vars, + critic_vars, + &candle_nn::VarMap::new(), // curiosity placeholder + 35_000.0, // initial_capital + self.tick_size as f32 * self.spread_ticks as f32, // avg_spread + 0.20, // cash_reserve_pct + ) { + Ok(collector) => { + info!("PPO GPU experience collector initialized (zero-roundtrip CUDA kernel)"); + self.gpu_collector = Some(collector); + } + Err(e) => { + warn!("PPO GPU collector init failed, using CPU fallback: {e}"); + } + } + } else { + // Sync weights from new PPO model + if let Some(ref mut collector) = self.gpu_collector { + if let Err(e) = collector.sync_weights( + ppo_agent.actor.vars(), + ppo_agent.critic.vars(), + ) { + warn!("PPO GPU weight sync failed: {e}"); + } + debug!("PPO GPU weights synced for new trial"); + } + } + } + + /// Collect experiences using GPU kernel, returning a TrajectoryBatch. + /// + /// Returns None if GPU collection is unavailable or fails (caller should + /// fall back to CPU trajectory generation). + #[cfg(feature = "cuda")] + fn gpu_collect_trajectories( + &mut self, + num_episodes: usize, + ) -> Option { + use crate::cuda_pipeline::gpu_ppo_collector::PpoCollectorConfig; + + let collector = self.gpu_collector.as_mut()?; + let features_buf = self.features_cuda.as_ref()?; + let targets_buf = self.targets_cuda.as_ref()?; + 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() + }; + + if let Err(e) = collector.reset_episodes(35_000.0, self.tick_size as f32 * self.spread_ticks as f32, 0.20) { + warn!("PPO GPU reset_episodes failed: {e}"); + return None; + } + + match collector.collect_experiences(features_buf, targets_buf, &episode_starts, &config) { + Ok(batch) => { + info!("GPU PPO collected {} experiences ({} episodes × {} timesteps)", + batch.n_episodes * batch.timesteps, batch.n_episodes, batch.timesteps); + Some(gpu_ppo_batch_to_trajectory_batch(&batch)) + } + Err(e) => { + warn!("PPO GPU experience collection failed, falling back to CPU: {e}"); + None + } + } + } +} + +/// Convert a GPU PPO experience batch to a TrajectoryBatch for `PPO::update()`. +/// +/// The GPU kernel produces 54-dim states (51 features + 3 portfolio state) but +/// the PPO model in hyperopt uses state_dim=51. We truncate to match. +#[cfg(feature = "cuda")] +fn gpu_ppo_batch_to_trajectory_batch( + batch: &crate::cuda_pipeline::gpu_ppo_collector::PpoExperienceBatch, +) -> TrajectoryBatch { + let kernel_state_dim = 54; // GPU kernel output: 51 features + 3 portfolio state + let model_state_dim = 51; // PPO hyperopt model state_dim + let total = batch.n_episodes * batch.timesteps; + + // Truncate states from 54 → 51 dims (strip portfolio state appended by kernel) + let states: Vec> = batch + .states + .chunks(kernel_state_dim) + .take(total) + .map(|c| c.get(..model_state_dim).unwrap_or(c).to_vec()) + .collect(); + + let actions: Vec = batch + .actions + .iter() + .take(total) + .map(|&a| match a.clamp(0, 2) { + 0 => TradingAction::Buy, + 1 => TradingAction::Sell, + _ => TradingAction::Hold, + }) + .collect(); + + let log_probs: Vec = batch.log_probs.iter().take(total).copied().collect(); + + // V(s) = R(s) - A(s) + let values: Vec = batch + .returns + .iter() + .zip(&batch.advantages) + .take(total) + .map(|(r, a)| r - a) + .collect(); + + let dones: Vec = batch + .done_flags + .iter() + .take(total) + .map(|&d| d != 0) + .collect(); + + let rewards = vec![0.0_f32; states.len()]; + + TrajectoryBatch { + trajectories: vec![], + states, + actions, + log_probs, + values, + rewards, + dones, + advantages: batch.advantages.iter().take(total).copied().collect(), + returns: batch.returns.iter().take(total).copied().collect(), + } } /// Write a log entry to the training log file @@ -515,18 +800,34 @@ impl HyperparameterOptimizable for PPOTrainer { 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. + #[cfg(feature = "cuda")] + self.ensure_gpu_data(&training_data, &ppo_agent); + // 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); - // Generate trajectories from real market data using current policy - let mut trajectory_batch = self - .generate_trajectories_from_data(&ppo_agent, train_data, episodes_this_batch) - .map_err(|e| { - MLError::TrainingError(format!("Failed to generate trajectories: {}", e)) - })?; + + // GPU-first experience collection with CPU fallback + #[cfg(feature = "cuda")] + let gpu_batch = self.gpu_collect_trajectories(episodes_this_batch); + #[cfg(not(feature = "cuda"))] + let gpu_batch: Option = None; + + let mut trajectory_batch = if let Some(batch) = gpu_batch { + batch + } else { + // CPU fallback: generate trajectories from real market data + self.generate_trajectories_from_data(&ppo_agent, train_data, episodes_this_batch) + .map_err(|e| { + MLError::TrainingError(format!("Failed to generate trajectories: {}", e)) + })? + }; // Update PPO with trajectory batch let (policy_loss, value_loss) = ppo_agent diff --git a/crates/ml/src/real_data_loader.rs b/crates/ml/src/real_data_loader.rs index 3f069f131..8f61c7929 100644 --- a/crates/ml/src/real_data_loader.rs +++ b/crates/ml/src/real_data_loader.rs @@ -616,6 +616,7 @@ mod tests { use super::*; #[tokio::test] + #[ignore] // Requires local DBN data files async fn test_load_symbol_data() -> Result<()> { let mut loader = RealDataLoader::new_from_workspace()?; @@ -638,6 +639,7 @@ mod tests { } #[tokio::test] + #[ignore] // Requires local DBN data files async fn test_extract_features() -> Result<()> { let mut loader = RealDataLoader::new_from_workspace()?; let bars = loader.load_symbol_data("ZN.FUT").await?; @@ -663,6 +665,7 @@ mod tests { } #[tokio::test] + #[ignore] // Requires local DBN data files async fn test_calculate_indicators() -> Result<()> { let mut loader = RealDataLoader::new_from_workspace()?; let bars = loader.load_symbol_data("ZN.FUT").await?; diff --git a/infra/k8s/training/job-template.yaml b/infra/k8s/training/job-template.yaml index 663b0d3cc..19e072c36 100644 --- a/infra/k8s/training/job-template.yaml +++ b/infra/k8s/training/job-template.yaml @@ -9,7 +9,7 @@ metadata: foxhunt/job-type: training spec: backoffLimit: 1 - activeDeadlineSeconds: 3600 + activeDeadlineSeconds: 21600 # 6 hours — hyperopt runs 20 trials × 8 epochs ttlSecondsAfterFinished: 600 template: metadata: diff --git a/infra/scripts/train.sh b/infra/scripts/train.sh index c1590cd44..bfc77b8d0 100755 --- a/infra/scripts/train.sh +++ b/infra/scripts/train.sh @@ -19,7 +19,7 @@ set -euo pipefail SYMBOL="ES.FUT" TRIALS=20 MAX_STEPS=2000 -TIMEOUT=3600 +TIMEOUT=3600 # default; hyperopt preset overrides to 6h below DATA_DIR="/data/futures-baseline" NAMESPACE="foxhunt" IMAGE="rg.fr-par.scw.cloud/foxhunt/training:latest" @@ -114,6 +114,10 @@ case "$PRESET" in hyperopt) [[ -z "$MODEL" ]] && die "hyperopt preset requires --model" validate_model "$MODEL" + # Hyperopt runs 20 trials × 8 epochs — needs 6h unless user specified --timeout + if [[ "$TIMEOUT" -eq 3600 ]]; then + TIMEOUT=21600 + fi ;; evaluate) [[ -z "$RUN_ID" ]] && die "evaluate preset requires --run-id (from a previous training run)" diff --git a/services/trading_service/src/core/order_manager.rs b/services/trading_service/src/core/order_manager.rs index bd6f18546..3fd46e4ee 100644 --- a/services/trading_service/src/core/order_manager.rs +++ b/services/trading_service/src/core/order_manager.rs @@ -497,7 +497,7 @@ impl OrderManager { if is_match { let match_quality = self.calculate_match_quality(price, entry_price, side); - if best_match.map_or(true, |(_, qual)| match_quality > qual) { + if best_match.is_none_or(|(_, qual)| match_quality > qual) { best_match = Some((i, match_quality)); } } @@ -516,7 +516,7 @@ impl OrderManager { if is_match { let match_quality = self.calculate_match_quality(price, entry.price, side); - if best_match.map_or(true, |(_, qual)| match_quality > qual) { + if best_match.is_none_or(|(_, qual)| match_quality > qual) { best_match = Some((i, match_quality)); } } diff --git a/services/trading_service/src/services/risk.rs b/services/trading_service/src/services/risk.rs index ac631b9e9..c37d4f2c3 100644 --- a/services/trading_service/src/services/risk.rs +++ b/services/trading_service/src/services/risk.rs @@ -594,12 +594,12 @@ impl RiskService for RiskServiceImpl { .filter(|p| { req.symbol .as_ref() - .map_or(true, |s| s.is_empty() || p.symbol == *s) + .is_none_or(|s| s.is_empty() || p.symbol == *s) }) .filter(|p| { req.account_id .as_ref() - .map_or(true, |a| a.is_empty() || p.account_id == *a) + .is_none_or(|a| a.is_empty() || p.account_id == *a) }) .collect();