diff --git a/crates/ml-backtesting/cuda/pnl_track.cu b/crates/ml-backtesting/cuda/pnl_track.cu new file mode 100644 index 000000000..f32dbab44 --- /dev/null +++ b/crates/ml-backtesting/cuda/pnl_track.cu @@ -0,0 +1,105 @@ +// pnl_track.cu — segment_complete detector + TradeRecord emission. +// +// Runs after every matching pass. Compares previous position state +// (persisted in OpenTradeState scratch) against current Pos.position_lots +// to detect transitions: +// prev == 0, now != 0 → entry: record entry context into scratch +// prev != 0, now == 0 → close: emit a TradeRecord with the spread +// between entry and exit P&L delta +// +// Single-writer (thread 0) per block. +// +// OpenTradeState layout (24 bytes per backtest): +// 0..8 entry_ts_ns (u64) +// 8..12 entry_px_x100 (i32 — entry VWAP × 100, fixed-point) +// 12..16 entry_size (i32 — signed, +long -short) +// 16..20 entry_realized_at_open (f32 — pos.realized_pnl when entry happened) +// 20..21 horizon_idx (u8) +// 21..24 padding +// +// TradeRecord layout (40 bytes — MUST match src/order.rs::TradeRecord): +// 0..8 entry_ts_ns +// 8..16 exit_ts_ns +// 16..20 entry_px_ticks (×100) +// 20..24 exit_px_ticks (×100) +// 24..28 size_lots (signed) +// 28..32 fees_usd_fp (placeholder, zero in v1) +// 32..36 realised_pnl_usd_fp (×100; price-units × $50 × 100 = ×5000) +// 36..37 horizon_idx +// 37..38 strategy_id +// 38..40 padding + +#include "lob_state.cuh" + +#define OPEN_TRADE_STATE_BYTES 24 +#define TRADE_RECORD_BYTES 40 + +extern "C" __global__ void pnl_track_step( + unsigned char* pos_base, // [n_backtests * 24 bytes] + unsigned char* open_trade_state, // [n_backtests * 24 bytes] + unsigned char* trade_log_base, // [n_backtests * cap * 40 bytes] + unsigned int* trade_log_head, // [n_backtests] + unsigned long long current_ts_ns, + int trade_log_cap, + int n_backtests +) { + int b = blockIdx.x; + if (b >= n_backtests || threadIdx.x != 0) return; + + Pos* pos = reinterpret_cast(pos_base + (size_t)b * sizeof(Pos)); + unsigned char* st = open_trade_state + (size_t)b * OPEN_TRADE_STATE_BYTES; + int prev_size = *reinterpret_cast(st + 12); + int now_size = pos->position_lots; + + if (prev_size == 0 && now_size != 0) { + // Open entry — snapshot context. + *reinterpret_cast(st + 0) = current_ts_ns; + *reinterpret_cast(st + 8) = (int)(pos->vwap_entry * 100.0f + 0.5f); + *reinterpret_cast(st + 12) = now_size; + *reinterpret_cast(st + 16) = pos->realized_pnl; + st[20] = 0; // horizon_idx; set by decision kernel in C7 + } else if (prev_size != 0 && now_size == 0) { + // Close — emit TradeRecord. + const unsigned int idx = trade_log_head[b] % (unsigned int)trade_log_cap; + trade_log_head[b] += 1; + unsigned char* rec = trade_log_base + ((size_t)b * trade_log_cap + idx) * TRADE_RECORD_BYTES; + + const unsigned long long entry_ts = *reinterpret_cast(st + 0); + const int entry_px_x100 = *reinterpret_cast(st + 8); + const int entry_size = *reinterpret_cast(st + 12); + const float realized_at_open = *reinterpret_cast(st + 16); + const unsigned char horizon_idx = st[20]; + + // Realized P&L delta = pos.realized_pnl_now − pos.realized_pnl_at_open. + // In price-units × lots. + const float segment_realized = pos->realized_pnl - realized_at_open; + // Convert to USD ×100 fixed-point: × $50/index-point × 100 = ×5000. + const int realised_usd_fp = (int)(segment_realized * 5000.0f + (segment_realized >= 0.0f ? 0.5f : -0.5f)); + // Exit price reconstructed from realized delta: realized = (exit − entry) × dir × |size|. + // For sanity we record pos.vwap_entry at close time (the residual avg if any), + // but for a clean open→close that's the same as entry_px. We store the + // implied exit price instead, derived from the realized delta. + const float entry_px = (float)entry_px_x100 / 100.0f; + const float size_abs = (entry_size > 0) ? (float)entry_size : (float)(-entry_size); + const float dir = (entry_size > 0) ? 1.0f : -1.0f; + const float exit_px = (size_abs > 0.0f) ? (entry_px + segment_realized * dir / size_abs) : entry_px; + + *reinterpret_cast(rec + 0) = entry_ts; + *reinterpret_cast(rec + 8) = current_ts_ns; + *reinterpret_cast(rec + 16) = entry_px_x100; + *reinterpret_cast(rec + 20) = (int)(exit_px * 100.0f + 0.5f); + *reinterpret_cast(rec + 24) = entry_size; + *reinterpret_cast(rec + 28) = 0; // fees_usd_fp placeholder + *reinterpret_cast(rec + 32) = realised_usd_fp; + rec[36] = horizon_idx; + rec[37] = 0; // strategy_id placeholder + rec[38] = 0; + rec[39] = 0; + + // Reset open-trade scratch. + #pragma unroll + for (int i = 0; i < OPEN_TRADE_STATE_BYTES; ++i) st[i] = 0; + } + // prev != 0 AND now != 0 (scale-in or partial close): leave entry + // scratch in place. Multi-fill averaging is a v2 refinement. +} diff --git a/crates/ml-backtesting/src/lob/mod.rs b/crates/ml-backtesting/src/lob/mod.rs index 0a76650c9..fc75f3929 100644 --- a/crates/ml-backtesting/src/lob/mod.rs +++ b/crates/ml-backtesting/src/lob/mod.rs @@ -5,6 +5,13 @@ pub const BOOK_LEVELS: usize = 10; pub const N_HORIZONS: usize = 5; pub const MAX_LIMITS: usize = 32; pub const MAX_STOPS: usize = 16; +/// Max closed-trade records buffered per backtest before the host +/// must drain. Sized for a single fixture day at ~30s decision cadence. +pub const TRADE_LOG_CAP: usize = 1024; +/// Bytes per TradeRecord (must match crates/ml-backtesting/src/order.rs). +pub const TRADE_RECORD_BYTES: usize = 40; +/// Bytes per OpenTradeState (kernel-side tracking of currently-open entry). +pub const OPEN_TRADE_STATE_BYTES: usize = 24; /// Flat host-side mirror of `cuda/lob_state.cuh::Book`. #[derive(Clone, Copy, Debug, Default, PartialEq)] diff --git a/crates/ml-backtesting/src/sim.rs b/crates/ml-backtesting/src/sim.rs index 8a1db073b..31c90cc85 100644 --- a/crates/ml-backtesting/src/sim.rs +++ b/crates/ml-backtesting/src/sim.rs @@ -20,26 +20,28 @@ const BOOK_UPDATE_CUBIN: &[u8] = include_bytes!(concat!(env!("OUT_DIR"), "/book_update.cubin")); 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")); pub struct LobSimCuda { n_backtests: usize, stream: Arc, book_update_fn: cudarc::driver::CudaFunction, submit_market_fn: cudarc::driver::CudaFunction, + pnl_track_fn: cudarc::driver::CudaFunction, - // Per-backtest book state on device. Laid out - // [n_backtests, 4 fields × 10 levels] = N × 40 floats. books_d: CudaSlice, - // Per-backtest Pos state. Pod struct = 24 bytes; allocated as u8 buffer. pos_d: CudaSlice, - // Mapped-pinned input snapshot buffers (size 10 each) — broadcast - // across all N backtests in v1. bid_px_d: CudaSlice, bid_sz_d: CudaSlice, ask_px_d: CudaSlice, ask_sz_d: CudaSlice, - // Per-backtest market-order target {side, size} buffer. market_targets_d: CudaSlice, + + // Trade-log + open-trade-state buffers (C6). + open_trade_state_d: CudaSlice, // [n_backtests * 24] + trade_log_d: CudaSlice, // [n_backtests * TRADE_LOG_CAP * 40] + trade_log_head_d: CudaSlice, // [n_backtests] } impl LobSimCuda { @@ -60,6 +62,12 @@ impl LobSimCuda { let submit_market_fn = match_module .load_function("submit_market_immediate") .context("load submit_market_immediate")?; + let pnl_track_module = ctx + .load_cubin(PNL_TRACK_CUBIN.to_vec()) + .context("load pnl_track cubin")?; + let pnl_track_fn = pnl_track_module + .load_function("pnl_track_step") + .context("load pnl_track_step")?; let books_d = stream .alloc_zeros::(n_backtests * BOOK_FIELDS * BOOK_LEVELS) @@ -75,11 +83,22 @@ impl LobSimCuda { .alloc_zeros::(n_backtests * 2) .context("alloc market_targets_d")?; + let open_trade_state_d = stream + .alloc_zeros::(n_backtests * crate::lob::OPEN_TRADE_STATE_BYTES) + .context("alloc open_trade_state_d")?; + let trade_log_d = stream + .alloc_zeros::(n_backtests * crate::lob::TRADE_LOG_CAP * crate::lob::TRADE_RECORD_BYTES) + .context("alloc trade_log_d")?; + let trade_log_head_d = stream + .alloc_zeros::(n_backtests) + .context("alloc trade_log_head_d")?; + Ok(Self { n_backtests, stream, book_update_fn, submit_market_fn, + pnl_track_fn, books_d, pos_d, bid_px_d, @@ -87,6 +106,9 @@ impl LobSimCuda { ask_px_d, ask_sz_d, market_targets_d, + open_trade_state_d, + trade_log_d, + trade_log_head_d, }) } @@ -131,22 +153,29 @@ impl LobSimCuda { } /// Submit a market order to one specific backtest. side: 0=buy, 1=sell. - /// Walks the book + updates Pos. Convenience entry for fixture testing; - /// production path goes through the decision-policy kernel (C7). - pub fn submit_market(&mut self, backtest_idx: usize, side: u8, size: u16) -> Result<()> { + /// Walks the book + updates Pos, then runs pnl_track to detect + /// segment_complete and emit TradeRecord if the position closed. + /// Convenience entry for fixture testing; production path goes + /// through the decision-policy kernel (C7). + pub fn submit_market( + &mut self, + backtest_idx: usize, + side: u8, + size: u16, + current_ts_ns: u64, + ) -> Result<()> { anyhow::ensure!( backtest_idx < self.n_backtests, "backtest_idx {} >= n_backtests {}", backtest_idx, self.n_backtests ); - // Build targets array: side+size for the target backtest, no-op (side=2) for others. let mut targets = vec![0i32; self.n_backtests * 2]; for b in 0..self.n_backtests { - targets[b * 2 + 0] = 2; // no-op + targets[b * 2] = 2; // no-op targets[b * 2 + 1] = 0; } - targets[backtest_idx * 2 + 0] = side as i32; + targets[backtest_idx * 2] = side as i32; targets[backtest_idx * 2 + 1] = size as i32; self.stream .memcpy_htod(targets.as_slice(), &mut self.market_targets_d)?; @@ -167,9 +196,73 @@ impl LobSimCuda { .launch(cfg)?; } self.stream.synchronize()?; + + // Run pnl_track to detect close + emit TradeRecord. + self.step_pnl_track(current_ts_ns)?; Ok(()) } + /// Run pnl_track_step kernel: detect segment_complete, emit TradeRecord. + /// Called automatically by submit_market; exposed for callers that + /// orchestrate matching themselves. + pub fn step_pnl_track(&mut self, current_ts_ns: u64) -> Result<()> { + 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 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(¤t_ts_ns) + .arg(&cap) + .arg(&n) + .launch(cfg)?; + } + self.stream.synchronize()?; + Ok(()) + } + + /// Read all closed-trade records for a backtest. Returns up to + /// TRADE_LOG_CAP records (oldest dropped if the ring wrapped, which + /// the v1 head counter exposes as `head_count`). + pub fn read_trade_records( + &self, + backtest_idx: usize, + ) -> Result> { + anyhow::ensure!( + backtest_idx < self.n_backtests, + "backtest_idx {} >= n_backtests {}", + backtest_idx, + self.n_backtests + ); + let mut heads = vec![0u32; self.n_backtests]; + self.stream.memcpy_dtoh(&self.trade_log_head_d, heads.as_mut_slice())?; + let head = heads[backtest_idx] as usize; + if head == 0 { + return Ok(vec![]); + } + let n_to_read = head.min(crate::lob::TRADE_LOG_CAP); + let rec_bytes = crate::lob::TRADE_RECORD_BYTES; + let off = backtest_idx * crate::lob::TRADE_LOG_CAP * rec_bytes; + let mut raw = vec![0u8; self.n_backtests * crate::lob::TRADE_LOG_CAP * rec_bytes]; + self.stream.memcpy_dtoh(&self.trade_log_d, raw.as_mut_slice())?; + let mut out = Vec::with_capacity(n_to_read); + for i in 0..n_to_read { + let rec_off = off + i * rec_bytes; + let r: crate::order::TradeRecord = + bytemuck::pod_read_unaligned(&raw[rec_off..rec_off + rec_bytes]); + out.push(r); + } + Ok(out) + } + /// Read back the Pos state for a specific backtest. pub fn read_pos(&self, backtest_idx: usize) -> Result { anyhow::ensure!( diff --git a/crates/ml-backtesting/tests/fixtures/pnl_accounting_buy_close.json b/crates/ml-backtesting/tests/fixtures/pnl_accounting_buy_close.json new file mode 100644 index 000000000..72a75d0fb --- /dev/null +++ b/crates/ml-backtesting/tests/fixtures/pnl_accounting_buy_close.json @@ -0,0 +1,50 @@ +{ + "name": "pnl_accounting_buy_close", + "n_backtests": 1, + "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": [12.0, 18.0, 24.0, 30.0, 36.0, 42.0, 48.0, 54.0, 60.0, 66.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": "submit_market", + "backtest_idx": 0, + "side": "buy", + "size": 4, + "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": "submit_market", + "backtest_idx": 0, + "side": "sell", + "size": 4, + "ts_ns": 2000000000 + } + ], + "expected_pos": [ + { + "position_lots": 0, + "realized_pnl": 20.0, + "realized_pnl_tol": 0.01 + } + ], + "expected_trade_records": [ + { + "entry_ts_ns": 1000000000, + "exit_ts_ns": 2000000000, + "size_lots": 4, + "realised_pnl_usd_fp": 100000, + "realised_pnl_tol_usd_fp": 100 + } + ] +} diff --git a/crates/ml-backtesting/tests/lob_sim_fixtures.rs b/crates/ml-backtesting/tests/lob_sim_fixtures.rs index 771200928..f6b036ca8 100644 --- a/crates/ml-backtesting/tests/lob_sim_fixtures.rs +++ b/crates/ml-backtesting/tests/lob_sim_fixtures.rs @@ -23,9 +23,13 @@ enum FixtureEvent { backtest_idx: usize, side: String, size: u16, + #[serde(default = "default_ts_ns")] + ts_ns: u64, }, } +fn default_ts_ns() -> u64 { 1 } + #[derive(Debug, Deserialize)] struct ExpectedPos { position_lots: i32, @@ -49,6 +53,17 @@ struct FixtureBook { ask_sz: [f32; BOOK_LEVELS], } +#[derive(Debug, Deserialize)] +struct ExpectedTradeRecord { + entry_ts_ns: u64, + exit_ts_ns: u64, + size_lots: i32, + realised_pnl_usd_fp: i32, + #[serde(default = "default_pnl_fp_tol")] + realised_pnl_tol_usd_fp: i32, +} +fn default_pnl_fp_tol() -> i32 { 100 } + #[derive(Debug, Deserialize)] struct Fixture { name: String, @@ -60,6 +75,8 @@ struct Fixture { expected_books_apply_to_all: bool, #[serde(default)] expected_pos: Vec, + #[serde(default)] + expected_trade_records: Vec, } fn run_book_fixture(path: &Path) -> Result<()> { @@ -77,13 +94,13 @@ fn run_book_fixture(path: &Path) -> Result<()> { FixtureEvent::Snapshot { bid_px, bid_sz, ask_px, ask_sz } => { sim.apply_snapshot(bid_px, bid_sz, ask_px, ask_sz)?; } - FixtureEvent::SubmitMarket { backtest_idx, side, size } => { + FixtureEvent::SubmitMarket { backtest_idx, side, size, ts_ns } => { let side_u8 = match side.as_str() { "buy" => 0u8, "sell" => 1u8, other => anyhow::bail!("submit_market: unknown side {other}"), }; - sim.submit_market(*backtest_idx, side_u8, *size)?; + sim.submit_market(*backtest_idx, side_u8, *size, *ts_ns)?; } } } @@ -112,6 +129,29 @@ fn run_book_fixture(path: &Path) -> Result<()> { } } + if !fx.expected_trade_records.is_empty() { + let records = sim.read_trade_records(0)?; + assert_eq!( + records.len(), + fx.expected_trade_records.len(), + "trade record count mismatch: got {}, want {}", + records.len(), + fx.expected_trade_records.len() + ); + for (i, want) in fx.expected_trade_records.iter().enumerate() { + let got = &records[i]; + assert_eq!(got.entry_ts_ns, want.entry_ts_ns, "trade[{i}].entry_ts_ns"); + assert_eq!(got.exit_ts_ns, want.exit_ts_ns, "trade[{i}].exit_ts_ns"); + assert_eq!(got.size_lots, want.size_lots, "trade[{i}].size_lots"); + let diff = (got.realised_pnl_usd_fp - want.realised_pnl_usd_fp).abs(); + assert!( + diff <= want.realised_pnl_tol_usd_fp, + "trade[{i}].realised_pnl_usd_fp got {} want {} (tol {})", + got.realised_pnl_usd_fp, want.realised_pnl_usd_fp, want.realised_pnl_tol_usd_fp + ); + } + } + if !fx.expected_pos.is_empty() { anyhow::ensure!( fx.expected_pos.len() == fx.n_backtests, @@ -168,3 +208,9 @@ fn fix_book_update_n_backtests() { fn fix_market_order_consumes_top() { run_book_fixture(Path::new("tests/fixtures/market_order_consumes_top.json")).unwrap(); } + +#[test] +#[ignore = "requires CUDA"] +fn fix_pnl_accounting_buy_close() { + run_book_fixture(Path::new("tests/fixtures/pnl_accounting_buy_close.json")).unwrap(); +}