From 84caa99b6561750407ced6971736a3fbab5d7b03 Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Sun, 31 May 2026 18:01:16 +0200 Subject: [PATCH] =?UTF-8?q?feat(lobsim):=20per-backtest=20trade-count=20re?= =?UTF-8?q?ad=20+=204=C3=97=20TRADE=5FLOG=5FCAP=20(E.1-E.3)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds two new public methods to LobSimCuda + bumps TRADE_LOG_CAP to prevent eval-phase ring-buffer wrap at cluster scale. - read_per_backtest_trade_counts() -> Vec: cumulative trade counters per backtest (length n_backtests). Replaces the broken pattern in alpha_rl_train.rs where head_before_eval = aggregate across batch was compared against all_records = single-account ring. - read_trade_records_all() -> Vec>: all backtests' rings in one call. Mapped-pinned staging per feedback_no_htod_htoh_only_mapped_pinned: allocate MappedRecordBuffer for the payload + MappedRecordBuffer for heads, DtoD copy from device buffers into mapped-pinned dev_ptrs, sync, read host_ptrs. - TRADE_LOG_CAP 1024 → 4096: cluster v11 (alpha-rl-8ll7j) showed ~342 eval trades/account mean with peaks toward 1000. 4096 gives 4× headroom; memory cost b=1024 × cap × 40 B = 167 MB (was 41 MB), comfortable on L40S 48GB / H100 80GB. Both methods synchronize after DtoD so host reads see the data. E.4 (alpha_rl_train.rs aggregation block replacement) lands separately after Phase A cluster validates to avoid alpha_rl_train.rs conflict. See spec docs/superpowers/specs/2026-05-31-eval-summary-trade-aggregation-design.md and plan docs/superpowers/plans/2026-05-31-eval-summary-trade-aggregation.md. Co-Authored-By: Claude Opus 4.7 --- crates/ml-backtesting/src/lob/mod.rs | 9 +- crates/ml-backtesting/src/sim/mod.rs | 126 +++++++++++++++++++++++++++ 2 files changed, 133 insertions(+), 2 deletions(-) diff --git a/crates/ml-backtesting/src/lob/mod.rs b/crates/ml-backtesting/src/lob/mod.rs index 813407280..fbef1bc28 100644 --- a/crates/ml-backtesting/src/lob/mod.rs +++ b/crates/ml-backtesting/src/lob/mod.rs @@ -16,8 +16,13 @@ pub const STOP_SLOT_BYTES: usize = 32; /// Bytes per Orders struct (limits[32] + stops[16]). pub const ORDERS_BYTES: usize = MAX_LIMITS * LIMIT_SLOT_BYTES + MAX_STOPS * STOP_SLOT_BYTES; /// 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; +/// must drain. Bumped from 1024 → 4096 on 2026-05-31 to prevent +/// eval-phase ring wrap at cluster scale (b=1024 × 5000 eval steps +/// produces ~342 dones/account mean, ~1000 peak; 4096 gives 4× +/// headroom). Memory cost at b=1024: 1024 × 4096 × 40 B = 167 MB +/// device (was 41 MB) — comfortable on L40S 48 GB / H100 80 GB. +/// See spec `docs/superpowers/specs/2026-05-31-eval-summary-trade-aggregation-design.md`. +pub const TRADE_LOG_CAP: usize = 4096; /// 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). diff --git a/crates/ml-backtesting/src/sim/mod.rs b/crates/ml-backtesting/src/sim/mod.rs index 38c0dedbb..987bbc777 100644 --- a/crates/ml-backtesting/src/sim/mod.rs +++ b/crates/ml-backtesting/src/sim/mod.rs @@ -1515,6 +1515,132 @@ impl LobSimCuda { Ok(out) } + /// Read per-backtest cumulative trade counters. Returns a `Vec` + /// of length `n_backtests` where entry `i` is the total number of + /// closed-trade records ever pushed to backtest `i`'s ring buffer + /// (may exceed `TRADE_LOG_CAP` if the ring wrapped). + /// + /// Used by `alpha_rl_train` to snapshot the eval-phase boundary per + /// account: `head_before_per_b[i] = read_per_backtest_trade_counts()[i]` + /// at eval start; post-eval `head_after_per_b[i] - head_before_per_b[i]` + /// gives the true eval-phase trade count for backtest `i`. + /// + /// Uses mapped-pinned staging per `feedback_no_htod_htoh_only_mapped_pinned`: + /// allocate a `MappedRecordBuffer`, DtoD copy from + /// `trade_log_head_d` into its dev_ptr, sync, read via host_ptr. + /// + /// Replaces the (`head_before_eval = read_total_trade_count()`, + /// `all_records = read_trade_records(0)`) pattern that conflated + /// aggregate counts with per-account ring contents and produced + /// meaningless eval_summary metrics — see spec + /// `docs/superpowers/specs/2026-05-31-eval-summary-trade-aggregation-design.md`. + pub fn read_per_backtest_trade_counts(&self) -> Result> { + use ml_alpha::pinned_mem::MappedRecordBuffer; + use ml_alpha::trainer::raw_launch::raw_memcpy_dtod_async; + let raw_stream = self.stream.cu_stream(); + let staging: MappedRecordBuffer = unsafe { MappedRecordBuffer::new(self.n_backtests) } + .map_err(|e| anyhow::anyhow!("alloc trade_head staging: {e}"))?; + unsafe { + raw_memcpy_dtod_async( + staging.dev_ptr, + self.trade_log_head_d.raw_ptr(), + self.n_backtests * std::mem::size_of::(), + raw_stream, + ) + .map_err(|e| anyhow::anyhow!("DtoD trade_log_head: {e:?}"))?; + } + self.stream + .synchronize() + .context("sync after trade_log_head DtoD")?; + let mut out = Vec::with_capacity(self.n_backtests); + for i in 0..self.n_backtests { + // SAFETY: mapped-pinned host_ptr is valid for [0, len); sync + // above ensures the DtoD writes are visible to host. + unsafe { + out.push(std::ptr::read_volatile(staging.host_ptr.add(i))); + } + } + Ok(out) + } + + /// Read all backtests' trade record rings. Returns + /// `Vec>` of length `n_backtests`; entry `i` is the + /// most recent up-to-`TRADE_LOG_CAP` records for backtest `i` (oldest + /// dropped on wrap). + /// + /// Uses mapped-pinned staging for both the per-backtest head counters + /// and the trade-log payload (per `feedback_no_htod_htoh_only_mapped_pinned`). + /// At cluster scale (b=1024, cap=4096) the payload staging buffer + /// is 167 MB — allocated once per call (end-of-run). Acceptable + /// because this is invoked once per fold-eval, not in the hot path. + /// + /// Callers compose this with + /// [`read_per_backtest_trade_counts`](Self::read_per_backtest_trade_counts) + /// to identify each account's eval-phase-only slice. + pub fn read_trade_records_all( + &self, + ) -> Result>> { + use ml_alpha::pinned_mem::MappedRecordBuffer; + use ml_alpha::trainer::raw_launch::raw_memcpy_dtod_async; + let raw_stream = self.stream.cu_stream(); + let rec_bytes = crate::lob::TRADE_RECORD_BYTES; + let cap = crate::lob::TRADE_LOG_CAP; + let payload_len_u8 = self.n_backtests * cap * rec_bytes; + + let head_staging: MappedRecordBuffer = + unsafe { MappedRecordBuffer::new(self.n_backtests) } + .map_err(|e| anyhow::anyhow!("alloc head staging: {e}"))?; + let payload_staging: MappedRecordBuffer = + unsafe { MappedRecordBuffer::new(payload_len_u8) } + .map_err(|e| anyhow::anyhow!("alloc payload staging ({payload_len_u8} B): {e}"))?; + + unsafe { + raw_memcpy_dtod_async( + head_staging.dev_ptr, + self.trade_log_head_d.raw_ptr(), + self.n_backtests * std::mem::size_of::(), + raw_stream, + ) + .map_err(|e| anyhow::anyhow!("DtoD trade_log_head: {e:?}"))?; + raw_memcpy_dtod_async( + payload_staging.dev_ptr, + self.trade_log_d.raw_ptr(), + payload_len_u8, + raw_stream, + ) + .map_err(|e| anyhow::anyhow!("DtoD trade_log payload: {e:?}"))?; + } + self.stream + .synchronize() + .context("sync after trade_log DtoD")?; + + let mut out = Vec::with_capacity(self.n_backtests); + for b in 0..self.n_backtests { + // SAFETY: mapped-pinned host_ptr is valid + post-sync the + // device-side DtoD writes are visible to the host. + let head = unsafe { std::ptr::read_volatile(head_staging.host_ptr.add(b)) } as usize; + let n_to_read = head.min(cap); + let off = b * cap * rec_bytes; + let mut records = Vec::with_capacity(n_to_read); + for i in 0..n_to_read { + let rec_off = off + i * rec_bytes; + // SAFETY: bounds-checked above (i < n_to_read ≤ cap; rec_off + rec_bytes ≤ payload_len_u8). + let bytes: [u8; crate::lob::TRADE_RECORD_BYTES] = unsafe { + let mut buf = [0u8; crate::lob::TRADE_RECORD_BYTES]; + for k in 0..rec_bytes { + buf[k] = std::ptr::read_volatile(payload_staging.host_ptr.add(rec_off + k)); + } + buf + }; + let r: crate::order::TradeRecord = + bytemuck::pod_read_unaligned(&bytes); + records.push(r); + } + out.push(records); + } + Ok(out) + } + /// Read back the Pos state for a specific backtest. pub fn read_pos(&self, backtest_idx: usize) -> Result { anyhow::ensure!(