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

feat(dqn): val plan_isv parity — hot-path integration

Browse Source 
Final commit toward task #94 val plan_isv parity. Wires the plan
machinery inside the chunked val backtest loop so state positions
[86..92) carry real plan signals matching training instead of zero-
fill. This is the structural distribution-shift fix behind the
observed val trade count of 21 / 214,654 bars (0.0098%).

Phase 6 inside `submit_dqn_step_loop_cublas`, per chunk, after env
step advances portfolio:

  1. Forward on the LAST-STEP N rows of chunked_states — a targeted
     `compute_q_values_to(last_step_states, N, scratch_q)` rewrites
     `save_h_s2` to exactly [N, SH2] for the final step. Needed
     because the batched `compute_q_values_to` for N*chunk_len rows
     leaves save_h_s2 at an arbitrary last-sub-batch slice.
  2. `compute_plan_params(plan_params_buf, N)` runs the trade-plan
     MLP on that clean save_h_s2, producing [N, 6] plan_params.
  3. `backtest_plan_state_isv` — Flat↔Positioned activation /
     deactivation on plan_state[N, 7] and writes plan_isv_buf[N, 6]
     consumed by the next chunk's first `launch_gather` for state
     positions [86..92).

ch_q_values is reused as the Q-scratch for step (1) — its original
chunk Q-values were already consumed by Phase 4 action-select.

Epoch-boundary diagnostic `launch_plan_diag_and_log` after the step
loop: launches `backtest_plan_diag_reduce` (1-block, 256 threads),
DtoH-copies the 8-float summary, syncs, and emits a single
`tracing::info!(target: "val_plan_diag", ...)` line with the six
labelled plan_isv means plus active_frac / n_active.

Scalars passed with exact i32 types (current_step, max_len,
feat_dim, n_windows) per feedback_cudarc_f64_f32_abi. Kernels
pulled via `plan_state_isv_kernel.as_ref().ok_or_else(...)` — they
are loaded in `ensure_action_select_ready` on first eval.

Not captured inside a CUDA Graph: the backtest step loop uses direct
kernel launches (evaluator comment: graphs SIGSEGV on 500K+ launches).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
jgrusewski
2026-04-24 01:20:30 +02:00
parent 39a353495d
commit b12483d91b
1 changed files with 133 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

133
crates/ml/src/cuda_pipeline/gpu_backtest_evaluator.rs
View File

