feat(ml-backtesting): detect_close_transitions_batched kernel (P3)

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) <noreply@anthropic.com>

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;
+}

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<f32>,
@@ -89,6 +93,11 @@ pub struct LobSimCuda {
     latency_ns_d:               CudaSlice<u32>,    // [n_backtests] — used by P2 kernel
     kelly_frac_floor_d:         CudaSlice<f32>,    // [n_backtests]
     sharpe_weight_floor_d:      CudaSlice<f32>,    // [n_backtests]
+
+    // P3: pre-submit Pos snapshot buffers for GPU close-transition detection.
+    prev_pos_lots_d:            CudaSlice<i32>,    // [n_backtests]
+    prev_realized_pnl_d:        CudaSlice<f32>,    // [n_backtests]
+    prev_open_horizon_mask_d:   CudaSlice<u32>,    // [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::<f32>(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::<i32>(n_backtests)
+            .context("alloc prev_pos_lots_d")?;
+        let prev_realized_pnl_d     = stream.alloc_zeros::<f32>(n_backtests)
+            .context("alloc prev_realized_pnl_d")?;
+        let prev_open_horizon_mask_d = stream.alloc_zeros::<u32>(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.
+        // Step 2: P3 — GPU-side snapshot of pre-submit Pos state for
-        // We need pre-submit position to detect close transitions;
+        // close-detection. Replaces the host-side snapshot trio.
-        // realised P&L delta requires pre and post (the post comes
+        let mut launch = self.stream.launch_builder(&self.snapshot_pos_state_fn);
-        // from read_pos after step 4 below).
+        unsafe {
-        let prev_pnl_per_block = self.snapshot_realized_pnl()?;
+            launch
-        let prev_pos_per_block = self.snapshot_position_lots()?;
+                .arg(&self.pos_d)
-        let prev_mask_per_block = self.snapshot_open_horizon_mask()?;
+                .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
+        // Step 5: P3 — GPU-side close-transition detection + on-device
-        // realised_return, dispatch isv_kelly_update_on_close.
+        // dispatch of isv_kelly_update_on_close. No host roundtrip needed;
-        let mut closed_masks = vec![0u32; self.n_backtests];
+        // kernel writes closed_horizon_mask_d + realised_return_d directly.
-        let mut realised_returns = vec![0.0f32; self.n_backtests];
+        let mut launch = self.stream.launch_builder(&self.detect_close_fn);
-        for b in 0..self.n_backtests {
+        unsafe {
-            // A close occurred if prev_pos_lots was non-zero and now is zero.
+            launch
-            if prev_pos_per_block[b] != 0 {
+                .arg(&self.pos_d)
-                let now_pos = self.read_pos(b)?;
+                .arg(&self.prev_pos_lots_d)
-                if now_pos.position_lots == 0 {
+                .arg(&self.prev_realized_pnl_d)
-                    let abs_size = prev_pos_per_block[b].unsigned_abs() as f32;
+                .arg(&self.prev_open_horizon_mask_d)
-                    let pnl_delta = now_pos.realized_pnl - prev_pnl_per_block[b];
+                .arg(&mut self.closed_horizon_mask_d)
-                    let ret_per_lot = if abs_size > 0.0 { pnl_delta / abs_size } else { 0.0 };
+                .arg(&mut self.realised_return_d)
-                    closed_masks[b] = prev_mask_per_block[b];
+                .arg(&n)
-                    realised_returns[b] = ret_per_lot;
+                .launch(cfg)?;
-                }
-            }
         }
-        let any_close = closed_masks.iter().any(|&m| m != 0);
+        // isv_kelly_update_on_close always launches; kernel skips backtests
-        if any_close {
+        // whose mask is 0 internally. Cheaper than a host any_close branch
-            self.stream
+        // that would require a memcpy_dtoh.
-                .memcpy_htod(closed_masks.as_slice(), &mut self.closed_horizon_mask_d)?;
+        let mut launch = self.stream.launch_builder(&self.isv_kelly_update_fn);
-            self.stream
+        unsafe {
-                .memcpy_htod(realised_returns.as_slice(), &mut self.realised_return_d)?;
+            launch
-            let mut launch = self.stream.launch_builder(&self.isv_kelly_update_fn);
+                .arg(&mut self.isv_kelly_d)
-            unsafe {
+                .arg(&self.closed_horizon_mask_d)
-                launch
+                .arg(&self.realised_return_d)
-                    .arg(&mut self.isv_kelly_d)
+                .arg(&n)
-                    .arg(&self.closed_horizon_mask_d)
+                .launch(cfg)?;
-                    .arg(&self.realised_return_d)
-                    .arg(&n)
-                    .launch(cfg)?;
-            }
-            self.stream.synchronize()?;
         }
+        self.stream.synchronize()?;
 
         Ok(())
     }