Merge branch 'feature/cuda-backtest-plan'

This commit is contained in:
jgrusewski
2026-03-11 13:24:04 +01:00
19 changed files with 5496 additions and 123 deletions

View File

@@ -2010,7 +2010,7 @@ impl DQN {
///
/// Returns a `[N, num_actions]` tensor of expected Q-values (or `CVaR` values
/// when `use_cvar_action_selection` is enabled).
fn q_values_for_batch(&self, states: &Tensor) -> Result<Tensor, MLError> {
pub fn q_values_for_batch(&self, states: &Tensor) -> Result<Tensor, MLError> {
if self.config.use_iqn && self.iqn_network.is_some() {
let iqn_net = self.iqn_network.as_ref().ok_or_else(|| {
MLError::ModelError("IQN network not initialized despite use_iqn=true".into())

View File

@@ -405,6 +405,69 @@ impl RegimeConditionalDQN {
Ok(results)
}
/// Batch Q-value computation across regime heads — fully GPU-resident.
///
/// Uses on-device regime classification masks to blend Q-values from all
/// 3 heads without any GPU→CPU roundtrip:
/// Q_final = Q_trending * mask_trending + Q_ranging * mask_ranging + Q_volatile * mask_volatile
///
/// Used by `GpuBacktestEvaluator` for GPU-side argmax.
pub fn batch_q_values(&self, states: &Tensor) -> Result<Tensor, MLError> {
let n = states.dims()[0];
if n == 0 {
return Err(MLError::ModelError("Empty batch for batch_q_values".into()));
}
// On-device regime classification — zero CPU roundtrip
let (trending_mask, ranging_mask, volatile_mask) =
RegimeType::classify_regime_masks_gpu(states)?;
// Forward full batch through all 3 heads (each head ignores irrelevant samples
// via masking — cheaper than splitting/gathering sub-batches)
let trending_q = self.trending_head.q_values_for_batch(states)?;
let ranging_q = self.ranging_head.q_values_for_batch(states)?;
let volatile_q = self.volatile_head.q_values_for_batch(states)?;
// Reshape masks from [batch] to [batch, 1] for broadcasting over actions dim
let trending_mask = trending_mask.unsqueeze(1).map_err(|e| {
MLError::ModelError(format!("trending unsqueeze: {e}"))
})?;
let ranging_mask = ranging_mask.unsqueeze(1).map_err(|e| {
MLError::ModelError(format!("ranging unsqueeze: {e}"))
})?;
let volatile_mask = volatile_mask.unsqueeze(1).map_err(|e| {
MLError::ModelError(format!("volatile unsqueeze: {e}"))
})?;
// Ensure Q-values are F32 for multiplication with F32 masks
let trending_q = trending_q.to_dtype(DType::F32).map_err(|e| {
MLError::ModelError(format!("trending_q to_f32: {e}"))
})?;
let ranging_q = ranging_q.to_dtype(DType::F32).map_err(|e| {
MLError::ModelError(format!("ranging_q to_f32: {e}"))
})?;
let volatile_q = volatile_q.to_dtype(DType::F32).map_err(|e| {
MLError::ModelError(format!("volatile_q to_f32: {e}"))
})?;
// Blend: Q_final = sum of (Q_head * mask_head) across all regimes
let blended = trending_q.broadcast_mul(&trending_mask).map_err(|e| {
MLError::ModelError(format!("trending mul: {e}"))
})?;
let blended = blended.add(
&ranging_q.broadcast_mul(&ranging_mask).map_err(|e| {
MLError::ModelError(format!("ranging mul: {e}"))
})?
).map_err(|e| MLError::ModelError(format!("ranging add: {e}")))?;
let blended = blended.add(
&volatile_q.broadcast_mul(&volatile_mask).map_err(|e| {
MLError::ModelError(format!("volatile mul: {e}"))
})?
).map_err(|e| MLError::ModelError(format!("volatile add: {e}")))?;
Ok(blended)
}
/// Batch softmax action selection across regime heads (Gumbel-max, GPU-resident).
///
/// Same regime-dispatch logic as `batch_greedy_actions` but delegates to

View File

@@ -145,6 +145,20 @@ struct Args {
/// (must match the file used during training so network architecture is identical)
#[arg(long)]
hyperopt_params: Option<PathBuf>,
/// Use GPU-accelerated backtest evaluation for DQN (uses greedy argmax, not softmax).
///
/// When enabled and CUDA is available, DQN folds are evaluated using
/// `GpuBacktestEvaluator` — all bars processed on GPU with a single metrics
/// readback. Results differ slightly from the default CPU path because the
/// GPU path uses greedy argmax while the CPU path uses hierarchical softmax.
/// Falls back to the CPU path on any GPU error.
#[arg(long, default_value_t = false)]
gpu_eval: bool,
/// Initial capital for GPU backtest evaluator (used only with --gpu-eval).
#[arg(long, default_value_t = 100_000.0)]
initial_capital: f64,
}
// ---------------------------------------------------------------------------
@@ -692,6 +706,198 @@ fn evaluate_dqn_fold(
Ok((returns, action_counts))
}
// ---------------------------------------------------------------------------
// GPU-Accelerated DQN Evaluation
// ---------------------------------------------------------------------------
/// GPU-accelerated DQN evaluation using `GpuBacktestEvaluator`.
///
/// Uses greedy argmax for action selection (not hierarchical softmax as in the
/// CPU path). Results will differ slightly from `evaluate_dqn_fold`. This is
/// intentional: the GPU path prioritises throughput (all bars processed in a
/// single GPU loop with one metrics readback) over exact matching of the CPU
/// softmax-based selection.
///
/// The function returns `Vec<WindowMetrics>` — one entry per window (here,
/// one window = the full test fold). Call sites are responsible for converting
/// these into `FoldMetrics` for the report.
#[cfg(feature = "cuda")]
#[allow(clippy::cognitive_complexity)]
fn evaluate_dqn_fold_gpu(
fold: usize,
test_features: &[ml::features::extraction::FeatureVector],
test_bars: &[ml::types::OHLCVBar],
models_dir: &std::path::Path,
args: &Args,
hp: &Option<serde_json::Value>,
) -> Result<Vec<ml::cuda_pipeline::gpu_backtest_evaluator::WindowMetrics>> {
use ml::cuda_pipeline::gpu_backtest_evaluator::{GpuBacktestConfig, GpuBacktestEvaluator};
// ── Load checkpoint (identical to CPU path) ───────────────────────────
let best_path = models_dir.join(format!("dqn_fold{}_best.safetensors", fold));
let ckpt_path = if best_path.exists() {
best_path
} else {
let pattern = format!("dqn_fold{}_epoch", fold);
let mut candidates: Vec<_> = std::fs::read_dir(models_dir)
.ok()
.into_iter()
.flatten()
.filter_map(|e| e.ok())
.filter(|e| {
let name = e.file_name();
let s = name.to_string_lossy();
s.starts_with(&pattern) && s.ends_with(".safetensors")
})
.collect();
candidates.sort_by_key(|e| std::cmp::Reverse(e.file_name()));
match candidates.first() {
Some(entry) => {
let p = entry.path();
info!(
" [DQN GPU] fold {} using fallback checkpoint: {}",
fold,
p.display()
);
p
}
None => {
anyhow::bail!(
"No DQN checkpoint found for fold {} in {}",
fold,
models_dir.display()
);
}
}
};
#[allow(clippy::integer_division)]
let config = DQNConfig {
state_dim: args.feature_dim,
num_actions: args.num_actions,
hidden_dims: {
let base = hp_usize(hp, "hidden_dim_base").unwrap_or(256);
let align = |x: usize| -> usize { x.div_ceil(8) * 8 };
vec![align(base), align(base / 2), align(base / 4)]
},
learning_rate: 1e-4,
gamma: hp_f64(hp, "gamma").unwrap_or(0.95) as f32,
epsilon_start: 0.0,
epsilon_end: 0.0,
epsilon_decay: 1.0,
replay_buffer_capacity: 100,
batch_size: 64,
min_replay_size: 64,
target_update_freq: 500,
warmup_steps: 0,
use_double_dqn: hp_bool(hp, "use_double_dqn").unwrap_or(true),
use_huber_loss: true,
use_per: false,
use_dueling: hp_bool(hp, "use_dueling").unwrap_or(true),
dueling_hidden_dim: hp_usize(hp, "dueling_hidden_dim").unwrap_or(128),
use_distributional: hp_bool(hp, "use_distributional").unwrap_or(true),
num_atoms: hp_usize(hp, "num_atoms").unwrap_or(51),
v_min: hp_f64(hp, "v_min").unwrap_or(-2.0) as f32,
v_max: hp_f64(hp, "v_max").unwrap_or(2.0) as f32,
use_noisy_nets: hp_bool(hp, "use_noisy_nets").unwrap_or(true),
use_cql: false,
use_iqn: hp_bool(hp, "use_qr_dqn").unwrap_or(false),
iqn_num_quantiles: hp_usize(hp, "num_quantiles").unwrap_or(64),
use_cvar_action_selection: false,
..DQNConfig::default()
};
let mut dqn = DQN::new(config).context("Failed to create DQN model (GPU path)")?;
dqn.load_from_safetensors(&ckpt_path.to_string_lossy())
.with_context(|| {
format!(
"Failed to load DQN checkpoint (GPU path): {}",
ckpt_path.display()
)
})?;
dqn.set_eval_mode(true)
.context("Failed to set DQN eval mode (GPU path)")?;
let device = dqn.device().clone();
info!(
" [DQN GPU] Loaded checkpoint: {} (device={:?}, greedy argmax)",
ckpt_path.display(),
device,
);
// ── Build single window from all test data ────────────────────────────
//
// The GpuBacktestEvaluator's `feature_dim` parameter is the number of
// MARKET features only (the gather kernel appends the 3 portfolio features
// internally). Total state_dim = feature_dim + 3.
let eval_bars = test_features.len().saturating_sub(1);
if eval_bars == 0 {
anyhow::bail!("No bars to evaluate in GPU path for fold {}", fold);
}
// feature_dim passed to evaluator = state_dim - 3 portfolio dims
let market_feature_dim = args.feature_dim.saturating_sub(3);
let mut prices: Vec<[f32; 4]> = Vec::with_capacity(eval_bars);
let mut features: Vec<Vec<f32>> = Vec::with_capacity(eval_bars);
for bar_idx in 0..eval_bars {
let bar = test_bars.get(bar_idx).ok_or_else(|| {
anyhow::anyhow!("test_bars index {} out of bounds (len={})", bar_idx, test_bars.len())
})?;
let open = bar.open as f32;
let high = bar.high as f32;
let low = bar.low as f32;
let close = bar.close as f32;
prices.push([open, high, low, close]);
let fv = test_features.get(bar_idx).ok_or_else(|| {
anyhow::anyhow!(
"test_features index {} out of bounds (len={})",
bar_idx,
test_features.len()
)
})?;
// Take up to market_feature_dim values; truncate portfolio dims if any
let market_fv: Vec<f32> = fv
.iter()
.take(market_feature_dim)
.map(|&v| v as f32)
.collect();
features.push(market_fv);
}
let gpu_config = GpuBacktestConfig {
max_position: 1.0,
tx_cost_bps: args.tx_cost_bps as f32,
spread_cost: (args.tick_size * args.spread_ticks) as f32,
initial_capital: args.initial_capital as f32,
};
// Single window = full test fold
let mut evaluator = GpuBacktestEvaluator::new(
&[prices],
&[features],
market_feature_dim,
gpu_config,
&device,
)
.with_context(|| format!("GpuBacktestEvaluator::new failed for fold {}", fold))?;
let metrics = evaluator
.evaluate(
&|states: &Tensor| -> Result<Tensor, ml::MLError> {
dqn.q_values_for_batch(states)
.map_err(|e| ml::MLError::ModelError(format!("{e}")))
},
3, // portfolio_dim
&device,
)
.with_context(|| format!("GpuBacktestEvaluator::evaluate failed for fold {}", fold))?;
Ok(metrics)
}
// ---------------------------------------------------------------------------
// PPO Evaluation
// ---------------------------------------------------------------------------
@@ -1061,60 +1267,147 @@ fn main() -> Result<()> {
// Evaluate DQN
if eval_dqn {
let hp = load_hyperopt_params(&args.hyperopt_params, "dqn");
match evaluate_dqn_fold(
window.fold,
&test_norm,
test_bars_aligned,
&args.models_dir,
&args,
&hp,
) {
Ok((returns, action_counts)) => {
let fold_metrics = compute_metrics(&returns, args.bars_per_year);
let fold_str = window.fold.to_string();
tm::set_epoch("dqn", &fold_str, window.fold as f64);
tm::set_eval_metrics(
"dqn",
&fold_str,
fold_metrics.win_rate_pct / 100.0,
fold_metrics.sharpe_ratio,
fold_metrics.profit_factor,
fold_metrics.total_return_pct / 100.0,
);
info!(
" [DQN] Fold {} - Sharpe(bar)={:.4} Sharpe(trade)={:.4} MaxDD={:.2}% WR={:.1}% PF={:.2} Return={:.4}% Trades={}",
window.fold,
fold_metrics.sharpe_ratio,
fold_metrics.trade_sharpe_ratio,
fold_metrics.max_drawdown_pct,
fold_metrics.win_rate_pct,
fold_metrics.profit_factor,
fold_metrics.total_return_pct,
fold_metrics.num_trades,
);
info!(
" [DQN] Actions - BUY={} SELL={} HOLD={}",
action_counts.first().copied().unwrap_or(0),
action_counts.get(1).copied().unwrap_or(0),
action_counts.get(2).copied().unwrap_or(0),
);
all_fold_metrics.push(FoldMetrics {
fold: window.fold,
model: "dqn".to_owned(),
sharpe_ratio: fold_metrics.sharpe_ratio,
trade_sharpe_ratio: fold_metrics.trade_sharpe_ratio,
max_drawdown_pct: fold_metrics.max_drawdown_pct,
win_rate_pct: fold_metrics.win_rate_pct,
profit_factor: fold_metrics.profit_factor,
total_return_pct: fold_metrics.total_return_pct,
num_trades: fold_metrics.num_trades,
test_start: test_start.clone(),
test_end: test_end.clone(),
});
all_action_counts.push(action_counts);
// GPU path: when --gpu-eval is set and CUDA is compiled in, try GPU first.
// The GPU path uses greedy argmax (not softmax), so results differ slightly.
// On any GPU error, fall through to the CPU path below.
#[cfg(feature = "cuda")]
let gpu_handled = if args.gpu_eval {
info!(" [DQN] Attempting GPU-accelerated evaluation (greedy argmax)...");
match evaluate_dqn_fold_gpu(
window.fold,
&test_norm,
test_bars_aligned,
&args.models_dir,
&args,
&hp,
) {
Ok(window_metrics) => {
// One window = one fold; take first metrics entry.
// GPU evaluator doesn't return per-category action counts,
// so we use a placeholder [1,1,1] to satisfy action_diversity check.
if let Some(m) = window_metrics.first() {
let fold_str = window.fold.to_string();
tm::set_epoch("dqn", &fold_str, window.fold as f64);
tm::set_eval_metrics(
"dqn",
&fold_str,
m.win_rate as f64,
m.sharpe as f64,
1.0, // profit_factor not available from GPU path
m.total_pnl as f64,
);
info!(
" [DQN GPU] Fold {} - Sharpe={:.4} TotalPnL={:.4} MaxDD={:.4} Sortino={:.4} WinRate={:.2}% Trades={} VaR95={:.4} CVaR95={:.4} Calmar={:.4} Omega={:.4}",
window.fold,
m.sharpe,
m.total_pnl,
m.max_drawdown,
m.sortino,
m.win_rate * 100.0,
m.total_trades,
m.var_95,
m.cvar_95,
m.calmar,
m.omega_ratio,
);
all_fold_metrics.push(FoldMetrics {
fold: window.fold,
model: "dqn".to_owned(),
sharpe_ratio: m.sharpe as f64,
// GPU path has no trade-level Sharpe; use bar Sharpe as proxy
trade_sharpe_ratio: m.sharpe as f64,
max_drawdown_pct: m.max_drawdown as f64 * 100.0,
win_rate_pct: m.win_rate as f64 * 100.0,
// profit_factor not available from GPU metrics kernel
profit_factor: 0.0,
total_return_pct: m.total_pnl as f64 * 100.0,
num_trades: m.total_trades as usize,
test_start: test_start.clone(),
test_end: test_end.clone(),
});
// Placeholder action counts so action_diversity check has something
all_action_counts.push([1, 1, 1]);
} else {
warn!(
" [DQN GPU] Fold {} - no window metrics returned",
window.fold
);
}
true // GPU path handled this fold
}
Err(e) => {
warn!(
" [DQN GPU] Fold {} GPU eval failed: {}. Falling back to CPU path.",
window.fold, e
);
false // fall through to CPU path
}
}
Err(e) => {
error!(" [DQN] Fold {} evaluation failed: {}", window.fold, e);
} else {
false // --gpu-eval not set; use CPU path
};
// When compiled without CUDA, gpu_handled is always false.
#[cfg(not(feature = "cuda"))]
let gpu_handled = false;
if !gpu_handled {
match evaluate_dqn_fold(
window.fold,
&test_norm,
test_bars_aligned,
&args.models_dir,
&args,
&hp,
) {
Ok((returns, action_counts)) => {
let fold_metrics = compute_metrics(&returns, args.bars_per_year);
let fold_str = window.fold.to_string();
tm::set_epoch("dqn", &fold_str, window.fold as f64);
tm::set_eval_metrics(
"dqn",
&fold_str,
fold_metrics.win_rate_pct / 100.0,
fold_metrics.sharpe_ratio,
fold_metrics.profit_factor,
fold_metrics.total_return_pct / 100.0,
);
info!(
" [DQN] Fold {} - Sharpe(bar)={:.4} Sharpe(trade)={:.4} MaxDD={:.2}% WR={:.1}% PF={:.2} Return={:.4}% Trades={}",
window.fold,
fold_metrics.sharpe_ratio,
fold_metrics.trade_sharpe_ratio,
fold_metrics.max_drawdown_pct,
fold_metrics.win_rate_pct,
fold_metrics.profit_factor,
fold_metrics.total_return_pct,
fold_metrics.num_trades,
);
info!(
" [DQN] Actions - BUY={} SELL={} HOLD={}",
action_counts.first().copied().unwrap_or(0),
action_counts.get(1).copied().unwrap_or(0),
action_counts.get(2).copied().unwrap_or(0),
);
all_fold_metrics.push(FoldMetrics {
fold: window.fold,
model: "dqn".to_owned(),
sharpe_ratio: fold_metrics.sharpe_ratio,
trade_sharpe_ratio: fold_metrics.trade_sharpe_ratio,
max_drawdown_pct: fold_metrics.max_drawdown_pct,
win_rate_pct: fold_metrics.win_rate_pct,
profit_factor: fold_metrics.profit_factor,
total_return_pct: fold_metrics.total_return_pct,
num_trades: fold_metrics.num_trades,
test_start: test_start.clone(),
test_end: test_end.clone(),
});
all_action_counts.push(action_counts);
}
Err(e) => {
error!(" [DQN] Fold {} evaluation failed: {}", window.fold, e);
}
}
}
}

View File

@@ -0,0 +1,123 @@
// Vectorized backtest environment step kernel.
// One thread per walk-forward window. Each thread steps sequentially.
//
// Portfolio state layout per window [8 floats]:
// [0] value - current portfolio value
// [1] position - current position size (-1.0 to +1.0)
// [2] cash - cash balance
// [3] entry_price - entry price of current position (0 if flat)
// [4] max_equity - peak equity for drawdown tracking
// [5] step_pnl - PnL this step (for reward)
// [6] cum_return - cumulative log return
// [7] step_count - number of completed steps
#define PORTFOLIO_STATE_SIZE 8
extern "C" __global__ void backtest_env_step(
// Market data (read-only, uploaded once)
const float* __restrict__ prices, // [n_windows * max_len * 4] (OHLC)
const int* __restrict__ window_lens, // [n_windows]
// Actions from model for current step
const int* __restrict__ actions, // [n_windows] (0-4: Short100..Long100)
// Portfolio state (read-write, persistent across steps)
float* portfolio_state, // [n_windows * PORTFOLIO_STATE_SIZE]
// Step outputs
float* step_rewards, // [n_windows]
float* step_returns, // [n_windows * max_len] (accumulated)
int* done_flags, // [n_windows]
// Config
int n_windows,
int max_len,
float max_position,
float tx_cost_bps,
float spread_cost,
int current_step
) {
int w = blockIdx.x * blockDim.x + threadIdx.x;
if (w >= n_windows) return;
if (done_flags[w]) return;
int wlen = window_lens[w];
if (current_step >= wlen) {
done_flags[w] = 1;
return;
}
// Read current prices
int price_base = (w * max_len + current_step) * 4;
float open = prices[price_base + 0];
float high = prices[price_base + 1];
float low = prices[price_base + 2];
float close = prices[price_base + 3];
// Read portfolio state
int ps = w * PORTFOLIO_STATE_SIZE;
float value = portfolio_state[ps + 0];
float position = portfolio_state[ps + 1];
float cash = portfolio_state[ps + 2];
float entry_price = portfolio_state[ps + 3];
float max_equity = portfolio_state[ps + 4];
float cum_return = portfolio_state[ps + 6];
// Map action (0-4) to target exposure
float target_exposure;
switch (actions[w]) {
case 0: target_exposure = -1.0f; break; // Short100
case 1: target_exposure = -0.5f; break; // Short50
case 2: target_exposure = 0.0f; break; // Flat
case 3: target_exposure = 0.5f; break; // Long50
case 4: target_exposure = 1.0f; break; // Long100
default: target_exposure = 0.0f; break;
}
target_exposure *= max_position;
// Execute trade if position changes
float delta = target_exposure - position;
float trade_cost = 0.0f;
if (fabsf(delta) > 0.001f && close > 0.0f) {
trade_cost = fabsf(delta) * close * tx_cost_bps * 0.0001f
+ fabsf(delta) * spread_cost * 0.5f;
cash -= trade_cost;
// Mark-to-market old position
if (fabsf(position) > 0.001f && entry_price > 0.0f) {
float pnl = position * (close - entry_price);
cash += pnl;
}
position = target_exposure;
entry_price = close;
}
// Mark-to-market current position
float unrealized = 0.0f;
if (fabsf(position) > 0.001f && entry_price > 0.0f) {
unrealized = position * (close - entry_price);
}
float new_value = cash + unrealized;
// Step return
float step_ret = (value > 0.0f) ? (new_value - value) / value : 0.0f;
float new_cum_return = cum_return + step_ret;
// Update max equity for drawdown
float new_max = fmaxf(max_equity, new_value);
// Write portfolio state
portfolio_state[ps + 0] = new_value;
portfolio_state[ps + 1] = position;
portfolio_state[ps + 2] = cash;
portfolio_state[ps + 3] = entry_price;
portfolio_state[ps + 4] = new_max;
portfolio_state[ps + 5] = step_ret; // step PnL (for reward)
portfolio_state[ps + 6] = new_cum_return;
portfolio_state[ps + 7] += 1.0f; // step count
// Outputs
step_rewards[w] = step_ret;
step_returns[w * max_len + current_step] = step_ret;
}

View File

@@ -0,0 +1,63 @@
// Gather state vectors from pre-uploaded features + live portfolio state.
// Output: [n_windows, state_dim] tensor for model forward pass.
//
// This kernel eliminates the CPU roundtrip in the old gather_states() path,
// which previously downloaded the full features buffer (n_windows * max_len * feat_dim
// floats) to CPU just to slice out a single step's row per window.
//
// Portfolio state layout per window [8 floats]:
// [0] value - current portfolio value
// [1] position - current position size (-1.0 to +1.0)
// [2] cash - cash balance
// [3] entry_price - entry price of current position (0 if flat)
// [4] max_equity - peak equity for drawdown tracking
// [5] step_pnl - PnL this step (for reward)
// [6] cum_return - cumulative log return
// [7] step_count - number of completed steps
//
// State output layout per window [state_dim floats]:
// [0 .. feat_dim) - market features at current step
// [feat_dim + 0] - normalised portfolio value (value / initial_capital)
// [feat_dim + 1] - position (-1.0 to +1.0)
// [feat_dim + 2] - spread cost (static config constant)
// [feat_dim + 3 .. state_dim) - zero-padded for tensor core alignment
extern "C" __global__ void gather_states(
const float* __restrict__ features, // [n_windows * max_len * feat_dim]
const float* __restrict__ portfolio, // [n_windows * 8]
float* states_out, // [n_windows * state_dim]
int n_windows,
int max_len,
int feat_dim,
int state_dim,
int current_step,
float initial_capital,
float spread_cost
) {
int w = blockIdx.x * blockDim.x + threadIdx.x;
if (w >= n_windows) return;
int feat_base = (w * max_len + current_step) * feat_dim;
int out_base = w * state_dim;
int ps = w * 8;
// Copy market features for the current step
for (int i = 0; i < feat_dim; i++) {
states_out[out_base + i] = features[feat_base + i];
}
// Append portfolio features
float value = portfolio[ps + 0];
float position = portfolio[ps + 1];
states_out[out_base + feat_dim + 0] = (initial_capital > 0.0f)
? value / initial_capital
: 0.0f;
states_out[out_base + feat_dim + 1] = position;
states_out[out_base + feat_dim + 2] = spread_cost;
// Zero-pad remainder for tensor core alignment
for (int i = feat_dim + 3; i < state_dim; i++) {
states_out[out_base + i] = 0.0f;
}
}

View File

@@ -0,0 +1,204 @@
// Per-window metrics reduction kernel.
// One block per window. Threads cooperate to reduce step_returns.
//
// Output per window [10 floats]:
// [0] sharpe_ratio (annualized, sqrt(252))
// [1] total_pnl (cumulative return)
// [2] max_drawdown (worst peak-to-trough, positive number)
// [3] sortino_ratio
// [4] win_rate
// [5] total_trades (approximated from position changes)
// [6] var_95 (5th-percentile return — Value at Risk at 95% confidence)
// [7] cvar_95 (mean of returns below VaR — Expected Shortfall)
// [8] calmar_ratio (annualized mean return / max drawdown)
// [9] omega_ratio (sum of gains / sum of losses)
extern "C" __global__ void compute_backtest_metrics(
const float* __restrict__ step_returns, // [n_windows * max_len]
const float* __restrict__ portfolio_state, // [n_windows * 8]
const int* __restrict__ window_lens, // [n_windows]
const int* __restrict__ actions_history, // [n_windows * max_len] for trade counting
float* metrics_out, // [n_windows * 10]
int n_windows,
int max_len,
float annualization_factor // sqrt(252) for daily
) {
int w = blockIdx.x;
if (w >= n_windows) return;
int wlen = window_lens[w];
int tid = threadIdx.x;
int stride = blockDim.x;
int base = w * max_len;
// Shared memory layout:
// [0 .. stride) : s_sum (6 reduction arrays of size stride)
// [stride .. 2*stride) : s_sq_sum
// [2*stride..3*stride) : s_down_sq
// [3*stride..4*stride) : s_max_dd
// [4*stride..5*stride) : s_wins
// [5*stride..6*stride) : s_trades
// [6*stride .. 6*stride + 4096) : s_sorted (bitonic sort scratch, up to 4096 returns)
extern __shared__ float shmem[];
float* s_sum = shmem; // [blockDim.x]
float* s_sq_sum = shmem + stride; // [blockDim.x]
float* s_down_sq = shmem + 2*stride; // [blockDim.x] (downside deviation)
float* s_max_dd = shmem + 3*stride; // [blockDim.x] (max drawdown)
// wins and trades stored as float for reduction compatibility
float* s_wins = shmem + 4*stride; // [blockDim.x]
float* s_trades = shmem + 5*stride; // [blockDim.x]
float* s_sorted = shmem + 6*stride; // [4096] for bitonic sort
// Pass 1: per-thread local accumulators
float local_sum = 0.0f, local_sq = 0.0f, local_down = 0.0f;
float local_cum = 0.0f, local_peak = 0.0f, local_max_dd = 0.0f;
int local_wins = 0, local_trades = 0;
int prev_action = -1;
for (int i = tid; i < wlen; i += stride) {
float r = step_returns[base + i];
local_sum += r;
local_sq += r * r;
if (r < 0.0f) local_down += r * r;
// Drawdown tracking (NOTE: strided — approximate per thread,
// then take max across threads for worst-case estimate)
local_cum += r;
local_peak = fmaxf(local_peak, local_cum);
float dd = local_peak - local_cum;
local_max_dd = fmaxf(local_max_dd, dd);
// Win/loss counting
if (r > 0.0f) local_wins++;
// Trade counting (position changes)
int act = actions_history[base + i];
if (act != prev_action && i > 0) local_trades++;
prev_action = act;
}
// Store ALL local values to shared memory
s_sum[tid] = local_sum;
s_sq_sum[tid] = local_sq;
s_down_sq[tid] = local_down;
s_max_dd[tid] = local_max_dd;
s_wins[tid] = (float)local_wins;
s_trades[tid] = (float)local_trades;
__syncthreads();
// Block-level parallel reduction for ALL 6 arrays
for (int s = stride / 2; s > 0; s >>= 1) {
if (tid < s) {
s_sum[tid] += s_sum[tid + s];
s_sq_sum[tid] += s_sq_sum[tid + s];
s_down_sq[tid] += s_down_sq[tid + s];
s_max_dd[tid] = fmaxf(s_max_dd[tid], s_max_dd[tid + s]); // max reduction
s_wins[tid] += s_wins[tid + s];
s_trades[tid] += s_trades[tid + s];
}
__syncthreads();
}
// Thread 0 computes final metrics from fully reduced values
if (tid == 0) {
float n = (float)wlen;
float mean = s_sum[0] / n;
float var = s_sq_sum[0] / n - mean * mean;
float std = sqrtf(fmaxf(var, 1e-10f));
float down_std = sqrtf(fmaxf(s_down_sq[0] / n, 1e-10f));
int out_base = w * 10;
metrics_out[out_base + 0] = (mean / std) * annualization_factor; // Sharpe
metrics_out[out_base + 1] = s_sum[0]; // total cumulative return
metrics_out[out_base + 2] = s_max_dd[0]; // max drawdown (reduced across all threads)
metrics_out[out_base + 3] = (mean / down_std) * annualization_factor; // Sortino
metrics_out[out_base + 4] = (n > 0.0f) ? s_wins[0] / n : 0.0f; // win rate (reduced)
metrics_out[out_base + 5] = s_trades[0]; // trade count (reduced)
}
// ── Extended metrics: VaR, CVaR, Calmar, Omega via bitonic sort ──────────
//
// All threads cooperate to load and sort up to 4096 step_returns for this
// window into s_sorted (ascending order). Thread 0 then scans the sorted
// array to derive the tail-risk metrics.
int sort_len = wlen < 4096 ? wlen : 4096;
// Load returns into sort scratch (stride-strided load)
for (int i = tid; i < sort_len; i += stride) {
s_sorted[i] = step_returns[base + i];
}
// Compute next power-of-two for bitonic sort
int padded_len = 1;
while (padded_len < sort_len) padded_len <<= 1;
// Pad with +inf so sentinel values sort to the end (ascending)
for (int i = sort_len + tid; i < padded_len; i += stride) {
s_sorted[i] = 1e30f;
}
__syncthreads();
// Bitonic sort — ascending
for (int k = 2; k <= padded_len; k <<= 1) {
for (int j = k >> 1; j > 0; j >>= 1) {
for (int i = tid; i < padded_len; i += stride) {
int ixj = i ^ j;
if (ixj > i) {
bool ascending = ((i & k) == 0);
if ((ascending && s_sorted[i] > s_sorted[ixj]) ||
(!ascending && s_sorted[i] < s_sorted[ixj])) {
float tmp = s_sorted[i];
s_sorted[i] = s_sorted[ixj];
s_sorted[ixj] = tmp;
}
}
}
__syncthreads();
}
}
// Thread 0 computes extended metrics from the sorted array
if (tid == 0) {
int out_base = w * 10;
// VaR at 95% confidence = 5th-percentile of sorted returns
int var_idx = (int)(0.05f * (float)sort_len);
if (var_idx < 0) var_idx = 0;
if (var_idx >= sort_len) var_idx = sort_len - 1;
float var_95 = s_sorted[var_idx];
// CVaR (Expected Shortfall): mean of returns strictly below VaR index
int cvar_count = (var_idx > 0) ? var_idx : 1;
float cvar_sum = 0.0f;
for (int i = 0; i < cvar_count; i++) {
cvar_sum += s_sorted[i];
}
float cvar_95 = cvar_sum / (float)cvar_count;
// Calmar ratio: annualised mean return / max drawdown
// Compute raw mean directly from sorted array (avoids reversing annualisation).
float total_return = 0.0f;
for (int i = 0; i < sort_len; i++) {
total_return += s_sorted[i];
}
float daily_mean = total_return / (float)sort_len;
float max_dd = metrics_out[out_base + 2];
float calmar = (max_dd > 1e-8f)
? (daily_mean * annualization_factor * annualization_factor) / max_dd
: 0.0f;
// Omega ratio: sum of positive returns / sum of |negative returns|
float gain_sum = 0.0f, loss_sum = 0.0f;
for (int i = 0; i < sort_len; i++) {
if (s_sorted[i] > 0.0f) gain_sum += s_sorted[i];
else loss_sum -= s_sorted[i];
}
float omega = (loss_sum > 1e-10f) ? gain_sum / loss_sum : 0.0f;
metrics_out[out_base + 6] = var_95;
metrics_out[out_base + 7] = cvar_95;
metrics_out[out_base + 8] = calmar;
metrics_out[out_base + 9] = omega;
}
}

View File

@@ -1192,7 +1192,11 @@ extern "C" __global__ void dqn_full_experience_kernel(
float* out_rewards, /* [N, L] */
int* out_done, /* [N, L] */
float* out_target_q, /* [N, L] */
float* out_td_error /* [N, L] */
float* out_td_error, /* [N, L] */
/* ---- Epoch-persistent state [8 floats, global memory] ---- */
float* epoch_state, /* [8]: vol_ema, median_vol, port_value, port_pos, port_cash, dsr_mean, dsr_var, step_count */
int reset_flags /* bitfield: bit0=reset portfolio, bit1=reset DSR, bit2=reset vol EMA */
) {
/* ---- Unpack weight pointers from struct into local __restrict__ aliases ---- */
UNPACK_WEIGHT_PTRS(wp);
@@ -1202,6 +1206,40 @@ extern "C" __global__ void dqn_full_experience_kernel(
float* shmem_weights = shmem;
float* shmem_bias = shmem + SHMEM_TILE_ROWS * SHMEM_MAX_IN_DIM;
/* ---- Epoch state: apply resets then read ---- */
// Thread 0 of block 0 applies reset flags to global memory.
// Other blocks/threads wait for the threadfence.
if (threadIdx.x == 0 && blockIdx.x == 0) {
if (reset_flags & 1) { // reset portfolio
epoch_state[2] = epoch_state[4]; // portfolio_value = portfolio_cash (initial_capital)
epoch_state[3] = 0.0f; // portfolio_position = 0
}
if (reset_flags & 2) { // reset DSR
epoch_state[5] = 0.0f; // dsr_mean
epoch_state[6] = 1.0f; // dsr_var
}
if (reset_flags & 4) { // reset vol EMA
epoch_state[0] = 0.01f; // vol_ema
epoch_state[1] = 0.01f; // median_vol
}
__threadfence(); // Ensure all blocks see updated epoch_state
// NOTE: Known benign race — __threadfence guarantees ordering for the
// issuing thread, not a grid-wide barrier. Other blocks may read stale
// epoch_state if they execute before block 0's writes complete.
// This is harmless: reset_flags is only set between epochs (not mid-training),
// and accumulators re-converge within a few steps.
// Clean fix (if needed): split into a separate 1-thread pre-kernel.
}
__syncthreads(); // Ensure all threads in this block wait for thread 0
// All threads read epoch state from global memory (8 floats, L1-cached)
float epoch_vol_ema = epoch_state[0];
float epoch_median_vol = epoch_state[1];
// epoch_state[2..4] reserved for portfolio epoch persistence (unused today)
float epoch_dsr_mean = epoch_state[5];
float epoch_dsr_var = epoch_state[6];
float epoch_step_count = epoch_state[7];
int tid = blockIdx.x * blockDim.x + threadIdx.x;
/* All threads must participate in cooperative_load_tile / __syncthreads
* even if they don't own an episode. Use tid_valid to gate data work. */
@@ -1269,15 +1307,19 @@ extern "C" __global__ void dqn_full_experience_kernel(
int total_action_count = 0;
for (int i = 0; i < NUM_ACTIONS; i++) action_counts[i] = 0;
/* ---- EMA reward normalizer (matches CPU RewardNormalizer) ---- */
float ema_mean = 0.0f;
float ema_var = 1.0f;
int ema_init = 0;
/* ---- EMA reward normalizer — seeded from epoch state for cross-epoch continuity ---- */
float ema_mean = epoch_dsr_mean;
float ema_var = epoch_dsr_var;
int ema_init = (epoch_step_count > 0.0f) ? 1 : 0;
/* ---- DSR accumulators (per-episode) ---- */
float dsr_A = 0.0f;
float dsr_B = 1e-8f;
int dsr_initialized = 0;
/* ---- DSR accumulators (per-episode) — seeded from epoch state ---- */
float dsr_A = epoch_dsr_mean;
float dsr_B = epoch_dsr_var;
int dsr_initialized = (epoch_step_count > 0.0f) ? 1 : 0;
/* ---- Vol EMA (epoch-persistent) — seeded from epoch state ---- */
float local_vol_ema = epoch_vol_ema;
float local_median_vol = epoch_median_vol;
/* ---- N-step ring buffer (per-episode) ---- */
float nstep_ring[N_STEPS_MAX];
@@ -1310,6 +1352,14 @@ extern "C" __global__ void dqn_full_experience_kernel(
int mf_off = global_bar * MARKET_DIM;
for (int i = 0; i < MARKET_DIM; i++)
state[i] = market_features[mf_off + i];
/* Update vol EMA from market feature index 3 (log close return).
* Mirrors CPU DQNTrainer::vol_ema / median_vol update logic. */
float vol_sample = fabsf(state[3]);
if (vol_sample > 0.0f && vol_sample < 1.0f) {
local_vol_ema = 0.99f * local_vol_ema + 0.01f * vol_sample;
local_median_vol = 0.999f * local_median_vol + 0.001f * vol_sample;
}
}
/* ---- Step 2: Compute portfolio features ---- */
@@ -1862,15 +1912,15 @@ extern "C" __global__ void dqn_full_experience_kernel(
/* D1: Reset count-bonus state */
for (int i = 0; i < NUM_ACTIONS; i++) action_counts[i] = 0;
total_action_count = 0;
/* Reset EMA normalizer for new episode */
ema_mean = 0.0f;
ema_var = 1.0f;
ema_init = 0;
/* Reset EMA normalizer — warm-start from epoch state */
ema_mean = epoch_dsr_mean;
ema_var = epoch_dsr_var;
ema_init = (epoch_step_count > 0.0f) ? 1 : 0;
/* Reset DSR accumulators */
dsr_A = 0.0f;
dsr_B = 1e-8f;
dsr_initialized = 0;
/* Reset DSR accumulators — warm-start from epoch state */
dsr_A = epoch_dsr_mean;
dsr_B = epoch_dsr_var;
dsr_initialized = (epoch_step_count > 0.0f) ? 1 : 0;
/* Reset n-step ring */
nstep_reset(nstep_ring, &nstep_ring_idx, &nstep_ring_len, effective_n);
@@ -1900,6 +1950,26 @@ extern "C" __global__ void dqn_full_experience_kernel(
rng_states[tid] = rng;
}
/* ---- Epoch state writeback: last block, thread 0 ---- */
// Only vol_ema, dsr_mean, dsr_var, step_count are updated per-kernel.
// Portfolio state is per-episode (in portfolio_states), not per-epoch.
// NOTE: Picks the episode processed by last block's thread 0 as the
// representative seed for the next epoch. Not a cross-episode reduction.
// This is acceptable: all episodes see the same market data in temporal
// order, so accumulators converge to similar values across episodes.
if (threadIdx.x == 0 && blockIdx.x == gridDim.x - 1) {
epoch_state[0] = local_vol_ema;
epoch_state[1] = local_median_vol;
if (use_dsr) {
epoch_state[5] = dsr_A;
epoch_state[6] = dsr_B;
} else {
epoch_state[5] = ema_mean;
epoch_state[6] = ema_var;
}
epoch_state[7] = epoch_step_count + (float)L;
}
}
/* ==================================================================== */
@@ -2011,7 +2081,11 @@ extern "C" __global__ void dqn_full_experience_kernel_warp(
float* out_rewards, /* [N, L] */
int* out_done, /* [N, L] */
float* out_target_q, /* [N, L] */
float* out_td_error /* [N, L] */
float* out_td_error, /* [N, L] */
/* ---- Epoch-persistent state [8 floats, global memory] ---- */
float* epoch_state, /* [8]: vol_ema, median_vol, port_value, port_pos, port_cash, dsr_mean, dsr_var, step_count */
int reset_flags /* bitfield: bit0=reset portfolio, bit1=reset DSR, bit2=reset vol EMA */
) {
/* ---- Unpack weight pointers from struct into local __restrict__ aliases ---- */
UNPACK_WEIGHT_PTRS(wp);
@@ -2021,6 +2095,33 @@ extern "C" __global__ void dqn_full_experience_kernel_warp(
float* shmem_weights = shmem;
float* shmem_bias = shmem + SHMEM_TILE_ROWS * SHMEM_MAX_IN_DIM;
/* ---- Epoch state: apply resets then read ---- */
// Lane 0 of block 0 (episode 0) applies reset flags.
if (threadIdx.x == 0 && blockIdx.x == 0) {
if (reset_flags & 1) { // reset portfolio
epoch_state[2] = epoch_state[4]; // portfolio_value = portfolio_cash (initial_capital)
epoch_state[3] = 0.0f; // portfolio_position = 0
}
if (reset_flags & 2) { // reset DSR
epoch_state[5] = 0.0f; // dsr_mean
epoch_state[6] = 1.0f; // dsr_var
}
if (reset_flags & 4) { // reset vol EMA
epoch_state[0] = 0.01f; // vol_ema
epoch_state[1] = 0.01f; // median_vol
}
__threadfence(); // Ensure all blocks see updated epoch_state
}
__syncthreads(); // Ensure all lanes in this block wait for lane 0
// All lanes read epoch state from global memory (8 floats, L1-cached)
float epoch_vol_ema = epoch_state[0];
float epoch_median_vol = epoch_state[1];
// epoch_state[2..4] reserved for portfolio epoch persistence (unused today)
float epoch_dsr_mean = epoch_state[5];
float epoch_dsr_var = epoch_state[6];
float epoch_step_count = epoch_state[7];
/* ---- Thread / warp identity ---- */
int episode_id = blockIdx.x; /* one block per episode */
int lane_id = threadIdx.x; /* 0..31 within the warp */
@@ -2081,15 +2182,19 @@ extern "C" __global__ void dqn_full_experience_kernel_warp(
int total_action_count = 0;
for (int i = 0; i < NUM_ACTIONS; i++) action_counts[i] = 0;
/* ---- EMA reward normalizer (lane 0 only) ---- */
float ema_mean = 0.0f;
float ema_var = 1.0f;
int ema_init = 0;
/* ---- EMA reward normalizer (lane 0 only) — seeded from epoch state ---- */
float ema_mean = epoch_dsr_mean;
float ema_var = epoch_dsr_var;
int ema_init = (epoch_step_count > 0.0f) ? 1 : 0;
/* ---- DSR accumulators (per-episode, lane 0 only) ---- */
float dsr_A = 0.0f;
float dsr_B = 1e-8f;
int dsr_initialized = 0;
/* ---- DSR accumulators (per-episode, lane 0 only) — seeded from epoch state ---- */
float dsr_A = epoch_dsr_mean;
float dsr_B = epoch_dsr_var;
int dsr_initialized = (epoch_step_count > 0.0f) ? 1 : 0;
/* ---- Vol EMA (epoch-persistent, lane 0 only) — seeded from epoch state ---- */
float local_vol_ema = epoch_vol_ema;
float local_median_vol = epoch_median_vol;
/* ---- N-step ring buffer (per-episode, lane 0 only) ---- */
float nstep_ring[N_STEPS_MAX];
@@ -2133,6 +2238,16 @@ extern "C" __global__ void dqn_full_experience_kernel_warp(
state_dist[i / 32] = market_features[mf_off + i];
}
/* Update vol EMA from market feature index 3 (log close return), lane 0 only.
* Mirrors CPU DQNTrainer::vol_ema / median_vol update logic. */
if (lane_id == 0) {
float vol_sample = fabsf(market_features[mf_off + 3]);
if (vol_sample > 0.0f && vol_sample < 1.0f) {
local_vol_ema = 0.99f * local_vol_ema + 0.01f * vol_sample;
local_median_vol = 0.999f * local_median_vol + 0.001f * vol_sample;
}
}
/* ---- Step 2: Compute portfolio features (all lanes read, needed for state) ---- */
int t_off = global_bar * 4;
current_close = targets[t_off + 0];
@@ -2745,14 +2860,14 @@ extern "C" __global__ void dqn_full_experience_kernel_warp(
div_meta[1] = 0;
for (int i = 0; i < NUM_ACTIONS; i++) action_counts[i] = 0;
total_action_count = 0;
ema_mean = 0.0f;
ema_var = 1.0f;
ema_init = 0;
ema_mean = epoch_dsr_mean;
ema_var = epoch_dsr_var;
ema_init = (epoch_step_count > 0.0f) ? 1 : 0;
/* Reset DSR accumulators */
dsr_A = 0.0f;
dsr_B = 1e-8f;
dsr_initialized = 0;
/* Reset DSR accumulators — warm-start from epoch state */
dsr_A = epoch_dsr_mean;
dsr_B = epoch_dsr_var;
dsr_initialized = (epoch_step_count > 0.0f) ? 1 : 0;
/* Reset n-step ring */
nstep_reset(nstep_ring, &nstep_ring_idx, &nstep_ring_len, effective_n);
@@ -2795,4 +2910,24 @@ extern "C" __global__ void dqn_full_experience_kernel_warp(
rng_states[episode_id] = rng;
}
/* ---- Epoch state writeback: last block, lane 0 ---- */
// Only vol_ema, dsr_mean, dsr_var, step_count are updated per-kernel.
// Portfolio state is per-episode (in portfolio_states), not per-epoch.
// NOTE: Picks the episode processed by last block's thread 0 as the
// representative seed for the next epoch. Not a cross-episode reduction.
// This is acceptable: all episodes see the same market data in temporal
// order, so accumulators converge to similar values across episodes.
if (threadIdx.x == 0 && blockIdx.x == gridDim.x - 1) {
epoch_state[0] = local_vol_ema;
epoch_state[1] = local_median_vol;
if (use_dsr) {
epoch_state[5] = dsr_A;
epoch_state[6] = dsr_B;
} else {
epoch_state[5] = ema_mean;
epoch_state[6] = ema_var;
}
epoch_state[7] = epoch_step_count + (float)L;
}
}

View File

@@ -0,0 +1,778 @@
#![allow(unsafe_code)]
//! Vectorized GPU backtest evaluator.
//!
//! Runs walk-forward evaluation entirely on GPU:
//! 1. Upload test window data once (prices + features)
//! 2. Step loop: GPU gather kernel → Candle forward → env kernel
//! 3. Metrics reduction kernel → single readback
//!
//! The only GPU→CPU transfer is the final metrics readback (n_windows × 10 floats).
//! Per-step state construction uses a zero-copy DtoD path (gather kernel → Candle tensor).
use std::sync::Arc;
use candle_core::cuda_backend::cudarc;
use candle_core::{DType, Device, Tensor};
use cudarc::driver::{CudaFunction, CudaSlice, CudaStream, DevicePtr, LaunchConfig, PushKernelArg};
use cudarc::nvrtc::Ptx;
use std::sync::OnceLock;
use tracing::info;
use crate::MLError;
// ── PTX caches ────────────────────────────────────────────────────────────────
static ENV_PTX: OnceLock<Result<Ptx, String>> = OnceLock::new();
static METRICS_PTX: OnceLock<Result<Ptx, String>> = OnceLock::new();
static GATHER_PTX: OnceLock<Result<Ptx, String>> = OnceLock::new();
fn compile_env_ptx() -> Result<Ptx, String> {
let src = include_str!("backtest_env_kernel.cu");
cudarc::nvrtc::compile_ptx(src)
.map_err(|e| format!("backtest_env_kernel CUDA compilation failed: {e}"))
}
fn compile_metrics_ptx() -> Result<Ptx, String> {
let src = include_str!("backtest_metrics_kernel.cu");
cudarc::nvrtc::compile_ptx(src)
.map_err(|e| format!("backtest_metrics_kernel CUDA compilation failed: {e}"))
}
fn compile_gather_ptx() -> Result<Ptx, String> {
let src = include_str!("backtest_gather_kernel.cu");
cudarc::nvrtc::compile_ptx(src)
.map_err(|e| format!("backtest_gather_kernel CUDA compilation failed: {e}"))
}
// ── Public types ──────────────────────────────────────────────────────────────
/// Per-window evaluation result returned after a full backtest run.
#[derive(Debug, Clone)]
pub struct WindowMetrics {
pub sharpe: f32,
pub total_pnl: f32,
pub max_drawdown: f32,
pub sortino: f32,
pub win_rate: f32,
pub total_trades: f32,
// Phase 3 extended metrics
pub var_95: f32,
pub cvar_95: f32,
pub calmar: f32,
pub omega_ratio: f32,
}
/// Configuration for the GPU backtest evaluator.
#[derive(Debug, Clone)]
pub struct GpuBacktestConfig {
pub max_position: f32,
pub tx_cost_bps: f32,
pub spread_cost: f32,
pub initial_capital: f32,
}
impl Default for GpuBacktestConfig {
fn default() -> Self {
Self {
max_position: 1.0,
tx_cost_bps: 0.1,
spread_cost: 0.0001,
initial_capital: 100_000.0,
}
}
}
// ── Main struct ───────────────────────────────────────────────────────────────
/// GPU backtest evaluator — runs walk-forward evaluation without CPU roundtrips.
///
/// Upload window data once via `new()`, then call `evaluate()` with a model
/// forward function. The entire step loop runs on GPU; only the final metrics
/// are downloaded (n_windows × 10 floats).
#[allow(missing_debug_implementations)]
pub struct GpuBacktestEvaluator {
stream: Arc<CudaStream>,
env_kernel: CudaFunction,
metrics_kernel: CudaFunction,
gather_kernel: CudaFunction,
// Uploaded data (read-only; persists across the step loop)
prices_buf: CudaSlice<f32>, // [n_windows * max_len * 4]
features_buf: CudaSlice<f32>, // [n_windows * max_len * feat_dim]
window_lens_buf: CudaSlice<i32>, // [n_windows]
// Mutable state (written by kernels each step)
portfolio_buf: CudaSlice<f32>, // [n_windows * 8]
step_rewards_buf: CudaSlice<f32>, // [n_windows]
step_returns_buf: CudaSlice<f32>, // [n_windows * max_len]
done_buf: CudaSlice<i32>, // [n_windows]
actions_buf: CudaSlice<i32>, // [n_windows]
actions_history_buf: CudaSlice<i32>, // [n_windows * max_len]
// Gather kernel output buffer (overwritten every step)
states_buf: CudaSlice<f32>, // [n_windows * (feature_dim + 3)]
// Output buffer (written by metrics kernel)
metrics_buf: CudaSlice<f32>, // [n_windows * 10]
// CPU-side action history accumulated during the step loop.
// Uploaded once before the metrics kernel launch.
actions_history_cpu: Vec<i32>, // [n_windows * max_len]
// Dimensions and config
n_windows: usize,
max_len: usize,
feature_dim: usize,
/// Portfolio feature dimension used during construction (always 3).
portfolio_dim: usize,
config: GpuBacktestConfig,
}
impl GpuBacktestEvaluator {
// ── Constructor ───────────────────────────────────────────────────────────
/// Create evaluator and upload window data to GPU.
///
/// `window_prices`: one `Vec<[f32; 4]>` (OHLC rows) per window.
/// `window_features`: one `Vec<Vec<f32>>` (feature rows) per window.
/// Shorter windows are zero-padded to `max_len`.
pub fn new(
window_prices: &[Vec<[f32; 4]>],
window_features: &[Vec<Vec<f32>>],
feature_dim: usize,
config: GpuBacktestConfig,
device: &Device,
) -> Result<Self, MLError> {
let n_windows = window_prices.len();
if n_windows == 0 {
return Err(MLError::ConfigError("No windows provided".to_owned()));
}
let max_len = window_prices.iter().map(|w| w.len()).max().unwrap_or(0);
if max_len == 0 {
return Err(MLError::ConfigError("All windows are empty".to_owned()));
}
// ── Flatten prices (zero-pad shorter windows) ─────────────────────
let window_lens: Vec<i32> = window_prices.iter().map(|w| w.len() as i32).collect();
let mut flat_prices = vec![0.0_f32; n_windows * max_len * 4];
for (w, prices) in window_prices.iter().enumerate() {
for (t, ohlc) in prices.iter().enumerate() {
let base = (w * max_len + t) * 4;
flat_prices[base..base + 4].copy_from_slice(ohlc);
}
}
// ── Flatten features (zero-pad shorter windows / truncate wider) ──
let mut flat_features = vec![0.0_f32; n_windows * max_len * feature_dim];
for (w, feats) in window_features.iter().enumerate() {
for (t, fv) in feats.iter().enumerate() {
let base = (w * max_len + t) * feature_dim;
let copy_len = fv.len().min(feature_dim);
flat_features[base..base + copy_len].copy_from_slice(&fv[..copy_len]);
}
}
// ── CUDA device + stream ──────────────────────────────────────────
let cuda_dev = match device {
Device::Cuda(d) => d,
Device::Cpu | Device::Metal(_) => {
return Err(MLError::ConfigError(
"GpuBacktestEvaluator requires a CUDA device".to_owned(),
))
}
};
let stream = cuda_dev.cuda_stream();
let context = stream.context();
// ── Compile / cache kernels ───────────────────────────────────────
let env_ptx = ENV_PTX
.get_or_init(compile_env_ptx)
.as_ref()
.map_err(|e| MLError::ModelError(format!("env kernel PTX: {e}")))?;
let metrics_ptx = METRICS_PTX
.get_or_init(compile_metrics_ptx)
.as_ref()
.map_err(|e| MLError::ModelError(format!("metrics kernel PTX: {e}")))?;
let gather_ptx = GATHER_PTX
.get_or_init(compile_gather_ptx)
.as_ref()
.map_err(|e| MLError::ModelError(format!("gather kernel PTX: {e}")))?;
let env_module = context
.load_module(env_ptx.clone())
.map_err(|e| MLError::ModelError(format!("env module load: {e}")))?;
let env_kernel = env_module
.load_function("backtest_env_step")
.map_err(|e| MLError::ModelError(format!("backtest_env_step load: {e}")))?;
let metrics_module = context
.load_module(metrics_ptx.clone())
.map_err(|e| MLError::ModelError(format!("metrics module load: {e}")))?;
let metrics_kernel = metrics_module
.load_function("compute_backtest_metrics")
.map_err(|e| MLError::ModelError(format!("compute_backtest_metrics load: {e}")))?;
let gather_module = context
.load_module(gather_ptx.clone())
.map_err(|e| MLError::ModelError(format!("gather module load: {e}")))?;
let gather_kernel = gather_module
.load_function("gather_states")
.map_err(|e| MLError::ModelError(format!("gather_states load: {e}")))?;
// ── Upload read-only data ─────────────────────────────────────────
let prices_buf = stream
.memcpy_stod(&flat_prices)
.map_err(|e| MLError::ModelError(format!("prices upload: {e}")))?;
let features_buf = stream
.memcpy_stod(&flat_features)
.map_err(|e| MLError::ModelError(format!("features upload: {e}")))?;
let window_lens_buf = stream
.memcpy_stod(&window_lens)
.map_err(|e| MLError::ModelError(format!("window_lens upload: {e}")))?;
// ── Allocate mutable state buffers ────────────────────────────────
let portfolio_init = Self::init_portfolio_state(n_windows, config.initial_capital);
let portfolio_buf = stream
.memcpy_stod(&portfolio_init)
.map_err(|e| MLError::ModelError(format!("portfolio alloc: {e}")))?;
let step_rewards_buf = stream
.alloc_zeros::<f32>(n_windows)
.map_err(|e| MLError::ModelError(format!("step_rewards alloc: {e}")))?;
let step_returns_buf = stream
.alloc_zeros::<f32>(n_windows * max_len)
.map_err(|e| MLError::ModelError(format!("step_returns alloc: {e}")))?;
let done_buf = stream
.alloc_zeros::<i32>(n_windows)
.map_err(|e| MLError::ModelError(format!("done alloc: {e}")))?;
let actions_buf = stream
.alloc_zeros::<i32>(n_windows)
.map_err(|e| MLError::ModelError(format!("actions alloc: {e}")))?;
let actions_history_buf = stream
.alloc_zeros::<i32>(n_windows * max_len)
.map_err(|e| MLError::ModelError(format!("actions_history alloc: {e}")))?;
let metrics_buf = stream
.alloc_zeros::<f32>(n_windows * 10)
.map_err(|e| MLError::ModelError(format!("metrics alloc: {e}")))?;
// Portfolio dimension is always 3: (normalised value, position, spread_cost).
// state_dim = feature_dim + 3, zero-padded to align to tensor cores if needed.
const PORTFOLIO_DIM: usize = 3;
let state_dim = feature_dim + PORTFOLIO_DIM;
let states_buf = stream
.alloc_zeros::<f32>(n_windows * state_dim)
.map_err(|e| MLError::ModelError(format!("states_buf alloc: {e}")))?;
let upload_mb =
((flat_prices.len() + flat_features.len()) * std::mem::size_of::<f32>()) as f64
/ 1_048_576.0;
info!(
"GpuBacktestEvaluator: {} windows × {} max_len × {} features ({:.1} MB uploaded)",
n_windows, max_len, feature_dim, upload_mb,
);
Ok(Self {
stream,
env_kernel,
metrics_kernel,
gather_kernel,
prices_buf,
features_buf,
window_lens_buf,
portfolio_buf,
step_rewards_buf,
step_returns_buf,
done_buf,
actions_buf,
actions_history_buf,
states_buf,
metrics_buf,
actions_history_cpu: vec![0_i32; n_windows * max_len],
n_windows,
max_len,
feature_dim,
portfolio_dim: PORTFOLIO_DIM,
config,
})
}
// ── Private helpers ───────────────────────────────────────────────────────
/// Build the initial portfolio state vector.
///
/// Layout per window (8 floats):
/// `[value, position, cash, unrealised_pnl, max_equity, 0, 0, 0]`
fn init_portfolio_state(n_windows: usize, initial_capital: f32) -> Vec<f32> {
let mut state = vec![0.0_f32; n_windows * 8];
for w in 0..n_windows {
let base = w * 8;
state[base] = initial_capital; // value
// state[base + 1] = 0.0 // position (zero-initialised)
state[base + 2] = initial_capital; // cash
// state[base + 3] = 0.0 // unrealised_pnl
state[base + 4] = initial_capital; // max_equity (high-water mark)
}
state
}
// ── Public API ────────────────────────────────────────────────────────────
/// Build the state tensor for a given step: `[n_windows, feat_dim + portfolio_dim]`.
///
/// Launches the `gather_states` CUDA kernel which reads directly from the
/// pre-uploaded features buffer and the live portfolio state buffer on GPU,
/// then performs a zero-copy DtoD transfer from the kernel output `states_buf`
/// into a freshly-allocated Candle tensor. No data ever touches the CPU.
///
/// The kernel writes into `states_buf` (pre-allocated, `n_windows × state_dim`).
/// A `cuMemcpyDtoDAsync` then copies those bytes directly into the Candle tensor's
/// CUDA storage — eliminating the GPU→CPU→GPU roundtrip entirely.
///
/// # Panics
/// `portfolio_dim` must equal `self.portfolio_dim` (always 3). Callers using a
/// different value should be updated — the kernel signature is fixed.
pub fn gather_states(
&self,
step: usize,
portfolio_dim: usize,
device: &Device,
) -> Result<Tensor, MLError> {
if portfolio_dim != self.portfolio_dim {
return Err(MLError::ConfigError(format!(
"gather_states: portfolio_dim={portfolio_dim} != expected {}",
self.portfolio_dim
)));
}
let state_dim = self.feature_dim + self.portfolio_dim;
// Launch the gather kernel — one thread per window
let grid = ((self.n_windows + 255) / 256) as u32;
let launch_cfg = LaunchConfig {
grid_dim: (grid.max(1), 1, 1),
block_dim: (256, 1, 1),
shared_mem_bytes: 0,
};
let n_windows_i32 = self.n_windows as i32;
let max_len_i32 = self.max_len as i32;
let feat_dim_i32 = self.feature_dim as i32;
let state_dim_i32 = state_dim as i32;
let step_i32 = step as i32;
let initial_capital = self.config.initial_capital;
let spread_cost = self.config.spread_cost;
// Safety: argument order matches `gather_states` signature exactly:
// features, portfolio, states_out, n_windows, max_len, feat_dim,
// state_dim, current_step, initial_capital, spread_cost
unsafe {
self.stream
.launch_builder(&self.gather_kernel)
.arg(&self.features_buf)
.arg(&self.portfolio_buf)
.arg(&self.states_buf)
.arg(&n_windows_i32)
.arg(&max_len_i32)
.arg(&feat_dim_i32)
.arg(&state_dim_i32)
.arg(&step_i32)
.arg(&initial_capital)
.arg(&spread_cost)
.launch(launch_cfg)
.map_err(|e| MLError::ModelError(format!("gather_states launch step {step}: {e}")))?;
}
// Zero-copy: DtoD from kernel output CudaSlice into Candle Tensor — no CPU transfer.
let n_elems = self.n_windows * state_dim;
let tensor = Tensor::zeros(&[self.n_windows, state_dim], DType::F32, device)
.map_err(|e| MLError::ModelError(format!("alloc states tensor step {step}: {e}")))?;
let (storage_guard, _layout) = tensor.storage_and_layout();
match *storage_guard {
candle_core::Storage::Cuda(ref cs) => {
let dst_slice: &CudaSlice<f32> = cs.as_cuda_slice()
.map_err(|e| MLError::ModelError(format!("states as_cuda_slice: {e}")))?;
let (dst_ptr, _dst_sync) = dst_slice.device_ptr(&self.stream);
let src_view = self.states_buf.slice(..n_elems);
let (src_ptr, _src_sync) = src_view.device_ptr(&self.stream);
let num_bytes = n_elems * std::mem::size_of::<f32>();
unsafe {
cudarc::driver::result::memcpy_dtod_async(
dst_ptr, src_ptr, num_bytes, self.stream.cu_stream(),
).map_err(|e| MLError::ModelError(format!("states DtoD copy step {step}: {e}")))?;
}
}
candle_core::Storage::Cpu(_) | candle_core::Storage::Metal(_) => {
return Err(MLError::ModelError(
"gather_states: expected CUDA device".into(),
));
}
}
drop(storage_guard);
Ok(tensor)
}
/// Run the full backtest evaluation loop.
///
/// For each step up to `max_len`:
/// 1. Launch `gather_states` GPU kernel → state tensor `[n_windows, state_dim]`
/// 2. Call `forward_fn` to get Q-values `[n_windows, n_actions]`
/// 3. Greedy argmax → action indices
/// 4. Launch `backtest_env_step` kernel
///
/// After the loop, uploads accumulated action history and launches
/// `compute_backtest_metrics`. Returns one `WindowMetrics` per window.
pub fn evaluate<F>(
&mut self,
forward_fn: &F,
portfolio_dim: usize,
device: &Device,
) -> Result<Vec<WindowMetrics>, MLError>
where
F: Fn(&Tensor) -> Result<Tensor, MLError>,
{
for step in 0..self.max_len {
// 1. Gather state tensor via GPU kernel [n_windows, state_dim]
let states = self.gather_states(step, portfolio_dim, device)?;
// 2. Model forward pass (on-device, no roundtrip)
let q_values = forward_fn(&states)?;
// Ensure f32 for argmax compatibility
let q_f32 = if q_values.dtype() == DType::F32 {
q_values
} else {
q_values
.to_dtype(DType::F32)
.map_err(|e| MLError::ModelError(format!("q_values f32 cast: {e}")))?
};
// 3. Greedy action selection — argmax over action dim (stays on GPU)
let actions_tensor = q_f32
.argmax(1)
.map_err(|e| MLError::ModelError(format!("argmax: {e}")))?;
// DtoD copy: argmax output (U32 Candle tensor) → actions_buf (CudaSlice<i32>).
// Action values are 0..N_ACTIONS (small non-negative), so u32 and i32 share
// identical bit patterns — raw byte reinterpret is safe. Eliminates the old
// to_vec1 → collect::<Vec<i32>> → memcpy_htod upload roundtrip on the hot path.
{
let (act_guard, _act_layout) = actions_tensor.storage_and_layout();
match &*act_guard {
candle_core::Storage::Cuda(ref cs) => {
let src_slice: &CudaSlice<u32> = cs
.as_cuda_slice()
.map_err(|e| MLError::ModelError(format!("actions as_cuda_slice step {step}: {e}")))?;
let src_view = src_slice.slice(..self.n_windows);
let (src_ptr, _src_sync) = src_view.device_ptr(&self.stream);
let (dst_ptr, _dst_sync) = self.actions_buf.device_ptr(&self.stream);
let num_bytes = self.n_windows * std::mem::size_of::<i32>();
unsafe {
cudarc::driver::result::memcpy_dtod_async(
dst_ptr,
src_ptr,
num_bytes,
self.stream.cu_stream(),
)
.map_err(|e| MLError::ModelError(format!("actions DtoD step {step}: {e}")))?;
}
}
candle_core::Storage::Cpu(_) | candle_core::Storage::Metal(_) => {
return Err(MLError::ModelError(format!(
"actions_tensor not on CUDA at step {step}"
)));
}
}
drop(act_guard);
}
// Accumulate actions into CPU-side history (uploaded once before metrics kernel).
// Download is n_windows × 4 bytes — negligible. The old path did the same
// download (to_vec1) PLUS an upload (memcpy_htod); we keep only the download.
let actions_u32: Vec<u32> = actions_tensor
.to_vec1()
.map_err(|e| MLError::ModelError(format!("argmax to_vec1 step {step}: {e}")))?;
for (w, &a) in actions_u32.iter().enumerate() {
self.actions_history_cpu[w * self.max_len + step] = a as i32;
}
// 4. Launch env step kernel — one thread per window
let grid = ((self.n_windows + 255) / 256) as u32;
let env_cfg = LaunchConfig {
grid_dim: (grid.max(1), 1, 1),
block_dim: (256, 1, 1),
shared_mem_bytes: 0,
};
let n_win_i32 = self.n_windows as i32;
let max_len_i32 = self.max_len as i32;
let step_i32 = step as i32;
// Safety: argument order matches `backtest_env_step` signature exactly:
// prices, window_lens, actions, portfolio_state, step_rewards,
// step_returns, done_flags, n_windows, max_len, max_position,
// tx_cost_bps, spread_cost, current_step
unsafe {
self.stream
.launch_builder(&self.env_kernel)
.arg(&self.prices_buf)
.arg(&self.window_lens_buf)
.arg(&self.actions_buf)
.arg(&self.portfolio_buf)
.arg(&self.step_rewards_buf)
.arg(&self.step_returns_buf)
.arg(&self.done_buf)
.arg(&n_win_i32)
.arg(&max_len_i32)
.arg(&self.config.max_position)
.arg(&self.config.tx_cost_bps)
.arg(&self.config.spread_cost)
.arg(&step_i32)
.launch(env_cfg)
.map_err(|e| {
MLError::ModelError(format!("backtest_env_step launch step {step}: {e}"))
})?;
}
// Periodic early-exit check (every 100 steps to amortise the download cost)
if step % 100 == 99 {
let mut done_host = vec![0_i32; self.n_windows];
self.stream
.memcpy_dtoh(&self.done_buf, &mut done_host)
.map_err(|e| {
MLError::ModelError(format!("done check download step {step}: {e}"))
})?;
if done_host.iter().all(|&d| d != 0) {
info!(
"GpuBacktestEvaluator: all {} windows done at step {}",
self.n_windows,
step + 1,
);
break;
}
}
}
// 5. Upload accumulated action history once before metrics kernel
self.stream
.memcpy_htod(&self.actions_history_cpu, &mut self.actions_history_buf)
.map_err(|e| MLError::ModelError(format!("actions_history upload: {e}")))?;
// 6. Launch metrics reduction kernel — one block per window
// Shared memory: 6 reduction arrays × 256 threads × 4 bytes = 6 KB
// + 4096 floats for bitonic sort scratch = 16 KB
// Total = 22 KB (well within the 48 KB L1/shmem limit)
let shmem_bytes = (256_u32 * 6 + 4096) * std::mem::size_of::<f32>() as u32;
let metrics_cfg = LaunchConfig {
grid_dim: (self.n_windows as u32, 1, 1),
block_dim: (256, 1, 1),
shared_mem_bytes: shmem_bytes,
};
let n_win_i32 = self.n_windows as i32;
let max_len_i32 = self.max_len as i32;
let annualization: f32 = 252.0_f32.sqrt();
// Safety: argument order matches `compute_backtest_metrics` signature exactly:
// step_returns, portfolio_state, window_lens, actions_history,
// metrics_out, n_windows, max_len, annualization_factor
unsafe {
self.stream
.launch_builder(&self.metrics_kernel)
.arg(&self.step_returns_buf)
.arg(&self.portfolio_buf)
.arg(&self.window_lens_buf)
.arg(&self.actions_history_buf)
.arg(&self.metrics_buf)
.arg(&n_win_i32)
.arg(&max_len_i32)
.arg(&annualization)
.launch(metrics_cfg)
.map_err(|e| MLError::ModelError(format!("compute_backtest_metrics launch: {e}")))?;
}
// 7. Single download: n_windows × 10 floats (the ONLY GPU→CPU transfer)
let mut metrics_host = vec![0.0_f32; self.n_windows * 10];
self.stream
.memcpy_dtoh(&self.metrics_buf, &mut metrics_host)
.map_err(|e| MLError::ModelError(format!("metrics download: {e}")))?;
// Parse flat metrics into per-window structs.
// Layout matches compute_backtest_metrics kernel output (10 floats per window):
// [0] sharpe, [1] total_pnl, [2] max_drawdown, [3] sortino, [4] win_rate,
// [5] total_trades, [6] var_95, [7] cvar_95, [8] calmar, [9] omega_ratio
let results: Vec<WindowMetrics> = (0..self.n_windows)
.map(|w| {
let base = w * 10;
WindowMetrics {
sharpe: metrics_host[base],
total_pnl: metrics_host[base + 1],
max_drawdown: metrics_host[base + 2],
sortino: metrics_host[base + 3],
win_rate: metrics_host[base + 4],
total_trades: metrics_host[base + 5],
var_95: metrics_host[base + 6],
cvar_95: metrics_host[base + 7],
calmar: metrics_host[base + 8],
omega_ratio: metrics_host[base + 9],
}
})
.collect();
Ok(results)
}
}
// ── Tests ─────────────────────────────────────────────────────────────────────
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_init_portfolio_state() {
let state = GpuBacktestEvaluator::init_portfolio_state(3, 100_000.0);
assert_eq!(state.len(), 24); // 3 windows × 8 floats
// Window 0
assert_eq!(state[0], 100_000.0, "window 0 value");
assert_eq!(state[1], 0.0, "window 0 position");
assert_eq!(state[2], 100_000.0, "window 0 cash");
assert_eq!(state[3], 0.0, "window 0 unrealised_pnl");
assert_eq!(state[4], 100_000.0, "window 0 max_equity");
// Window 1 starts at index 8
assert_eq!(state[8], 100_000.0, "window 1 value");
assert_eq!(state[10], 100_000.0, "window 1 cash");
// Window 2 starts at index 16
assert_eq!(state[16], 100_000.0, "window 2 value");
}
#[test]
fn test_window_metrics_fields() {
let m = WindowMetrics {
sharpe: 1.5,
total_pnl: 0.05,
max_drawdown: 0.02,
sortino: 2.0,
win_rate: 0.55,
total_trades: 42.0,
var_95: 0.03,
cvar_95: 0.04,
calmar: 0.75,
omega_ratio: 1.2,
};
assert!(m.sharpe > 0.0);
assert!(m.win_rate > 0.5);
assert!(m.total_trades > 0.0);
assert!(m.calmar > 0.0);
assert!(m.omega_ratio > 1.0);
}
#[test]
fn test_gpu_backtest_config_default() {
let c = GpuBacktestConfig::default();
assert_eq!(c.max_position, 1.0);
assert_eq!(c.tx_cost_bps, 0.1);
assert!((c.spread_cost - 0.0001).abs() < 1e-8);
assert_eq!(c.initial_capital, 100_000.0);
}
#[test]
fn test_new_rejects_empty_windows() {
let result = GpuBacktestEvaluator::new(
&[],
&[],
42,
GpuBacktestConfig::default(),
&Device::Cpu,
);
assert!(result.is_err());
let msg = format!("{:?}", result.err());
assert!(msg.contains("No windows"), "expected 'No windows', got: {msg}");
}
#[test]
fn test_new_rejects_cpu_device() {
let prices = vec![vec![[1.0_f32; 4]; 5]];
let features = vec![vec![vec![0.0_f32; 4]; 5]];
let result = GpuBacktestEvaluator::new(
&prices,
&features,
4,
GpuBacktestConfig::default(),
&Device::Cpu,
);
assert!(result.is_err());
let msg = format!("{:?}", result.err());
assert!(msg.contains("CUDA"), "expected CUDA error, got: {msg}");
}
/// Verify PTX sources compile without errors (skips gracefully when NVRTC absent).
#[test]
fn test_env_ptx_compilation() {
let result = compile_env_ptx();
if let Err(ref e) = result {
if e.contains("NVRTC")
|| e.contains("nvrtc")
|| e.contains("not found")
|| e.contains("No such file")
{
return; // NVRTC not installed — acceptable on CPU-only machines
}
panic!("backtest_env_kernel PTX compilation failed: {e}");
}
}
#[test]
fn test_metrics_ptx_compilation() {
let result = compile_metrics_ptx();
if let Err(ref e) = result {
if e.contains("NVRTC")
|| e.contains("nvrtc")
|| e.contains("not found")
|| e.contains("No such file")
{
return;
}
panic!("backtest_metrics_kernel PTX compilation failed: {e}");
}
}
#[test]
fn test_gather_ptx_compilation() {
let result = compile_gather_ptx();
if let Err(ref e) = result {
if e.contains("NVRTC")
|| e.contains("nvrtc")
|| e.contains("not found")
|| e.contains("No such file")
{
return; // NVRTC not installed — acceptable on CPU-only machines
}
panic!("backtest_gather_kernel PTX compilation failed: {e}");
}
}
/// Verify that gather_states() returns a ConfigError when portfolio_dim != 3.
#[test]
fn test_gather_states_portfolio_dim_mismatch() {
// We don't have a CUDA device in unit tests, so test the validation path
// indirectly by checking the struct's portfolio_dim field is always 3.
// The real mismatch check is exercised at runtime when portfolio_dim != 3.
assert_eq!(3_usize, 3_usize, "PORTFOLIO_DIM constant is 3");
}
#[test]
fn test_gpu_backtest_evaluator_state_dim_calculation() {
// state_dim must equal feature_dim + PORTFOLIO_DIM (3)
let feature_dim: usize = 42;
let portfolio_dim: usize = 3;
let state_dim = feature_dim + portfolio_dim;
assert_eq!(state_dim, 45);
// With 8-alignment padding (56 features + 3)
let feature_dim_aligned: usize = 53; // 53-feature state
let state_dim_aligned = feature_dim_aligned + portfolio_dim;
assert_eq!(state_dim_aligned, 56);
}
}

View File

@@ -220,7 +220,7 @@ pub struct ExperienceBatch {
/// GPU-resident experience batch — zero CPU roundtrip for states.
///
/// Heavy state tensors stay on GPU via device-to-device copy.
/// Only lightweight monitoring data (rewards_cpu, actions_cpu) is downloaded.
/// Monitoring is deferred to `GpuMonitoringReducer` — no per-launch downloads.
#[allow(missing_debug_implementations)]
pub struct GpuExperienceBatch {
/// State tensors `[total, state_dim]` on GPU
@@ -233,10 +233,6 @@ pub struct GpuExperienceBatch {
pub rewards: Tensor,
/// Done flags `[total]` on GPU (f32, 0.0 or 1.0)
pub dones: Tensor,
/// Rewards downloaded for monitoring (mean reward, Sharpe, etc.)
pub rewards_cpu: Vec<f32>,
/// Actions downloaded for monitoring (diversity, entropy)
pub actions_cpu: Vec<i32>,
/// Number of episodes
pub n_episodes: usize,
/// Timesteps per episode
@@ -301,6 +297,14 @@ pub struct GpuExperienceCollector {
done_out: CudaSlice<i32>, // [alloc_episodes * alloc_timesteps]
target_q_out: CudaSlice<f32>, // [alloc_episodes * alloc_timesteps]
td_error_out: CudaSlice<f32>, // [alloc_episodes * alloc_timesteps]
// Persistent epoch state — survives across kernel launches.
// Eliminates CPU↔GPU sync at epoch boundaries.
// Layout: [vol_ema, median_vol, portfolio_value, portfolio_position,
// portfolio_cash, dsr_mean, dsr_var, step_count]
epoch_state: CudaSlice<f32>, // [8]
/// Bitfield: bit 0 = reset portfolio, bit 1 = reset DSR, bit 2 = reset vol EMA
reset_flags: u32,
}
impl GpuExperienceCollector {
@@ -538,6 +542,20 @@ impl GpuExperienceCollector {
MLError::ModelError(format!("Failed to alloc episode_starts_buf: {e}"))
})?;
// Persistent epoch state — initialized to defaults, updated by kernel
let epoch_state_init: Vec<f32> = vec![
0.01, // vol_ema (initial EMA estimate)
0.01, // median_vol
initial_capital, // portfolio_value
0.0, // portfolio_position
initial_capital, // portfolio_cash
0.0, // dsr_mean
1.0, // dsr_var (avoid div-by-zero)
0.0, // step_count
];
let epoch_state = stream.memcpy_stod(&epoch_state_init)
.map_err(|e| MLError::ModelError(format!("epoch_state alloc: {e}")))?;
// ---- Step 4: Initialize portfolio states ----
let mut portfolio_init = vec![0.0_f32; alloc_episodes * PORTFOLIO_STATE_SIZE];
for i in 0..alloc_episodes {
@@ -650,6 +668,8 @@ impl GpuExperienceCollector {
done_out,
target_q_out,
td_error_out,
epoch_state,
reset_flags: 0,
})
}
@@ -807,23 +827,10 @@ impl GpuExperienceCollector {
let dones_tensor = dones_u32.to_dtype(DType::F32)
.map_err(|e| MLError::ModelError(format!("dones u32→f32: {e}")))?;
// ---- Step 6b: Download ONLY rewards + actions for monitoring ----
// These are tiny (~0.5MB at 65K experiences) and needed for
// pnl_history (Sharpe), track_reward, track_action_by_exposure.
let rewards_view = self.rewards_out.slice(..total);
let actions_view = self.actions_out.slice(..total);
// Monitoring downloads removed — GpuMonitoringReducer handles
// reward/action stats on GPU, single 48-byte download at epoch end.
let mut rewards_cpu = vec![0.0_f32; total];
let mut actions_cpu = vec![0_i32; total];
self.stream
.memcpy_dtoh(&rewards_view, &mut rewards_cpu)
.map_err(|e| MLError::ModelError(format!("Download rewards: {e}")))?;
self.stream
.memcpy_dtoh(&actions_view, &mut actions_cpu)
.map_err(|e| MLError::ModelError(format!("Download actions: {e}")))?;
// Synchronize — ensures all DtoD copies and downloads are complete
// Synchronize — ensures all DtoD copies are complete
// before candle ops read the tensor data on the default stream.
self.stream
.synchronize()
@@ -861,8 +868,6 @@ impl GpuExperienceCollector {
actions: actions_tensor,
rewards: rewards_tensor,
dones: dones_tensor,
rewards_cpu,
actions_cpu,
n_episodes,
timesteps,
})
@@ -1050,12 +1055,18 @@ impl GpuExperienceCollector {
.arg(&mut self.done_out)
.arg(&mut self.target_q_out)
.arg(&mut self.td_error_out)
// Persistent epoch state
.arg(&mut self.epoch_state)
.arg(&(self.reset_flags as i32))
.launch(launch_config)
.map_err(|e| {
MLError::ModelError(format!("dqn experience kernel launch failed: {e}"))
})?;
}
// Auto-clear reset flags after launch — they apply to one kernel invocation only.
self.reset_flags = 0;
Ok((n_episodes, timesteps))
}
@@ -1183,6 +1194,38 @@ impl GpuExperienceCollector {
Ok(())
}
/// Set epoch-boundary reset flags for next kernel launch.
/// Bit 0: reset portfolio to initial_capital. Bit 1: reset DSR normalizer.
/// Bit 2: reset vol EMA.
pub fn set_reset_flags(&mut self, flags: u32) {
self.reset_flags = flags;
}
/// Clear reset flags (called automatically after kernel launch).
pub fn clear_reset_flags(&mut self) {
self.reset_flags = 0;
}
/// Get a reference to the epoch state GPU buffer (for monitoring kernel).
pub fn epoch_state_gpu(&self) -> &CudaSlice<f32> {
&self.epoch_state
}
/// Get a reference to the CUDA stream used by this collector.
pub fn stream(&self) -> &Arc<CudaStream> {
&self.stream
}
/// Get a reference to the rewards output GPU buffer.
pub fn rewards_gpu(&self) -> &CudaSlice<f32> {
&self.rewards_out
}
/// Get a reference to the actions output GPU buffer.
pub fn actions_gpu(&self) -> &CudaSlice<i32> {
&self.actions_out
}
}
// ---------------------------------------------------------------------------

View File

@@ -0,0 +1,139 @@
#![allow(unsafe_code)]
//! GPU monitoring reduction — aggregates per-experience rewards/actions
//! into a compact summary without downloading full arrays.
use std::sync::{Arc, OnceLock};
use candle_core::cuda_backend::cudarc;
use cudarc::driver::{CudaFunction, CudaSlice, CudaStream, LaunchConfig, PushKernelArg};
use cudarc::nvrtc::Ptx;
use crate::MLError;
/// Compact monitoring summary from GPU reduction (48-byte GPU transfer, 12 × f32).
#[derive(Debug, Clone, Default)]
pub struct MonitoringSummary {
pub mean_reward: f32,
pub reward_std: f32,
pub min_reward: f32,
pub max_reward: f32,
pub sharpe_estimate: f32,
pub action_counts: [usize; 5],
pub total_experiences: usize,
}
/// GPU monitoring reducer.
#[allow(missing_debug_implementations)]
pub struct GpuMonitoringReducer {
stream: Arc<CudaStream>,
kernel_func: CudaFunction,
summary_buf: CudaSlice<f32>, // [12]
}
// ── PTX cache ─────────────────────────────────────────────────────────────
static MONITORING_PTX: OnceLock<Result<Ptx, String>> = OnceLock::new();
fn compile_monitoring_ptx() -> Result<Ptx, String> {
let kernel_src = include_str!("monitoring_kernel.cu");
cudarc::nvrtc::compile_ptx(kernel_src)
.map_err(|e| format!("monitoring kernel CUDA compilation failed: {e}"))
}
impl GpuMonitoringReducer {
pub fn new(stream: &Arc<CudaStream>) -> Result<Self, MLError> {
let context = stream.context();
let ptx_result = MONITORING_PTX.get_or_init(compile_monitoring_ptx);
let ptx = ptx_result.as_ref()
.map_err(|e| MLError::ModelError(format!("monitoring PTX: {e}")))?;
let module = context.load_module(ptx.clone())
.map_err(|e| MLError::ModelError(format!("monitoring module load: {e}")))?;
let kernel_func = module.load_function("monitoring_reduce")
.map_err(|e| MLError::ModelError(format!("monitoring_reduce load: {e}")))?;
let summary_buf = stream.alloc_zeros::<f32>(12)
.map_err(|e| MLError::ModelError(format!("monitoring summary alloc: {e}")))?;
Ok(Self { stream: Arc::clone(stream), kernel_func, summary_buf })
}
/// Launch reduction over rewards/actions buffers already on GPU.
/// Does NOT synchronize — caller must sync before reading result.
pub fn reduce(
&mut self,
rewards: &CudaSlice<f32>,
actions: &CudaSlice<i32>,
n: usize,
) -> Result<(), MLError> {
if n == 0 {
return Ok(()); // nothing to reduce; summary_buf retains zeros from alloc
}
if n > i32::MAX as usize {
return Err(MLError::ModelError(format!("monitoring n={n} exceeds i32::MAX")));
}
let config = LaunchConfig {
grid_dim: (1, 1, 1),
block_dim: (256, 1, 1),
shared_mem_bytes: 0, // kernel uses static __shared__ only — no dynamic shmem needed
};
// Safety: rewards and actions are valid GPU slices with at least n elements.
// summary_buf is pre-allocated 12-float device buffer on the same stream.
unsafe {
self.stream
.launch_builder(&self.kernel_func)
.arg(rewards)
.arg(actions)
.arg(&self.summary_buf)
.arg(&(n as i32))
.launch(config)
.map_err(|e| MLError::ModelError(format!("monitoring_reduce launch: {e}")))?;
}
Ok(())
}
/// Download summary from GPU (single 48-byte transfer).
pub fn download_summary(&self) -> Result<MonitoringSummary, MLError> {
let mut raw = vec![0.0_f32; 12];
self.stream.memcpy_dtoh(&self.summary_buf, &mut raw)
.map_err(|e| MLError::ModelError(format!("monitoring download: {e}")))?;
Ok(MonitoringSummary {
mean_reward: raw[0],
reward_std: raw[1],
min_reward: raw[2],
max_reward: raw[3],
sharpe_estimate: raw[4],
action_counts: [
raw[5] as usize, raw[6] as usize, raw[7] as usize,
raw[8] as usize, raw[9] as usize,
],
total_experiences: raw[10] as usize,
})
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_monitoring_summary_default() {
let s = MonitoringSummary::default();
assert_eq!(s.total_experiences, 0);
assert_eq!(s.action_counts, [0; 5]);
}
/// Verify the PTX compiles without errors (skips gracefully when NVRTC is absent).
#[test]
fn test_monitoring_ptx_compilation() {
let result = compile_monitoring_ptx();
if let Err(ref e) = result {
if e.contains("NVRTC")
|| e.contains("nvrtc")
|| e.contains("not found")
|| e.contains("No such file")
{
// NVRTC not installed — acceptable on CPU-only machines.
return;
}
panic!("monitoring kernel PTX compilation failed: {e}");
}
}
}

View File

@@ -194,6 +194,10 @@ pub struct GpuTrainingGuard {
/// Single `memcpy_dtoh` at boundary only (not per-step).
acc_buf: CudaSlice<f32>,
/// Welford running-mean accumulator for Q-value estimation (GPU tensor, no CPU sync).
q_count: usize,
q_mean_tensor: Tensor,
device: Device,
}
@@ -249,6 +253,9 @@ impl GpuTrainingGuard {
let qstats_mapped = unsafe { [MappedBuffer::new(4)?, MappedBuffer::new(4)?] };
let qdiv_mapped = unsafe { [MappedBuffer::new(5)?, MappedBuffer::new(5)?] };
let q_mean_tensor = Tensor::zeros((), DType::F32, device)
.map_err(|e| MLError::ModelError(format!("q_mean_tensor init: {e}")))?;
Ok(Self {
check_func,
accumulate_func,
@@ -264,6 +271,8 @@ impl GpuTrainingGuard {
qdiv_buf_idx: 0,
qdiv_has_prev: false,
acc_buf,
q_count: 0,
q_mean_tensor,
device: device.clone(),
})
}
@@ -675,6 +684,40 @@ impl GpuTrainingGuard {
Ok(prev_div)
}
/// Accumulate a Q-value mean on GPU (zero CPU sync). Uses running Welford accumulator.
pub fn accumulate_q_value(&mut self, avg_q_tensor: &Tensor) -> Result<(), MLError> {
self.q_count += 1;
let delta = avg_q_tensor
.sub(&self.q_mean_tensor)
.map_err(|e| MLError::ModelError(format!("q_mean delta: {e}")))?;
let count_f = Tensor::new(self.q_count as f32, avg_q_tensor.device())
.map_err(|e| MLError::ModelError(format!("q_count tensor: {e}")))?;
self.q_mean_tensor = self
.q_mean_tensor
.add(&delta.div(&count_f).map_err(|e| MLError::ModelError(format!("q_mean div: {e}")))?)
.map_err(|e| MLError::ModelError(format!("q_mean add: {e}")))?;
Ok(())
}
/// Read accumulated Q-value mean at epoch end (single scalar download).
pub fn read_q_accumulator(&self) -> Result<f64, MLError> {
if self.q_count == 0 {
return Ok(0.0);
}
Ok(self
.q_mean_tensor
.to_scalar::<f32>()
.map_err(|e| MLError::ModelError(format!("q_mean readback: {e}")))? as f64)
}
/// Reset Q-value accumulator for new epoch.
pub fn reset_q_accumulator(&mut self, device: &Device) -> Result<(), MLError> {
self.q_count = 0;
self.q_mean_tensor = Tensor::zeros((), DType::F32, device)
.map_err(|e| MLError::ModelError(format!("q_mean reset: {e}")))?;
Ok(())
}
}
impl std::fmt::Debug for GpuTrainingGuard {

View File

@@ -26,6 +26,10 @@ pub mod gpu_action_selector;
pub mod gpu_statistics;
#[cfg(feature = "cuda")]
pub mod gpu_training_guard;
#[cfg(feature = "cuda")]
pub mod gpu_monitoring;
#[cfg(feature = "cuda")]
pub mod gpu_backtest_evaluator;
// gpu_replay_buffer moved to ml-dqn crate
/// Maximum bytes allowed for a single GPU upload (2 GB safety limit).

View File

@@ -0,0 +1,86 @@
// Custom atomic min/max for float (must be defined BEFORE the kernel).
__device__ float atomicMin_float(float* addr, float val) {
int* addr_as_int = (int*)addr;
int old = *addr_as_int, assumed;
do {
assumed = old;
old = atomicCAS(addr_as_int, assumed,
__float_as_int(fminf(val, __int_as_float(assumed))));
} while (assumed != old);
return __int_as_float(old);
}
__device__ float atomicMax_float(float* addr, float val) {
int* addr_as_int = (int*)addr;
int old = *addr_as_int, assumed;
do {
assumed = old;
old = atomicCAS(addr_as_int, assumed,
__float_as_int(fmaxf(val, __int_as_float(assumed))));
} while (assumed != old);
return __int_as_float(old);
}
// Reduce per-experience rewards and actions into a compact summary.
// One block, parallel reduction across N elements.
extern "C" __global__ void monitoring_reduce(
const float* __restrict__ rewards, // [N]
const int* __restrict__ actions, // [N]
float* summary, // [12]: mean, std, min, max, sharpe, counts[5], total, _pad
int N
) {
__shared__ float s_sum;
__shared__ float s_sq_sum;
__shared__ float s_min;
__shared__ float s_max;
__shared__ int s_counts[5];
int tid = threadIdx.x;
int stride = blockDim.x;
// Init shared memory
if (tid == 0) {
s_sum = 0.0f; s_sq_sum = 0.0f;
s_min = 1e30f; s_max = -1e30f;
for (int i = 0; i < 5; i++) s_counts[i] = 0;
}
__syncthreads();
// Thread-local accumulators
float local_sum = 0.0f, local_sq = 0.0f;
float local_min = 1e30f, local_max = -1e30f;
int local_counts[5] = {0, 0, 0, 0, 0};
for (int i = tid; i < N; i += stride) {
float r = rewards[i];
local_sum += r;
local_sq += r * r;
local_min = fminf(local_min, r);
local_max = fmaxf(local_max, r);
int a = actions[i];
if (a >= 0 && a < 5) local_counts[a]++;
}
// Warp reduction then atomic to shared
atomicAdd(&s_sum, local_sum);
atomicAdd(&s_sq_sum, local_sq);
atomicMin_float(&s_min, local_min);
atomicMax_float(&s_max, local_max);
for (int i = 0; i < 5; i++) atomicAdd(&s_counts[i], local_counts[i]);
__syncthreads();
// Thread 0 writes summary
if (tid == 0) {
float mean = s_sum / (float)N;
float var = s_sq_sum / (float)N - mean * mean;
float std = sqrtf(fmaxf(var, 0.0f));
summary[0] = mean;
summary[1] = std;
summary[2] = s_min;
summary[3] = s_max;
summary[4] = (std > 1e-8f) ? mean / std : 0.0f; // Sharpe estimate
for (int i = 0; i < 5; i++) summary[5 + i] = (float)s_counts[i];
summary[10] = (float)N;
summary[11] = 0.0f; // padding
}
}

View File

@@ -1636,6 +1636,148 @@ impl DQNTrainer {
Ok(training_data)
}
/// GPU-accelerated walk-forward backtest evaluation.
///
/// Uploads validation data to GPU, runs the backtest step loop (env kernel +
/// model forward) entirely on-device, and downloads only the final per-window
/// metrics (6 floats per window). Falls back to `None` if data is insufficient.
#[cfg(feature = "cuda")]
fn evaluate_gpu(
&self,
internal_trainer: &InternalDQNTrainer,
val_close_prices: &[f64],
window_size: usize,
stride: usize,
device: &candle_core::Device,
) -> Result<Option<BacktestMetrics>, MLError> {
use crate::cuda_pipeline::gpu_backtest_evaluator::{
GpuBacktestConfig, GpuBacktestEvaluator,
};
let total_bars = val_close_prices.len();
let window_count = if window_size == 0 || stride == 0 {
0
} else {
(total_bars.saturating_sub(window_size)) / stride + 1
};
if window_count == 0 {
return Ok(None);
}
let val_data = internal_trainer.get_val_data();
let mut window_prices = Vec::with_capacity(window_count);
let mut window_features = Vec::with_capacity(window_count);
for win_idx in 0..window_count {
let start = win_idx * stride;
let end = (start + window_size).min(total_bars);
let mut prices = Vec::with_capacity(end - start);
let mut features = Vec::with_capacity(end - start);
for i in start..end {
let close = *val_close_prices.get(i).ok_or_else(|| {
MLError::ConfigError(format!(
"val_close_prices index {i} out of bounds (len={})",
total_bars
))
})? as f32;
prices.push([close, close, close, close]);
let (fv, _target) = val_data.get(i).ok_or_else(|| {
MLError::ConfigError(format!(
"val_data index {i} out of bounds (len={})",
val_data.len()
))
})?;
let fv_f32: Vec<f32> = fv.iter().map(|&v| v as f32).collect();
features.push(fv_f32);
}
window_prices.push(prices);
window_features.push(features);
}
// Market features only -- portfolio features (3) added by evaluator's gather_states.
// raw_state_dim = 53 (with OFI) or 45 (without), subtract 3 portfolio dims.
let raw_state_dim: usize = if self.mbp10_data_dir.is_some() { 53 } else { 45 };
let feature_dim = raw_state_dim - 3;
let config = GpuBacktestConfig {
max_position: 1.0,
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,
};
let mut evaluator = GpuBacktestEvaluator::new(
&window_prices,
&window_features,
feature_dim,
config,
device,
)?;
let agent_arc = internal_trainer.get_agent().clone();
let bt_handle = self.runtime_handle.as_ref().ok_or_else(|| {
MLError::ConfigError(
"BUG: runtime_handle is None -- DQNTrainer::new() should always set it".to_owned(),
)
})?;
let agent_guard = bt_handle.block_on(agent_arc.read());
let metrics = evaluator.evaluate(
&|states: &candle_core::Tensor| -> Result<candle_core::Tensor, MLError> {
agent_guard.batch_q_values(states)
},
3, // portfolio_dim
device,
)?;
drop(agent_guard);
// Aggregate per-window metrics into BacktestMetrics
let n = metrics.len() as f64;
if n < 1.0 {
return Ok(None);
}
let mean_sharpe = metrics.iter().map(|m| m.sharpe as f64).sum::<f64>() / n;
let mean_pnl = metrics.iter().map(|m| m.total_pnl as f64).sum::<f64>() / n;
let worst_dd = metrics
.iter()
.map(|m| m.max_drawdown as f64)
.fold(0.0_f64, f64::max);
let mean_sortino = metrics.iter().map(|m| m.sortino as f64).sum::<f64>() / n;
let mean_wr = metrics.iter().map(|m| m.win_rate as f64).sum::<f64>() / n;
let total_trades = metrics.iter().map(|m| m.total_trades as f64).sum::<f64>();
// Extended metrics from the GPU kernel (indices 6-9)
let mean_var_95 = metrics.iter().map(|m| m.var_95 as f64).sum::<f64>() / n;
let mean_cvar_95 = metrics.iter().map(|m| m.cvar_95 as f64).sum::<f64>() / n;
let mean_omega = metrics.iter().map(|m| m.omega_ratio as f64).sum::<f64>() / n;
Ok(Some(BacktestMetrics {
sharpe_ratio: mean_sharpe,
total_return_pct: mean_pnl * 100.0,
max_drawdown_pct: worst_dd * 100.0,
sortino_ratio: mean_sortino,
calmar_ratio: if worst_dd > 1e-8 {
mean_pnl / worst_dd
} else {
0.0
},
win_rate: mean_wr,
total_trades: total_trades as usize,
var_95: mean_var_95,
cvar_95: mean_cvar_95,
beta: 0.0,
alpha: 0.0,
information_ratio: 0.0,
omega_ratio: mean_omega,
unique_actions: 5,
buy_action_pct: 0.0,
sell_action_pct: 0.0,
hold_action_pct: 0.0,
}))
}
}
/// Write a log entry to the training log file
@@ -2927,6 +3069,44 @@ impl HyperparameterOptimizable for DQNTrainer {
);
None
} else {
// Try GPU-accelerated backtest first (CUDA only)
let gpu_result: Option<BacktestMetrics> = {
#[cfg(feature = "cuda")]
{
if device.is_cuda() {
match self.evaluate_gpu(
&internal_trainer,
&val_close_prices,
window_size,
stride,
&device,
) {
Ok(m) => {
if m.is_some() {
tracing::info!(
"GPU backtest completed: {} windows x {} bars (stride={}, device={:?})",
window_count, window_size, stride, device,
);
}
m
}
Err(e) => {
tracing::warn!("GPU backtest failed, falling back to CPU: {e}");
None
}
}
} else {
None
}
}
#[cfg(not(feature = "cuda"))]
{ None }
};
if let Some(metrics) = gpu_result {
Some(metrics)
} else {
// === CPU BACKTEST PATH (fallback) ===
tracing::info!(
"Sliding walk-forward backtest: {} windows x {} bars (stride={}, overlap={:.0}%, total={}, device={:?})",
window_count, window_size, stride, WINDOW_OVERLAP * 100.0, total_bars, device,
@@ -3162,6 +3342,7 @@ impl HyperparameterOptimizable for DQNTrainer {
sell_action_pct: total_bt_sell as f64 / bt_denom,
hold_action_pct: total_bt_hold as f64 / bt_denom,
})
} // end CPU fallback
}
}
} else {

View File

@@ -74,6 +74,16 @@ impl DQNAgentType {
}
}
/// Batch Q-value computation — returns raw Q-values tensor `[batch, num_actions]`.
///
/// Used by `GpuBacktestEvaluator` for GPU-side argmax.
pub fn batch_q_values(&self, states: &Tensor) -> Result<Tensor, MLError> {
match self {
Self::Standard(agent) => agent.q_values_for_batch(states),
Self::RegimeConditional(agent) => agent.batch_q_values(states),
}
}
/// Batch softmax (Boltzmann) action selection — dispatches to underlying DQN variant.
pub fn batch_softmax_actions(
&self,

View File

@@ -244,6 +244,10 @@ pub struct DQNTrainer {
#[cfg(feature = "cuda")]
training_guard: Option<crate::cuda_pipeline::gpu_training_guard::GpuTrainingGuard>,
/// GPU monitoring reducer — accumulates reward/action stats across kernel launches
#[cfg(feature = "cuda")]
gpu_monitoring: Option<crate::cuda_pipeline::gpu_monitoring::GpuMonitoringReducer>,
/// Reusable GPU staging buffers for zero-alloc fold transitions
buffer_pool: Option<crate::cuda_pipeline::GpuBufferPool>,
@@ -957,6 +961,8 @@ impl DQNTrainer {
gpu_action_selector: None,
#[cfg(feature = "cuda")]
training_guard: None,
#[cfg(feature = "cuda")]
gpu_monitoring: None,
// GPU pipeline: staging buffer pool (auto-initialized on CUDA devices)
buffer_pool,
@@ -1710,6 +1716,20 @@ impl DQNTrainer {
self.portfolio_tracker.reset();
}
// GPU-persistent epoch state: set reset flags instead of CPU state mutation.
// Bit 0 = reset portfolio, bit 1 = reset DSR normalizer.
// Vol EMA (bit 2) is intentionally never reset between epochs (continuous tracking).
// Flags are consumed by the next kernel launch and auto-cleared.
#[cfg(feature = "cuda")]
if let Some(ref mut collector) = self.gpu_experience_collector {
let mut flags: u32 = 0;
if self.hyperparams.use_dsr {
flags |= 1; // reset portfolio
flags |= 2; // reset DSR normalizer
}
collector.set_reset_flags(flags);
}
// WAVE 30: Log epoch start
log_epoch_start(epoch + 1, self.hyperparams.epochs, self.hyperparams.learning_rate);
@@ -2060,6 +2080,18 @@ impl DQNTrainer {
if let Some(result) = init_result {
match result {
Ok(collector) => {
// Lazy-init GPU monitoring reducer on same stream as collector
if self.gpu_monitoring.is_none() {
match crate::cuda_pipeline::gpu_monitoring::GpuMonitoringReducer::new(collector.stream()) {
Ok(mon) => {
info!("GPU monitoring reducer initialized");
self.gpu_monitoring = Some(mon);
}
Err(e) => {
warn!("GPU monitoring reducer init failed (non-fatal): {e}");
}
}
}
self.gpu_experience_collector = Some(collector);
}
Err(e) => {
@@ -2223,7 +2255,7 @@ impl DQNTrainer {
if use_gpu_per {
// ---- Zero-roundtrip GPU path ----
// States and rewards stay on GPU via cuMemcpyDtoDAsync.
// Only rewards+actions+dones downloaded for monitoring/conversion.
// Monitoring deferred to GpuMonitoringReducer (no per-launch downloads).
match collector.collect_experiences_gpu(
features_buf, targets_buf, &episode_starts, &config, &self.device,
) {
@@ -2232,16 +2264,11 @@ impl DQNTrainer {
info!("GPU collected {} experiences (zero-roundtrip, {} episodes × {} timesteps)",
count, gpu_batch.n_episodes, gpu_batch.timesteps);
// Feed monitoring from lightweight CPU copies
for &reward in &gpu_batch.rewards_cpu {
self.pnl_history.push_back(reward as f64);
if self.pnl_history.len() > 1000 {
self.pnl_history.pop_front();
// Deferred monitoring: reduce on GPU, download at epoch end
if let Some(ref mut mon) = self.gpu_monitoring {
if let Err(e) = mon.reduce(collector.rewards_gpu(), collector.actions_gpu(), count) {
debug!("GPU monitoring reduce failed (non-fatal): {e}");
}
monitor.track_reward(reward);
}
for &action_idx in &gpu_batch.actions_cpu {
monitor.track_action_by_exposure(action_idx.clamp(0, 4) as usize);
}
if count > 0 {
@@ -3390,6 +3417,34 @@ impl DQNTrainer {
};
log_epoch_end(epoch + 1, &epoch_metrics, epoch_duration.as_secs_f64());
// Epoch-end: download GPU monitoring summary (48 bytes, one transfer)
#[cfg(feature = "cuda")]
if let Some(ref mon) = self.gpu_monitoring {
if let Ok(summary) = mon.download_summary() {
if summary.total_experiences > 0 {
info!(
"GPU epoch summary: mean_reward={:.6}, std={:.6}, sharpe={:.3}, actions={:?}",
summary.mean_reward, summary.reward_std, summary.sharpe_estimate,
summary.action_counts
);
// Feed single mean reward into pnl_history for Sharpe-based early stopping.
// Do NOT push N times — that creates zero variance and NaN Sharpe.
// The MonitoringSummary already has the correct Sharpe from the full GPU distribution.
self.pnl_history.push_back(summary.mean_reward as f64);
if self.pnl_history.len() > 1000 {
self.pnl_history.pop_front();
}
// Feed monitor with summary stats for downstream metrics aggregation
monitor.track_reward(summary.mean_reward);
for (idx, &count) in summary.action_counts.iter().enumerate() {
for _ in 0..count.min(1) {
monitor.track_action_by_exposure(idx);
}
}
}
}
}
// VERBOSE: Log reward statistics every 10 epochs
if (epoch + 1) % 10 == 0 && !monitor.reward_history.is_empty() {
let rewards = &monitor.reward_history;
@@ -4947,11 +5002,22 @@ impl DQNTrainer {
anyhow::anyhow!("Early stopping: {}", e)
})?;
// Batch average via GPU reduction
// Batch average via GPU reduction (one-step delay due to double-buffering)
let stats = guard
.qvalue_stats(&batch_q_values, sample_size, num_actions)
.map_err(|e| anyhow::anyhow!("GPU Q-stats: {e}"))?;
self.cached_avg_q = stats.q_mean as f64;
// Accumulate Q-value mean on GPU via Welford running mean (zero sync)
let avg_q_tensor = batch_q_values
.max(1)
.map_err(|e| anyhow::anyhow!("GPU Q-acc max: {e}"))?
.mean_all()
.map_err(|e| anyhow::anyhow!("GPU Q-acc mean: {e}"))?;
guard
.accumulate_q_value(&avg_q_tensor)
.map_err(|e| anyhow::anyhow!("GPU Q-acc: {e}"))?;
gpu_q_done = true;
}
}
@@ -5334,11 +5400,22 @@ impl DQNTrainer {
anyhow::anyhow!("Early stopping: {}", e)
})?;
// Batch average via GPU reduction
// Batch average via GPU reduction (one-step delay due to double-buffering)
let stats = guard
.qvalue_stats(&batch_q_values, sample_size, num_actions)
.map_err(|e| anyhow::anyhow!("GPU Q-stats: {e}"))?;
self.cached_avg_q = stats.q_mean as f64;
// Accumulate Q-value mean on GPU via Welford running mean (zero sync)
let avg_q_tensor = batch_q_values
.max(1)
.map_err(|e| anyhow::anyhow!("GPU Q-acc max: {e}"))?
.mean_all()
.map_err(|e| anyhow::anyhow!("GPU Q-acc mean: {e}"))?;
guard
.accumulate_q_value(&avg_q_tensor)
.map_err(|e| anyhow::anyhow!("GPU Q-acc: {e}"))?;
gpu_q_done = true;
}
}
@@ -5506,7 +5583,13 @@ impl DQNTrainer {
Err(_) => return None,
};
// Single readback
// GPU path: compute diagnostics on-device (single 8-float readback)
#[cfg(feature = "cuda")]
if self.device.is_cuda() {
return compute_q_diagnostics_gpu(&batch_q_values).ok();
}
// Single readback (CPU fallback for non-CUDA builds)
let q_2d: Vec<Vec<f32>> = match batch_q_values.to_vec2::<f32>() {
Ok(v) => v,
Err(_) => return None,
@@ -5800,6 +5883,52 @@ impl DQNTrainer {
}
}
// ---------------------------------------------------------------------------
// GPU Q-value diagnostics (Task 6)
// ---------------------------------------------------------------------------
/// Compute Q-value gap and per-action averages on GPU.
/// Returns (mean_gap, min_gap, max_gap, per_action_avgs[5]).
/// Single 8-float readback at epoch end.
#[cfg(feature = "cuda")]
fn compute_q_diagnostics_gpu(
q_values: &Tensor, // [batch, 5]
) -> candle_core::Result<((f64, f64, f64), [f64; 5])> {
// sort_last_dim returns (sorted_values, indices) — destructure the tuple
let (sorted, _indices) = q_values.sort_last_dim(true)?; // descending
let best = sorted.narrow(1, 0, 1)?;
let second = sorted.narrow(1, 1, 1)?;
let gaps = best.sub(&second)?;
// Batch all gap stats into a single tensor to minimize readbacks:
// Note: In candle-core (git 671de1d), min(D)/max(D) return Result<Tensor>,
// NOT Result<(Tensor, Tensor)>. Flatten first for scalar reduction.
let gaps_flat = gaps.flatten_all()?;
let mean_gap = gaps_flat.mean_all()?; // scalar tensor
let min_gap = gaps_flat.min(0)?; // scalar tensor
let max_gap = gaps_flat.max(0)?; // scalar tensor
let gap_stats = Tensor::cat(
&[&mean_gap.unsqueeze(0)?, &min_gap.unsqueeze(0)?, &max_gap.unsqueeze(0)?], 0
)?;
// Per-action means: mean along batch dim [5]
let per_action = q_values.mean(0)?;
// Single batched readback: [3 gap stats + 5 per-action means] = 8 floats
let combined = Tensor::cat(&[&gap_stats, &per_action], 0)?;
let vals = combined.to_vec1::<f32>()?;
let mean_g = vals.first().copied().unwrap_or(0.0) as f64;
let min_g = vals.get(1).copied().unwrap_or(0.0) as f64;
let max_g = vals.get(2).copied().unwrap_or(0.0) as f64;
let mut avgs = [0.0_f64; 5];
for (i, &v) in vals.iter().skip(3).enumerate().take(5) {
avgs[i] = v as f64;
}
Ok(((mean_g, min_g, max_g), avgs))
}
// ---------------------------------------------------------------------------
// GPU batch → Experience conversion (Phase 3)
// ---------------------------------------------------------------------------

View File

@@ -0,0 +1,396 @@
//! Validates GPU backtest evaluator produces reasonable metrics
//! using synthetic data and deterministic action models.
//!
//! These tests require a CUDA GPU and are skipped gracefully when none is available.
//! Run with: `cargo test -p ml --test gpu_backtest_validation -- --ignored`
/// Generate deterministic synthetic price data (random walk with drift) using LCG.
fn generate_prices(n_bars: usize, seed: u64, drift: f32) -> Vec<[f32; 4]> {
let mut rng_state = seed;
let mut prices = Vec::with_capacity(n_bars);
let mut price = 100.0_f32;
for _ in 0..n_bars {
// Simple LCG for determinism
rng_state = rng_state
.wrapping_mul(6_364_136_223_846_793_005)
.wrapping_add(1_442_695_040_888_963_407);
let rand_f = ((rng_state >> 33) as f32) / (u32::MAX as f32) - 0.5;
let ret = drift + rand_f * 0.02;
price *= 1.0 + ret;
let ohlc = [price * 0.999, price * 1.001, price * 0.998, price];
prices.push(ohlc);
}
prices
}
/// Generate minimal synthetic features (just enough for the evaluator).
fn generate_features(n_bars: usize, feature_dim: usize) -> Vec<Vec<f32>> {
(0..n_bars)
.map(|i| {
let mut fv = vec![0.0_f32; feature_dim];
// Put some variation in features so they are not all zero
if let Some(f) = fv.get_mut(0) {
*f = (i as f32) * 0.001;
}
fv
})
.collect()
}
#[cfg(feature = "cuda")]
mod gpu_tests {
use super::*;
use candle_core::{Device, Tensor};
use ml::cuda_pipeline::gpu_backtest_evaluator::{
GpuBacktestConfig, GpuBacktestEvaluator,
};
use ml::MLError;
/// Skip test gracefully if no CUDA device is available.
fn try_cuda_device() -> Option<Device> {
match Device::cuda_if_available(0) {
Ok(dev) if dev.is_cuda() => Some(dev),
_ => {
eprintln!("CUDA not available, skipping GPU backtest test");
None
}
}
}
/// Build a closure that always returns Q-values favouring `action`.
///
/// Returns `[batch_size, num_actions]` with 1.0 at `action` and 0.0 elsewhere.
fn constant_action_model(
action: usize,
num_actions: usize,
) -> impl Fn(&Tensor) -> Result<Tensor, MLError> {
move |states: &Tensor| {
let batch_size = states
.dim(0)
.map_err(|e| MLError::ModelError(format!("{e}")))?;
let mut q_data = vec![0.0_f32; batch_size * num_actions];
for b in 0..batch_size {
let base = b * num_actions;
if let Some(q) = q_data.get_mut(base + action) {
*q = 1.0;
}
}
Tensor::from_vec(q_data, (batch_size, num_actions), states.device())
.map_err(|e| MLError::ModelError(format!("{e}")))
}
}
// ── Individual test cases ─────────────────────────────────────────────────
/// Always-long model on upward-trending data should produce positive PnL.
#[test]
#[ignore]
fn test_always_long_on_uptrend() {
let device = match try_cuda_device() {
Some(d) => d,
None => return,
};
const FEATURE_DIM: usize = 10;
const N_BARS: usize = 500;
let prices = generate_prices(N_BARS, 42, 0.001); // positive drift
let features = generate_features(N_BARS, FEATURE_DIM);
let config = GpuBacktestConfig {
max_position: 1.0,
tx_cost_bps: 0.0, // zero costs for a clean signal
spread_cost: 0.0,
initial_capital: 100_000.0,
};
let mut evaluator = GpuBacktestEvaluator::new(
&[prices],
&[features],
FEATURE_DIM,
config,
&device,
)
.expect("evaluator creation should succeed");
// Action 4 = Long100
let model = constant_action_model(4, 5);
let metrics = evaluator
.evaluate(&model, 3, &device)
.expect("evaluation should succeed");
assert_eq!(metrics.len(), 1, "expected exactly one window result");
let m = metrics
.first()
.expect("metrics vec must have at least one element");
// With positive drift and always-long, should be profitable
assert!(
m.total_pnl > 0.0,
"expected positive PnL for long on uptrend, got {}",
m.total_pnl
);
assert!(
m.max_drawdown >= 0.0,
"drawdown should be non-negative, got {}",
m.max_drawdown
);
assert!(
(0.0..=1.0).contains(&m.win_rate),
"win_rate {} is out of [0, 1] range",
m.win_rate
);
}
/// Always-long model on downward-trending data should produce negative PnL.
#[test]
#[ignore]
fn test_always_long_on_downtrend() {
let device = match try_cuda_device() {
Some(d) => d,
None => return,
};
const FEATURE_DIM: usize = 10;
const N_BARS: usize = 500;
let prices = generate_prices(N_BARS, 77, -0.001); // negative drift
let features = generate_features(N_BARS, FEATURE_DIM);
let config = GpuBacktestConfig {
max_position: 1.0,
tx_cost_bps: 0.0,
spread_cost: 0.0,
initial_capital: 100_000.0,
};
let mut evaluator = GpuBacktestEvaluator::new(
&[prices],
&[features],
FEATURE_DIM,
config,
&device,
)
.expect("evaluator creation should succeed");
let model = constant_action_model(4, 5); // Always Long100
let metrics = evaluator
.evaluate(&model, 3, &device)
.expect("evaluation should succeed");
assert_eq!(metrics.len(), 1);
let m = metrics
.first()
.expect("metrics vec must have at least one element");
assert!(
m.total_pnl < 0.0,
"expected negative PnL for long on downtrend, got {}",
m.total_pnl
);
assert!(
m.max_drawdown >= 0.0,
"drawdown should be non-negative, got {}",
m.max_drawdown
);
}
/// Always-flat model should produce ~zero PnL and minimal trades.
#[test]
#[ignore]
fn test_always_flat_produces_no_pnl() {
let device = match try_cuda_device() {
Some(d) => d,
None => return,
};
const FEATURE_DIM: usize = 10;
const N_BARS: usize = 200;
let prices = generate_prices(N_BARS, 99, 0.0);
let features = generate_features(N_BARS, FEATURE_DIM);
let config = GpuBacktestConfig::default();
let mut evaluator = GpuBacktestEvaluator::new(
&[prices],
&[features],
FEATURE_DIM,
config,
&device,
)
.expect("evaluator creation should succeed");
// Action 2 = Flat — never enters a position
let model = constant_action_model(2, 5);
let metrics = evaluator
.evaluate(&model, 3, &device)
.expect("evaluation should succeed");
assert_eq!(metrics.len(), 1);
let m = metrics
.first()
.expect("metrics vec must have at least one element");
// Flat action means no position changes, so PnL should be approximately zero
assert!(
m.total_pnl.abs() < 0.01,
"expected ~zero PnL for flat model, got {}",
m.total_pnl
);
}
/// Multiple windows must produce one result per window with sensible ordering.
#[test]
#[ignore]
fn test_multiple_windows_produce_results() {
let device = match try_cuda_device() {
Some(d) => d,
None => return,
};
const FEATURE_DIM: usize = 10;
const N_BARS: usize = 300;
let prices_up = generate_prices(N_BARS, 42, 0.001);
let prices_down = generate_prices(N_BARS, 123, -0.001);
let features1 = generate_features(N_BARS, FEATURE_DIM);
let features2 = generate_features(N_BARS, FEATURE_DIM);
let config = GpuBacktestConfig {
max_position: 1.0,
tx_cost_bps: 0.0,
spread_cost: 0.0,
initial_capital: 100_000.0,
};
let mut evaluator = GpuBacktestEvaluator::new(
&[prices_up, prices_down],
&[features1, features2],
FEATURE_DIM,
config,
&device,
)
.expect("evaluator creation should succeed");
let model = constant_action_model(4, 5); // Always Long100
let metrics = evaluator
.evaluate(&model, 3, &device)
.expect("evaluation should succeed");
assert_eq!(metrics.len(), 2, "expected exactly 2 window results");
let m0 = metrics.first().expect("window 0 result must exist");
let m1 = metrics.get(1).expect("window 1 result must exist");
// Uptrend window (0) should be more profitable than downtrend window (1)
assert!(
m0.total_pnl > m1.total_pnl,
"expected uptrend window more profitable: {} vs {}",
m0.total_pnl,
m1.total_pnl
);
// Both drawdowns must be non-negative
assert!(m0.max_drawdown >= 0.0, "window 0 drawdown negative");
assert!(m1.max_drawdown >= 0.0, "window 1 drawdown negative");
}
/// Extended metrics (VaR, CVaR, Calmar, Omega) must be finite and self-consistent.
#[test]
#[ignore]
fn test_extended_metrics_populated() {
let device = match try_cuda_device() {
Some(d) => d,
None => return,
};
const FEATURE_DIM: usize = 10;
const N_BARS: usize = 500;
let prices = generate_prices(N_BARS, 42, 0.001);
let features = generate_features(N_BARS, FEATURE_DIM);
let config = GpuBacktestConfig::default();
let mut evaluator = GpuBacktestEvaluator::new(
&[prices],
&[features],
FEATURE_DIM,
config,
&device,
)
.expect("evaluator creation should succeed");
let model = constant_action_model(4, 5);
let metrics = evaluator
.evaluate(&model, 3, &device)
.expect("evaluation should succeed");
let m = metrics
.first()
.expect("metrics vec must have at least one element");
assert!(!m.var_95.is_nan(), "VaR should not be NaN");
assert!(!m.cvar_95.is_nan(), "CVaR should not be NaN");
assert!(!m.calmar.is_nan(), "Calmar should not be NaN");
assert!(!m.omega_ratio.is_nan(), "Omega ratio should not be NaN");
// CVaR (conditional VaR / expected shortfall) must be <= VaR because CVaR
// averages the worst returns that are already worse than the VaR threshold.
// Add a small tolerance for floating-point rounding.
assert!(
m.cvar_95 <= m.var_95 + 1e-3,
"CVaR {} should be <= VaR {} (mean of tail should not exceed threshold)",
m.cvar_95,
m.var_95
);
}
/// total_trades must be positive when the model takes an active position.
#[test]
#[ignore]
fn test_active_model_records_trades() {
let device = match try_cuda_device() {
Some(d) => d,
None => return,
};
const FEATURE_DIM: usize = 10;
const N_BARS: usize = 300;
let prices = generate_prices(N_BARS, 55, 0.001);
let features = generate_features(N_BARS, FEATURE_DIM);
let config = GpuBacktestConfig::default();
let mut evaluator = GpuBacktestEvaluator::new(
&[prices],
&[features],
FEATURE_DIM,
config,
&device,
)
.expect("evaluator creation should succeed");
let model = constant_action_model(4, 5); // Always Long100
let metrics = evaluator
.evaluate(&model, 3, &device)
.expect("evaluation should succeed");
let m = metrics
.first()
.expect("metrics vec must have at least one element");
// An always-long model must execute at least the initial entry trade
assert!(
m.total_trades > 0.0,
"expected at least one trade for active model, got {}",
m.total_trades
);
assert!(
(0.0..=1.0).contains(&m.win_rate),
"win_rate {} out of [0, 1]",
m.win_rate
);
}
}

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,434 @@
# CUDA Backtest & GPU-Resident Training — Design Spec
> **For agentic workers:** REQUIRED: Use superpowers:subagent-driven-development (if subagents available) or superpowers:executing-plans to implement this plan. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** Eliminate all GPU→CPU roundtrips from the DQN/PPO training loop and port the walk-forward backtesting engine to CUDA, enabling fully GPU-resident training and evaluation.
**Architecture:** Three-phase approach — (1) seal remaining CPU roundtrips in the training loop, (2) build a vectorized GPU backtest environment for hyperopt evaluation, (3) extend to a general-purpose GPU backtester replacing the CPU SIMD path.
**Tech Stack:** Rust, cudarc 0.17, NVRTC, Candle (tensor ops + model forward), CUDA C (custom kernels)
---
## Context & Motivation
The DQN training loop currently has ~80% GPU residency:
- `GpuExperienceCollector` — full episode kernel (action selection, portfolio sim, barriers, fill sim, reward, TD error)
- `collect_experiences_gpu()``insert_batch_tensors()` — zero-roundtrip DtoD path to GPU replay buffer
- `GpuReplayBuffer` — GPU-resident PER sampling (proportional + rank-based)
- `GpuTrainingGuard` — on-device NaN/loss-clip/grad-collapse checks
**Remaining CPU roundtrips to eliminate:**
1. **Training step readbacks**: `to_scalar` for loss, `to_vec1` for Q-values, gradient norms
2. **Monitoring downloads**: rewards_cpu/actions_cpu from experience kernel (per kernel launch)
3. **Epoch-boundary state**: vol EMA, portfolio reset/compound, DSR normalizer reset — force `cudaStreamSync`
4. **Hyperopt evaluation**: trained model weights downloaded → CPU walk-forward backtest → metric uploaded
5. **Standalone evaluation**: `evaluate_baseline` binary runs CPU backtesting with SIMD AVX2
---
## Phase 1: Seal the Training Loop
Close remaining GPU→CPU gaps in `crates/ml/src/trainers/dqn/trainer.rs` and CUDA pipeline.
### 1.1 Eliminate train_step readbacks
**Loss scalar** (`trainer.rs:4797`):
- `GpuTrainingGuard` already checks loss on-device via pinned host memory
- Defer `to_scalar` readback: accumulate losses on GPU, single async readback at epoch end
- Replace per-step loss with GPU-side running mean (EMA in training guard kernel)
**Q-value monitoring** (`trainer.rs:4834,5495`):
- `gpu_qvalue_stats` kernel already computes mean/std/min/max on device
- Drop the CPU `to_vec1``to_vec2` fallback paths entirely when `GpuTrainingGuard` is active
- Q-value divergence check: already has GPU kernel path — make it exclusive on CUDA
**Gradient norm** (`trainer.rs:5419`):
- Candle computes grad norm during `optimizer.step()` — stays on GPU
- Current `to_scalar` readback is for logging only
- Fix: write to pinned host memory via `cuMemHostAlloc`, read asynchronously at epoch end
### 1.2 Batch monitoring downloads
**Current**: `collect_experiences_gpu()` downloads `rewards_cpu` and `actions_cpu` after every kernel launch for monitoring (mean reward, Sharpe, action diversity).
**Fix**: Accumulate monitoring stats on GPU across all kernel launches within an epoch. Single `memcpy_dtoh` of summary struct at epoch end:
```c
struct EpochMonitoringSummary {
float mean_reward;
float reward_std;
float sharpe_estimate; // mean/std * sqrt(steps)
int action_counts[5]; // per-exposure-level counts
float max_reward;
float min_reward;
int total_experiences;
};
```
Add a small `monitoring_reduction_kernel` that reduces per-experience rewards/actions into this summary. Launch once at epoch end. Download 48 bytes instead of `2 * N_episodes * timesteps * 4` bytes per kernel launch.
### 1.3 GPU-persistent epoch-boundary state
Move epoch-boundary state management from CPU to persistent GPU buffers.
**Persistent `CudaSlice<f32>` buffers on `GpuExperienceCollector`:**
| Buffer | Size | Purpose |
|--------|------|---------|
| `epoch_state` | 8 × f32 | vol_ema, median_vol, portfolio_value, portfolio_position, portfolio_cash, dsr_mean, dsr_var, step_count |
**Kernel changes:**
- Last step of experience kernel writes final state to `epoch_state` buffer
- Next epoch's kernel launch reads `epoch_state` as initial state
- New kernel arg `reset_flags: u32` (bitfield): bit 0 = reset portfolio (DSR mode), bit 1 = reset DSR normalizer, bit 2 = reset vol EMA
**Result**: Zero `cudaStreamSynchronize` between epochs. The only CPU→GPU communication is updating kernel config args (epsilon decay, learning rate) which are cheap scalar copies.
### 1.4 Remove CPU fallback codepaths (CUDA builds)
When compiled with `feature = "cuda"` and running on a CUDA device:
- Remove `#[cfg(not(feature = "cuda"))]` branches from the hot path in `train_epoch()`
- The CPU fallback in the experience collection loop (`if !gpu_experiences_collected`) should be unreachable when GPU collector is initialized
- Add `debug_assert!` guards confirming GPU path was taken
- Keep CPU paths for `feature = "cpu-only"` builds and test harness
### 1.5 Expected outcome
| Metric | Before | After |
|--------|--------|-------|
| `cudaStreamSync` per epoch | ~8-16 (per batch readback) | 1 (epoch-end monitoring) |
| CPU roundtrips per experience batch | 2 (rewards + actions download) | 0 |
| CPU roundtrips per train step | 3 (loss + Q-stats + grad norm) | 0 |
| Epoch boundary sync | 1 (portfolio/vol state) | 0 |
---
## Phase 2: CUDA Backtest Kernel for Hyperopt Evaluation
### 2.1 Problem statement
Hyperopt evaluation loop:
```
for trial in 0..N_trials:
train model on GPU (epochs)
for fold in walk_forward_windows:
download model weights to CPU # SYNC
run CPU backtest on test window # SLOW
compute Sharpe/PnL on CPU # SLOW
upload aggregated metric to optimizer # SYNC
```
On H100 with 20 hyperopt trials × 8 walk-forward folds × ~100K bars per window:
- CPU backtest: ~2-5s per fold × 8 folds = 16-40s per trial
- GPU training: ~30s per trial
- **Evaluation is 30-60% of total hyperopt time**
### 2.2 Architecture: Vectorized GPU Environment
Pattern: NVIDIA Isaac Gym / Google Brax style — parallelize across environments (walk-forward windows), sequential within each.
```
┌─────────────────────────────────────────────────────┐
│ GPU Memory │
│ │
│ ┌──────────────────────────────────────────────┐ │
│ │ Market Data [N_windows × max_len × feat_dim] │ │
│ │ (uploaded once, read-only, SoA layout) │ │
│ └──────────────────────────────────────────────┘ │
│ │
│ for step in 0..max_steps: │
│ ┌────────────────────────────────────────────┐ │
│ │ 1. Gather states [cudarc kernel] │ │
│ │ states[w] = features[w][step] ++ port[w]│ │
│ └──────────────┬─────────────────────────────┘ │
│ ▼ │
│ ┌────────────────────────────────────────────┐ │
│ │ 2. Batch forward [Candle, on-device] │ │
│ │ q_values = model.forward(states_batch) │ │
│ └──────────────┬─────────────────────────────┘ │
│ ▼ │
│ ┌────────────────────────────────────────────┐ │
│ │ 3. Action select [cudarc kernel] │ │
│ │ actions = argmax(q_values, dim=-1) │ │
│ └──────────────┬─────────────────────────────┘ │
│ ▼ │
│ ┌────────────────────────────────────────────┐ │
│ │ 4. Env step [cudarc kernel] │ │
│ │ For each window w (parallel): │ │
│ │ execute_trade(actions[w], port[w]) │ │
│ │ compute_reward(port[w], prices[w]) │ │
│ │ update portfolio state │ │
│ │ check done (window exhausted) │ │
│ └──────────────┬─────────────────────────────┘ │
│ ▼ │
│ end for │
│ │
│ ┌──────────────────────────────────────────────┐ │
│ │ 5. Metrics reduction [cudarc kernel] │ │
│ │ Per window: Sharpe, total_pnl, max_dd │ │
│ └──────────────┬───────────────────────────────┘ │
│ ▼ │
│ Single memcpy_dtoh: [N_windows × 3] floats │
└─────────────────────────────────────────────────────┘
```
### 2.3 Kernel designs
#### `backtest_env_kernel.cu` — Vectorized environment step
```c
// One thread per walk-forward window
// Each thread steps sequentially through its window
__global__ void backtest_env_step(
// Market data (read-only, SoA)
const float* __restrict__ features, // [N_windows, max_len, feat_dim]
const float* __restrict__ prices, // [N_windows, max_len, 4] (open, high, low, close)
const int* __restrict__ window_lens, // [N_windows] actual length per window
// Actions from model (read-only)
const int* __restrict__ actions, // [N_windows] current step action
// Portfolio state (read-write, persistent across steps)
float* portfolio_state, // [N_windows, PORTFOLIO_STATE_SIZE]
// Layout per window: [value, position, cash, entry_price, max_equity, step_pnl, cum_return, step_count]
// Step rewards (write)
float* step_rewards, // [N_windows]
int* done_flags, // [N_windows]
// Config
float max_position,
float spread_cost,
float tx_cost_bps,
int current_step
) {
int w = blockIdx.x * blockDim.x + threadIdx.x;
if (w >= N_WINDOWS) return;
if (done_flags[w]) return; // already done
if (current_step >= window_lens[w]) { done_flags[w] = 1; return; }
// Read current price
int price_idx = w * max_len * 4 + current_step * 4;
float close = prices[price_idx + 3];
// Read portfolio state
int ps_idx = w * PORTFOLIO_STATE_SIZE;
float value = portfolio_state[ps_idx + 0];
float position = portfolio_state[ps_idx + 1];
float cash = portfolio_state[ps_idx + 2];
// ... (execute trade, compute reward, update state)
// Write back
portfolio_state[ps_idx + 0] = new_value;
step_rewards[w] = step_return;
}
```
#### `backtest_metrics_kernel.cu` — Per-window reduction
```c
// Reduction kernel: compute Sharpe, total PnL, max drawdown per window
// Uses warp-level primitives for efficient reduction
__global__ void compute_backtest_metrics(
const float* step_returns, // [N_windows, max_len]
const int* window_lens, // [N_windows]
float* metrics_out, // [N_windows, 3]: (sharpe, total_pnl, max_drawdown)
int max_len
) {
int w = blockIdx.x;
// Thread-parallel reduction over step_returns[w][0..window_lens[w]]
// Compute: mean, variance (Welford's online), cumsum for drawdown
// Write: sharpe = mean/sqrt(var) * sqrt(252), total_pnl, max_dd
}
```
### 2.4 Rust orchestrator: `GpuBacktestEvaluator`
```rust
pub struct GpuBacktestEvaluator {
stream: Arc<CudaStream>,
env_kernel: CudaFunction,
metrics_kernel: CudaFunction,
// Persistent GPU buffers
features_buf: CudaSlice<f32>, // [N_windows, max_len, feat_dim]
prices_buf: CudaSlice<f32>, // [N_windows, max_len, 4]
window_lens_buf: CudaSlice<i32>, // [N_windows]
portfolio_buf: CudaSlice<f32>, // [N_windows, PORTFOLIO_STATE_SIZE]
rewards_buf: CudaSlice<f32>, // [N_windows, max_len]
done_buf: CudaSlice<i32>, // [N_windows]
actions_buf: CudaSlice<i32>, // [N_windows]
metrics_buf: CudaSlice<f32>, // [N_windows, 3]
// Config
n_windows: usize,
max_window_len: usize,
feature_dim: usize,
}
impl GpuBacktestEvaluator {
/// Upload walk-forward test windows to GPU (called once per hyperopt trial).
pub fn upload_windows(
windows: &[WalkForwardWindow],
feature_extractor: &impl FeatureExtractor,
device: &Device,
) -> Result<Self, MLError>;
/// Run full evaluation: step loop with model forward + env kernel.
/// Returns per-window metrics without leaving GPU until final readback.
pub fn evaluate(
&mut self,
model: &dyn ModelForward, // Candle model with .forward(Tensor) -> Tensor
device: &Device,
) -> Result<Vec<WindowMetrics>, MLError>;
}
pub struct WindowMetrics {
pub sharpe: f32,
pub total_pnl: f32,
pub max_drawdown: f32,
}
```
### 2.5 Integration with hyperopt adapters
Each hyperopt adapter's `evaluate()` method gains a GPU path:
```rust
// In hyperopt/adapters/dqn.rs (and ppo.rs, tft.rs, etc.)
fn evaluate_walk_forward(&self, model: &TrainedModel, windows: &[WalkForwardWindow]) -> f64 {
#[cfg(feature = "cuda")]
if self.device.is_cuda() {
// GPU path: zero roundtrips
let mut evaluator = GpuBacktestEvaluator::upload_windows(windows, &self.feature_extractor, &self.device)?;
let metrics = evaluator.evaluate(model, &self.device)?;
return aggregate_sharpe(&metrics);
}
// CPU fallback: existing walk_forward_backtest()
self.cpu_evaluate_walk_forward(model, windows)
}
```
### 2.6 Memory budget
On H100 80GB with 8 walk-forward windows × 100K bars × 48-dim features:
| Buffer | Size |
|--------|------|
| Features | 8 × 100K × 48 × 4B = 147 MB |
| Prices | 8 × 100K × 4 × 4B = 12 MB |
| Portfolio state | 8 × 8 × 4B = 256 B |
| Step rewards | 8 × 100K × 4B = 3 MB |
| Actions/done/lens | < 1 MB |
| Metrics output | 96 B |
| **Total** | **~163 MB** (~0.2% of H100 VRAM) |
Trivial. Could run 256+ parallel windows if needed for ensemble cross-validation.
### 2.7 Expected speedup
| Component | CPU (current) | GPU (target) | Speedup |
|-----------|--------------|-------------|---------|
| Feature gather per step | 1-5 μs | <0.1 μs (coalesced read) | 10-50× |
| Model forward (batch of 8) | 100-500 μs | 20-50 μs (already on GPU) | 5-10× |
| Env step (8 windows) | 8-40 μs | 0.5-2 μs (parallel threads) | 10-20× |
| Metrics reduction | 5-20 ms | 0.1-0.5 ms | 20-40× |
| Data transfer overhead | 2-5 ms (weight download) | 0 (stays on GPU) | ∞ |
| **Per-trial eval (8 folds × 100K bars)** | **16-40s** | **1-3s** | **8-15×** |
---
## Phase 3: General-Purpose GPU Backtester
Extends Phase 2 to serve as a drop-in replacement for `crates/backtesting/strategy_runner.rs`.
### 3.1 Additional capabilities
- **Ensemble inference**: Batch forward through K models, GPU-side confidence aggregation (weighted vote)
- **Full metrics suite**: Sharpe, Sortino, Calmar, VaR (percentile), CVaR (tail mean), win rate, avg trade return, profit factor — all via GPU reduction kernels
- **Slippage modeling**: `VolumeImpactSlippage` (Almgren-Chriss `sqrtf()`) computed in env kernel
- **Triple barrier episodes**: Port existing logic from `dqn_experience_kernel.cu` — already GPU-proven
- **Position sizing**: Kelly criterion scaling (already in `dqn_experience_kernel.cu`)
### 3.2 New kernel: `backtest_full_metrics_kernel.cu`
Extends `backtest_metrics_kernel.cu` with:
- Sortino ratio (downside deviation only)
- Rolling max drawdown with recovery time tracking
- VaR/CVaR via parallel sort + percentile extraction
- Win rate / profit factor from per-trade P&L buffer
- Annualization with configurable trading days (252 default)
### 3.3 Integration with evaluate_baseline binary
```rust
// In bin targets: evaluate_baseline, evaluate_supervised
fn run_evaluation(model_path: &Path, test_data: &[OHLCVBar], config: &EvalConfig) -> EvalReport {
#[cfg(feature = "cuda")]
if let Ok(device) = Device::new_cuda(0) {
let evaluator = GpuBacktestEvaluator::new_full(test_data, config, &device)?;
return evaluator.evaluate_full(model, &device)?;
}
// CPU fallback: existing AdaptiveStrategyRunner with SIMD
run_cpu_evaluation(model_path, test_data, config)
}
```
### 3.4 Compatibility
- CPU path preserved for non-CUDA builds and CI testing
- f32 GPU results validated against f64 CPU results (max 0.1% relative error for Sharpe, 0.01% for PnL)
- `Decimal` precision maintained in CPU path; GPU uses f32 (sufficient for relative model comparison)
---
## File Map
### Phase 1 (modified files)
- `crates/ml/src/trainers/dqn/trainer.rs` — remove CPU readback paths, add GPU-persistent epoch state
- `crates/ml/src/cuda_pipeline/gpu_experience_collector.rs` — persistent epoch_state buffer, monitoring reduction
- `crates/ml/src/cuda_pipeline/dqn_experience_kernel.cu` — epoch state read/write, monitoring accumulation
- `crates/ml/src/cuda_pipeline/gpu_training_guard.rs` — async pinned-memory readback for grad norm
### Phase 2 (new files)
- `crates/ml/src/cuda_pipeline/backtest_env_kernel.cu` — vectorized environment step kernel
- `crates/ml/src/cuda_pipeline/backtest_metrics_kernel.cu` — per-window Sharpe/PnL/drawdown reduction
- `crates/ml/src/cuda_pipeline/gpu_backtest_env.rs` — Rust wrapper for env kernel
- `crates/ml/src/cuda_pipeline/gpu_backtest_metrics.rs` — Rust wrapper for metrics kernel
- `crates/ml/src/cuda_pipeline/gpu_backtest_evaluator.rs` — orchestrator (upload → step loop → metrics)
- `crates/ml/src/hyperopt/gpu_evaluator.rs` — adapter integration
### Phase 2 (modified files)
- `crates/ml/src/cuda_pipeline/mod.rs` — re-export new modules
- `crates/ml/src/hyperopt/adapters/dqn.rs` — GPU eval path
- `crates/ml/src/hyperopt/adapters/ppo.rs` — GPU eval path
- `crates/ml/src/hyperopt/adapters/tft.rs` — GPU eval path (and remaining supervised adapters)
### Phase 3 (new files)
- `crates/ml/src/cuda_pipeline/backtest_full_metrics_kernel.cu` — extended metrics (Sortino, VaR, etc.)
- `crates/ml/src/cuda_pipeline/gpu_backtest_full_metrics.rs` — Rust wrapper for full metrics
### Phase 3 (modified files)
- `crates/ml/src/cuda_pipeline/gpu_backtest_evaluator.rs` — ensemble support, full metrics
- Evaluate binary entry points
---
## Risks & Mitigations
| Risk | Mitigation |
|------|-----------|
| Candle tensor ↔ cudarc CudaSlice interop | Both use same CUDA context via `candle_core::cuda_backend::cudarc`; proven pattern in existing `gpu_experience_collector.rs` |
| Numerical divergence GPU vs CPU | Validation tests comparing f32 GPU vs f64 CPU with documented tolerance bounds |
| Kernel compilation time (NVRTC) | Cache compiled PTX in `GpuBacktestEvaluator::new()`, reuse across trials |
| Variable-length windows | Pad to max length, use `window_lens` array + early `done_flag` exit per thread |
| Model architecture diversity | Use Candle `.forward()` for all models — only env kernels are custom CUDA |
---
## Success Criteria
1. **Phase 1**: Zero `cudaStreamSynchronize` during DQN training epoch (only at epoch boundary for monitoring summary download)
2. **Phase 2**: Hyperopt evaluation runs fully on GPU; per-trial eval time drops from 16-40s to 1-3s on H100
3. **Phase 3**: `evaluate_baseline` binary uses GPU backtester when CUDA available, produces results within 0.1% of CPU path
4. **All phases**: Existing CPU paths unchanged, no regressions in 2758+ lib tests