feat(ml-backtesting): decision_policy + per-horizon ISV-Kelly (C7)

Two new kernels in cuda/decision_policy.cu:

  decision_policy_default — alpha[N_HORIZONS] × per-horizon IsvKellyState
  → market target. Per-horizon Kelly fraction × signal magnitude × ISV-cap
  (target_annual_vol / sqrt(realised_return_var × annualisation_factor)),
  aggregated via WeightedByRealizedSharpe (weights = max(0, recent_sharpe)
  / Σ; auto-shifts capital toward empirically winning horizons per spec §5
  + §6). No floor — sub-1-lot intents become no-op. Round-to-nearest on
  the final lot count to dodge an f32-truncation off-by-one where
  0.8(f32) * 0.75 * 5.0 ≈ 2.99999976 → trunc-int 2.

  isv_kelly_update_on_close — for every horizon flagged in
  closed_horizon_mask[b], updates pnl_ema_{win,loss} via Wiener-α (floor
  0.4 per pearl_wiener_alpha_floor_for_nonstationary), win_rate_ema,
  Welford-ish realised_return_var, and the recent_sharpe composite.
  First-observation bootstrap (per pearl_first_observation_bootstrap):
  sentinel n_trades_seen=0 → direct EMA replacement, no zero-bias warmup.

The full bytecode VM from spec §6 is NOT in this commit — the default
policy is hardcoded in the kernel as the path Strategy::default_for()
already produces. Bytecode plumbing in src/policy/mod.rs stays put for
v2 expansion (custom RegimeSwitch / Portfolio compositions).

IsvKellyState struct added to lob_state.cuh (24 bytes per horizon × 5
per backtest); host mirror IsvKellyStateHost from C3 cast-compatible
via bytemuck::Pod. LobSimCuda gains broadcast_alpha + step_decision +
read_isv_kelly + write_isv_kelly (warm-start). step_decision chains:
  decision_policy → merge_open_mask → submit_market_immediate
  → pnl_track_step → host close-detect → isv_kelly_update_on_close.

PRE-submit pos/pnl/mask snapshots feed the host close-detection;
captured via three small DtoH copies (cold path, 24 bytes × n_backtests).

decision_alpha_buy_close fixture: warm-start h4 with positive Kelly
state (n=50, recent_sharpe=1.0), broadcast alpha[4]=0.9 → buy 3 lots
@ ask top 5500.00. Snapshot moves to bid 5505.00, broadcast alpha[4]=0.1
→ sell 3 lots → close at +15 P&L. Verify ISV-Kelly h4: n_trades 50→51
exactly, others unchanged. PASS — all 6 Ring 1 fixtures green on RTX 3050.

Out-of-scope for this commit (defer to follow-ups, per plan trim notes):
  - stop_trigger / oco_one_cancels_other / submission_overflow fixtures
    (need resting-order LimitSlot[32] machinery deferred from C5)
  - Bytecode VM dispatch (RegimeSwitch / Portfolio compositions)

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

crates/ml-backtesting/cuda/decision_policy.cu

