diff --git a/bin/fxt-backtest/src/main.rs b/bin/fxt-backtest/src/main.rs index f5f4d00fd..84db27edb 100644 --- a/bin/fxt-backtest/src/main.rs +++ b/bin/fxt-backtest/src/main.rs @@ -110,6 +110,15 @@ struct RunArgs { /// Output directory; per-cell artifacts written to /cell_NNNN/. #[arg(long)] out: PathBuf, + /// Diagnostic: sample the 5 per-horizon `alpha_probs` every N events + /// (post-`forward_step_into`), DtoD into a mapped-pinned buffer, and + /// accumulate per-horizon mean / stddev / min / max / 10-bin histogram. + /// Emitted at end-of-run as `per_horizon_logit_diag h{N}: ...` for each + /// of (h30, h100, h300, h1000, h6000). Default: disabled (zero overhead). + /// Recommended diagnostic value: 1000 (≈100 ms overhead on a 2 M-event + /// run). Mutually exclusive with stride=0 (rejected at parse time). + #[arg(long)] + per_horizon_logit_sampling_stride: Option, } #[derive(Parser, Debug)] @@ -209,6 +218,10 @@ struct SweepBase { /// seeding when |target_signed − effective| < delta_floor. Default 1.0 = /// 1 lot (ES single-contract sizing). #[serde(default = "default_delta_floor")] delta_floor: f32, + /// Diagnostic: per-horizon alpha_probs sampling stride (events). When + /// set, each cell emits a `per_horizon_logit_diag h{N}: ...` line per + /// horizon at end-of-run. None = disabled (default). + #[serde(default)] per_horizon_logit_sampling_stride: Option, } /// P6: per-sim-variant overrides for the batched-cell sweep flow. @@ -269,6 +282,9 @@ struct SweepCell { #[serde(default)] checkpoint: Option, #[serde(default)] kelly_frac_floor: Option, #[serde(default)] sharpe_weight_floor: Option, + /// Per-cell override for the diagnostic alpha_probs sampler. When + /// unset, falls back to `SweepBase::per_horizon_logit_sampling_stride`. + #[serde(default)] per_horizon_logit_sampling_stride: Option, } fn main() -> Result<()> { @@ -388,6 +404,9 @@ fn sweep(args: SweepArgs) -> Result<()> { sharpe_weight_floor: cell.sharpe_weight_floor.unwrap_or(grid.base.sharpe_weight_floor), out: cell_out.clone(), + per_horizon_logit_sampling_stride: + cell.per_horizon_logit_sampling_stride + .or(grid.base.per_horizon_logit_sampling_stride), }; run(run_args).with_context(|| format!("sweep cell {}", cell.name))?; tracing::info!(cell = cell.name.as_str(), "cell complete"); @@ -456,6 +475,11 @@ fn run_batched_cell( PerceptionTrainer::new(&dev, &cfg2).context("PerceptionTrainer::new (random init)")? }; + let logit_stride = cell.per_horizon_logit_sampling_stride + .or(base.per_horizon_logit_sampling_stride); + if let Some(s) = logit_stride { + anyhow::ensure!(s > 0, "per_horizon_logit_sampling_stride must be > 0 (got 0)"); + } let cfg = BacktestHarnessConfig { data_root: cell_data.to_path_buf(), predecoded_dir: base.predecoded_dir.clone().unwrap_or_else(|| cell_data.to_path_buf()), @@ -472,6 +496,8 @@ fn run_batched_cell( variant_names: Some(names), sim_config_override: Some(batched), strategies: Vec::new(), // batched flow uses default policy per backtest + per_horizon_logit_sampling_stride: logit_stride, + kernel_step_trace_path: None, }; std::fs::create_dir_all(cell_out) @@ -542,6 +568,9 @@ fn run(args: RunArgs) -> Result<()> { .context("PerceptionTrainer::new")? }; + if let Some(s) = args.per_horizon_logit_sampling_stride { + anyhow::ensure!(s > 0, "--per-horizon-logit-sampling-stride must be > 0 (got 0)"); + } let cfg = BacktestHarnessConfig { data_root: args.data.clone(), predecoded_dir: args.predecoded_dir.clone().unwrap_or(args.data), @@ -558,6 +587,8 @@ fn run(args: RunArgs) -> Result<()> { variant_names: None, // P6: legacy single-cell flow uses cell_NNNN naming sim_config_override: None, // P6: legacy single-cell flow uses from_uniform strategies: strategy_grid, + per_horizon_logit_sampling_stride: args.per_horizon_logit_sampling_stride, + kernel_step_trace_path: None, }; std::fs::create_dir_all(&args.out) diff --git a/crates/ml-backtesting/src/harness.rs b/crates/ml-backtesting/src/harness.rs index c31c72d36..ce6159f26 100644 --- a/crates/ml-backtesting/src/harness.rs +++ b/crates/ml-backtesting/src/harness.rs @@ -10,11 +10,18 @@ //! sliding K-window for the recurrent Mamba2 context, drive the sim's //! decision loop with the trainer's per-horizon probability output. +// The diagnostic per-horizon alpha_probs sampler calls `MappedF32Buffer::new`, +// a cudarc `cuMemHostAlloc(DEVICEMAP)` FFI that requires `unsafe`. The +// workspace-wide `-W unsafe-code` lint is therefore noise here. Same rationale +// as `sim/mod.rs`'s file-level allow for kernel-launch unsafes. +#![allow(unsafe_code)] + use anyhow::{Context, Result}; use ml_alpha::cfc::snap_features::Mbp10RawInput; use ml_alpha::data::loader::{ discover_mbp10_files_sorted, MultiHorizonLoader, MultiHorizonLoaderConfig, }; +use ml_alpha::pinned_mem::MappedF32Buffer; use ml_alpha::trainer::perception::PerceptionTrainer; use ml_core::device::MlDevice; use std::collections::VecDeque; @@ -22,6 +29,69 @@ use std::path::PathBuf; use crate::sim::LobSimCuda; +/// Number of forward-horizons used by the per-horizon heads. Mirrors +/// `ml_alpha::heads::N_HORIZONS` (locally constant so we don't depend on +/// ml-alpha's identifier at runtime). Used by the per-horizon alpha-probs +/// sampler. +const N_HORIZONS: usize = 5; + +/// Number of histogram bins covering [0.0, 1.0] for sigmoid'd probs. +const HIST_BINS: usize = 10; + +/// Welford running stats for the per-horizon alpha_probs sampler. +/// Tracks count, mean, M2 (sum of squared diffs from mean) for std, +/// running min/max, and a 10-bin histogram over [0, 1]. +#[derive(Clone, Debug, Default)] +struct PerHorizonStats { + count: u64, + mean: f64, + m2: f64, + min: f32, + max: f32, + hist: [u64; HIST_BINS], +} + +impl PerHorizonStats { + fn new() -> Self { + Self { + count: 0, + mean: 0.0, + m2: 0.0, + min: f32::INFINITY, + max: f32::NEG_INFINITY, + hist: [0; HIST_BINS], + } + } + + fn update(&mut self, x: f32) { + self.count += 1; + let delta = x as f64 - self.mean; + self.mean += delta / self.count as f64; + let delta2 = x as f64 - self.mean; + self.m2 += delta * delta2; + if x < self.min { self.min = x; } + if x > self.max { self.max = x; } + // Bin assignment: clamp to [0, HIST_BINS-1] for any prob slightly + // outside [0, 1] (defensive — sigmoid output is bounded but stray + // NaN/Inf would otherwise panic on usize cast). + let scaled = (x * HIST_BINS as f32).floor(); + let bin = if scaled.is_nan() { + 0 + } else if scaled < 0.0 { + 0 + } else if scaled >= HIST_BINS as f32 { + HIST_BINS - 1 + } else { + scaled as usize + }; + self.hist[bin] += 1; + } + + fn std(&self) -> f64 { + if self.count < 2 { 0.0 } else { (self.m2 / (self.count - 1) as f64).sqrt() } + } +} + #[derive(Clone, Debug)] pub struct BacktestHarnessConfig { pub data_root: PathBuf, @@ -72,6 +142,35 @@ pub struct BacktestHarnessConfig { /// of building from_uniform off the scalar fields above. Length MUST /// equal n_parallel when set. pub sim_config_override: Option, + /// Diagnostic: every N events (post-forward_step_into), DtoD the 5 + /// per-horizon `alpha_probs` into a mapped-pinned buffer and update + /// running mean / variance / min / max / 10-bin histogram for each + /// of the 5 horizons. None = sampling disabled (zero overhead; + /// default). Some(N) emits a `per_horizon_logit_diag h{H}: ...` line + /// per horizon at end of run, alongside the existing `crt_diag` + /// battery. + /// + /// Per `feedback_no_htod_htoh_only_mapped_pinned`: the sample buffer + /// is allocated once in `BacktestHarness::new` (5×f32 mapped-pinned) + /// and reused on every tick. At stride=1000 on 2M events, ~2k DtoD + /// stops × ~50µs ≈ 100ms total overhead. + /// + /// Goal (Smoke 2 follow-up): distinguish whether the observed + /// 1.04× uniform `mean_run_len` across horizons is caused by uniform + /// per-horizon LOGITS (→ block-diagonal heads needed) or by + /// DIFFERENTIATED logits with collapsed downstream computation. + pub per_horizon_logit_sampling_stride: Option, + /// Diagnostic mirror of `PerceptionTrainerConfig.kernel_step_trace_path`. + /// `Some(path)` records that this backtest was built with the trainer's + /// kernel-step-trace JSONL drain pointed at `path`; the harness emits + /// a startup `tracing::info!` so post-mortem log scrapes can correlate + /// trace files to backtest cells. The trainer is constructed by the + /// caller (`bin/fxt-backtest/src/main.rs`) and owns the actual ring + + /// drain task; the field here is the contract record. When `None` + /// (default), no trace file is referenced and the trainer carries + /// zero overhead even if it was built with the `kernel-step-trace` + /// Cargo feature compiled in. + pub kernel_step_trace_path: Option, } pub struct BacktestHarness { @@ -119,6 +218,17 @@ pub struct BacktestHarness { /// Equal to `decision_count` after `run()` completes. Used to size /// the end-of-run DtoV correctly. convictions_recorded: u32, + /// Diagnostic: optional mapped-pinned buffer (5 × f32) used by the + /// per-horizon alpha_probs sampler. `Some` iff + /// `cfg.per_horizon_logit_sampling_stride.is_some()`. Allocated once + /// in `new()` so the hot path only pays a DtoD copy + sync per + /// sampling tick (no per-tick host allocation). + logit_sample_buf: Option, + /// Diagnostic: per-horizon Welford accumulators + 10-bin histograms + /// for the sampled alpha_probs. Always allocated (cheap: ~80 bytes per + /// horizon) — populated only when sampling is enabled. Emitted at + /// end-of-run in the existing CRT.diag block as `per_horizon_logit_diag`. + logit_stats: [PerHorizonStats; N_HORIZONS], } impl BacktestHarness { @@ -202,6 +312,22 @@ impl BacktestHarness { sim.upload_price_range(&sim_config.min_reasonable_px, &sim_config.max_reasonable_px)?; let pnl_curves = (0..cfg.n_parallel).map(|_| Vec::with_capacity(1024)).collect(); + + // Allocate the per-horizon alpha_probs sampling buffer iff the + // CLI flag is set. `MappedF32Buffer::new` requires an active CUDA + // context, which is guaranteed because `LobSimCuda::new(...)` + // above already initialised one on this thread. The `unsafe` here + // is the cudarc `cuMemHostAlloc` FFI call — identical pattern to + // `crates/ml-alpha/src/heads.rs::download` and `sim/mod.rs`'s + // file-level FFI policy. + let logit_sample_buf = if cfg.per_horizon_logit_sampling_stride.is_some() { + let buf = unsafe { MappedF32Buffer::new(N_HORIZONS) } + .map_err(|e| anyhow::anyhow!("logit-sample MappedF32Buffer alloc: {e}"))?; + Some(buf) + } else { + None + }; + Ok(Self { cfg, sim_config, @@ -214,6 +340,8 @@ impl BacktestHarness { event_count: 0, pnl_curves, convictions_recorded: 0, + logit_sample_buf, + logit_stats: std::array::from_fn(|_| PerHorizonStats::new()), }) } @@ -257,6 +385,28 @@ impl BacktestHarness { .forward_step_into(&raw, self.sim.alpha_probs_d_mut()) .context("trainer.forward_step_into")?; + // Diagnostic: per-horizon alpha_probs sampler. Stride-gated + // DtoD copy into a mapped-pinned 5-f32 buffer + host-side + // Welford accumulation. Runs BEFORE step_decision_with_latency + // so the sample reflects the same alpha_probs the decision + // kernel will consume on this event (the sync is acceptable + // here — sampling is gated to ~1-in-1000 events). + if let (Some(stride), Some(buf)) = ( + self.cfg.per_horizon_logit_sampling_stride, + self.logit_sample_buf.as_ref(), + ) { + debug_assert!(stride > 0, "stride==0 is rejected at CLI parse time"); + if self.event_count % stride as u64 == 0 { + self.sim + .sample_alpha_probs(buf) + .context("sample_alpha_probs")?; + for h in 0..N_HORIZONS { + let p = buf.read_record(h); + self.logit_stats[h].update(p); + } + } + } + // Side-channel: record this decision's max_conviction // on-device for the threshold-tuning percentile // computation. Kernel reads `alpha_probs_d` and writes @@ -466,6 +616,37 @@ impl BacktestHarness { lo, hi, n_total, win_rate * 100.0, mean_pnl, ); } + + // Per-horizon alpha_probs sampler diagnostic. Only emitted + // when --per-horizon-logit-sampling-stride was set; tells us + // whether Smoke 2's uniform `mean_run_len` (~1.04× across all + // 5 horizons) is caused by uniform per-horizon LOGITS + // (→ heads_w1/w2/gate/main need block-diagonal too) or by + // DIFFERENTIATED logits with collapsed downstream + // (→ different fix). + if self.cfg.per_horizon_logit_sampling_stride.is_some() { + for h in 0..n_horizons { + let s = &self.logit_stats[h]; + let mut hist_str = String::from("["); + for bk in 0..HIST_BINS { + if bk > 0 { hist_str.push(' '); } + hist_str.push_str(&format!("bin{}:{}", bk, s.hist[bk])); + } + hist_str.push(']'); + // Defensive: if no samples (e.g., max_events < seq_len), + // min/max are still ±INFINITY sentinels — emit NaN + // so downstream tooling doesn't latch onto sentinels. + let (min_disp, max_disp) = if s.count == 0 { + (f32::NAN, f32::NAN) + } else { + (s.min, s.max) + }; + eprintln!( + "per_horizon_logit_diag h{}: count={} mean={:.4} std={:.4} min={:.4} max={:.4} hist={}", + horizons[h], s.count, s.mean, s.std(), min_disp, max_disp, hist_str, + ); + } + } } Err(e) => eprintln!("crt_diag read failed: {e}"), } diff --git a/crates/ml-backtesting/src/sim/mod.rs b/crates/ml-backtesting/src/sim/mod.rs index 4b0648ddb..79aaca369 100644 --- a/crates/ml-backtesting/src/sim/mod.rs +++ b/crates/ml-backtesting/src/sim/mod.rs @@ -17,7 +17,8 @@ mod batched_config; pub use batched_config::{BatchedSimConfig, ResolvedSimVariant, UniformSimParams}; use anyhow::{Context, Result}; -use cudarc::driver::{CudaSlice, CudaStream, LaunchConfig, PushKernelArg}; +use cudarc::driver::{CudaSlice, CudaStream, DevicePtr, LaunchConfig, PushKernelArg}; +use ml_alpha::pinned_mem::MappedF32Buffer; use ml_core::device::MlDevice; use std::sync::Arc; @@ -1312,6 +1313,41 @@ impl LobSimCuda { &mut self.alpha_probs_d } + /// Diagnostic-only sampler: copy the current `alpha_probs_d` (5 f32s, + /// broadcast across backtests) into a caller-owned mapped-pinned + /// buffer and synchronise the stream so the host view is coherent. + /// + /// Per `feedback_no_htod_htoh_only_mapped_pinned`: the buffer is + /// allocated once in the harness state via `MappedF32Buffer::new(5)` + /// and reused on every sampling tick. The transfer is DtoD into the + /// mapped-pinned device pointer (the only permitted cross-direction + /// channel); a single `stream.synchronize()` makes the host-side + /// `read_record` calls deterministic. + /// + /// Intended for low-frequency sampling (e.g. every 1000 events). The + /// per-event production path keeps reading `alpha_probs_d` on-device + /// and is unaffected. + pub fn sample_alpha_probs(&self, dst: &MappedF32Buffer) -> Result<()> { + anyhow::ensure!( + dst.len >= N_HORIZONS, + "sample_alpha_probs: dst buffer len {} < N_HORIZONS {}", + dst.len, N_HORIZONS, + ); + let nbytes = N_HORIZONS * std::mem::size_of::(); + unsafe { + let (src_ptr, _g) = self.alpha_probs_d.device_ptr(&self.stream); + cudarc::driver::result::memcpy_dtod_async( + dst.dev_ptr, + src_ptr, + nbytes, + self.stream.cu_stream(), + ) + .context("sample_alpha_probs DtoD")?; + } + self.stream.synchronize().context("sample_alpha_probs sync")?; + Ok(()) + } + /// CRT Phase A0.5 corrective: record one max-conviction f32 at /// `convictions_d[write_idx]` from the current `alpha_probs_d`. /// Single-thread kernel; stream-ordered, no sync. Caller advances