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>
This commit is contained in:
@@ -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;
|
||||
}
|
||||
|
||||
@@ -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.
|
||||
// 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(())
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user