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

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

Browse Source 
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:
jgrusewski
2026-05-19 17:04:42 +02:00
parent b741f0c5ce
commit 3836e25783
2 changed files with 114 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

48
crates/ml-backtesting/cuda/pnl_track.cu
View File

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

105
crates/ml-backtesting/src/sim/mod.rs
View File

@@ -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(())
}