From 3836e25783baae0e466a2d686ae4d1d89437a249 Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Tue, 19 May 2026 17:04:42 +0200 Subject: [PATCH] feat(ml-backtesting): detect_close_transitions_batched kernel (P3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replaces TWO host loops in step_decision_with_latency with two GPU kernels: 1. snapshot_pos_state — replaces the host-side snapshot_realized_pnl + snapshot_position_lots + snapshot_open_horizon_mask trio (3 separate memcpy_dtoh per decision). Now one kernel launch writes prev_pos_lots_d / prev_realized_pnl_d / prev_open_horizon_mask_d directly on the device. 2. detect_close_transitions_batched — replaces the host close-detect loop that called read_pos per close-eligible backtest (up to n_backtests memcpy_dtoh per decision). Now one kernel writes closed_horizon_mask_d + realised_return_d on the device, and isv_kelly_update_on_close consumes them with no host roundtrip. At n_parallel=140 these two loops together accounted for ~350M+ small host roundtrips per quarter. Combined with P2 the latency-path of step_decision_with_latency is now fully GPU-resident. isv_kelly_update_on_close kernel always launches (skips backtests with mask=0 internally) rather than gating via a host any_close check. All P1+P2 regression tests pass through the new GPU close-detect path (at n=1 with uniform config + immediate-fill latency, no close happens in the cold-start tests since they only check market_target; the close path is exercised indirectly). Co-Authored-By: Claude Opus 4.7 (1M context) --- crates/ml-backtesting/cuda/pnl_track.cu | 48 +++++++++++ crates/ml-backtesting/src/sim/mod.rs | 105 +++++++++++++++--------- 2 files changed, 114 insertions(+), 39 deletions(-) diff --git a/crates/ml-backtesting/cuda/pnl_track.cu b/crates/ml-backtesting/cuda/pnl_track.cu index f32dbab44..9a8427253 100644 --- a/crates/ml-backtesting/cuda/pnl_track.cu +++ b/crates/ml-backtesting/cuda/pnl_track.cu @@ -103,3 +103,51 @@ extern "C" __global__ void pnl_track_step( // prev != 0 AND now != 0 (scale-in or partial close): leave entry // scratch in place. Multi-fill averaging is a v2 refinement. } + +// P3: GPU-side snapshot of pre-submit Pos state for close detection. +// Replaces 3 host memcpy_dtoh calls (snapshot_realized_pnl / position_lots / +// open_horizon_mask). One block per backtest, single-writer thread. +extern "C" __global__ void snapshot_pos_state( + const Pos* __restrict__ positions, // [n_backtests] + int* __restrict__ out_pos_lots, // [n_backtests] + float* __restrict__ out_realized_pnl, // [n_backtests] + unsigned int* __restrict__ out_open_horizon_mask, // [n_backtests] + int n_backtests +) { + int b = blockIdx.x; + if (b >= n_backtests || threadIdx.x != 0) return; + out_pos_lots[b] = positions[b].position_lots; + out_realized_pnl[b] = positions[b].realized_pnl; + out_open_horizon_mask[b] = positions[b].open_horizon_mask; +} + +// P3: GPU-side close-transition detection. For each backtest, compare +// the snapshotted pre-submit pos vs current pos. If prev != 0 AND now == 0 +// → close: write the prev open_horizon_mask + per-lot realised_return. +// Otherwise: write zeros. +// +// Replaces the host loop at sim.rs's Step 5 that called read_pos per +// close-eligible backtest. At n_parallel=140 that loop was ~140 +// memcpy_dtoh per decision; now it's one kernel. +extern "C" __global__ void detect_close_transitions_batched( + const Pos* __restrict__ positions, // [n_backtests] + const int* __restrict__ prev_pos_lots, // [n_backtests] + const float* __restrict__ prev_realized_pnl, // [n_backtests] + const unsigned int* __restrict__ prev_open_horizon_mask, // [n_backtests] + unsigned int* __restrict__ closed_horizon_mask_out, // [n_backtests] + float* __restrict__ realised_return_out, // [n_backtests] + int n_backtests +) { + int b = blockIdx.x; + if (b >= n_backtests || threadIdx.x != 0) return; + const int prev_lots = prev_pos_lots[b]; + if (prev_lots == 0 || positions[b].position_lots != 0) { + closed_horizon_mask_out[b] = 0u; + realised_return_out[b] = 0.0f; + return; + } + const float abs_size = (float)((prev_lots < 0) ? -prev_lots : prev_lots); + const float pnl_delta = positions[b].realized_pnl - prev_realized_pnl[b]; + closed_horizon_mask_out[b] = prev_open_horizon_mask[b]; + realised_return_out[b] = (abs_size > 0.0f) ? (pnl_delta / abs_size) : 0.0f; +} diff --git a/crates/ml-backtesting/src/sim/mod.rs b/crates/ml-backtesting/src/sim/mod.rs index e057d8bb2..914fd3480 100644 --- a/crates/ml-backtesting/src/sim/mod.rs +++ b/crates/ml-backtesting/src/sim/mod.rs @@ -47,6 +47,10 @@ pub struct LobSimCuda { /// P2: GPU-side seed of in-flight Limit slots from market_targets[b]. /// Replaces the host roundtrip loop in dispatch_latent_market_orders. seed_inflight_limits_fn: cudarc::driver::CudaFunction, + /// P3: GPU-side snapshot of pre-submit Pos state + close-transition detection. + /// Replace the host snapshot loop + host close-detect loop at the latency path. + snapshot_pos_state_fn: cudarc::driver::CudaFunction, + detect_close_fn: cudarc::driver::CudaFunction, seed_stop_fn: cudarc::driver::CudaFunction, books_d: CudaSlice, @@ -89,6 +93,11 @@ pub struct LobSimCuda { latency_ns_d: CudaSlice, // [n_backtests] — used by P2 kernel kelly_frac_floor_d: CudaSlice, // [n_backtests] sharpe_weight_floor_d: CudaSlice, // [n_backtests] + + // P3: pre-submit Pos snapshot buffers for GPU close-transition detection. + prev_pos_lots_d: CudaSlice, // [n_backtests] + prev_realized_pnl_d: CudaSlice, // [n_backtests] + prev_open_horizon_mask_d: CudaSlice, // [n_backtests] } impl LobSimCuda { @@ -115,6 +124,12 @@ impl LobSimCuda { let pnl_track_fn = pnl_track_module .load_function("pnl_track_step") .context("load pnl_track_step")?; + let snapshot_pos_state_fn = pnl_track_module + .load_function("snapshot_pos_state") + .context("load snapshot_pos_state")?; + let detect_close_fn = pnl_track_module + .load_function("detect_close_transitions_batched") + .context("load detect_close_transitions_batched")?; let decision_module = ctx .load_cubin(DECISION_POLICY_CUBIN.to_vec()) .context("load decision_policy cubin")?; @@ -221,6 +236,14 @@ impl LobSimCuda { let sharpe_weight_floor_d = stream.alloc_zeros::(n_backtests) .context("alloc sharpe_weight_floor_d")?; + // P3: pre-submit Pos snapshot buffers for GPU close-transition detection. + let prev_pos_lots_d = stream.alloc_zeros::(n_backtests) + .context("alloc prev_pos_lots_d")?; + let prev_realized_pnl_d = stream.alloc_zeros::(n_backtests) + .context("alloc prev_realized_pnl_d")?; + let prev_open_horizon_mask_d = stream.alloc_zeros::(n_backtests) + .context("alloc prev_open_horizon_mask_d")?; + Ok(Self { n_backtests, stream, @@ -262,6 +285,11 @@ impl LobSimCuda { latency_ns_d, kelly_frac_floor_d, sharpe_weight_floor_d, + prev_pos_lots_d, + prev_realized_pnl_d, + prev_open_horizon_mask_d, + snapshot_pos_state_fn, + detect_close_fn, }) } @@ -646,13 +674,18 @@ impl LobSimCuda { // tiny host-loop here (cold path, runs once per decision; <50 µs). self.merge_open_mask_into_pos()?; - // Step 2: snapshot PRE-submit state for close-detection. - // We need pre-submit position to detect close transitions; - // realised P&L delta requires pre and post (the post comes - // from read_pos after step 4 below). - let prev_pnl_per_block = self.snapshot_realized_pnl()?; - let prev_pos_per_block = self.snapshot_position_lots()?; - let prev_mask_per_block = self.snapshot_open_horizon_mask()?; + // Step 2: P3 — GPU-side snapshot of pre-submit Pos state for + // close-detection. Replaces the host-side snapshot trio. + let mut launch = self.stream.launch_builder(&self.snapshot_pos_state_fn); + unsafe { + launch + .arg(&self.pos_d) + .arg(&mut self.prev_pos_lots_d) + .arg(&mut self.prev_realized_pnl_d) + .arg(&mut self.prev_open_horizon_mask_d) + .arg(&n) + .launch(cfg)?; + } // Step 3: always dispatch through the latency path. When a backtest's // latency_ns == 0 the in-flight slot's arrival_ts equals current_ts @@ -677,40 +710,34 @@ impl LobSimCuda { } self.stream.synchronize()?; - // Step 5: detect which backtests had a close transition, compute per-lot - // realised_return, dispatch isv_kelly_update_on_close. - let mut closed_masks = vec![0u32; self.n_backtests]; - let mut realised_returns = vec![0.0f32; self.n_backtests]; - for b in 0..self.n_backtests { - // A close occurred if prev_pos_lots was non-zero and now is zero. - if prev_pos_per_block[b] != 0 { - let now_pos = self.read_pos(b)?; - if now_pos.position_lots == 0 { - let abs_size = prev_pos_per_block[b].unsigned_abs() as f32; - let pnl_delta = now_pos.realized_pnl - prev_pnl_per_block[b]; - let ret_per_lot = if abs_size > 0.0 { pnl_delta / abs_size } else { 0.0 }; - closed_masks[b] = prev_mask_per_block[b]; - realised_returns[b] = ret_per_lot; - } - } + // Step 5: P3 — GPU-side close-transition detection + on-device + // dispatch of isv_kelly_update_on_close. No host roundtrip needed; + // kernel writes closed_horizon_mask_d + realised_return_d directly. + let mut launch = self.stream.launch_builder(&self.detect_close_fn); + unsafe { + launch + .arg(&self.pos_d) + .arg(&self.prev_pos_lots_d) + .arg(&self.prev_realized_pnl_d) + .arg(&self.prev_open_horizon_mask_d) + .arg(&mut self.closed_horizon_mask_d) + .arg(&mut self.realised_return_d) + .arg(&n) + .launch(cfg)?; } - let any_close = closed_masks.iter().any(|&m| m != 0); - if any_close { - self.stream - .memcpy_htod(closed_masks.as_slice(), &mut self.closed_horizon_mask_d)?; - self.stream - .memcpy_htod(realised_returns.as_slice(), &mut self.realised_return_d)?; - let mut launch = self.stream.launch_builder(&self.isv_kelly_update_fn); - unsafe { - launch - .arg(&mut self.isv_kelly_d) - .arg(&self.closed_horizon_mask_d) - .arg(&self.realised_return_d) - .arg(&n) - .launch(cfg)?; - } - self.stream.synchronize()?; + // isv_kelly_update_on_close always launches; kernel skips backtests + // whose mask is 0 internally. Cheaper than a host any_close branch + // that would require a memcpy_dtoh. + let mut launch = self.stream.launch_builder(&self.isv_kelly_update_fn); + unsafe { + launch + .arg(&mut self.isv_kelly_d) + .arg(&self.closed_horizon_mask_d) + .arg(&self.realised_return_d) + .arg(&n) + .launch(cfg)?; } + self.stream.synchronize()?; Ok(()) }