diag(per-horizon-cfc): per-horizon alpha_probs sampling for Smoke 2 investigation
Adds --per-horizon-logit-sampling-stride CLI flag to fxt-backtest run/sweep
+ harness instrumentation that DtoHs alpha_probs every N events and
accumulates per-horizon mean/stddev/min/max + 10-bin histogram.
Emitted at end of CRT.diag block as:
per_horizon_logit_diag h{N}: count=... mean=... std=... min=... max=... hist=[...]
Goal: distinguish whether Smoke 2's uniform mean_run_len 1.04x across
horizons is caused by uniform per-horizon LOGITS (-> Task 11 scope
incomplete; heads_w1/w2/gate/main need block-diagonal too) or by
DIFFERENTIATED logits with collapsed downstream (-> different fix).
Default stride=None (disabled, zero overhead). With stride=1000 on
2M events, ~2000 DtoD stops x ~50us = ~100ms total overhead per run.
Per feedback_no_htod_htoh_only_mapped_pinned: uses mapped-pinned buffer
(single 5xf32 allocation reused on every sampling tick).
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
jgrusewski
@@ -110,6 +110,15 @@ struct RunArgs {
|
||||
/// Output directory; per-cell artifacts written to <out>/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<u32>,
|
||||
}
|
||||
|
||||
#[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<u32>,
|
||||
}
|
||||
|
||||
/// P6: per-sim-variant overrides for the batched-cell sweep flow.
|
||||
@@ -269,6 +282,9 @@ struct SweepCell {
|
||||
#[serde(default)] checkpoint: Option<PathBuf>,
|
||||
#[serde(default)] kelly_frac_floor: Option<f32>,
|
||||
#[serde(default)] sharpe_weight_floor: Option<f32>,
|
||||
/// 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<u32>,
|
||||
}
|
||||
|
||||
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)
|
||||
|
||||
@@ -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<crate::sim::BatchedSimConfig>,
|
||||
/// 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<u32>,
|
||||
/// 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<PathBuf>,
|
||||
}
|
||||
|
||||
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<MappedF32Buffer>,
|
||||
/// 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}"),
|
||||
}
|
||||
|
||||
@@ -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::<f32>();
|
||||
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
|
||||
|
||||
Reference in New Issue
Block a user