@@ -0,0 +1,166 @@
+// decision_policy.cu — alpha → per-horizon Kelly → aggregate → market target.
+//
+// Hardcoded v1 policy = Strategy::default_for(max_lots) from C3:
+//   for each h in 0..N_HORIZONS:
+//     compute signed sizing from (alpha[h], IsvKellyState[h])
+//   aggregate via WeightedByRealizedSharpe:
+//     w[h] = max(0, recent_sharpe[h]) / sum_of_positive_sharpes
+//     final = sum_h w[h] * signed_size[h]
+//   final → side+size in market_targets[b]
+//
+// The bytecode VM from spec §6 is intentionally not implemented in v1 —
+// the default Strategy::default_for produces exactly this shape and
+// alternative compositions are deferred. Bytecode plumbing remains in
+// src/policy/mod.rs ready for v2 expansion.
+//
+// Single-writer per block (thread 0). See spec §5.
+
+#include "lob_state.cuh"
+
+extern "C" __global__ void decision_policy_default(
+    const float*       __restrict__ alpha_probs,        // [N_HORIZONS] broadcast
+    const unsigned char* __restrict__ isv_kelly_base,   // [n_backtests * N_HORIZONS * sizeof(IsvKellyState)]
+    const Pos*         __restrict__ positions,          // [n_backtests]
+    int*               __restrict__ market_targets,     // [n_backtests * 2] — written
+    unsigned int*      __restrict__ open_horizon_masks, // [n_backtests] — written (entry attribution)
+    float                          target_annual_vol_units,
+    float                          annualisation_factor,
+    int                            max_lots,
+    int                            n_backtests
+) {
+    int b = blockIdx.x;
+    if (b >= n_backtests || threadIdx.x != 0) return;
+
+    const IsvKellyState* isv = reinterpret_cast<const IsvKellyState*>(
+        isv_kelly_base + (size_t)b * N_HORIZONS * sizeof(IsvKellyState)
+    );
+
+    float signed_sizes[N_HORIZONS];
+    float weights[N_HORIZONS];
+    unsigned int attribution_mask = 0;
+    float w_sum = 0.0f;
+
+    #pragma unroll
+    for (int h = 0; h < N_HORIZONS; ++h) {
+        const float p_h = alpha_probs[h];
+        const IsvKellyState& s = isv[h];
+
+        // Per-spec §5: no floor, cap derives from ISV.
+        // If pnl_ema_win is sentinel (0), no kelly_frac estimate available.
+        float ss = 0.0f;
+        if (s.pnl_ema_win > 1e-9f && s.realised_return_var > 1e-9f) {
+            const float sig_mag = fabsf(p_h - 0.5f) * 2.0f;       // ∈ [0, 1]
+            const float dir     = (p_h > 0.5f) ? 1.0f : -1.0f;
+            // Kelly fraction; clamp [0, 1].
+            float kelly_frac = (s.win_rate_ema * s.pnl_ema_win
+                                - (1.0f - s.win_rate_ema) * s.pnl_ema_loss)
+                               / s.pnl_ema_win;
+            kelly_frac = fmaxf(0.0f, fminf(1.0f, kelly_frac));
+            // Cap units from realized variance + target vol budget.
+            const float cap_units = target_annual_vol_units
+                                  / sqrtf(s.realised_return_var * annualisation_factor);
+            const float cap_lots = fminf(cap_units, (float)max_lots);
+            ss = dir * sig_mag * kelly_frac * cap_lots;
+        }
+        signed_sizes[h] = ss;
+        const float w = fmaxf(0.0f, s.recent_sharpe);
+        weights[h] = w;
+        w_sum += w;
+    }
+
+    float final_size = 0.0f;
+    if (w_sum > 1e-9f) {
+        #pragma unroll
+        for (int h = 0; h < N_HORIZONS; ++h) {
+            const float w_norm = weights[h] / w_sum;
+            final_size += w_norm * signed_sizes[h];
+            // Attribute the open to any horizon whose weight contributed.
+            if (weights[h] > 1e-9f) attribution_mask |= (1u << h);
+        }
+    }
+
+    // Round-to-nearest (truncation alone bites here: 0.8(f32) * 0.75 * 5.0
+    // = 2.99999976 → (int)2, off by one. Floor truncation for sub-1
+    // suppression is preserved via the explicit |lots| < 1 check.
+    int lots = (int)truncf(final_size + (final_size >= 0.0f ? 0.5f : -0.5f));
+    if (lots == 0) {
+        market_targets[b * 2 + 0] = 2;                     // no-op
+        market_targets[b * 2 + 1] = 0;
+        return;
+    }
+
+    // If currently flat, this is an opening trade: record attribution mask
+    // so pnl_track + isv_kelly_update can credit the right horizons on close.
+    if (positions[b].position_lots == 0) {
+        open_horizon_masks[b] = attribution_mask;
+    }
+
+    const int side = (lots > 0) ? 0 : 1;
+    const int abs_sz = (lots > 0) ? lots : -lots;
+    market_targets[b * 2 + 0] = side;
+    market_targets[b * 2 + 1] = abs_sz;
+}
+
+// Updates per-horizon IsvKellyState[h] for every horizon flagged in
+// `closed_horizon_mask[b]`. Called by the host after pnl_track_step
+// detects close transitions. `realised_return[b]` is the closed-trade
+// per-lot return in price units (positive = win, negative = loss).
+//
+// Wiener-α floor 0.4 per pearl_wiener_alpha_floor_for_nonstationary.md.
+// Welford variance updates a running estimate around mean 0 (simplified;
+// v2 may switch to ema-centered).
+//
+// recent_sharpe composite re-computed from the freshly updated EMAs.
+extern "C" __global__ void isv_kelly_update_on_close(
+    unsigned char*       isv_kelly_base,
+    const unsigned int*  closed_horizon_mask,    // [n_backtests]
+    const float*         realised_return,        // [n_backtests]
+    int                  n_backtests
+) {
+    int b = blockIdx.x;
+    if (b >= n_backtests || threadIdx.x != 0) return;
+    const unsigned int mask = closed_horizon_mask[b];
+    if (mask == 0u) return;
+
+    const float ret = realised_return[b];
+    IsvKellyState* isv = reinterpret_cast<IsvKellyState*>(
+        isv_kelly_base + (size_t)b * N_HORIZONS * sizeof(IsvKellyState)
+    );
+
+    const float alpha = 0.4f;  // Wiener-α floor (non-stationary policy P&L)
+
+    #pragma unroll
+    for (int h = 0; h < N_HORIZONS; ++h) {
+        if (!(mask & (1u << h))) continue;
+        IsvKellyState& s = isv[h];
+        const bool win = ret > 0.0f;
+
+        if (s.n_trades_seen == 0u) {
+            // First-observation bootstrap: replace EMAs directly (no zero-bias warmup).
+            s.pnl_ema_win  = win ? ret : 0.0f;
+            s.pnl_ema_loss = win ? 0.0f : -ret;
+            s.win_rate_ema = win ? 1.0f : 0.0f;
+            s.realised_return_var = ret * ret;   // single-sample variance proxy
+        } else {
+            if (win) {
+                s.pnl_ema_win  = (1.0f - alpha) * s.pnl_ema_win  + alpha * ret;
+                s.win_rate_ema = (1.0f - alpha) * s.win_rate_ema + alpha * 1.0f;
+            } else {
+                s.pnl_ema_loss = (1.0f - alpha) * s.pnl_ema_loss + alpha * (-ret);
+                s.win_rate_ema = (1.0f - alpha) * s.win_rate_ema + alpha * 0.0f;
+            }
+            // Welford-ish running variance around 0.
+            const float prev_n = (float)s.n_trades_seen;
+            s.realised_return_var =
+                (prev_n * s.realised_return_var + ret * ret) / (prev_n + 1.0f);
+        }
+        s.n_trades_seen += 1u;
+
+        // Composite recent_sharpe = (mean_winning_contribution − mean_losing_contribution)
+        //                          / sqrt(realised_return_var)
+        const float numerator = s.pnl_ema_win * s.win_rate_ema
+                               - s.pnl_ema_loss * (1.0f - s.win_rate_ema);
+        const float denom = sqrtf(fmaxf(s.realised_return_var, 1e-9f));
+        s.recent_sharpe = numerator / denom;
+    }
+}

