feat(lobsim): per-backtest trade-count read + 4× TRADE_LOG_CAP (E.1-E.3)
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<u32>: 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<Vec<TradeRecord>>: all backtests' rings in one call. Mapped-pinned staging per feedback_no_htod_htoh_only_mapped_pinned: allocate MappedRecordBuffer<u8> for the payload + MappedRecordBuffer<u32> 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 <noreply@anthropic.com>
This commit is contained in:
@@ -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).
|
||||
|
||||
@@ -1515,6 +1515,132 @@ impl LobSimCuda {
|
||||
Ok(out)
|
||||
}
|
||||
|
||||
/// Read per-backtest cumulative trade counters. Returns a `Vec<u32>`
|
||||
/// 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<u32>`, 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<Vec<u32>> {
|
||||
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<u32> = 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::<u32>(),
|
||||
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<Vec<TradeRecord>>` 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<Vec<Vec<crate::order::TradeRecord>>> {
|
||||
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<u32> =
|
||||
unsafe { MappedRecordBuffer::new(self.n_backtests) }
|
||||
.map_err(|e| anyhow::anyhow!("alloc head staging: {e}"))?;
|
||||
let payload_staging: MappedRecordBuffer<u8> =
|
||||
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::<u32>(),
|
||||
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<PosFlat> {
|
||||
anyhow::ensure!(
|
||||
|
||||
Reference in New Issue
Block a user