From b12483d91bedccad95beb190093c857412e5bc20 Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Fri, 24 Apr 2026 01:20:30 +0200 Subject: [PATCH] =?UTF-8?q?feat(dqn):=20val=20plan=5Fisv=20parity=20?= =?UTF-8?q?=E2=80=94=20hot-path=20integration?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- .../cuda_pipeline/gpu_backtest_evaluator.rs | 133 ++++++++++++++++++ 1 file changed, 133 insertions(+) diff --git a/crates/ml/src/cuda_pipeline/gpu_backtest_evaluator.rs b/crates/ml/src/cuda_pipeline/gpu_backtest_evaluator.rs index d7a1913ce..ef1311bd5 100644 --- a/crates/ml/src/cuda_pipeline/gpu_backtest_evaluator.rs +++ b/crates/ml/src/cuda_pipeline/gpu_backtest_evaluator.rs @@ -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 {