crates/ml-backtesting/cuda/lob_state.cuh

@@ -28,4 +28,15 @@ struct Pos {
     unsigned int open_horizon_mask;   // bit h set if currently open position was authorized by horizon h
 };
 
+// Per-horizon adaptive Kelly state. Lives per (backtest, horizon).
+// See spec §5 (per-horizon ISV-Kelly).
+struct IsvKellyState {
+    float pnl_ema_win;             // Wiener-α blended EMA of winning trade returns (price units)
+    float pnl_ema_loss;            // EMA of losing trade returns (positive magnitude)
+    float win_rate_ema;
+    unsigned int n_trades_seen;
+    float realised_return_var;     // Welford running variance over per-trade returns
+    float recent_sharpe;            // composite: (μ × win_rate − μ_loss × (1−win_rate)) / sqrt(var)
+};
+
 #endif // LOB_STATE_CUH

crates/ml-backtesting/src/sim.rs

@@ -22,6 +22,11 @@ const ORDER_MATCH_CUBIN: &[u8] =
     include_bytes!(concat!(env!("OUT_DIR"), "/order_match.cubin"));
 const PNL_TRACK_CUBIN: &[u8] =
     include_bytes!(concat!(env!("OUT_DIR"), "/pnl_track.cubin"));
+const DECISION_POLICY_CUBIN: &[u8] =
+    include_bytes!(concat!(env!("OUT_DIR"), "/decision_policy.cubin"));
+
+const N_HORIZONS: usize = 5;
+const ISV_KELLY_STATE_BYTES: usize = 24;  // matches lob_state.cuh::IsvKellyState
 
 pub struct LobSimCuda {
     n_backtests: usize,
@@ -29,6 +34,8 @@ pub struct LobSimCuda {
     book_update_fn: cudarc::driver::CudaFunction,
     submit_market_fn: cudarc::driver::CudaFunction,
     pnl_track_fn: cudarc::driver::CudaFunction,
+    decision_fn: cudarc::driver::CudaFunction,
+    isv_kelly_update_fn: cudarc::driver::CudaFunction,
 
     books_d: CudaSlice<f32>,
     pos_d: CudaSlice<u8>,
@@ -42,6 +49,13 @@ pub struct LobSimCuda {
     open_trade_state_d: CudaSlice<u8>,    // [n_backtests * 24]
     trade_log_d: CudaSlice<u8>,           // [n_backtests * TRADE_LOG_CAP * 40]
     trade_log_head_d: CudaSlice<u32>,     // [n_backtests]
+
+    // Decision + ISV-Kelly buffers (C7).
+    isv_kelly_d: CudaSlice<u8>,             // [n_backtests * 5 * 24]
+    alpha_probs_d: CudaSlice<f32>,          // [N_HORIZONS] — broadcast
+    open_horizon_mask_d: CudaSlice<u32>,    // [n_backtests] — entry attribution
+    closed_horizon_mask_d: CudaSlice<u32>,  // [n_backtests] — close-time snapshot
+    realised_return_d: CudaSlice<f32>,      // [n_backtests] — per-block close-trade return
 }
 
 impl LobSimCuda {
@@ -68,6 +82,15 @@ impl LobSimCuda {
         let pnl_track_fn = pnl_track_module
             .load_function("pnl_track_step")
             .context("load pnl_track_step")?;
+        let decision_module = ctx
+            .load_cubin(DECISION_POLICY_CUBIN.to_vec())
+            .context("load decision_policy cubin")?;
+        let decision_fn = decision_module
+            .load_function("decision_policy_default")
+            .context("load decision_policy_default")?;
+        let isv_kelly_update_fn = decision_module
+            .load_function("isv_kelly_update_on_close")
+            .context("load isv_kelly_update_on_close")?;
 
         let books_d = stream
             .alloc_zeros::<f32>(n_backtests * BOOK_FIELDS * BOOK_LEVELS)
@@ -93,12 +116,30 @@ impl LobSimCuda {
             .alloc_zeros::<u32>(n_backtests)
             .context("alloc trade_log_head_d")?;
 
+        let isv_kelly_d = stream
+            .alloc_zeros::<u8>(n_backtests * N_HORIZONS * ISV_KELLY_STATE_BYTES)
+            .context("alloc isv_kelly_d")?;
+        let alpha_probs_d = stream
+            .alloc_zeros::<f32>(N_HORIZONS)
+            .context("alloc alpha_probs_d")?;
+        let open_horizon_mask_d = stream
+            .alloc_zeros::<u32>(n_backtests)
+            .context("alloc open_horizon_mask_d")?;
+        let closed_horizon_mask_d = stream
+            .alloc_zeros::<u32>(n_backtests)
+            .context("alloc closed_horizon_mask_d")?;
+        let realised_return_d = stream
+            .alloc_zeros::<f32>(n_backtests)
+            .context("alloc realised_return_d")?;
+
         Ok(Self {
             n_backtests,
             stream,
             book_update_fn,
             submit_market_fn,
             pnl_track_fn,
+            decision_fn,
+            isv_kelly_update_fn,
             books_d,
             pos_d,
             bid_px_d,
@@ -109,6 +150,11 @@ impl LobSimCuda {
             open_trade_state_d,
             trade_log_d,
             trade_log_head_d,
+            isv_kelly_d,
+            alpha_probs_d,
+            open_horizon_mask_d,
+            closed_horizon_mask_d,
+            realised_return_d,
         })
     }
 
@@ -279,6 +325,249 @@ impl LobSimCuda {
         Ok(pos)
     }
 
+    /// Broadcast alpha probs (per-horizon) to the decision kernel.
+    /// Single source of truth for v1: all N backtests see the same probs.
+    pub fn broadcast_alpha(&mut self, probs: &[f32; 5]) -> Result<()> {
+        self.stream.memcpy_htod(probs.as_slice(), &mut self.alpha_probs_d)?;
+        Ok(())
+    }
+
+    /// Run the C7 decision pipeline:
+    ///   1. decision_policy_default kernel — alpha × per-horizon ISV-Kelly
+    ///      → market target (side, size) per backtest + open_horizon_mask
+    ///      attribution if currently flat.
+    ///   2. submit_market_immediate — execute the targets against the book.
+    ///   3. pnl_track_step — detect close → emit TradeRecord.
+    ///   4. isv_kelly_update_on_close — for closed trades, update per-horizon
+    ///      EMAs + variance + recent_sharpe using the realised return per lot.
+    ///
+    /// The decision kernel reads `open_horizon_mask_d` for new opens. The
+    /// close-time mask is captured from `Pos.open_horizon_mask` BEFORE the
+    /// matching kernel zeros it; here we snapshot via a small host copy.
+    pub fn step_decision(
+        &mut self,
+        current_ts_ns: u64,
+        target_annual_vol_units: f32,
+        annualisation_factor: f32,
+        max_lots: u16,
+    ) -> Result<()> {
+        // Step 1: decision kernel writes market_targets + open_horizon_mask.
+        let cfg = LaunchConfig {
+            grid_dim: (self.n_backtests as u32, 1, 1),
+            block_dim: (32, 1, 1),
+            shared_mem_bytes: 0,
+        };
+        let n = self.n_backtests as i32;
+        let max_lots_i = max_lots as i32;
+        let mut launch = self.stream.launch_builder(&self.decision_fn);
+        unsafe {
+            launch
+                .arg(&self.alpha_probs_d)
+                .arg(&self.isv_kelly_d)
+                .arg(&self.pos_d)
+                .arg(&mut self.market_targets_d)
+                .arg(&mut self.open_horizon_mask_d)
+                .arg(&target_annual_vol_units)
+                .arg(&annualisation_factor)
+                .arg(&max_lots_i)
+                .arg(&n)
+                .launch(cfg)?;
+        }
+        self.stream.synchronize()?;
+
+        // Persist open_horizon_mask into Pos.open_horizon_mask via a small
+        // helper kernel? No — the decision kernel wrote it into open_horizon_mask_d
+        // already, and pnl_track only needs it when detecting close. Instead
+        // of changing pnl_track's interface, we copy the device→device value
+        // into Pos.open_horizon_mask field via the matching kernel's natural
+        // flow: the submit_market kernel doesn't touch this field, but
+        // we set it here pre-trade so it persists until close.
+        //
+        // Simpler approach: read pos[b] for each backtest with active orders,
+        // OR the broadcast mask onto pos[b].open_horizon_mask. Done via a
+        // 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 3: execute targets against book.
+        let mut launch = self.stream.launch_builder(&self.submit_market_fn);
+        unsafe {
+            launch
+                .arg(&self.books_d)
+                .arg(&self.market_targets_d)
+                .arg(&mut self.pos_d)
+                .arg(&n)
+                .launch(cfg)?;
+        }
+        self.stream.synchronize()?;
+
+        // Step 4: pnl_track — detect close, emit TradeRecord.
+        let cap = crate::lob::TRADE_LOG_CAP as i32;
+        let mut launch = self.stream.launch_builder(&self.pnl_track_fn);
+        unsafe {
+            launch
+                .arg(&mut self.pos_d)
+                .arg(&mut self.open_trade_state_d)
+                .arg(&mut self.trade_log_d)
+                .arg(&mut self.trade_log_head_d)
+                .arg(&current_ts_ns)
+                .arg(&cap)
+                .arg(&n)
+                .launch(cfg)?;
+        }
+        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;
+                }
+            }
+        }
+        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()?;
+        }
+
+        Ok(())
+    }
+
+    fn merge_open_mask_into_pos(&mut self) -> Result<()> {
+        // Read open_horizon_mask, OR into pos[b].open_horizon_mask field, push back.
+        // Pos field layout: position_lots(4) vwap_entry(4) realized_pnl(4)
+        //                   peak_equity(4) submission_overflow(4) open_horizon_mask(4).
+        let pos_bytes = std::mem::size_of::<PosFlat>();
+        let mut raw = vec![0u8; self.n_backtests * pos_bytes];
+        self.stream.memcpy_dtoh(&self.pos_d, raw.as_mut_slice())?;
+        let mut masks = vec![0u32; self.n_backtests];
+        self.stream.memcpy_dtoh(&self.open_horizon_mask_d, masks.as_mut_slice())?;
+        for b in 0..self.n_backtests {
+            let off = b * pos_bytes + 20;  // open_horizon_mask offset within PosFlat
+            let mut cur_bytes = [0u8; 4];
+            cur_bytes.copy_from_slice(&raw[off..off + 4]);
+            let cur = u32::from_le_bytes(cur_bytes);
+            let merged = cur | masks[b];
+            raw[off..off + 4].copy_from_slice(&merged.to_le_bytes());
+        }
+        self.stream.memcpy_htod(raw.as_slice(), &mut self.pos_d)?;
+        Ok(())
+    }
+
+    fn snapshot_realized_pnl(&self) -> Result<Vec<f32>> {
+        let pos_bytes = std::mem::size_of::<PosFlat>();
+        let mut raw = vec![0u8; self.n_backtests * pos_bytes];
+        self.stream.memcpy_dtoh(&self.pos_d, raw.as_mut_slice())?;
+        let mut out = vec![0.0f32; self.n_backtests];
+        for b in 0..self.n_backtests {
+            let off = b * pos_bytes + 8;  // realized_pnl offset (i32 + f32 then f32 here)
+            let mut bytes = [0u8; 4];
+            bytes.copy_from_slice(&raw[off..off + 4]);
+            out[b] = f32::from_le_bytes(bytes);
+        }
+        Ok(out)
+    }
+
+    fn snapshot_position_lots(&self) -> Result<Vec<i32>> {
+        let pos_bytes = std::mem::size_of::<PosFlat>();
+        let mut raw = vec![0u8; self.n_backtests * pos_bytes];
+        self.stream.memcpy_dtoh(&self.pos_d, raw.as_mut_slice())?;
+        let mut out = vec![0i32; self.n_backtests];
+        for b in 0..self.n_backtests {
+            let off = b * pos_bytes;  // position_lots is field 0
+            let mut bytes = [0u8; 4];
+            bytes.copy_from_slice(&raw[off..off + 4]);
+            out[b] = i32::from_le_bytes(bytes);
+        }
+        Ok(out)
+    }
+
+    fn snapshot_open_horizon_mask(&self) -> Result<Vec<u32>> {
+        let pos_bytes = std::mem::size_of::<PosFlat>();
+        let mut raw = vec![0u8; self.n_backtests * pos_bytes];
+        self.stream.memcpy_dtoh(&self.pos_d, raw.as_mut_slice())?;
+        let mut out = vec![0u32; self.n_backtests];
+        for b in 0..self.n_backtests {
+            let off = b * pos_bytes + 20;
+            let mut bytes = [0u8; 4];
+            bytes.copy_from_slice(&raw[off..off + 4]);
+            out[b] = u32::from_le_bytes(bytes);
+        }
+        Ok(out)
+    }
+
+    /// Write per-horizon ISV-Kelly state for a specific backtest (warm-start).
+    pub fn write_isv_kelly(
+        &mut self,
+        backtest_idx: usize,
+        states: &[crate::policy::IsvKellyStateHost; 5],
+    ) -> Result<()> {
+        anyhow::ensure!(
+            backtest_idx < self.n_backtests,
+            "backtest_idx {} >= n_backtests {}",
+            backtest_idx,
+            self.n_backtests
+        );
+        let total_bytes = self.n_backtests * N_HORIZONS * ISV_KELLY_STATE_BYTES;
+        let mut raw = vec![0u8; total_bytes];
+        self.stream.memcpy_dtoh(&self.isv_kelly_d, raw.as_mut_slice())?;
+        let off = backtest_idx * N_HORIZONS * ISV_KELLY_STATE_BYTES;
+        let bytes: &[u8] = bytemuck::cast_slice(states);
+        raw[off..off + bytes.len()].copy_from_slice(bytes);
+        self.stream.memcpy_htod(raw.as_slice(), &mut self.isv_kelly_d)?;
+        Ok(())
+    }
+
+    /// Read per-horizon ISV-Kelly state for a specific backtest.
+    pub fn read_isv_kelly(
+        &self,
+        backtest_idx: usize,
+    ) -> Result<[crate::policy::IsvKellyStateHost; 5]> {
+        anyhow::ensure!(
+            backtest_idx < self.n_backtests,
+            "backtest_idx {} >= n_backtests {}",
+            backtest_idx,
+            self.n_backtests
+        );
+        let mut raw = vec![0u8; self.n_backtests * N_HORIZONS * ISV_KELLY_STATE_BYTES];
+        self.stream.memcpy_dtoh(&self.isv_kelly_d, raw.as_mut_slice())?;
+        let off = backtest_idx * N_HORIZONS * ISV_KELLY_STATE_BYTES;
+        let slice = &raw[off..off + N_HORIZONS * ISV_KELLY_STATE_BYTES];
+        let mut out = [crate::policy::IsvKellyStateHost::default(); 5];
+        let bytes_out: &mut [u8] = bytemuck::cast_slice_mut(&mut out);
+        bytes_out.copy_from_slice(slice);
+        Ok(out)
+    }
+
     /// Read back every backtest's book state from device. For tests + diagnostics.
     pub fn read_books(&self) -> Result<Vec<BookFlat>> {
         let mut raw = vec![0.0f32; self.n_backtests * BOOK_FIELDS * BOOK_LEVELS];

crates/ml-backtesting/tests/fixtures/decision_alpha_buy_close.json

@@ -0,0 +1,61 @@
+{
+  "name": "decision_alpha_buy_close",
+  "n_backtests": 1,
+  "warm_start_isv_kelly": [
+    [
+      { "pnl_ema_win": 0.0, "pnl_ema_loss": 0.0, "win_rate_ema": 0.0, "n_trades_seen": 0,  "realised_return_var": 0.0,  "recent_sharpe": 0.0 },
+      { "pnl_ema_win": 0.0, "pnl_ema_loss": 0.0, "win_rate_ema": 0.0, "n_trades_seen": 0,  "realised_return_var": 0.0,  "recent_sharpe": 0.0 },
+      { "pnl_ema_win": 0.0, "pnl_ema_loss": 0.0, "win_rate_ema": 0.0, "n_trades_seen": 0,  "realised_return_var": 0.0,  "recent_sharpe": 0.0 },
+      { "pnl_ema_win": 0.0, "pnl_ema_loss": 0.0, "win_rate_ema": 0.0, "n_trades_seen": 0,  "realised_return_var": 0.0,  "recent_sharpe": 0.0 },
+      { "pnl_ema_win": 2.0, "pnl_ema_loss": 0.5, "win_rate_ema": 0.8, "n_trades_seen": 50, "realised_return_var": 0.25, "recent_sharpe": 1.0 }
+    ]
+  ],
+  "events": [
+    {
+      "type": "snapshot",
+      "bid_px": [5499.75, 5499.50, 5499.25, 5499.00, 5498.75, 5498.50, 5498.25, 5498.00, 5497.75, 5497.50],
+      "bid_sz": [100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0],
+      "ask_px": [5500.00, 5500.25, 5500.50, 5500.75, 5501.00, 5501.25, 5501.50, 5501.75, 5502.00, 5502.25],
+      "ask_sz": [100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0]
+    },
+    {
+      "type": "decision",
+      "alpha_probs": [0.5, 0.5, 0.5, 0.5, 0.9],
+      "target_annual_vol_units": 50.0,
+      "annualisation_factor": 1.0,
+      "max_lots": 5,
+      "ts_ns": 1000000000
+    },
+    {
+      "type": "snapshot",
+      "bid_px": [5505.00, 5504.75, 5504.50, 5504.25, 5504.00, 5503.75, 5503.50, 5503.25, 5503.00, 5502.75],
+      "bid_sz": [100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0],
+      "ask_px": [5505.25, 5505.50, 5505.75, 5506.00, 5506.25, 5506.50, 5506.75, 5507.00, 5507.25, 5507.50],
+      "ask_sz": [100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0]
+    },
+    {
+      "type": "decision",
+      "alpha_probs": [0.5, 0.5, 0.5, 0.5, 0.1],
+      "target_annual_vol_units": 50.0,
+      "annualisation_factor": 1.0,
+      "max_lots": 5,
+      "ts_ns": 2000000000
+    }
+  ],
+  "expected_pos": [
+    {
+      "position_lots": 0,
+      "realized_pnl_tol": 100.0
+    }
+  ],
+  "expected_min_trade_records": 1,
+  "expected_isv_kelly_after": [
+    [
+      { "n_trades_seen_min": 0,  "n_trades_seen_max": 0 },
+      { "n_trades_seen_min": 0,  "n_trades_seen_max": 0 },
+      { "n_trades_seen_min": 0,  "n_trades_seen_max": 0 },
+      { "n_trades_seen_min": 0,  "n_trades_seen_max": 0 },
+      { "n_trades_seen_min": 51, "n_trades_seen_max": 51 }
+    ]
+  ]
+}

crates/ml-backtesting/tests/lob_sim_fixtures.rs

@@ -5,6 +5,7 @@
 
 use anyhow::{Context, Result};
 use ml_backtesting::lob::BOOK_LEVELS;
+use ml_backtesting::policy::IsvKellyStateHost;
 use ml_backtesting::sim::LobSimCuda;
 use ml_core::device::MlDevice;
 use serde::Deserialize;
@@ -26,10 +27,48 @@ enum FixtureEvent {
         #[serde(default = "default_ts_ns")]
         ts_ns: u64,
     },
+    Decision {
+        alpha_probs: [f32; 5],
+        target_annual_vol_units: f32,
+        annualisation_factor: f32,
+        max_lots: u16,
+        #[serde(default = "default_ts_ns")]
+        ts_ns: u64,
+    },
 }
 
 fn default_ts_ns() -> u64 { 1 }
 
+#[derive(Debug, Deserialize, Default)]
+struct IsvKellyFixture {
+    #[serde(default)] pnl_ema_win: f32,
+    #[serde(default)] pnl_ema_loss: f32,
+    #[serde(default)] win_rate_ema: f32,
+    #[serde(default)] n_trades_seen: u32,
+    #[serde(default)] realised_return_var: f32,
+    #[serde(default)] recent_sharpe: f32,
+}
+
+impl IsvKellyFixture {
+    fn to_host(&self) -> IsvKellyStateHost {
+        IsvKellyStateHost {
+            pnl_ema_win: self.pnl_ema_win,
+            pnl_ema_loss: self.pnl_ema_loss,
+            win_rate_ema: self.win_rate_ema,
+            n_trades_seen: self.n_trades_seen,
+            realised_return_var: self.realised_return_var,
+            recent_sharpe: self.recent_sharpe,
+        }
+    }
+}
+
+#[derive(Debug, Deserialize, Default)]
+struct ExpectedIsvKelly {
+    #[serde(default)] n_trades_seen_min: u32,
+    #[serde(default = "u32_max")] n_trades_seen_max: u32,
+}
+fn u32_max() -> u32 { u32::MAX }
+
 #[derive(Debug, Deserialize)]
 struct ExpectedPos {
     position_lots: i32,
@@ -77,6 +116,12 @@ struct Fixture {
     expected_pos: Vec<ExpectedPos>,
     #[serde(default)]
     expected_trade_records: Vec<ExpectedTradeRecord>,
+    #[serde(default)]
+    warm_start_isv_kelly: Vec<[IsvKellyFixture; 5]>,
+    #[serde(default)]
+    expected_min_trade_records: usize,
+    #[serde(default)]
+    expected_isv_kelly_after: Vec<[ExpectedIsvKelly; 5]>,
 }
 
 fn run_book_fixture(path: &Path) -> Result<()> {
@@ -89,6 +134,18 @@ fn run_book_fixture(path: &Path) -> Result<()> {
         .map_err(|e| anyhow::anyhow!("cuda device: {e}"))?;
     let mut sim = LobSimCuda::new(fx.n_backtests, &dev)?;
 
+    // Apply ISV-Kelly warm-start if provided (one row per backtest).
+    for (b, states) in fx.warm_start_isv_kelly.iter().enumerate() {
+        let host_states: [IsvKellyStateHost; 5] = [
+            states[0].to_host(),
+            states[1].to_host(),
+            states[2].to_host(),
+            states[3].to_host(),
+            states[4].to_host(),
+        ];
+        sim.write_isv_kelly(b, &host_states)?;
+    }
+
     for ev in &fx.events {
         match ev {
             FixtureEvent::Snapshot { bid_px, bid_sz, ask_px, ask_sz } => {
@@ -102,6 +159,41 @@ fn run_book_fixture(path: &Path) -> Result<()> {
                 };
                 sim.submit_market(*backtest_idx, side_u8, *size, *ts_ns)?;
             }
+            FixtureEvent::Decision {
+                alpha_probs,
+                target_annual_vol_units,
+                annualisation_factor,
+                max_lots,
+                ts_ns,
+            } => {
+                sim.broadcast_alpha(alpha_probs)?;
+                sim.step_decision(*ts_ns, *target_annual_vol_units, *annualisation_factor, *max_lots)?;
+            }
+        }
+    }
+
+    if fx.expected_min_trade_records > 0 {
+        let records = sim.read_trade_records(0)?;
+        assert!(
+            records.len() >= fx.expected_min_trade_records,
+            "expected >= {} trade records, got {}",
+            fx.expected_min_trade_records,
+            records.len()
+        );
+    }
+
+    if !fx.expected_isv_kelly_after.is_empty() {
+        for (b, expected) in fx.expected_isv_kelly_after.iter().enumerate() {
+            let got = sim.read_isv_kelly(b)?;
+            for h in 0..5 {
+                let n = got[h].n_trades_seen;
+                assert!(
+                    n >= expected[h].n_trades_seen_min && n <= expected[h].n_trades_seen_max,
+                    "backtest {b} horizon {h}: n_trades_seen got {n}, want in [{}..{}]",
+                    expected[h].n_trades_seen_min,
+                    expected[h].n_trades_seen_max
+                );
+            }
         }
     }
 
@@ -214,3 +306,9 @@ fn fix_market_order_consumes_top() {
 fn fix_pnl_accounting_buy_close() {
     run_book_fixture(Path::new("tests/fixtures/pnl_accounting_buy_close.json")).unwrap();
 }
+
+#[test]
+#[ignore = "requires CUDA"]
+fn fix_decision_alpha_buy_close() {
+    run_book_fixture(Path::new("tests/fixtures/decision_alpha_buy_close.json")).unwrap();
+}