From 25f8d514b1fce1507358b8cb74dec1472987f5b2 Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Tue, 3 Mar 2026 13:37:12 +0100 Subject: [PATCH] perf(ml): batched GPU inference for hyperopt backtest evaluation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace per-bar GPU forward passes with chunked batch inference (1024 bars per chunk). This reduces ~90K individual CUDA kernel launches to ~88 batched forward passes — ~1000× fewer GPU round-trips. Changes: - Add DQN::batch_greedy_actions(&self) for immutable batched forward+argmax - Add RegimeConditionalDQN::batch_greedy_actions with per-regime-head batching - Add DQNTrainer::convert_to_state_vec (CPU-only, skips GPU tensor allocation) - Add PortfolioTracker::set_position_direct for backtest state sync - Rewrite hyperopt backtest loop: chunked batching with portfolio state updates between chunks (1024-bar granularity ≈ 17h of 1-min data) Portfolio features (last 3 of 54 dims) are refreshed between chunks via set_portfolio_for_backtest(), keeping position/PnL/exposure accurate at chunk boundaries while batching inference within each chunk. 2640 tests pass, 0 clippy warnings. Co-Authored-By: Claude Opus 4.6 --- crates/ml/src/dqn/dqn.rs | 26 ++ crates/ml/src/dqn/portfolio_tracker.rs | 27 +++ crates/ml/src/dqn/regime_conditional.rs | 49 ++++ crates/ml/src/hyperopt/adapters/dqn.rs | 305 +++++++++++++----------- crates/ml/src/trainers/dqn/config.rs | 10 + crates/ml/src/trainers/dqn/trainer.rs | 41 ++++ 6 files changed, 318 insertions(+), 140 deletions(-) diff --git a/crates/ml/src/dqn/dqn.rs b/crates/ml/src/dqn/dqn.rs index f7c1fe415..28b659ec0 100644 --- a/crates/ml/src/dqn/dqn.rs +++ b/crates/ml/src/dqn/dqn.rs @@ -1349,6 +1349,32 @@ impl DQN { Ok(action) } + /// Batch greedy action selection — single GPU forward pass for N states. + /// + /// Runs a batched forward pass through the Q-network and returns the greedy + /// (argmax) action index for each state. No exploration, no side effects. + /// + /// # Arguments + /// * `states` - Tensor of shape `[N, state_dim]` + /// + /// # Returns + /// `Vec` of length N containing greedy action indices. + /// + /// # Performance + /// Reduces N individual GPU kernel launches to a single batched forward pass. + /// With EVAL_CHUNK_SIZE=1024, this gives ~1000× fewer kernel launches vs per-bar inference. + pub fn batch_greedy_actions(&self, states: &Tensor) -> Result, MLError> { + let q_values = self.forward(states)?; + let action_indices = q_values + .argmax(1) + .map_err(|e| MLError::ModelError(format!("Batch argmax failed: {}", e)))? + .to_vec1::() + .map_err(|e| { + MLError::ModelError(format!("Failed to transfer action indices to CPU: {}", e)) + })?; + Ok(action_indices.into_iter().map(|a| a as usize).collect()) + } + /// Track action for entropy penalty calculation /// /// This should be called after action selection (both single and batch modes) diff --git a/crates/ml/src/dqn/portfolio_tracker.rs b/crates/ml/src/dqn/portfolio_tracker.rs index e56800871..12cd21593 100644 --- a/crates/ml/src/dqn/portfolio_tracker.rs +++ b/crates/ml/src/dqn/portfolio_tracker.rs @@ -538,6 +538,33 @@ impl PortfolioTracker { self.cumulative_transaction_costs = 0.0; } + /// Get initial capital + pub fn initial_capital(&self) -> f32 { + self.initial_capital + } + + /// Get average spread + pub fn spread(&self) -> f32 { + self.avg_spread + } + + /// Set position directly for backtest portfolio state synchronization. + /// + /// Used between evaluation chunks to sync the tracker with the + /// EvaluationEngine's current position without executing a trade. + pub fn set_position_direct( + &mut self, + position_size: f32, + entry_price: f32, + current_price: f32, + ) { + self.position_size = position_size; + self.position_entry_price = entry_price; + self.last_price = current_price; + // Reconstruct cash: initial_capital - position cost + self.cash = self.initial_capital - (position_size * entry_price).abs(); + } + // ========== Public Accessors for TradeExecutor Integration ========== /// Get current cash balance diff --git a/crates/ml/src/dqn/regime_conditional.rs b/crates/ml/src/dqn/regime_conditional.rs index 34d542113..a76521af3 100644 --- a/crates/ml/src/dqn/regime_conditional.rs +++ b/crates/ml/src/dqn/regime_conditional.rs @@ -296,6 +296,55 @@ impl RegimeConditionalDQN { Ok(action) } + /// Batch greedy action selection across regime heads. + /// + /// Classifies each state into a regime, groups by regime, batches per head, + /// then reassembles results in original order. + pub fn batch_greedy_actions(&self, states: &Tensor) -> Result, MLError> { + let state_vecs = states.to_vec2::().map_err(|e| { + MLError::ModelError(format!("Failed to extract states for regime: {}", e)) + })?; + let n = state_vecs.len(); + let mut results = vec![0_usize; n]; + + let mut trending_idx = Vec::new(); + let mut ranging_idx = Vec::new(); + let mut volatile_idx = Vec::new(); + for (i, sv) in state_vecs.iter().enumerate() { + match RegimeType::classify_from_features(sv) { + RegimeType::Trending => trending_idx.push(i), + RegimeType::Ranging => ranging_idx.push(i), + RegimeType::Volatile => volatile_idx.push(i), + } + } + + let device = &self.device; + for (indices, head) in [ + (&trending_idx, &self.trending_head), + (&ranging_idx, &self.ranging_head), + (&volatile_idx, &self.volatile_head), + ] { + if indices.is_empty() { + continue; + } + let dim = state_vecs[0].len(); + let mut flat = Vec::with_capacity(indices.len() * dim); + for &i in indices { + flat.extend_from_slice(&state_vecs[i]); + } + let sub_tensor = + Tensor::from_vec(flat, (indices.len(), dim), device).map_err(|e| { + MLError::ModelError(format!("Regime sub-batch failed: {}", e)) + })?; + let actions = head.batch_greedy_actions(&sub_tensor)?; + for (j, &idx) in indices.iter().enumerate() { + results[idx] = actions[j]; + } + } + + Ok(results) + } + /// Store experience in all head buffers /// /// Since we no longer have shared memory, each head maintains its own buffer. diff --git a/crates/ml/src/hyperopt/adapters/dqn.rs b/crates/ml/src/hyperopt/adapters/dqn.rs index 99d6a7af5..6ca901ee2 100644 --- a/crates/ml/src/hyperopt/adapters/dqn.rs +++ b/crates/ml/src/hyperopt/adapters/dqn.rs @@ -2761,169 +2761,194 @@ impl HyperparameterOptimizable for DQNTrainer { tracing::debug!("Backtest is enabled, starting backtest..."); tracing::info!("Running backtest on validation data..."); - // Get validation data from trainer - let val_data = internal_trainer.get_val_data(); + // Pre-extract close prices from validation data (drops borrow after this block) + let val_close_prices: Vec = { + let val_data = internal_trainer.get_val_data(); + tracing::debug!("Got {} validation samples for backtest", val_data.len()); + val_data + .iter() + .map(|(fv, target)| { + if target.len() >= 2 { + target[0] + } else { + fv[3] + } + }) + .collect() + }; - // Log validation data details - tracing::debug!("Got {} validation samples for backtest", val_data.len()); - if !val_data.is_empty() { - tracing::debug!("Sample validation data - features.len={}, target_len={}", - val_data[0].0.len(), val_data[0].1.len()); - } - - if val_data.is_empty() { + if val_close_prices.is_empty() { tracing::warn!("No validation data available for backtest"); None } else { - // Run backtest in dedicated runtime (clean separation from training) + // ── Chunked batch inference backtest ────────────────────────── + // + // Process bars in chunks of EVAL_CHUNK_SIZE (1024). Each chunk: + // 1. Extracts state vectors on CPU (with current portfolio features) + // 2. Single GPU forward pass → all Q-values at once + // 3. Sequential trade simulation on CPU + // 4. Syncs portfolio state to trainer for next chunk + // + // vs. the old per-bar loop: ~1000× fewer GPU kernel launches. + const EVAL_CHUNK_SIZE: usize = 1024; + + let kelly_fraction = internal_trainer.get_kelly_fraction(); + let mut engine = EvaluationEngine::new_with_kelly(10000.0, kelly_fraction); + + let agent_arc = internal_trainer.get_agent().clone(); + let device = internal_trainer.device().clone(); + + let total_bars = val_close_prices.len(); + let num_chunks = total_bars.div_ceil(EVAL_CHUNK_SIZE); + let mut ohlcv_bars = Vec::with_capacity(total_bars); + let mut conversion_failures: usize = 0; + + tracing::info!( + "Batched backtest: {} bars in {} chunks of {} (device: {:?})", + total_bars, num_chunks, EVAL_CHUNK_SIZE, device, + ); + + // Acquire read lock once — batch_greedy_actions is &self (no mutation) let runtime = tokio::runtime::Runtime::new().map_err(|e| { MLError::TrainingError(format!("Failed to create runtime for backtest: {}", e)) })?; + let agent_guard = runtime.block_on(agent_arc.read()); - let backtest_result = runtime.block_on(async { - // Get Kelly fraction from trainer for position sizing - let kelly_fraction = internal_trainer.get_kelly_fraction(); - eprintln!("DEBUG: Kelly fraction for backtest: {:.4}", kelly_fraction); + for chunk_idx in 0..num_chunks { + let chunk_start = chunk_idx * EVAL_CHUNK_SIZE; + let chunk_end = (chunk_start + EVAL_CHUNK_SIZE).min(total_bars); + let chunk_len = chunk_end - chunk_start; - // Create evaluation engine with $10K initial capital and Kelly position sizing - let mut engine = EvaluationEngine::new_with_kelly(10000.0, kelly_fraction); + // 1. Extract state vectors on CPU (portfolio features from trainer) + // Re-borrow val_data per chunk to allow mutable portfolio updates between chunks + let mut flat_states: Vec = Vec::with_capacity(chunk_len * 54); + let mut valid_mask: Vec = Vec::with_capacity(chunk_len); - // Get agent from trainer (Arc>) - let agent_arc = internal_trainer.get_agent(); + { + let val_data = internal_trainer.get_val_data(); + let chunk = &val_data[chunk_start..chunk_end]; - // Collect OHLCV bars for metrics calculation - let mut ohlcv_bars = Vec::with_capacity(val_data.len()); - - // DEBUG: Track conversion failures - eprintln!("DEBUG: Starting backtest loop with {} bars", val_data.len()); - let mut conversion_failures = 0; - let mut tensor_extraction_failures = 0; - let mut action_selection_failures = 0; - - // Run backtest using trained agent - for (bar_idx, (feature_vec, target)) in val_data.iter().enumerate() { - // Extract close price from target (target[0] is current close, target[1] is next close) - let close_price = if target.len() >= 2 { - target[0] - } else { - feature_vec[3] // Fallback to close log return feature - }; - - // Convert feature vector to state tensor - let state_tensor = match internal_trainer.convert_to_state(feature_vec, close_price) { - Ok(tensor) => tensor, - Err(e) => { + for (i, (feature_vec, _target)) in chunk.iter().enumerate() { + let close_price = val_close_prices[chunk_start + i]; + match internal_trainer.convert_to_state_vec(feature_vec, close_price) { + Ok(sv) => { + flat_states.extend_from_slice(&sv); + valid_mask.push(true); + } + Err(_) => { conversion_failures += 1; - if bar_idx < 5 { - eprintln!("DEBUG: Bar {}: State conversion failed - close_price={}, error={}", bar_idx, close_price, e); - } - continue; + flat_states.extend(std::iter::repeat_n(0.0_f32, 54)); + valid_mask.push(false); } - }; - - // Extract state as Vec from tensor - let state_vec: Vec = match state_tensor.to_vec1() { - Ok(vec) => { - if bar_idx < 5 { - eprintln!("DEBUG: Bar {}: State extracted successfully - dim={}", bar_idx, vec.len()); - } - vec - }, - Err(e) => { - tensor_extraction_failures += 1; - if bar_idx < 5 { - eprintln!("DEBUG: Bar {}: Tensor extraction failed - error={}", bar_idx, e); - } - continue; - } - }; - - // Select action (greedy, no epsilon) using write lock - // Note: agent is Arc> with tokio::sync::RwLock (async) - let trading_action = { - let mut agent = agent_arc.write().await; - match agent.select_action(&state_vec) { - Ok(action) => action, - Err(e) => { - action_selection_failures += 1; - if bar_idx < 5 { - eprintln!("DEBUG: Bar {}: Action selection failed - error={}", bar_idx, e); - } - continue; - } - } - }; - - // Convert FactoredAction to evaluation::Action via legacy TradingAction - let legacy_action = trading_action.to_legacy_action(); - let action = match legacy_action { - crate::dqn::TradingAction::Buy => Action::Buy, - crate::dqn::TradingAction::Sell => Action::Sell, - crate::dqn::TradingAction::Hold => Action::Hold, - }; - - // Create OHLCV bar (we only have close price from validation data) - // For backtest purposes, we set open=high=low=close - let bar = OHLCVBarF32 { - timestamp: bar_idx as i64, // Use bar index as timestamp - open: close_price as f32, - high: close_price as f32, - low: close_price as f32, - close: close_price as f32, - volume: 0.0, // Volume not available in validation data - }; - - // Process bar in evaluation engine - engine.process_bar(bar_idx, &bar, action); - ohlcv_bars.push(bar); + } } + } // val_data borrow dropped here - // DEBUG: Report failure statistics - eprintln!("DEBUG: Backtest loop complete - processed {} bars", ohlcv_bars.len()); - eprintln!("DEBUG: Failures - conversion: {}, tensor: {}, action: {}", - conversion_failures, tensor_extraction_failures, action_selection_failures); + // 2. Single GPU forward pass for entire chunk + let batch_tensor = + candle_core::Tensor::from_slice(&flat_states, (chunk_len, 54), &device) + .map_err(|e| { + MLError::ModelError(format!( + "Batch tensor creation failed: {}", + e + )) + })?; - // Close any open position at end of backtest - if let Some(last_bar) = ohlcv_bars.last() { - engine.close_position(ohlcv_bars.len() - 1, last_bar); + let action_indices = agent_guard.batch_greedy_actions(&batch_tensor)?; + + // 3. Sequential trade simulation on CPU + for (i, &action_idx) in action_indices.iter().enumerate() { + if !valid_mask[i] { + continue; } + let bar_idx = chunk_start + i; + let close = val_close_prices[bar_idx] as f32; - // Calculate performance metrics - let metrics = PerformanceMetrics::from_trades( - &engine.trades, - engine.initial_capital, - &ohlcv_bars, + let factored = crate::dqn::FactoredAction::from_index(action_idx)?; + let legacy = factored.to_legacy_action(); + let action = match legacy { + crate::dqn::TradingAction::Buy => Action::Buy, + crate::dqn::TradingAction::Sell => Action::Sell, + crate::dqn::TradingAction::Hold => Action::Hold, + }; + + let bar = OHLCVBarF32 { + timestamp: bar_idx as i64, + open: close, + high: close, + low: close, + close, + volume: 0.0, + }; + engine.process_bar(bar_idx, &bar, action); + ohlcv_bars.push(bar); + } + + // 4. Sync portfolio state to trainer for next chunk's features + if chunk_idx + 1 < num_chunks { + let (pos_size, entry_price, last_close) = + if let Some(ref pos) = engine.current_position { + let size = match pos.direction { + crate::evaluation::engine::PositionDirection::Long => 1.0_f32, + crate::evaluation::engine::PositionDirection::Short => -1.0_f32, + }; + let last = val_close_prices[chunk_end - 1] as f32; + (size, pos.entry_price, last) + } else { + (0.0_f32, 0.0_f32, 0.0_f32) + }; + internal_trainer.set_portfolio_for_backtest( + pos_size, + entry_price, + last_close, ); + } + } - eprintln!("DEBUG: Metrics - Sharpe: {:.4}, Trades: {}, Win Rate: {:.2}%, DD: {:.2}%, Return: {:.2}%", - metrics.sharpe_ratio, metrics.total_trades, metrics.win_rate, metrics.max_drawdown_pct, metrics.total_return_pct); + // Release read lock + drop(agent_guard); - tracing::info!("Backtest complete: {} trades, Sharpe {:.4}, Win Rate {:.2}%, Max DD {:.2}%, Total Return {:.2}%", - metrics.total_trades, - metrics.sharpe_ratio, - metrics.win_rate, - metrics.max_drawdown_pct, - metrics.total_return_pct - ); + // Close any open position at end + if let Some(last_bar) = ohlcv_bars.last() { + engine.close_position(ohlcv_bars.len() - 1, last_bar); + } - BacktestMetrics { - sharpe_ratio: metrics.sharpe_ratio, - win_rate: metrics.win_rate, - max_drawdown_pct: metrics.max_drawdown_pct, - total_return_pct: metrics.total_return_pct, - total_trades: metrics.total_trades, - sortino_ratio: metrics.sortino_ratio, - calmar_ratio: metrics.calmar_ratio, - var_95: metrics.var_95, - cvar_95: metrics.cvar_95, - beta: metrics.beta, - alpha: metrics.alpha, - information_ratio: metrics.information_ratio, - omega_ratio: metrics.omega_ratio, - } - }); + // Calculate performance metrics + let metrics = PerformanceMetrics::from_trades( + &engine.trades, + engine.initial_capital, + &ohlcv_bars, + ); - Some(backtest_result) + tracing::info!( + "Batched backtest complete: {} bars, {} chunks, {} trades, \ + Sharpe {:.4}, Win Rate {:.2}%, Max DD {:.2}%, Return {:.2}%, \ + conversion_failures={}", + ohlcv_bars.len(), + num_chunks, + metrics.total_trades, + metrics.sharpe_ratio, + metrics.win_rate, + metrics.max_drawdown_pct, + metrics.total_return_pct, + conversion_failures, + ); + + Some(BacktestMetrics { + sharpe_ratio: metrics.sharpe_ratio, + win_rate: metrics.win_rate, + max_drawdown_pct: metrics.max_drawdown_pct, + total_return_pct: metrics.total_return_pct, + total_trades: metrics.total_trades, + sortino_ratio: metrics.sortino_ratio, + calmar_ratio: metrics.calmar_ratio, + var_95: metrics.var_95, + cvar_95: metrics.cvar_95, + beta: metrics.beta, + alpha: metrics.alpha, + information_ratio: metrics.information_ratio, + omega_ratio: metrics.omega_ratio, + }) } } else { None diff --git a/crates/ml/src/trainers/dqn/config.rs b/crates/ml/src/trainers/dqn/config.rs index 539b703e4..15a42f0ad 100644 --- a/crates/ml/src/trainers/dqn/config.rs +++ b/crates/ml/src/trainers/dqn/config.rs @@ -61,6 +61,16 @@ impl DQNAgentType { } } + /// Batch greedy action selection — single forward pass per architecture head. + /// + /// Dispatches to the underlying DQN or RegimeConditionalDQN batch method. + pub fn batch_greedy_actions(&self, states: &Tensor) -> Result, MLError> { + match self { + Self::Standard(agent) => agent.batch_greedy_actions(states), + Self::RegimeConditional(agent) => agent.batch_greedy_actions(states), + } + } + /// Reset count-based exploration bonus (call at epoch boundary) pub fn reset_count_bonus(&mut self) { match self { diff --git a/crates/ml/src/trainers/dqn/trainer.rs b/crates/ml/src/trainers/dqn/trainer.rs index bbb33ad4f..2f2676cbf 100644 --- a/crates/ml/src/trainers/dqn/trainer.rs +++ b/crates/ml/src/trainers/dqn/trainer.rs @@ -3747,6 +3747,47 @@ impl DQNTrainer { .context("Failed to create state tensor from TradingState") } + /// Convert feature vector to flat state Vec (CPU only, no GPU tensor). + /// + /// Same as `convert_to_state` but returns the raw vector instead of a GPU tensor. + /// Used by chunked batch inference to avoid per-bar GPU allocations. + pub fn convert_to_state_vec( + &self, + feature_vec: &FeatureVector51, + close_price: f64, + ) -> Result> { + let close = rust_decimal::Decimal::try_from(close_price) + .map_err(|e| CommonError::validation(&format!("Invalid close price: {}", e)))?; + let trading_state = self.feature_vector_to_state(feature_vec, Some(close))?; + Ok(trading_state.to_vector()) + } + + /// Update portfolio tracker to reflect current position from backtest engine. + /// + /// Called between chunks so the next chunk's portfolio features + /// accurately reflect the current position (direction, value, exposure). + pub fn set_portfolio_for_backtest( + &mut self, + position_size: f32, + entry_price: f32, + current_price: f32, + ) { + self.portfolio_tracker = PortfolioTracker::new( + self.portfolio_tracker.initial_capital(), + self.portfolio_tracker.spread(), + 0.0, + ); + if position_size.abs() > f32::EPSILON { + self.portfolio_tracker + .set_position_direct(position_size, entry_price, current_price); + } + } + + /// Get the device used by this trainer + pub fn device(&self) -> &candle_core::Device { + &self.device + } + /// Get access to the DQN agent /// /// Returns a reference to the Arc> for checkpoint saving.