gitadmin/foxhunt
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

perf(ml): batched GPU inference for hyperopt backtest evaluation

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
jgrusewski
2026-03-03 13:37:12 +01:00
parent a7b7796147
commit 25f8d514b1
6 changed files with 318 additions and 140 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
crates/ml/src/dqn/dqn.rs
View File

@@ -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<usize>` 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<Vec<usize>, 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::<u32>()
.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)

27
crates/ml/src/dqn/portfolio_tracker.rs
View File

@@ -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

49
crates/ml/src/dqn/regime_conditional.rs
View File

@@ -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<Vec<usize>, MLError> {
let state_vecs = states.to_vec2::<f32>().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.

305
crates/ml/src/hyperopt/adapters/dqn.rs
View File

@@ -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<f64> = {
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<f32> = Vec::with_capacity(chunk_len * 54);
let mut valid_mask: Vec<bool> = Vec::with_capacity(chunk_len);
// Get agent from trainer (Arc<RwLock<DQN>>)
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<f32> from tensor
let state_vec: Vec<f32> = 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<RwLock<DQN>> 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

10
crates/ml/src/trainers/dqn/config.rs
View File

@@ -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<Vec<usize>, 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 {

41
crates/ml/src/trainers/dqn/trainer.rs
View File

@@ -3747,6 +3747,47 @@ impl DQNTrainer {
.context("Failed to create state tensor from TradingState")
}
/// Convert feature vector to flat state Vec<f32> (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<Vec<f32>> {
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<RwLock<DQNAgentType>> for checkpoint saving.