ff9207e7bb
Mirrors the alpha_train CLI flag (--kernel-step-trace <path>) into fxt-backtest. The PerceptionTrainerConfig already accepts the path; this commit just exposes the CLI surface and propagates through: fxt-backtest run --kernel-step-trace <path> -> BacktestHarnessConfig.kernel_step_trace_path -> PerceptionTrainerConfig.kernel_step_trace_path Sweep YAML gains a `kernel_step_trace: Option<PathBuf>` base field. Argo lob-backtest-sweep-template gains a `kernel-step-trace` workflow parameter (default disabled; when set, --features kernel-step-trace is passed to cargo build AND --kernel-step-trace to fxt-backtest). Gated behind the `kernel-step-trace` Cargo feature (matching alpha_train). When disabled (default), zero overhead. Enables per-step JSONL diagnostic emission from inference kernels -- needed for Smoke 2 deep-dive after sampling histograms (in parallel investigation) localize the per-horizon dynamics bottleneck. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
632 lines
28 KiB
Rust
632 lines
28 KiB
Rust
//! `fxt-backtest` — Real-LOB GPU backtest CLI.
|
||
//!
|
||
//! Two subcommands:
|
||
//!
|
||
//! fxt-backtest run --data <mbp10-dir> [--policy-grid <yaml>] ...
|
||
//! Runs the LOB backtest harness on a directory of MBP-10 dbn files.
|
||
//! Writes per-cell artifacts under --out/cell_<NNNN>/.
|
||
//!
|
||
//! fxt-backtest aggregate <sweep-dir>
|
||
//! Walks every cell directory under <sweep-dir>, parses summary.json,
|
||
//! emits aggregate.parquet + pareto_frontier.json at the sweep-dir root.
|
||
//!
|
||
//! See docs/superpowers/specs/2026-05-18-real-lob-integration-design.md §7.
|
||
|
||
use anyhow::{Context, Result};
|
||
use clap::{Parser, Subcommand};
|
||
use ml_alpha::cfc::trunk::CfcConfig;
|
||
use ml_alpha::heads::N_HORIZONS;
|
||
use ml_alpha::trainer::perception::{PerceptionTrainer, PerceptionTrainerConfig};
|
||
use ml_backtesting::aggregate::{aggregate_sweep_dir, emit_deployability_verdict};
|
||
use ml_backtesting::harness::{BacktestHarness, BacktestHarnessConfig};
|
||
use ml_backtesting::policy::Strategy;
|
||
use ml_core::device::MlDevice;
|
||
use std::path::PathBuf;
|
||
|
||
#[derive(Parser, Debug)]
|
||
#[command(name = "fxt-backtest", version, about = "Real-LOB GPU backtest CLI", long_about = None)]
|
||
struct Cli {
|
||
#[command(subcommand)]
|
||
cmd: Cmd,
|
||
}
|
||
|
||
#[derive(Subcommand, Debug)]
|
||
enum Cmd {
|
||
/// Run a single backtest over an MBP-10 dataset.
|
||
Run(RunArgs),
|
||
/// Aggregate per-cell summary.json files in a sweep directory into
|
||
/// aggregate.parquet + pareto_frontier.json.
|
||
Aggregate(AggregateArgs),
|
||
/// Run a parameter sweep: iterate over a grid of `Run` configs,
|
||
/// writing each cell's artifacts to `<out>/<cell_name>/`, then
|
||
/// automatically invoke `aggregate` on the resulting directory.
|
||
/// All cells run sequentially on the same GPU (single-machine);
|
||
/// for cluster fan-out use `argo-alpha-sweep.sh` (see infra/argo).
|
||
Sweep(SweepArgs),
|
||
/// Emit a deployability verdict from a completed sweep. Reads the
|
||
/// per-cell summary.json files at the realistic (1.0 tick, 200 ms)
|
||
/// and stress (1.5 tick, 400 ms) anchors against the supplied
|
||
/// pre-registered threshold + walk-forward window IDs, classifies
|
||
/// into Pass-robust / Pass-nominal / Fail-inconclusive / Fail /
|
||
/// Fail-degenerate per spec §3.5, and writes the audit JSON.
|
||
Verdict(VerdictArgs),
|
||
}
|
||
|
||
#[derive(Parser, Debug)]
|
||
struct RunArgs {
|
||
/// Root directory containing MBP-10 .dbn.zst files (and predecoded sidecars).
|
||
#[arg(long)]
|
||
data: PathBuf,
|
||
/// Directory containing predecoded sidecars; defaults to --data.
|
||
#[arg(long)]
|
||
predecoded_dir: Option<PathBuf>,
|
||
/// Number of parallel backtest cells (one CUDA block per cell).
|
||
#[arg(long, default_value_t = 1)]
|
||
n_parallel: usize,
|
||
/// Optional YAML file listing per-cell Strategy compositions.
|
||
/// If omitted, every cell gets Strategy::default_for(--max-lots).
|
||
#[arg(long)]
|
||
policy_grid: Option<PathBuf>,
|
||
/// Total submission→fill latency in nanoseconds (IBKR + Scaleway
|
||
/// baseline = 100_000_000 = 100ms). Propagated to the resting-order
|
||
/// engine so in-flight orders wait for arrival_ts_ns before crossing.
|
||
/// 0 = immediate-match path (legacy).
|
||
#[arg(long, default_value_t = 100_000_000)]
|
||
latency_ns: u32,
|
||
/// Target annual volatility in price-unit lot-equivalents
|
||
/// (see spec §5 IsvKellyState.target_annual_vol).
|
||
#[arg(long, default_value_t = 50.0)]
|
||
target_annual_vol_units: f32,
|
||
/// Trade-rate annualisation factor — non-overlapping convention,
|
||
/// 825 (K=6000 holding × 250 trading days) per pearl_phase1d4 default.
|
||
#[arg(long, default_value_t = 825.0)]
|
||
annualisation_factor: f32,
|
||
/// Per-leaf max concurrent lots (Strategy::default_for argument).
|
||
#[arg(long, default_value_t = 5)]
|
||
max_lots: u16,
|
||
/// Cap on events processed; 0 = exhaust the input stream.
|
||
#[arg(long, default_value_t = 0)]
|
||
max_events: u64,
|
||
/// Cold-start Kelly fraction floor (see BacktestHarnessConfig +
|
||
/// pearl_blend_formulas_must_have_permanent_floor). When isv_kelly
|
||
/// has no closed-trade history, kelly_frac defaults to this floor so
|
||
/// the first trade can fire and bootstrap state.
|
||
#[arg(long, default_value_t = 0.20)]
|
||
kelly_frac_floor: f32,
|
||
/// Cold-start Sharpe weight floor — applied to recent_sharpe in the
|
||
/// cross-horizon aggregation. Lets the aggregate produce a non-zero
|
||
/// signed size before recent_sharpe is populated.
|
||
#[arg(long, default_value_t = 0.10)]
|
||
sharpe_weight_floor: f32,
|
||
/// Random seed for trunk initialisation when --checkpoint is not given.
|
||
#[arg(long, default_value_t = 0xC0FFEE)]
|
||
seed: u64,
|
||
/// Path to a CfcTrunk bincode checkpoint (see ml-alpha
|
||
/// CfcTrunk::save_checkpoint). When set, overrides --seed and the
|
||
/// trunk runs with trained weights. Without it the trunk is
|
||
/// random-initialised — useful for plumbing tests, not real backtests.
|
||
#[arg(long)]
|
||
checkpoint: Option<PathBuf>,
|
||
/// Output directory; per-cell artifacts written to <out>/cell_NNNN/.
|
||
#[arg(long)]
|
||
out: PathBuf,
|
||
/// Enable per-step kernel-state JSONL trace to the given file.
|
||
///
|
||
/// Mirrors `alpha_train --kernel-step-trace`. When set, the
|
||
/// `PerceptionTrainer` constructed by the binary allocates the GPU
|
||
/// log ring + spawns a drain task that writes one JSONL record per
|
||
/// kernel emission to this path. Requires the `kernel-step-trace`
|
||
/// Cargo feature at compile time (proxied through to `ml-alpha`).
|
||
/// When unset (default), no ring allocated, zero overhead.
|
||
///
|
||
/// Example: `--kernel-step-trace /feature-cache/runs/<sha>/step_trace.jsonl`
|
||
#[arg(long)]
|
||
kernel_step_trace: Option<PathBuf>,
|
||
/// Per-horizon alpha_probs sampling stride for ALPHA Smoke 2
|
||
/// diagnostic. When set to N, every Nth inference call DtoHs the 5
|
||
/// alpha_probs and accumulates per-horizon mean/stddev/min/max + 10-bin
|
||
/// histogram (emitted at end of CRT.diag block).
|
||
#[arg(long)]
|
||
per_horizon_logit_sampling_stride: Option<u32>,
|
||
}
|
||
|
||
#[derive(Parser, Debug)]
|
||
struct AggregateArgs {
|
||
/// Sweep directory containing one subdirectory per cell, each with
|
||
/// a summary.json file.
|
||
sweep_dir: PathBuf,
|
||
}
|
||
|
||
#[derive(Parser, Debug)]
|
||
struct VerdictArgs {
|
||
/// Sweep directory containing one subdirectory per cell (named
|
||
/// `cell_<NNNN>_cost<C>_lat<L>_th<T>_<W>`), each with a summary.json.
|
||
sweep_dir: PathBuf,
|
||
/// Pre-registered threshold (from the threshold-tuning sweep on W0).
|
||
/// Read from `config/ml/v2_prod_thresholds.json` in production usage.
|
||
#[arg(long)]
|
||
threshold: f32,
|
||
/// Walk-forward window IDs, comma-separated. Typical:
|
||
/// "W1,W2,W3,W4" matching the 4 held-out quarters in
|
||
/// `config/ml/sweep_deployability.yaml`.
|
||
#[arg(long, default_value = "W1,W2,W3,W4")]
|
||
windows: String,
|
||
/// Training-run commit SHA (recorded in the audit JSON).
|
||
#[arg(long)]
|
||
training_sha: String,
|
||
/// Spec-revision commit SHA (recorded in the audit JSON).
|
||
#[arg(long)]
|
||
spec_sha: String,
|
||
/// Output JSON path. Convention: `deployability_verdict.json` at
|
||
/// repo root, committed as audit record.
|
||
#[arg(long)]
|
||
out: PathBuf,
|
||
}
|
||
|
||
#[derive(Parser, Debug)]
|
||
struct SweepArgs {
|
||
/// Sweep grid YAML file. See SweepGrid struct in main.rs for the
|
||
/// schema; each cell may override a subset of Run args.
|
||
#[arg(long)]
|
||
grid: PathBuf,
|
||
/// Root output directory for the sweep. One subdirectory created
|
||
/// per cell, plus aggregate.parquet + pareto_frontier.json at root.
|
||
#[arg(long)]
|
||
out: PathBuf,
|
||
/// Stop the sweep after this many cells (0 = run all). Useful for
|
||
/// smoke-testing a large grid.
|
||
#[arg(long, default_value_t = 0)]
|
||
max_cells: usize,
|
||
}
|
||
|
||
/// Sweep grid YAML schema. The `base` field provides defaults that
|
||
/// each `cells` entry may override on a per-field basis.
|
||
#[derive(Debug, serde::Deserialize)]
|
||
struct SweepGrid {
|
||
base: SweepBase,
|
||
cells: Vec<SweepCell>,
|
||
}
|
||
|
||
#[derive(Debug, serde::Deserialize, Clone)]
|
||
struct SweepBase {
|
||
/// Single data file path. Used by the legacy fan-out flow (one Argo
|
||
/// task per cell). For the P6 batched flow that varies window per
|
||
/// cell, prefer `data_template` with `{window}` interpolation.
|
||
#[serde(default)]
|
||
data: Option<PathBuf>,
|
||
/// P6: data path template with `{window}` placeholder. When set,
|
||
/// each cell's `window` field interpolates into this template to
|
||
/// produce the per-cell data path.
|
||
/// Example: /mnt/training-data/futures-baseline-mbp10/ES.FUT/ES.FUT_{window}.dbn.zst
|
||
#[serde(default)]
|
||
data_template: Option<String>,
|
||
#[serde(default)]
|
||
predecoded_dir: Option<PathBuf>,
|
||
#[serde(default = "default_n_parallel")] n_parallel: usize,
|
||
#[serde(default = "default_latency_ns")] latency_ns: u32,
|
||
#[serde(default = "default_target_vol")] target_annual_vol_units: f32,
|
||
#[serde(default = "default_ann_factor")] annualisation_factor: f32,
|
||
#[serde(default = "default_max_lots")] max_lots: u16,
|
||
#[serde(default)] max_events: u64,
|
||
#[serde(default = "default_seed")] seed: u64,
|
||
#[serde(default)] checkpoint: Option<PathBuf>,
|
||
#[serde(default = "default_kelly_frac_floor")] kelly_frac_floor: f32,
|
||
#[serde(default = "default_sharpe_weight_floor")] sharpe_weight_floor: f32,
|
||
/// P6: list of sim variants. When non-empty, each cell runs ONE
|
||
/// harness at n_parallel=variants.len() instead of fanning out one
|
||
/// harness per cell. Cells become a window axis; variants become the
|
||
/// (cost, latency, threshold) axis.
|
||
#[serde(default)] sim_variants: Vec<SimVariant>,
|
||
/// S1.19: instrument-aware price-range sanitization. Snapshots whose
|
||
/// top-of-book bid or ask falls outside [min, max] are skipped.
|
||
/// None = no lower bound (0.0 passed to kernel, which disables the check).
|
||
#[serde(default)] min_reasonable_px: Option<f32>,
|
||
/// None = no upper bound (f32::INFINITY passed to kernel).
|
||
#[serde(default)] max_reasonable_px: Option<f32>,
|
||
/// CRT.1 C1.3: no-trade band in lots. seed_inflight_limits_batched skips
|
||
/// seeding when |target_signed − effective| < delta_floor. Default 1.0 =
|
||
/// 1 lot (ES single-contract sizing).
|
||
#[serde(default = "default_delta_floor")] delta_floor: f32,
|
||
/// Optional per-step kernel-state JSONL trace path. Propagated to
|
||
/// every cell's `PerceptionTrainerConfig.kernel_step_trace_path` +
|
||
/// `BacktestHarnessConfig.kernel_step_trace_path`. Requires the
|
||
/// `kernel-step-trace` Cargo feature on the binary at compile time.
|
||
/// Sweep cells share the same trace file path — useful for single-cell
|
||
/// diagnostic sweeps; with N>1 cells the trace will interleave records
|
||
/// from every harness/trainer instance (per-cell isolation must be
|
||
/// handled by the operator setting distinct paths via a richer schema).
|
||
#[serde(default)] kernel_step_trace: Option<PathBuf>,
|
||
/// Per-horizon alpha_probs sampling stride (ALPHA Smoke 2 diagnostic).
|
||
/// `None` = disabled. When set, every Nth inference call DtoHs the 5
|
||
/// alpha_probs into a mapped-pinned buffer for per-horizon stats.
|
||
#[serde(default)] per_horizon_logit_sampling_stride: Option<u32>,
|
||
}
|
||
|
||
/// P6: per-sim-variant overrides for the batched-cell sweep flow.
|
||
/// `threshold` and `cost_per_lot_per_side` are required (no defaults —
|
||
/// these are the spec's primary sweep axes); other fields are optional
|
||
/// overrides on top of the SweepBase scalars.
|
||
#[derive(Debug, serde::Deserialize, Clone)]
|
||
struct SimVariant {
|
||
name: String,
|
||
threshold: f32,
|
||
cost_per_lot_per_side: f32,
|
||
#[serde(default)] latency_ns: Option<u32>,
|
||
#[serde(default)] target_annual_vol_units: Option<f32>,
|
||
#[serde(default)] annualisation_factor: Option<f32>,
|
||
#[serde(default)] max_lots: Option<u16>,
|
||
#[serde(default)] kelly_frac_floor: Option<f32>,
|
||
#[serde(default)] sharpe_weight_floor: Option<f32>,
|
||
/// Follow-up gp74n: max-hold force-close. 0 = disabled (default).
|
||
/// Existing YAMLs omit this field; serde(default) gives 0.
|
||
#[serde(default)] max_hold_ns: u64,
|
||
/// CRT.1 C1.3: per-variant override for the no-trade band. None = use
|
||
/// SweepBase.delta_floor.
|
||
#[serde(default)] delta_floor: Option<f32>,
|
||
}
|
||
|
||
fn default_n_parallel() -> usize { 1 }
|
||
fn default_latency_ns() -> u32 { 100_000_000 }
|
||
fn default_target_vol() -> f32 { 50.0 }
|
||
fn default_ann_factor() -> f32 { 825.0 }
|
||
fn default_max_lots() -> u16 { 5 }
|
||
fn default_seed() -> u64 { 0xC0FFEE }
|
||
// Sized so a strong-conviction signal (|p-0.5| ≥ 0.2 ⇒ sig_mag ≥ 0.4)
|
||
// produces ≥1 lot at cold-start for max_lots=5:
|
||
// ss = sig_mag * floor * cap_lots → 0.4 * 0.20 * 5 = 0.4 → rounds to 0
|
||
// ss = sig_mag * floor * cap_lots → 0.5 * 0.20 * 5 = 0.5 → rounds to 1
|
||
// Weaker signals stay at 0 lots cold-start. Once kelly state bootstraps
|
||
// (first close), `max(floor, computed_kelly)` keeps the floor as the
|
||
// minimum trading intensity.
|
||
fn default_kelly_frac_floor() -> f32 { 0.20 }
|
||
fn default_sharpe_weight_floor() -> f32 { 0.10 }
|
||
// CRT.1 C1.3: 1.0 lot — skip seeding when |delta| < 1 lot. Appropriate for
|
||
// ES single-contract sizing. Per-variant YAML can override.
|
||
fn default_delta_floor() -> f32 { 1.0 }
|
||
|
||
#[derive(Debug, serde::Deserialize, Clone)]
|
||
struct SweepCell {
|
||
name: String,
|
||
/// P6: window identifier (e.g., "2025-Q2") substituted into
|
||
/// SweepBase.data_template. When set, the cell's data path is
|
||
/// data_template.replace("{window}", window).
|
||
#[serde(default)] window: Option<String>,
|
||
#[serde(default)] latency_ns: Option<u32>,
|
||
#[serde(default)] target_annual_vol_units: Option<f32>,
|
||
#[serde(default)] annualisation_factor: Option<f32>,
|
||
#[serde(default)] max_lots: Option<u16>,
|
||
#[serde(default)] max_events: Option<u64>,
|
||
#[serde(default)] seed: Option<u64>,
|
||
#[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<()> {
|
||
tracing_subscriber::fmt::init();
|
||
let cli = Cli::parse();
|
||
match cli.cmd {
|
||
Cmd::Run(args) => run(args),
|
||
Cmd::Aggregate(args) => {
|
||
let out = aggregate_sweep_dir(&args.sweep_dir)?;
|
||
println!(
|
||
"aggregated {} cells; {} on Pareto frontier; parquet={}, frontier={}",
|
||
out.n_cells,
|
||
out.n_pareto,
|
||
out.parquet_path.display(),
|
||
out.frontier_path.display()
|
||
);
|
||
Ok(())
|
||
}
|
||
Cmd::Sweep(args) => sweep(args),
|
||
Cmd::Verdict(args) => {
|
||
let windows: Vec<&str> = args.windows.split(',').map(|s| s.trim()).collect();
|
||
let verdict = emit_deployability_verdict(
|
||
&args.sweep_dir,
|
||
args.threshold,
|
||
&windows,
|
||
&args.training_sha,
|
||
&args.spec_sha,
|
||
)
|
||
.context("emit_deployability_verdict")?;
|
||
let json = serde_json::to_string_pretty(&verdict)
|
||
.context("serialize deployability_verdict")?;
|
||
std::fs::write(&args.out, &json)
|
||
.with_context(|| format!("write {}", args.out.display()))?;
|
||
println!(
|
||
"verdict: {:?} (realistic median sharpe = {:.3}, max_dd_pct = {:.3}; stress median sharpe = {:.3}, max_dd_pct = {:.3}) -> {}",
|
||
verdict.verdict,
|
||
verdict.realistic.median_sharpe,
|
||
verdict.realistic.median_max_dd_pct,
|
||
verdict.stress.median_sharpe,
|
||
verdict.stress.median_max_dd_pct,
|
||
args.out.display(),
|
||
);
|
||
Ok(())
|
||
}
|
||
}
|
||
}
|
||
|
||
fn sweep(args: SweepArgs) -> Result<()> {
|
||
let grid_yaml = std::fs::read_to_string(&args.grid)
|
||
.with_context(|| format!("read grid yaml {}", args.grid.display()))?;
|
||
let grid: SweepGrid = serde_yaml::from_str(&grid_yaml).context("parse grid yaml")?;
|
||
let n_cells = if args.max_cells > 0 {
|
||
args.max_cells.min(grid.cells.len())
|
||
} else {
|
||
grid.cells.len()
|
||
};
|
||
anyhow::ensure!(n_cells > 0, "grid has no cells");
|
||
std::fs::create_dir_all(&args.out)
|
||
.with_context(|| format!("create sweep out {}", args.out.display()))?;
|
||
|
||
// P6: resolve sim variants once. When non-empty, each cell runs ONE
|
||
// harness at n_parallel=variants.len() with BatchedSimConfig::from_grid.
|
||
let resolved_variants = if grid.base.sim_variants.is_empty() {
|
||
Vec::new()
|
||
} else {
|
||
resolve_sim_variants(&grid.base)
|
||
};
|
||
let batched_mode = !resolved_variants.is_empty();
|
||
|
||
for (i, cell) in grid.cells.iter().take(n_cells).enumerate() {
|
||
let cell_out = args.out.join(&cell.name);
|
||
tracing::info!(
|
||
cell = cell.name.as_str(),
|
||
progress = format!("{}/{}", i + 1, n_cells),
|
||
"running sweep cell"
|
||
);
|
||
// Resolve per-cell data path. data_template + cell.window interpolation
|
||
// takes precedence over scalar `data` when both are present.
|
||
let cell_data = match (&grid.base.data_template, &cell.window) {
|
||
(Some(tpl), Some(window)) => PathBuf::from(tpl.replace("{window}", window)),
|
||
_ => grid.base.data.clone()
|
||
.ok_or_else(|| anyhow::anyhow!(
|
||
"sweep base must set either `data` (legacy) or `data_template + cell.window`"
|
||
))?,
|
||
};
|
||
if batched_mode {
|
||
// P6 batched flow: build BatchedSimConfig from grid, run ONE
|
||
// harness at n_parallel=variants.len() with variant_names plumbed
|
||
// through for sim_<name>/ output dirs.
|
||
run_batched_cell(
|
||
&resolved_variants,
|
||
cell,
|
||
&grid.base,
|
||
&cell_data,
|
||
&cell_out,
|
||
)?;
|
||
tracing::info!(cell = cell.name.as_str(), "batched cell complete");
|
||
continue;
|
||
}
|
||
let run_args = RunArgs {
|
||
data: cell_data,
|
||
predecoded_dir: grid.base.predecoded_dir.clone(),
|
||
n_parallel: grid.base.n_parallel,
|
||
policy_grid: None, // sweep cells use the default policy; nested
|
||
// grid-of-grids is a separate v2 concern.
|
||
latency_ns: cell.latency_ns.unwrap_or(grid.base.latency_ns),
|
||
target_annual_vol_units:
|
||
cell.target_annual_vol_units.unwrap_or(grid.base.target_annual_vol_units),
|
||
annualisation_factor:
|
||
cell.annualisation_factor.unwrap_or(grid.base.annualisation_factor),
|
||
max_lots: cell.max_lots.unwrap_or(grid.base.max_lots),
|
||
max_events: cell.max_events.unwrap_or(grid.base.max_events),
|
||
seed: cell.seed.unwrap_or(grid.base.seed),
|
||
checkpoint: cell.checkpoint.clone().or_else(|| grid.base.checkpoint.clone()),
|
||
kelly_frac_floor:
|
||
cell.kelly_frac_floor.unwrap_or(grid.base.kelly_frac_floor),
|
||
sharpe_weight_floor:
|
||
cell.sharpe_weight_floor.unwrap_or(grid.base.sharpe_weight_floor),
|
||
out: cell_out.clone(),
|
||
kernel_step_trace: grid.base.kernel_step_trace.clone(),
|
||
per_horizon_logit_sampling_stride: 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");
|
||
}
|
||
|
||
// Auto-aggregate at the end.
|
||
let out = aggregate_sweep_dir(&args.out)?;
|
||
println!(
|
||
"sweep complete: {} cells; {} on Pareto frontier; parquet={}, frontier={}",
|
||
out.n_cells,
|
||
out.n_pareto,
|
||
out.parquet_path.display(),
|
||
out.frontier_path.display()
|
||
);
|
||
Ok(())
|
||
}
|
||
|
||
/// P6: resolve sim_variants from SweepBase, layering per-variant overrides
|
||
/// over base scalars. Produces a flat Vec<ResolvedSimVariant> suitable for
|
||
/// BatchedSimConfig::from_grid.
|
||
fn resolve_sim_variants(base: &SweepBase) -> Vec<ml_backtesting::sim::ResolvedSimVariant> {
|
||
base.sim_variants.iter().map(|v| ml_backtesting::sim::ResolvedSimVariant {
|
||
name: v.name.clone(),
|
||
target_annual_vol_units: v.target_annual_vol_units.unwrap_or(base.target_annual_vol_units),
|
||
annualisation_factor: v.annualisation_factor.unwrap_or(base.annualisation_factor),
|
||
max_lots: v.max_lots.unwrap_or(base.max_lots),
|
||
latency_ns: v.latency_ns.unwrap_or(base.latency_ns),
|
||
kelly_frac_floor: v.kelly_frac_floor.unwrap_or(base.kelly_frac_floor),
|
||
sharpe_weight_floor: v.sharpe_weight_floor.unwrap_or(base.sharpe_weight_floor),
|
||
threshold: v.threshold,
|
||
cost_per_lot_per_side: v.cost_per_lot_per_side,
|
||
max_hold_ns: v.max_hold_ns,
|
||
min_reasonable_px: base.min_reasonable_px.unwrap_or(0.0),
|
||
max_reasonable_px: base.max_reasonable_px.unwrap_or(f32::INFINITY),
|
||
delta_floor: v.delta_floor.unwrap_or(base.delta_floor),
|
||
}).collect()
|
||
}
|
||
|
||
/// P6: run ONE cell with n_parallel=variants.len() backtests sharing the
|
||
/// forward pass. Each backtest gets distinct (cost, latency, threshold,
|
||
/// ...) from its ResolvedSimVariant. Output dirs are named `sim_<name>`
|
||
/// per spec §3.3.
|
||
fn run_batched_cell(
|
||
variants: &[ml_backtesting::sim::ResolvedSimVariant],
|
||
cell: &SweepCell,
|
||
base: &SweepBase,
|
||
cell_data: &std::path::Path,
|
||
cell_out: &std::path::Path,
|
||
) -> Result<()> {
|
||
use ml_alpha::trainer::perception::{PerceptionTrainer, PerceptionTrainerConfig};
|
||
use ml_core::device::MlDevice;
|
||
let n = variants.len();
|
||
let names: Vec<String> = variants.iter().map(|v| v.name.clone()).collect();
|
||
let batched = ml_backtesting::sim::BatchedSimConfig::from_grid(variants);
|
||
|
||
let dev = MlDevice::cuda(0).context("CUDA device unavailable")?;
|
||
let seed = cell.seed.unwrap_or(base.seed);
|
||
let trainer_cfg = PerceptionTrainerConfig {
|
||
seed,
|
||
n_batch: 1,
|
||
kernel_step_trace_path: base.kernel_step_trace.clone(),
|
||
..Default::default()
|
||
};
|
||
let checkpoint = cell.checkpoint.clone().or_else(|| base.checkpoint.clone());
|
||
let trainer = if let Some(ckpt_path) = &checkpoint {
|
||
PerceptionTrainer::from_checkpoint(&dev, &trainer_cfg, ckpt_path)
|
||
.with_context(|| format!("load checkpoint {}", ckpt_path.display()))?
|
||
} else {
|
||
let mut cfg2 = trainer_cfg.clone();
|
||
cfg2.seed = seed;
|
||
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()),
|
||
n_parallel: n,
|
||
target_annual_vol_units: base.target_annual_vol_units, // overridden by sim_config
|
||
annualisation_factor: base.annualisation_factor,
|
||
max_lots: base.max_lots,
|
||
max_events: cell.max_events.unwrap_or(base.max_events),
|
||
latency_ns: base.latency_ns,
|
||
kelly_frac_floor: base.kelly_frac_floor,
|
||
sharpe_weight_floor: base.sharpe_weight_floor,
|
||
threshold: 0.0, // overridden by sim_config
|
||
cost_per_lot_per_side: 0.0, // overridden by sim_config
|
||
variant_names: Some(names),
|
||
sim_config_override: Some(batched),
|
||
strategies: Vec::new(), // batched flow uses default policy per backtest
|
||
kernel_step_trace_path: base.kernel_step_trace.clone(),
|
||
per_horizon_logit_sampling_stride: base.per_horizon_logit_sampling_stride,
|
||
};
|
||
|
||
std::fs::create_dir_all(cell_out)
|
||
.with_context(|| format!("create cell out {}", cell_out.display()))?;
|
||
let mut harness = BacktestHarness::new(cfg, &dev, trainer).context("BacktestHarness::new (batched)")?;
|
||
let stats = harness.run().context("BacktestHarness::run (batched)")?;
|
||
harness.write_artifacts(cell_out).context("write_artifacts (batched)")?;
|
||
println!(
|
||
"batched cell {} done: {} events, {} decisions, {} variants → {}",
|
||
cell.name, stats.events_processed, stats.decisions_taken, n, cell_out.display(),
|
||
);
|
||
Ok(())
|
||
}
|
||
|
||
fn run(args: RunArgs) -> Result<()> {
|
||
// Resolve strategies — either from YAML or default_for.
|
||
let strategies: Vec<Strategy> = if let Some(grid_path) = &args.policy_grid {
|
||
let yaml = std::fs::read_to_string(grid_path)
|
||
.with_context(|| format!("read policy grid {}", grid_path.display()))?;
|
||
let parsed: Vec<Strategy> =
|
||
serde_yaml::from_str(&yaml).context("parse policy grid yaml")?;
|
||
anyhow::ensure!(
|
||
parsed.len() == args.n_parallel,
|
||
"policy grid has {} cells, --n-parallel = {}",
|
||
parsed.len(),
|
||
args.n_parallel
|
||
);
|
||
parsed
|
||
} else {
|
||
(0..args.n_parallel)
|
||
.map(|_| Strategy::default_for(args.max_lots))
|
||
.collect()
|
||
};
|
||
// If the user provided a policy grid YAML, strategies will be flattened
|
||
// and uploaded to the per-backtest bytecode-program slots in the sim;
|
||
// otherwise each cell uses the hardcoded default policy
|
||
// (WeightedByRealizedSharpe across all 5 horizons).
|
||
let strategy_grid = if args.policy_grid.is_some() {
|
||
strategies
|
||
} else {
|
||
Vec::new()
|
||
};
|
||
|
||
let dev = MlDevice::cuda(0).map_err(|e| anyhow::anyhow!("MlDevice::cuda(0): {e}"))?;
|
||
// X11 inference path: backtester drives forward via PerceptionTrainer
|
||
// (in inference role). The trainer wraps a CfcTrunk loaded from a
|
||
// Checkpoint file; forward kernels live on the trainer and read
|
||
// weights from `self.trunk` (the post-X1-X9 source of truth).
|
||
let trainer_cfg = PerceptionTrainerConfig {
|
||
n_batch: 1,
|
||
seq_len: 32,
|
||
kernel_step_trace_path: args.kernel_step_trace.clone(),
|
||
..Default::default()
|
||
};
|
||
let trainer = if let Some(ckpt_path) = &args.checkpoint {
|
||
tracing::info!(checkpoint = %ckpt_path.display(), "loading PerceptionTrainer from checkpoint");
|
||
PerceptionTrainer::from_checkpoint(&dev, &trainer_cfg, ckpt_path)
|
||
.with_context(|| format!("load checkpoint {}", ckpt_path.display()))?
|
||
} else {
|
||
tracing::warn!(
|
||
seed = args.seed,
|
||
"no --checkpoint provided; using random-init PerceptionTrainer (backtest results are NOISE)"
|
||
);
|
||
let _ = CfcConfig::default; // CfcConfig used transitively via trunk_cfg
|
||
let _ = N_HORIZONS;
|
||
let mut cfg2 = trainer_cfg.clone();
|
||
cfg2.seed = args.seed;
|
||
PerceptionTrainer::new(&dev, &cfg2)
|
||
.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),
|
||
n_parallel: args.n_parallel,
|
||
target_annual_vol_units: args.target_annual_vol_units,
|
||
annualisation_factor: args.annualisation_factor,
|
||
max_lots: args.max_lots,
|
||
max_events: args.max_events,
|
||
latency_ns: args.latency_ns,
|
||
kelly_frac_floor: args.kelly_frac_floor,
|
||
sharpe_weight_floor: args.sharpe_weight_floor,
|
||
threshold: 0.0, // P4: default = gate disabled
|
||
cost_per_lot_per_side: 0.0, // P4: default = frictionless
|
||
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,
|
||
kernel_step_trace_path: args.kernel_step_trace.clone(),
|
||
per_horizon_logit_sampling_stride: args.per_horizon_logit_sampling_stride,
|
||
};
|
||
|
||
std::fs::create_dir_all(&args.out)
|
||
.with_context(|| format!("create out dir {}", args.out.display()))?;
|
||
|
||
let mut harness = BacktestHarness::new(cfg, &dev, trainer).context("BacktestHarness::new")?;
|
||
let stats = harness.run().context("BacktestHarness::run")?;
|
||
harness.write_artifacts(&args.out).context("write_artifacts")?;
|
||
|
||
println!(
|
||
"done: {} events processed, {} decisions taken; artifacts under {}",
|
||
stats.events_processed,
|
||
stats.decisions_taken,
|
||
args.out.display()
|
||
);
|
||
Ok(())
|
||
}
|