@@ -922,9 +922,61 @@ impl GpuBacktestEvaluator {
// still uses CUDA Graphs — those capture ~20 operations per replay,
// well within driver limits.
self.submit_dqn_step_loop_cublas(dqn_cfg, q_provider)?;
// Val plan_isv parity diagnostic (task #94): reduce plan_state +
// plan_isv across windows → 8 floats, log for epoch-boundary
// visibility into plan activity. Cold path, runs once per eval.
self.launch_plan_diag_and_log()?;
self.launch_metrics_and_download()
}
/// Val plan_isv parity diagnostic emission (task #94).
///
/// Launches `backtest_plan_diag_reduce` on the final plan_state + plan_isv
/// buffers, DtoH-copies the 8-float summary, and logs a single
/// `val_plan_diag` line with labelled stats. Called once per
/// `evaluate_dqn_graphed` invocation at the epoch boundary — a stream
/// sync here is acceptable (already followed by the metrics readback
/// which syncs anyway).
fn launch_plan_diag_and_log(&mut self) -> Result<(), MLError> {
let kernel = self.plan_diag_reduce_kernel.as_ref().ok_or_else(|| {
MLError::ModelError("plan_diag_reduce_kernel not loaded".to_owned())
})?;
let n_i32 = self.n_windows as i32;
unsafe {
self.stream
.launch_builder(kernel)
.arg(&self.plan_state_buf)
.arg(&self.plan_isv_buf)
.arg(&self.plan_diag_buf)
.arg(&n_i32)
.launch(LaunchConfig {
grid_dim: (1, 1, 1),
block_dim: (256, 1, 1),
shared_mem_bytes: 0,
})
.map_err(|e| MLError::ModelError(format!(
"backtest_plan_diag_reduce launch: {e}"
)))?;
}
let mut diag = [0_f32; 8];
self.stream
.memcpy_dtoh(&self.plan_diag_buf, &mut diag)
.map_err(|e| MLError::ModelError(format!("plan_diag DtoH: {e}")))?;
self.stream.synchronize().map_err(|e| MLError::ModelError(format!(
"plan_diag sync: {e}"
)))?;
info!(
target: "val_plan_diag",
"val_plan_isv_diag: active_frac={:.3} n_active={:.0} \
mean|isv0|(progress)={:.4} mean|isv1|(pnl_vs_tgt)={:.4} mean|isv2|(pnl_vs_stop)={:.4} \
mean_isv3(conv_entry)={:.4} mean|isv4|(conv_drift)={:.4} mean|isv5|(regime_shift)={:.4}",
diag[0], diag[7], diag[1], diag[2], diag[3], diag[4], diag[5], diag[6],
);
Ok(())
}
/// Discard the cached CUDA graph.
///
@@ -1342,6 +1394,87 @@ impl GpuBacktestEvaluator {
)))?;
}
// ── Phase 6: Val plan_isv parity (task #94) ──────────────────────
//
// After Phase 5 has advanced portfolio_buf to the end-of-chunk
// state (last step = chunk_start + chunk_len - 1), compute:
// (a) plan_params[N, 6] for the LAST step's N windows via a
// targeted forward on ch_states[(chunk_len-1)*N..chunk_len*N]
// — this rewrites save_h_s2 = [N, SH2] for just those N
// rows (compute_q_values_to's earlier batched loop left
// save_h_s2 holding an arbitrary last-sub-batch slice, not
// aligned to the final-step N windows).
// (b) plan_state_isv: activation / deactivation / plan_isv
// compute so that next chunk's first launch_gather reads
// real plan_isv[6] at state positions [86..92).
//
// Using ch_q_values as scratch for the Q-output is safe — those
// Q-values were consumed by Phase 4 action-select and the Phase 5
// env step has already committed.
let plan_kernel = self.plan_state_isv_kernel.as_ref().ok_or_else(|| {
MLError::ModelError("plan_state_isv_kernel not loaded".to_owned())
})?;
let last_step = chunk_start + chunk_len - 1;
let last_step_row = chunk_len - 1;
let last_step_states_ptr = ch_states_base
+ (last_step_row * state_row_bytes) as u64;
let scratch_q_ptr = raw_f32_ptr(ch_q_values, &self.stream);
match q_provider {
Some(ref mut qp) => {
// (a.1) Forward on last-step N rows. Writes save_h_s2=[N,SH2].
qp.compute_q_values_to(last_step_states_ptr, n, scratch_q_ptr)
.map_err(|e| MLError::ModelError(format!(
"plan_params last-step forward chunk {chunk_idx}: {e}"
)))?;
// (a.2) Plan MLP on save_h_s2. Writes plan_params_buf[N,6].
let plan_params_ptr = {
let (p, g) = self.plan_params_buf.device_ptr(&self.stream);
let _ = ManuallyDrop::new(g);
p
};
qp.compute_plan_params(plan_params_ptr, n)
.map_err(|e| MLError::ModelError(format!(
"compute_plan_params chunk {chunk_idx}: {e}"
)))?;
}
None => {
return Err(MLError::ModelError(
"evaluate_dqn_graphed requires a QValueProvider for plan_params".to_owned()
));
}
}
// (b) Launch plan_state_isv kernel — updates plan_state in-place
// and writes plan_isv_buf[N, 6] consumed by the next chunk's
// first launch_gather() call.
let plan_blocks = ((n as u32) + 255) / 256;
let n_i32 = n as i32;
let last_step_i32 = last_step as i32;
let max_len_i32 = self.max_len as i32;
let feat_dim_i32 = self.feature_dim as i32;
unsafe {
self.stream
.launch_builder(plan_kernel)
.arg(&self.plan_params_buf)
.arg(&self.plan_state_buf)
.arg(&self.portfolio_buf)
.arg(&self.isv_signals_ptr)
.arg(&self.features_buf)
.arg(&last_step_i32)
.arg(&max_len_i32)
.arg(&feat_dim_i32)
.arg(&self.plan_isv_buf)
.arg(&n_i32)
.launch(LaunchConfig {
grid_dim: (plan_blocks.max(1), 1, 1),
block_dim: (256, 1, 1),
shared_mem_bytes: 0,
})
.map_err(|e| MLError::ModelError(format!(
"backtest_plan_state_isv chunk {chunk_idx}: {e}"
)))?;
}
// Periodic sync: catch deferred CUDA errors without stalling
// every chunk. Errors in chunk N surface within 10 chunks.
if chunk_idx % 10 == 0 || chunk_idx + 1 == total_chunks {