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>

bin/fxt-backtest/src/main.rs

@@ -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)

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<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}"),
         }

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::<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