0e17fd4f2d
Driven by ffr59 (commit b44a97ff9) findings: all 5 horizons flip
direction every 2.5 events; win rate flat 24% across conviction; mean
PnL anti-correlated with conviction. Hypothesis: per-event alpha output
is high-frequency noise on top of a slower signal. If true, smoothing
input alpha BEFORE the conviction formula should reduce direction flips
and recover signal.
Adds Wiener-α adaptive EMA on raw alpha_probs[h] for each horizon,
applied BEFORE the multi-horizon conviction formula. Floor at 0.1
(stronger than the 0.4 floor on the output-side conviction EMA — this
tests whether INPUT smoothing has different impact than OUTPUT smoothing).
Three new device slots:
- alpha_ema_per_b_per_h (per-horizon EMA state)
- alpha_diff_var_per_b_per_h (variance of changes)
- alpha_sample_var_per_b_per_h (variance of value)
The CRT.diag Group A direction-flip counter still reads RAW alpha_probs
so we have a head-to-head comparison: raw flip rate vs smoothed flip rate.
Group E adds the smoothed-direction counter + mean run length.
End-of-run log adds one line per horizon:
crt_diag h<X> smoothed: flips=Y mean_run_len=Z events (vs raw F / M)
If smoothed mean_run_len >> raw mean_run_len: hypothesis is RIGHT, the
input had signal under the noise. Next step would be to make this an
operational EMA in the controller.
If smoothed and raw are similar: hypothesis is WRONG, per-event output
is genuinely noisy. Next step would be to investigate model training
(horizon collapse) OR the AUC=0.66 measurement definition.
Either way, definitive result from one smoke run.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
576 lines
27 KiB
Rust
576 lines
27 KiB
Rust
//! BacktestHarness — orchestrator wiring ml-alpha's MultiHorizonLoader
|
|
//! and PerceptionTrainer (in inference role) into the LobSimCuda.
|
|
//!
|
|
//! Design: the harness reuses the trainer's loader + forward chain
|
|
//! (PerceptionTrainer::evaluate_batched, accessed via `forward_only`)
|
|
//! so train-vs-deploy skew is structurally impossible. The input struct
|
|
//! (`Mbp10RawInput`), feature-assembly cubin, and full perception
|
|
//! forward are the same code paths used in training. The harness's
|
|
//! only job is to walk the chronological snapshot stream, maintain a
|
|
//! sliding K-window for the recurrent Mamba2 context, drive the sim's
|
|
//! decision loop with the trainer's per-horizon probability output.
|
|
|
|
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::trainer::perception::PerceptionTrainer;
|
|
use ml_core::device::MlDevice;
|
|
use std::collections::VecDeque;
|
|
use std::path::PathBuf;
|
|
|
|
use crate::sim::LobSimCuda;
|
|
|
|
#[derive(Clone, Debug)]
|
|
pub struct BacktestHarnessConfig {
|
|
pub data_root: PathBuf,
|
|
pub predecoded_dir: PathBuf,
|
|
pub n_parallel: usize,
|
|
pub target_annual_vol_units: f32,
|
|
pub annualisation_factor: f32,
|
|
pub max_lots: u16,
|
|
/// Max number of inference inputs to consume. 0 = exhaust loader.
|
|
pub max_events: u64,
|
|
/// Per-backtest Strategy compositions. Empty = every cell uses the
|
|
/// hardcoded default policy (WeightedByRealizedSharpe across all 5
|
|
/// horizons). Non-empty: len MUST equal n_parallel; each entry is
|
|
/// flattened to bytecode and uploaded to its backtest slot.
|
|
pub strategies: Vec<crate::policy::Strategy>,
|
|
/// Total submission→fill latency in nanoseconds. 0 = immediate
|
|
/// (legacy path). 100_000_000 = 100ms IBKR + Scaleway baseline.
|
|
/// When non-zero, the orchestrator submits aggressive IOC limits
|
|
/// with `active=2`/`arrival_ts_ns = current + latency_ns` so the
|
|
/// book can move under them during the in-flight window.
|
|
pub latency_ns: u32,
|
|
/// Cold-start Kelly fraction floor — used when isv_kelly is at
|
|
/// sentinel (no closed trades yet). Per
|
|
/// pearl_blend_formulas_must_have_permanent_floor: max(floor, real),
|
|
/// not blend. Default 0.10 = trade conservatively until Kelly state
|
|
/// bootstraps.
|
|
pub kelly_frac_floor: f32,
|
|
/// Cold-start Sharpe weight floor — applied to recent_sharpe when
|
|
/// aggregating signed-sizes across horizons. Lets the cross-horizon
|
|
/// sum produce a non-zero size before recent_sharpe is populated.
|
|
/// Default 0.10 = uniform 1/5 weight per horizon at cold-start.
|
|
pub sharpe_weight_floor: f32,
|
|
/// P4: threshold gate — absolute conviction cutoff. When
|
|
/// `max_h |alpha[h] - 0.5| * 2 < threshold`, the decision kernels
|
|
/// emit noop. Default 0.0 = gate disabled (every signal passes).
|
|
pub threshold: f32,
|
|
/// P4: per-fill cost (price-units / lot / side). Deducted from
|
|
/// pos.realized_pnl and accumulated into total_fees_per_b_d at
|
|
/// every fill. Default 0.0 = no cost (frictionless).
|
|
pub cost_per_lot_per_side: f32,
|
|
/// P6: optional per-backtest variant names. When Some, write_artifacts
|
|
/// writes subdirs as `sim_<name>` instead of `cell_NNNN`. Length MUST
|
|
/// equal n_parallel when set. None preserves the legacy cell_NNNN
|
|
/// naming for non-grid-pack callers (smoke, fixtures).
|
|
pub variant_names: Option<Vec<String>>,
|
|
/// P6: optional pre-built BatchedSimConfig (one entry per backtest =
|
|
/// per variant). When Some, the harness uses this directly instead
|
|
/// of building from_uniform off the scalar fields above. Length MUST
|
|
/// equal n_parallel when set.
|
|
pub sim_config_override: Option<crate::sim::BatchedSimConfig>,
|
|
}
|
|
|
|
pub struct BacktestHarness {
|
|
cfg: BacktestHarnessConfig,
|
|
/// Per-backtest sim parameter arrays. P1 onwards is the source of
|
|
/// truth for sim-side scalar params; cfg's scalar fields are still
|
|
/// used to construct this via BatchedSimConfig::from_uniform at
|
|
/// harness new(), but the run-loop reads sim_config, not cfg.
|
|
sim_config: crate::sim::BatchedSimConfig,
|
|
loader: MultiHorizonLoader,
|
|
/// PerceptionTrainer in inference role — owns the trunk (loaded from
|
|
/// Checkpoint) and the kernel-launch scratches. CRT Phase A0.5: the
|
|
/// run loop now drives `forward_step_into` (incremental SSM advance)
|
|
/// every event so the encoder state stays current; decisions remain
|
|
/// stride-gated until A1 deletes the stride. `forward_only` is no
|
|
/// longer called from the harness's hot path; it remains accessible
|
|
/// for setup paths (e.g., A1's bit-identical state seeding via
|
|
/// `seed_step_state_from_forward_only`).
|
|
trainer: PerceptionTrainer,
|
|
/// Sliding K-window of recent snapshots — kept as a window-fill
|
|
/// gate so the harness can detect when the encoder has seen enough
|
|
/// events to produce a meaningful prediction (`len() == seq_len`).
|
|
/// Once full, the snapshot contents are no longer the source of
|
|
/// truth for the forward — `forward_step_into`'s persistent SSM
|
|
/// state is.
|
|
snapshot_window: VecDeque<Mbp10RawInput>,
|
|
/// Window capacity = trainer's seq_len, captured at construction.
|
|
seq_len: usize,
|
|
sim: LobSimCuda,
|
|
decision_count: u64,
|
|
event_count: u64,
|
|
/// Per-cell cumulative P&L curve in USD, sampled once per event.
|
|
/// `pnl_curves[b][i]` = realized_pnl USD for backtest b after event i.
|
|
pnl_curves: Vec<Vec<f32>>,
|
|
/// CRT Phase A0.5 corrective: conviction logging moved on-device.
|
|
/// Per-decision max-conviction values are written by the sim's
|
|
/// `record_max_conviction` kernel to `LobSimCuda::convictions_d`;
|
|
/// `BacktestHarness::run` returns the populated Vec at the end via a
|
|
/// single DtoV. The previous host-side `Vec<f32>` field + per-event
|
|
/// `last_probs` cache are removed because the probs no longer leave
|
|
/// device memory on the hot path (forward_step_into writes directly
|
|
/// into the sim's alpha_probs_d).
|
|
///
|
|
/// Host counter of decisions recorded into `sim.convictions_d`.
|
|
/// Equal to `decision_count` after `run()` completes. Used to size
|
|
/// the end-of-run DtoV correctly.
|
|
convictions_recorded: u32,
|
|
}
|
|
|
|
impl BacktestHarness {
|
|
/// Construct with an externally-built MlDevice + PerceptionTrainer.
|
|
/// Caller loads the trainer from a Checkpoint via
|
|
/// `PerceptionTrainer::from_checkpoint`. The harness reads
|
|
/// `trainer.config().seq_len` to size its sliding snapshot window.
|
|
pub fn new(
|
|
cfg: BacktestHarnessConfig,
|
|
dev: &MlDevice,
|
|
trainer: PerceptionTrainer,
|
|
) -> Result<Self> {
|
|
if !cfg.strategies.is_empty() {
|
|
anyhow::ensure!(
|
|
cfg.strategies.len() == cfg.n_parallel,
|
|
"strategies len {} ≠ n_parallel {}",
|
|
cfg.strategies.len(),
|
|
cfg.n_parallel
|
|
);
|
|
}
|
|
let files = discover_mbp10_files_sorted(&cfg.data_root)?;
|
|
let loader_cfg = MultiHorizonLoaderConfig {
|
|
files,
|
|
predecoded_dir: cfg.predecoded_dir.clone(),
|
|
seq_len: 1,
|
|
horizons: [30, 100, 300, 1000, 6000],
|
|
n_max_sequences: 0,
|
|
seed: 0,
|
|
// A1: decision_stride removed; inference mode walks every event at stride=1.
|
|
inference_only: true,
|
|
};
|
|
let loader = MultiHorizonLoader::new(&loader_cfg)?;
|
|
|
|
let seq_len = trainer.config().seq_len;
|
|
anyhow::ensure!(
|
|
trainer.config().n_batch == 1,
|
|
"BacktestHarness requires PerceptionTrainer with n_batch=1 (got {})",
|
|
trainer.config().n_batch
|
|
);
|
|
|
|
let mut sim = LobSimCuda::new(cfg.n_parallel, dev)?;
|
|
|
|
// P6: if sim_config_override is provided (sweep runner grid-pack
|
|
// flow), use it directly; otherwise build a uniform broadcast from
|
|
// scalar cfg fields (smoke + fixture flows).
|
|
let sim_config = match cfg.sim_config_override.clone() {
|
|
Some(bc) => {
|
|
anyhow::ensure!(
|
|
bc.n_backtests() == cfg.n_parallel,
|
|
"sim_config_override has {} variants but n_parallel = {}",
|
|
bc.n_backtests(), cfg.n_parallel,
|
|
);
|
|
bc
|
|
}
|
|
None => crate::sim::BatchedSimConfig::from_uniform(
|
|
cfg.n_parallel,
|
|
&crate::sim::UniformSimParams {
|
|
target_annual_vol_units: cfg.target_annual_vol_units,
|
|
annualisation_factor: cfg.annualisation_factor,
|
|
max_lots: cfg.max_lots,
|
|
latency_ns: cfg.latency_ns,
|
|
kelly_frac_floor: cfg.kelly_frac_floor,
|
|
sharpe_weight_floor: cfg.sharpe_weight_floor,
|
|
threshold: cfg.threshold,
|
|
cost_per_lot_per_side: cfg.cost_per_lot_per_side,
|
|
max_hold_ns: 0, // default disabled; set via sim_config_override
|
|
min_reasonable_px: 0.0,
|
|
max_reasonable_px: f32::INFINITY,
|
|
delta_floor: 1.0, // CRT.1 C1.3: default 1-lot band
|
|
},
|
|
),
|
|
};
|
|
|
|
// Upload caller-provided strategies (one program per backtest).
|
|
for (b, strat) in cfg.strategies.iter().enumerate() {
|
|
let prog = strat.flatten();
|
|
sim.upload_program(b, &prog)?;
|
|
}
|
|
|
|
// S1.19: upload per-backtest price-range bounds from sim_config.
|
|
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();
|
|
Ok(Self {
|
|
cfg,
|
|
sim_config,
|
|
loader,
|
|
trainer,
|
|
snapshot_window: VecDeque::with_capacity(seq_len),
|
|
seq_len,
|
|
sim,
|
|
decision_count: 0,
|
|
event_count: 0,
|
|
pnl_curves,
|
|
convictions_recorded: 0,
|
|
})
|
|
}
|
|
|
|
/// Drive the chronological event stream → trunk inference → sim
|
|
/// decision loop until either the stream is exhausted or
|
|
/// cfg.max_events is reached. See spec §7 orchestrator code.
|
|
pub fn run(&mut self) -> Result<RunStats> {
|
|
let mut total_decisions = 0u64;
|
|
// Periodic progress line — without this, multi-million-event
|
|
// runs look identical to a deadlock in pod logs.
|
|
const PROGRESS_EVERY: u64 = 1_000_000;
|
|
let mut next_progress_event = PROGRESS_EVERY;
|
|
let started = std::time::Instant::now();
|
|
|
|
while let Some(raw) = self.loader.next_inference_input()? {
|
|
// Apply snapshot to every backtest's book.
|
|
self.sim.apply_snapshot(&raw.bid_px, &raw.bid_sz, &raw.ask_px, &raw.ask_sz)?;
|
|
|
|
// Process resting orders (in-flight promotion, queue decay,
|
|
// marketability check, stop triggers, OCO). Trade-flow signal
|
|
// comes from the loader's L1-delta tick-rule inference (see
|
|
// ml_alpha::data::loader::infer_signed_trade_flow). Positive =
|
|
// buyer-initiated, negative = seller-initiated.
|
|
self.sim.step_resting_orders(raw.ts_ns, raw.trade_signed_vol)?;
|
|
|
|
// Maintain the sliding K-window for the recurrent forward.
|
|
// Push the current snapshot, evict the oldest if at capacity.
|
|
if self.snapshot_window.len() == self.seq_len {
|
|
self.snapshot_window.pop_front();
|
|
}
|
|
self.snapshot_window.push_back(raw.clone());
|
|
|
|
// CRT A1: once the window is full, fire BOTH forward_step_into AND
|
|
// step_decision_with_latency on every event. decision_stride is gone;
|
|
// the window-full guard is the only gate.
|
|
// forward_step_into (A0.5) advances the SSM state and writes
|
|
// per-horizon probs into alpha_probs_d. step_decision_with_latency
|
|
// reads alpha_probs_d immediately after — zero CPU roundtrip.
|
|
if self.snapshot_window.len() == self.seq_len {
|
|
self.trainer
|
|
.forward_step_into(&raw, self.sim.alpha_probs_d_mut())
|
|
.context("trainer.forward_step_into")?;
|
|
|
|
// Side-channel: record this decision's max_conviction
|
|
// on-device for the threshold-tuning percentile
|
|
// computation. Kernel reads `alpha_probs_d` and writes
|
|
// one f32 into `sim.convictions_d[convictions_recorded]`.
|
|
// Single end-of-run DtoV in `read_convictions` materialises
|
|
// the host log. Doing it BEFORE step_decision so the log
|
|
// captures every decision attempt, including those the
|
|
// threshold gate would skip.
|
|
self.sim.record_max_conviction(self.convictions_recorded)?;
|
|
self.convictions_recorded = self.convictions_recorded.saturating_add(1);
|
|
|
|
self.sim.step_decision_with_latency(raw.ts_ns, &self.sim_config)?;
|
|
self.decision_count += 1;
|
|
total_decisions += 1;
|
|
}
|
|
|
|
// Sample per-cell realised P&L (price-units → USD via $50/index-pt).
|
|
for b in 0..self.cfg.n_parallel {
|
|
let pos = self.sim.read_pos(b)?;
|
|
self.pnl_curves[b].push(pos.realized_pnl * 50.0);
|
|
}
|
|
|
|
self.event_count += 1;
|
|
if self.event_count >= next_progress_event {
|
|
let elapsed = started.elapsed().as_secs_f32();
|
|
let rate = self.event_count as f32 / elapsed.max(1e-3);
|
|
let trades = self.sim.read_total_trade_count().unwrap_or(0);
|
|
eprintln!(
|
|
"progress: events={} decisions={} trades={} elapsed={:.0}s rate={:.0}ev/s",
|
|
self.event_count, total_decisions, trades, elapsed, rate,
|
|
);
|
|
next_progress_event += PROGRESS_EVERY;
|
|
}
|
|
if self.cfg.max_events > 0 && self.event_count >= self.cfg.max_events {
|
|
break;
|
|
}
|
|
}
|
|
|
|
// S1.20: NaN instrumentation summary. Print ONCE per smoke so the
|
|
// cluster log shows which kernel-side arithmetic site produced the
|
|
// residual 97 zero + 85 i32::MAX sentinel trade records.
|
|
match self.sim.read_nan_counters() {
|
|
Ok(counters) => {
|
|
let sum_u32 = |v: &[u32]| -> u64 { v.iter().map(|&x| x as u64).sum() };
|
|
eprintln!(
|
|
"nan_counters: avg_px={} realised={} realized_pnl={} zero_vwap_open={} saturated_vwap_open={} defensive_clamp={}",
|
|
sum_u32(&counters.nan_avg_px),
|
|
sum_u32(&counters.nan_realised),
|
|
sum_u32(&counters.nan_realized_pnl),
|
|
sum_u32(&counters.zero_vwap_at_open),
|
|
sum_u32(&counters.saturated_vwap_at_open),
|
|
sum_u32(&counters.defensive_exit_clamp),
|
|
);
|
|
}
|
|
Err(e) => eprintln!("nan_counters read failed: {e}"),
|
|
}
|
|
|
|
// S1.21: per-vwap-write-site counters. Printed after v1 line so both
|
|
// appear together in the smoke log.
|
|
match self.sim.read_nan_counters_v2() {
|
|
Ok(c) => {
|
|
let sum_u32 = |v: &[u32]| -> u64 { v.iter().map(|&x| x as u64).sum() };
|
|
let last_bad_vwap_b0 = c.last_bad_vwap.first().copied().unwrap_or(0.0);
|
|
let last_bad_path_b0 = c.last_bad_path.first().copied().unwrap_or(0);
|
|
eprintln!(
|
|
"nan_counters_v2: zero_flat={} huge_flat={} zero_scale={} huge_scale={} zero_flip={} huge_flip={} session_gap_was_bad={} | b0_last_bad_vwap={} b0_last_bad_path={}",
|
|
sum_u32(&c.vwap_zero_open_flat),
|
|
sum_u32(&c.vwap_huge_open_flat),
|
|
sum_u32(&c.vwap_zero_scale_in),
|
|
sum_u32(&c.vwap_huge_scale_in),
|
|
sum_u32(&c.vwap_zero_flip),
|
|
sum_u32(&c.vwap_huge_flip),
|
|
sum_u32(&c.vwap_session_gap_was_bad),
|
|
last_bad_vwap_b0,
|
|
last_bad_path_b0,
|
|
);
|
|
}
|
|
Err(e) => eprintln!("nan_counters_v2 read failed: {e}"),
|
|
}
|
|
|
|
// S2.1/S2.2: generic stop-controller diagnostic counters. The 6
|
|
// max_hold-specific decision-rate counters were removed when max_hold
|
|
// enforcement moved to resting_orders_step (event-rate). Two remain:
|
|
// kernel_calls — stop_check_isv entries with open position (SL/trail active)
|
|
// seed_saw_force_flat — force-flat signals seen by seed_inflight (SL/trail)
|
|
match self.sim.read_max_hold_counters() {
|
|
Ok(c) => {
|
|
let sum_u32 = |v: &[u32]| -> u64 { v.iter().map(|&x| x as u64).sum() };
|
|
eprintln!(
|
|
"stop_ctrl_counters: kernel_calls={} seed_saw_force_flat={}",
|
|
sum_u32(&c.mh_kernel_calls),
|
|
sum_u32(&c.mh_force_flat_seen_by_seed),
|
|
);
|
|
}
|
|
Err(e) => eprintln!("stop_ctrl_counters read failed: {e}"),
|
|
}
|
|
|
|
// CRT.diag: empirical measurement battery (Groups A-D). Drives the
|
|
// next round of CRT.1 threshold design — measure structural truth
|
|
// BEFORE another tuning pass. Single end-of-run dump, observe-only.
|
|
match self.sim.read_diagnostics() {
|
|
Ok(diag) => {
|
|
let n_b = self.cfg.n_parallel;
|
|
// Horizons match crates/ml-alpha/src/heads.rs::HORIZONS.
|
|
let horizons: [&str; 5] = ["30", "100", "300", "1000", "6000"];
|
|
let n_horizons = horizons.len();
|
|
|
|
// Group A — per-horizon signal persistence.
|
|
for h in 0..n_horizons {
|
|
let flips: u64 = (0..n_b)
|
|
.map(|b| diag.flip_count[b * n_horizons + h] as u64)
|
|
.sum();
|
|
let sum_run: u64 = (0..n_b)
|
|
.map(|b| diag.sum_run_length[b * n_horizons + h])
|
|
.sum();
|
|
let mean_run = if flips > 0 {
|
|
sum_run as f64 / flips as f64
|
|
} else {
|
|
f64::NAN
|
|
};
|
|
eprintln!(
|
|
"crt_diag h{}: flips={} mean_run_len={:.1} events",
|
|
horizons[h], flips, mean_run,
|
|
);
|
|
let bucket_labels = ["1-9", "10-99", "100-999", "1k-9.9k", "10k+"];
|
|
let mut hist_str = String::new();
|
|
for bk in 0..5 {
|
|
let count: u64 = (0..n_b)
|
|
.map(|b| {
|
|
diag.run_length_hist[(b * n_horizons + h) * 5 + bk] as u64
|
|
})
|
|
.sum();
|
|
hist_str.push_str(&format!(" {}:{}", bucket_labels[bk], count));
|
|
}
|
|
eprintln!("crt_diag h{} run_length_hist:{}", horizons[h], hist_str);
|
|
|
|
// CRT.diag.2 Group E: SMOOTHED-direction flip rate head-to-head
|
|
// vs raw. If smoothed mean_run_len ≫ raw mean_run_len, the
|
|
// hypothesis (per-event output noise masks a slower signal)
|
|
// is supported. If similar, hypothesis is falsified.
|
|
let smoothed_flips: u64 = (0..n_b)
|
|
.map(|b| diag.smoothed_flip_count[b * n_horizons + h] as u64)
|
|
.sum();
|
|
let smoothed_sum_run: u64 = (0..n_b)
|
|
.map(|b| diag.smoothed_sum_run_length[b * n_horizons + h])
|
|
.sum();
|
|
let smoothed_mean_run = if smoothed_flips > 0 {
|
|
smoothed_sum_run as f64 / smoothed_flips as f64
|
|
} else {
|
|
f64::NAN
|
|
};
|
|
eprintln!(
|
|
"crt_diag h{} smoothed: flips={} mean_run_len={:.1} events (vs raw {} / {:.1})",
|
|
horizons[h], smoothed_flips, smoothed_mean_run,
|
|
flips, mean_run,
|
|
);
|
|
}
|
|
|
|
// Group B — smoothed-conviction histogram.
|
|
let mut conv_str = String::new();
|
|
for bk in 0..10 {
|
|
let count: u64 = (0..n_b)
|
|
.map(|b| diag.conv_hist[b * 10 + bk] as u64)
|
|
.sum();
|
|
let lo = bk as f32 / 10.0;
|
|
let hi = (bk + 1) as f32 / 10.0;
|
|
conv_str.push_str(&format!(" [{:.1}-{:.1}]:{}", lo, hi, count));
|
|
}
|
|
eprintln!("crt_diag conv_ema_hist:{}", conv_str);
|
|
|
|
// Group C — hold-time histogram.
|
|
let hold_labels = ["<1s", "1-10s", "10-60s", "1-10m", "10-60m", ">1h"];
|
|
let mut hold_str = String::new();
|
|
for bk in 0..6 {
|
|
let count: u64 = (0..n_b)
|
|
.map(|b| diag.hold_hist[b * 6 + bk] as u64)
|
|
.sum();
|
|
hold_str.push_str(&format!(" {}:{}", hold_labels[bk], count));
|
|
}
|
|
eprintln!("crt_diag hold_time_hist:{}", hold_str);
|
|
|
|
// Group D — outcome by entry-conviction.
|
|
for bk in 0..10 {
|
|
let n_total: u64 = (0..n_b)
|
|
.map(|b| diag.outcome_n[b * 10 + bk] as u64)
|
|
.sum();
|
|
let n_wins: u64 = (0..n_b)
|
|
.map(|b| diag.outcome_n_wins[b * 10 + bk] as u64)
|
|
.sum();
|
|
let sum_pnl: f64 = (0..n_b)
|
|
.map(|b| diag.outcome_sum_pnl[b * 10 + bk] as f64)
|
|
.sum();
|
|
let mean_pnl = if n_total > 0 {
|
|
sum_pnl / n_total as f64
|
|
} else {
|
|
f64::NAN
|
|
};
|
|
let win_rate = if n_total > 0 {
|
|
n_wins as f64 / n_total as f64
|
|
} else {
|
|
f64::NAN
|
|
};
|
|
let lo = bk as f32 / 10.0;
|
|
let hi = (bk + 1) as f32 / 10.0;
|
|
eprintln!(
|
|
"crt_diag outcome_by_entry_conv[{:.1}-{:.1}]: n={} win_rate={:.2}% mean_pnl_pu={:.3}",
|
|
lo, hi, n_total, win_rate * 100.0, mean_pnl,
|
|
);
|
|
}
|
|
}
|
|
Err(e) => eprintln!("crt_diag read failed: {e}"),
|
|
}
|
|
|
|
Ok(RunStats {
|
|
events_processed: self.event_count,
|
|
decisions_taken: total_decisions,
|
|
})
|
|
}
|
|
|
|
/// After `run()` completes, write per-cell artifacts to
|
|
/// `<out_dir>/cell_<b>/{summary.json,trades.csv,pnl_curve.bin}`.
|
|
pub fn write_artifacts(&self, out_dir: &std::path::Path) -> Result<()> {
|
|
std::fs::create_dir_all(out_dir)
|
|
.with_context(|| format!("create out dir {}", out_dir.display()))?;
|
|
|
|
// Write the threshold-tuning side-channel ONCE per cell (shared
|
|
// across all backtests in a batched cell because forward_step_into
|
|
// writes a single broadcast alpha_probs into the sim). Raw
|
|
// convictions.bin (little-endian f32) + conviction_percentiles.json
|
|
// with the pre-computed p60/p70/p80/p90/p95 values.
|
|
//
|
|
// CRT Phase A0.5 corrective: convictions live on-device during the
|
|
// run; one DtoV here materialises the populated prefix. Matches
|
|
// the lifecycle of the trade_log / max_hold counter reads also
|
|
// happening at end-of-run.
|
|
let conviction_log = self.sim
|
|
.read_convictions(self.convictions_recorded as usize)
|
|
.context("read on-device conviction history")?;
|
|
if !conviction_log.is_empty() {
|
|
let convictions_path = out_dir.join("convictions.bin");
|
|
let mut bytes = Vec::with_capacity(conviction_log.len() * 4);
|
|
for v in &conviction_log {
|
|
bytes.extend_from_slice(&v.to_le_bytes());
|
|
}
|
|
std::fs::write(&convictions_path, &bytes)
|
|
.with_context(|| format!("write {}", convictions_path.display()))?;
|
|
|
|
let mut sorted = conviction_log.clone();
|
|
sorted.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));
|
|
let pct = |q: f32| -> f32 {
|
|
let idx = ((sorted.len() - 1) as f32 * q).round() as usize;
|
|
sorted[idx]
|
|
};
|
|
let pcts = serde_json::json!({
|
|
"n_decisions": sorted.len(),
|
|
"min": sorted.first().copied().unwrap_or(0.0),
|
|
"max": sorted.last().copied().unwrap_or(0.0),
|
|
"mean": sorted.iter().sum::<f32>() / sorted.len() as f32,
|
|
"p10": pct(0.10),
|
|
"p25": pct(0.25),
|
|
"p50": pct(0.50),
|
|
"p60": pct(0.60),
|
|
"p70": pct(0.70),
|
|
"p80": pct(0.80),
|
|
"p90": pct(0.90),
|
|
"p95": pct(0.95),
|
|
"p99": pct(0.99),
|
|
});
|
|
let pcts_path = out_dir.join("conviction_percentiles.json");
|
|
std::fs::write(&pcts_path, serde_json::to_string_pretty(&pcts)?)
|
|
.with_context(|| format!("write {}", pcts_path.display()))?;
|
|
eprintln!(
|
|
"convictions: n={} min={:.4} p60={:.4} p70={:.4} p80={:.4} p90={:.4} p95={:.4} max={:.4}",
|
|
sorted.len(),
|
|
sorted.first().copied().unwrap_or(0.0),
|
|
pct(0.60), pct(0.70), pct(0.80), pct(0.90), pct(0.95),
|
|
sorted.last().copied().unwrap_or(0.0),
|
|
);
|
|
}
|
|
if let Some(names) = self.cfg.variant_names.as_ref() {
|
|
anyhow::ensure!(
|
|
names.len() == self.cfg.n_parallel,
|
|
"variant_names len {} != n_parallel {}",
|
|
names.len(), self.cfg.n_parallel,
|
|
);
|
|
}
|
|
for b in 0..self.cfg.n_parallel {
|
|
// P6: when sweep runner provides variant names, dir = sim_<name>
|
|
// per spec §3.3. Otherwise fall back to legacy cell_NNNN naming.
|
|
let subdir_name = match self.cfg.variant_names.as_ref() {
|
|
Some(names) => format!("sim_{}", names[b]),
|
|
None => format!("cell_{b:04}"),
|
|
};
|
|
let cell_dir = out_dir.join(subdir_name);
|
|
std::fs::create_dir_all(&cell_dir)
|
|
.with_context(|| format!("create cell dir {}", cell_dir.display()))?;
|
|
let records = self.sim.read_trade_records(b)?;
|
|
let curve = &self.pnl_curves[b];
|
|
let summary = crate::artifacts::compute_summary(&records, curve);
|
|
crate::artifacts::write_summary(&cell_dir.join("summary.json"), &summary)?;
|
|
crate::artifacts::write_trades_csv(&cell_dir.join("trades.csv"), &records)?;
|
|
crate::artifacts::write_pnl_curve_bin(&cell_dir.join("pnl_curve.bin"), curve)?;
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
pub fn sim(&self) -> &LobSimCuda { &self.sim }
|
|
pub fn event_count(&self) -> u64 { self.event_count }
|
|
pub fn decision_count(&self) -> u64 { self.decision_count }
|
|
}
|
|
|
|
#[derive(Clone, Copy, Debug)]
|
|
pub struct RunStats {
|
|
pub events_processed: u64,
|
|
pub decisions_taken: u64,
|
|
}
|