diff --git a/Cargo.toml b/Cargo.toml index 674c99f7b..c77d9894d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -146,6 +146,7 @@ members = [ "crates/training_uploader", # CLI binary "bin/fxt", + "bin/fxt-backtest", # Services "services/backtesting_service", "services/broker_gateway_service", diff --git a/bin/fxt-backtest/Cargo.toml b/bin/fxt-backtest/Cargo.toml new file mode 100644 index 000000000..3cecb0172 --- /dev/null +++ b/bin/fxt-backtest/Cargo.toml @@ -0,0 +1,27 @@ +[package] +name = "fxt-backtest" +version.workspace = true +edition.workspace = true +rust-version.workspace = true +authors.workspace = true +license.workspace = true +repository.workspace = true +homepage.workspace = true +documentation.workspace = true +publish.workspace = true +keywords.workspace = true +categories.workspace = true +description = "Real-LOB GPU backtest CLI (run + sweep aggregator)" + +[dependencies] +ml-backtesting = { path = "../../crates/ml-backtesting" } +ml-alpha = { path = "../../crates/ml-alpha" } +ml-core.workspace = true +anyhow.workspace = true +clap = { workspace = true, features = ["derive", "env"] } +serde_yaml.workspace = true +tracing.workspace = true +tracing-subscriber.workspace = true + +[lints] +workspace = true diff --git a/bin/fxt-backtest/src/main.rs b/bin/fxt-backtest/src/main.rs new file mode 100644 index 000000000..e09c8d045 --- /dev/null +++ b/bin/fxt-backtest/src/main.rs @@ -0,0 +1,167 @@ +//! `fxt-backtest` — Real-LOB GPU backtest CLI. +//! +//! Two subcommands: +//! +//! fxt-backtest run --data [--policy-grid ] ... +//! Runs the LOB backtest harness on a directory of MBP-10 dbn files. +//! Writes per-cell artifacts under --out/cell_/. +//! +//! fxt-backtest aggregate +//! Walks every cell directory under , 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, CfcTrunk}; +use ml_backtesting::aggregate::aggregate_sweep_dir; +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 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), +} + +#[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, + /// 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, + /// Decide every Sth event (matches training stride; default 4). + #[arg(long, default_value_t = 4)] + decision_stride: usize, + /// Total latency budget in nanoseconds (IBKR+Scaleway baseline = 100ms). + /// Not currently consumed by the v1 immediate-match path; reserved for + /// follow-up commits that add resting-order in-flight promotion. + #[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, + /// Random seed for v1 trunk initialisation (a checkpoint loader is + /// follow-up work — see harness.rs::new doc). + #[arg(long, default_value_t = 0xC0FFEE)] + seed: u64, + /// Output directory; per-cell artifacts written to /cell_NNNN/. + #[arg(long)] + out: PathBuf, +} + +#[derive(Parser, Debug)] +struct AggregateArgs { + /// Sweep directory containing one subdirectory per cell, each with + /// a summary.json file. + sweep_dir: PathBuf, +} + +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(()) + } + } +} + +fn run(args: RunArgs) -> Result<()> { + // Resolve strategies — either from YAML or default_for. + let strategies: Vec = 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 = + 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() + }; + // Strategies parsed but not yet plumbed into the v1 LobSimCuda which + // hardcodes the WeightedByRealizedSharpe default policy (see C7 + // commit message). Keep the parse step so the YAML path is validated + // end-to-end and ready when the v2 bytecode VM lands. + let _ = strategies; + + let dev = MlDevice::cuda(0).map_err(|e| anyhow::anyhow!("MlDevice::cuda(0): {e}"))?; + let trunk = CfcTrunk::new_random(&dev, &CfcConfig::default(), args.seed) + .context("CfcTrunk::new_random")?; + + let cfg = BacktestHarnessConfig { + data_root: args.data.clone(), + predecoded_dir: args.predecoded_dir.clone().unwrap_or(args.data), + n_parallel: args.n_parallel, + decision_stride: args.decision_stride, + target_annual_vol_units: args.target_annual_vol_units, + annualisation_factor: args.annualisation_factor, + max_lots: args.max_lots, + max_events: args.max_events, + }; + + let _ = args.latency_ns; // reserved for follow-up resting-limit work + std::fs::create_dir_all(&args.out) + .with_context(|| format!("create out dir {}", args.out.display()))?; + + let mut harness = BacktestHarness::new(cfg, &dev, trunk).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(()) +} diff --git a/crates/ml-backtesting/Cargo.toml b/crates/ml-backtesting/Cargo.toml index 6c46beba7..7a55a3f57 100644 --- a/crates/ml-backtesting/Cargo.toml +++ b/crates/ml-backtesting/Cargo.toml @@ -19,12 +19,18 @@ ml-alpha = { path = "../ml-alpha" } # loader + trunk reuse — see real-LO cudarc = { path = "../../vendor/cudarc" } # CUDA bindings — same vendored fork as ml-alpha serde = { workspace = true, features = ["derive"] } serde_json.workspace = true +serde_yaml.workspace = true chrono.workspace = true anyhow.workspace = true csv.workspace = true bytemuck = { workspace = true, features = ["derive"] } tracing.workspace = true thiserror.workspace = true +# Sweep aggregator artifact format (parquet + arrow) — see C9. +parquet.workspace = true +arrow.workspace = true +arrow-array.workspace = true +arrow-schema.workspace = true [dev-dependencies] tempfile = "3.12" diff --git a/crates/ml-backtesting/src/aggregate.rs b/crates/ml-backtesting/src/aggregate.rs new file mode 100644 index 000000000..65bfa9be7 --- /dev/null +++ b/crates/ml-backtesting/src/aggregate.rs @@ -0,0 +1,254 @@ +//! Sweep-directory aggregator: walks `//summary.json` +//! across all cells, emits a single `aggregate.parquet` with one row +//! per cell plus a top-level `pareto_frontier.json` highlighting +//! non-dominated cells in (sharpe_ann, max_drawdown_usd, total_fees_usd) +//! space. +//! +//! See docs/superpowers/specs/2026-05-18-real-lob-integration-design.md §7. + +use anyhow::{Context, Result}; +use arrow_array::{Float32Array, RecordBatch, StringArray, UInt64Array}; +use arrow_schema::{DataType, Field, Schema}; +use parquet::arrow::ArrowWriter; +use parquet::basic::Compression; +use parquet::file::properties::WriterProperties; +use std::path::Path; +use std::sync::Arc; + +use crate::artifacts::Summary; + +/// One row of the aggregate parquet: cell name + a flattened subset of +/// `Summary` fields the user is most likely to filter / sort by. +#[derive(Clone, Debug)] +struct AggregateRow { + cell: String, + total_pnl_usd: f32, + sharpe_ann: f32, + sortino_ann: f32, + max_drawdown_usd: f32, + calmar: f32, + n_trades: u64, + win_rate: f32, + profit_factor: f32, + total_fees_usd: f32, +} + +impl AggregateRow { + fn from_summary(cell: String, s: &Summary) -> Self { + AggregateRow { + cell, + total_pnl_usd: s.total_pnl_usd, + sharpe_ann: s.sharpe_ann, + sortino_ann: s.sortino_ann, + max_drawdown_usd: s.max_drawdown_usd, + calmar: s.calmar, + n_trades: s.n_trades, + win_rate: s.win_rate, + profit_factor: s.profit_factor, + total_fees_usd: s.total_fees_usd, + } + } +} + +/// Walk `sweep_dir` for subdirectories containing `summary.json`, +/// emit `/aggregate.parquet` and +/// `/pareto_frontier.json`. +pub fn aggregate_sweep_dir(sweep_dir: &Path) -> Result { + let mut rows: Vec = Vec::new(); + for entry in std::fs::read_dir(sweep_dir) + .with_context(|| format!("read sweep dir {}", sweep_dir.display()))? + { + let entry = entry?; + let p = entry.path(); + if !p.is_dir() { + continue; + } + let summary_path = p.join("summary.json"); + if !summary_path.exists() { + continue; + } + let s: Summary = serde_json::from_reader( + std::fs::File::open(&summary_path) + .with_context(|| format!("open {}", summary_path.display()))?, + ) + .with_context(|| format!("parse {}", summary_path.display()))?; + let cell = p + .file_name() + .map(|n| n.to_string_lossy().into_owned()) + .unwrap_or_else(|| p.display().to_string()); + rows.push(AggregateRow::from_summary(cell, &s)); + } + anyhow::ensure!( + !rows.is_empty(), + "no cell directories with summary.json under {}", + sweep_dir.display() + ); + + let parquet_path = sweep_dir.join("aggregate.parquet"); + write_aggregate_parquet(&parquet_path, &rows)?; + + let frontier = pareto_frontier(&rows); + let frontier_path = sweep_dir.join("pareto_frontier.json"); + serde_json::to_writer_pretty( + std::fs::File::create(&frontier_path) + .with_context(|| format!("create {}", frontier_path.display()))?, + &frontier, + ) + .context("write pareto_frontier.json")?; + + Ok(AggregateOutcome { + n_cells: rows.len(), + n_pareto: frontier.len(), + parquet_path, + frontier_path, + }) +} + +#[derive(Clone, Debug)] +pub struct AggregateOutcome { + pub n_cells: usize, + pub n_pareto: usize, + pub parquet_path: std::path::PathBuf, + pub frontier_path: std::path::PathBuf, +} + +/// Maximise sharpe_ann; minimise max_drawdown_usd + total_fees_usd. +/// A cell is on the Pareto frontier if no other cell weakly dominates +/// it in all three objectives AND strictly dominates in at least one. +fn pareto_frontier(rows: &[AggregateRow]) -> Vec { + let mut out = Vec::new(); + for (i, a) in rows.iter().enumerate() { + let dominated = rows.iter().enumerate().any(|(j, b)| { + i != j + && b.sharpe_ann >= a.sharpe_ann + && b.max_drawdown_usd <= a.max_drawdown_usd + && b.total_fees_usd <= a.total_fees_usd + && (b.sharpe_ann > a.sharpe_ann + || b.max_drawdown_usd < a.max_drawdown_usd + || b.total_fees_usd < a.total_fees_usd) + }); + if !dominated { + out.push(a.cell.clone()); + } + } + out +} + +fn write_aggregate_parquet(path: &Path, rows: &[AggregateRow]) -> Result<()> { + let schema = Arc::new(Schema::new(vec![ + Field::new("cell", DataType::Utf8, false), + Field::new("total_pnl_usd", DataType::Float32, false), + Field::new("sharpe_ann", DataType::Float32, false), + Field::new("sortino_ann", DataType::Float32, false), + Field::new("max_drawdown_usd", DataType::Float32, false), + Field::new("calmar", DataType::Float32, false), + Field::new("n_trades", DataType::UInt64, false), + Field::new("win_rate", DataType::Float32, false), + Field::new("profit_factor", DataType::Float32, false), + Field::new("total_fees_usd", DataType::Float32, false), + ])); + + let cells: Vec<&str> = rows.iter().map(|r| r.cell.as_str()).collect(); + let total_pnl: Vec = rows.iter().map(|r| r.total_pnl_usd).collect(); + let sharpe: Vec = rows.iter().map(|r| r.sharpe_ann).collect(); + let sortino: Vec = rows.iter().map(|r| r.sortino_ann).collect(); + let dd: Vec = rows.iter().map(|r| r.max_drawdown_usd).collect(); + let calmar: Vec = rows.iter().map(|r| r.calmar).collect(); + let n_tr: Vec = rows.iter().map(|r| r.n_trades).collect(); + let wr: Vec = rows.iter().map(|r| r.win_rate).collect(); + let pf: Vec = rows.iter().map(|r| r.profit_factor).collect(); + let fees: Vec = rows.iter().map(|r| r.total_fees_usd).collect(); + + let batch = RecordBatch::try_new( + schema.clone(), + vec![ + Arc::new(StringArray::from(cells)), + Arc::new(Float32Array::from(total_pnl)), + Arc::new(Float32Array::from(sharpe)), + Arc::new(Float32Array::from(sortino)), + Arc::new(Float32Array::from(dd)), + Arc::new(Float32Array::from(calmar)), + Arc::new(UInt64Array::from(n_tr)), + Arc::new(Float32Array::from(wr)), + Arc::new(Float32Array::from(pf)), + Arc::new(Float32Array::from(fees)), + ], + )?; + + let f = std::fs::File::create(path) + .with_context(|| format!("create {}", path.display()))?; + let props = WriterProperties::builder() + .set_compression(Compression::SNAPPY) + .build(); + let mut writer = ArrowWriter::try_new(f, schema, Some(props))?; + writer.write(&batch)?; + writer.close()?; + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn mk_summary(pnl: f32, sharpe: f32, dd: f32, n: u64) -> Summary { + Summary { + total_pnl_usd: pnl, + sharpe_ann: sharpe, + max_drawdown_usd: dd, + n_trades: n, + ..Default::default() + } + } + + fn cell_dir(root: &Path, name: &str, s: &Summary) { + let dir = root.join(name); + std::fs::create_dir_all(&dir).unwrap(); + crate::artifacts::write_summary(&dir.join("summary.json"), s).unwrap(); + } + + #[test] + fn aggregate_three_cells_produces_parquet_and_frontier() { + let tmp = tempfile::tempdir().unwrap(); + let root = tmp.path(); + cell_dir(root, "cellA", &mk_summary(100.0, 1.5, 50.0, 10)); + cell_dir(root, "cellB", &mk_summary(200.0, 2.0, 60.0, 12)); // dominates A on pnl+sharpe but worse dd + cell_dir(root, "cellC", &mk_summary(50.0, 0.5, 10.0, 5)); // best dd, lowest sharpe + + let out = aggregate_sweep_dir(root).expect("aggregate"); + assert_eq!(out.n_cells, 3); + assert!(out.parquet_path.exists()); + assert!(out.frontier_path.exists()); + + // Pareto frontier should include B (best sharpe) and C (best dd). + // A is dominated by B (B.sharpe>A.sharpe, B.dd>A.dd but B fees=A fees=0): + // Actually B has WORSE dd than A — neither dominates the other on dd. + // B dominates A only if sharpe AND dd AND fees are weakly better. Here + // B.sharpe>A.sharpe but B.dd>A.dd, so B does NOT dominate A. A is + // Pareto-optimal too. Final frontier = {A, B, C} for this fixture. + let frontier_json: Vec = + serde_json::from_reader(std::fs::File::open(&out.frontier_path).unwrap()).unwrap(); + assert_eq!(frontier_json.len(), 3); + assert!(frontier_json.contains(&"cellA".to_string())); + assert!(frontier_json.contains(&"cellB".to_string())); + assert!(frontier_json.contains(&"cellC".to_string())); + } + + #[test] + fn dominated_cell_excluded_from_frontier() { + let tmp = tempfile::tempdir().unwrap(); + let root = tmp.path(); + // A: pnl=100, sharpe=1.0, dd=50 — DOMINATED by B + // B: pnl=200, sharpe=2.0, dd=30 — strictly better on all three (fees=0 both) + // C: pnl=10, sharpe=0.1, dd=5 — Pareto for tiny dd + cell_dir(root, "cellA", &mk_summary(100.0, 1.0, 50.0, 10)); + cell_dir(root, "cellB", &mk_summary(200.0, 2.0, 30.0, 15)); + cell_dir(root, "cellC", &mk_summary(10.0, 0.1, 5.0, 3)); + + let out = aggregate_sweep_dir(root).unwrap(); + let frontier: Vec = + serde_json::from_reader(std::fs::File::open(&out.frontier_path).unwrap()).unwrap(); + assert!(!frontier.contains(&"cellA".to_string()), "A dominated by B"); + assert!(frontier.contains(&"cellB".to_string())); + assert!(frontier.contains(&"cellC".to_string())); + } +} diff --git a/crates/ml-backtesting/src/artifacts.rs b/crates/ml-backtesting/src/artifacts.rs new file mode 100644 index 000000000..aac876ea7 --- /dev/null +++ b/crates/ml-backtesting/src/artifacts.rs @@ -0,0 +1,303 @@ +//! Per-cell backtest output writers + summary statistics. +//! +//! See docs/superpowers/specs/2026-05-18-real-lob-integration-design.md §7 +//! "Output artifacts". +//! +//! summary.json top-level stats (Sharpe, drawdown, profit factor, etc.) +//! trades.csv per-trade audit log +//! pnl_curve.bin binary float32 cumulative P&L at each event timestamp +//! +//! Annualisation uses the non-overlapping convention per +//! `pearl_phase1d4_backtest_cost_edge_frontier`: σ_ann = σ_trade × √825 +//! (K=6000 holding period, 250 trading days = ~825 non-overlapping trades/yr). + +use anyhow::{Context, Result}; +use serde::{Deserialize, Serialize}; +use std::io::Write; +use std::path::Path; + +use crate::order::TradeRecord; + +/// Top-level per-cell summary. Serialised to summary.json. +#[derive(Clone, Debug, Default, Serialize, Deserialize)] +pub struct Summary { + pub total_pnl_usd: f32, + pub sharpe_ann: f32, + pub sortino_ann: f32, + pub max_drawdown_usd: f32, + pub calmar: f32, + pub n_trades: u64, + pub win_rate: f32, + pub avg_win_usd: f32, + pub avg_loss_usd: f32, + pub profit_factor: f32, + pub total_fees_usd: f32, + /// Fraction of decision points where any backtest held a non-flat + /// position. Populated by the harness if it tracks bars-in-position; + /// 0.0 placeholder when unmeasured. + pub exposure_pct: f32, + /// Downsampled per-horizon Kelly cap trace (entries × N_HORIZONS). + /// Empty if the harness doesn't record per-decision Kelly caps; + /// when populated, used by aggregate/diagnostics. + #[serde(default)] + pub kelly_cap_history_sample: Vec>, +} + +/// Non-overlapping annualisation factor (√825) per +/// `pearl_phase1d4_backtest_cost_edge_frontier.md`. Pinned here so +/// `aggregate` reads from the same constant if it ever recomputes. +pub const ANNUALISATION_SQRT_FACTOR: f32 = 28.722_815; + +/// Convert TradeRecord fixed-point USD ×100 to plain USD float. +#[inline] +fn fp_to_usd(fp: i32) -> f32 { + fp as f32 / 100.0 +} + +/// Compute summary statistics from the per-cell trade log + cumulative +/// P&L curve (USD). The curve is sampled at every event; the harness +/// passes the same vector it serialises to pnl_curve.bin. +pub fn compute_summary(records: &[TradeRecord], pnl_curve_usd: &[f32]) -> Summary { + if records.is_empty() { + return Summary { + total_pnl_usd: 0.0, + n_trades: 0, + ..Default::default() + }; + } + + let n_trades = records.len() as u64; + let mut wins = 0u64; + let mut losses = 0u64; + let mut win_sum = 0.0f32; + let mut loss_sum = 0.0f32; + let mut total_fees = 0.0f32; + let mut per_trade_returns: Vec = Vec::with_capacity(records.len()); + for r in records { + let p = fp_to_usd(r.realised_pnl_usd_fp); + total_fees += fp_to_usd(r.fees_usd_fp); + per_trade_returns.push(p); + if p > 0.0 { + wins += 1; + win_sum += p; + } else if p < 0.0 { + losses += 1; + loss_sum += -p; + } + } + + let total_pnl_usd: f32 = per_trade_returns.iter().sum(); + let win_rate = wins as f32 / n_trades as f32; + let avg_win = if wins > 0 { win_sum / wins as f32 } else { 0.0 }; + let avg_loss = if losses > 0 { loss_sum / losses as f32 } else { 0.0 }; + let profit_factor = if loss_sum > 0.0 { + win_sum / loss_sum + } else if win_sum > 0.0 { + f32::INFINITY + } else { + 0.0 + }; + + // Per-trade Sharpe + Sortino. + let mean = total_pnl_usd / n_trades as f32; + let var: f32 = per_trade_returns + .iter() + .map(|p| (p - mean).powi(2)) + .sum::() + / n_trades as f32; + let std_dev = var.sqrt().max(1e-9); + let sharpe_per_trade = mean / std_dev; + let sharpe_ann = sharpe_per_trade * ANNUALISATION_SQRT_FACTOR; + + let downside_var: f32 = per_trade_returns + .iter() + .filter_map(|p| if *p < mean { Some((p - mean).powi(2)) } else { None }) + .sum::() + / n_trades as f32; + let downside_std = downside_var.sqrt().max(1e-9); + let sortino_per_trade = mean / downside_std; + let sortino_ann = sortino_per_trade * ANNUALISATION_SQRT_FACTOR; + + // Max drawdown from the cumulative P&L curve. + let (max_drawdown_usd, _peak_at_dd) = if pnl_curve_usd.is_empty() { + (0.0, 0.0) + } else { + let mut peak = pnl_curve_usd[0]; + let mut dd = 0.0f32; + let mut peak_at_dd = peak; + for &v in pnl_curve_usd { + if v > peak { + peak = v; + } + let cur_dd = peak - v; + if cur_dd > dd { + dd = cur_dd; + peak_at_dd = peak; + } + } + (dd, peak_at_dd) + }; + + let calmar = if max_drawdown_usd > 0.0 { + total_pnl_usd / max_drawdown_usd + } else if total_pnl_usd > 0.0 { + f32::INFINITY + } else { + 0.0 + }; + + Summary { + total_pnl_usd, + sharpe_ann, + sortino_ann, + max_drawdown_usd, + calmar, + n_trades, + win_rate, + avg_win_usd: avg_win, + avg_loss_usd: avg_loss, + profit_factor, + total_fees_usd: total_fees, + exposure_pct: 0.0, // populated by harness if it tracks this + kelly_cap_history_sample: Vec::new(), + } +} + +/// Write `summary.json` to the given path. +pub fn write_summary(path: &Path, s: &Summary) -> Result<()> { + let f = std::fs::File::create(path) + .with_context(|| format!("create {}", path.display()))?; + serde_json::to_writer_pretty(f, s).context("write summary.json")?; + Ok(()) +} + +/// Write `trades.csv` to the given path. Side is rendered "buy"/"sell" +/// from the sign of `size_lots`; price columns are converted from +/// fixed-point ×100 ticks back to float-tick values for human readability. +pub fn write_trades_csv(path: &Path, records: &[TradeRecord]) -> Result<()> { + let mut f = std::fs::File::create(path) + .with_context(|| format!("create {}", path.display()))?; + writeln!( + f, + "entry_ts_ns,exit_ts_ns,side,size_lots,entry_px_ticks,exit_px_ticks,fees_usd,realised_pnl_usd,strategy_id,horizon_idx" + )?; + for r in records { + let side = if r.size_lots > 0 { "buy" } else { "sell" }; + let entry_px = r.entry_px_ticks as f32 / 100.0; + let exit_px = r.exit_px_ticks as f32 / 100.0; + writeln!( + f, + "{},{},{},{},{:.2},{:.2},{:.2},{:.2},{},{}", + r.entry_ts_ns, + r.exit_ts_ns, + side, + r.size_lots.abs(), + entry_px, + exit_px, + fp_to_usd(r.fees_usd_fp), + fp_to_usd(r.realised_pnl_usd_fp), + r.strategy_id, + r.horizon_idx, + )?; + } + Ok(()) +} + +/// Write the cumulative P&L curve as a packed binary float32 array. +pub fn write_pnl_curve_bin(path: &Path, curve_usd: &[f32]) -> Result<()> { + let mut f = std::fs::File::create(path) + .with_context(|| format!("create {}", path.display()))?; + let bytes: &[u8] = bytemuck::cast_slice(curve_usd); + f.write_all(bytes).context("write pnl_curve.bin")?; + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn trade(entry_ts: u64, exit_ts: u64, size: i32, pnl_fp: i32) -> TradeRecord { + TradeRecord { + entry_ts_ns: entry_ts, + exit_ts_ns: exit_ts, + entry_px_ticks: 550_000, + exit_px_ticks: 550_500, + size_lots: size, + fees_usd_fp: 0, + realised_pnl_usd_fp: pnl_fp, + horizon_idx: 4, + strategy_id: 0, + _pad: [0; 2], + } + } + + #[test] + fn empty_records_summary_is_zero() { + let s = compute_summary(&[], &[]); + assert_eq!(s.n_trades, 0); + assert_eq!(s.total_pnl_usd, 0.0); + assert_eq!(s.profit_factor, 0.0); + } + + #[test] + fn three_wins_one_loss_summary() { + let recs = vec![ + trade(1_000_000_000, 2_000_000_000, 1, 5000), // +$50 + trade(3_000_000_000, 4_000_000_000, 1, 7500), // +$75 + trade(5_000_000_000, 6_000_000_000, -1, -2500), // -$25 + trade(7_000_000_000, 8_000_000_000, 1, 10000), // +$100 + ]; + // Cumulative pnl curve sampled at trade-close events. + let curve = vec![0.0, 50.0, 125.0, 100.0, 200.0]; + let s = compute_summary(&recs, &curve); + assert_eq!(s.n_trades, 4); + assert!((s.total_pnl_usd - 200.0).abs() < 0.01); + assert!((s.win_rate - 0.75).abs() < 0.01); + assert!((s.avg_win_usd - 75.0).abs() < 0.01); + assert!((s.avg_loss_usd - 25.0).abs() < 0.01); + assert!((s.profit_factor - (225.0 / 25.0)).abs() < 0.01); + // Drawdown of $25 from peak $125 to trough $100. + assert!((s.max_drawdown_usd - 25.0).abs() < 0.01); + } + + #[test] + fn write_then_read_pnl_curve_bin_roundtrip() { + let dir = tempfile::tempdir().expect("tmpdir"); + let path = dir.path().join("pnl_curve.bin"); + let curve = vec![0.0f32, 10.0, 25.5, -3.25, 100.125]; + write_pnl_curve_bin(&path, &curve).unwrap(); + let raw = std::fs::read(&path).unwrap(); + let read_back: Vec = bytemuck::cast_slice(&raw).to_vec(); + assert_eq!(read_back, curve); + } + + #[test] + fn write_summary_then_parse_back() { + let dir = tempfile::tempdir().expect("tmpdir"); + let path = dir.path().join("summary.json"); + let s = Summary { + total_pnl_usd: 123.45, + sharpe_ann: 1.5, + n_trades: 7, + win_rate: 0.6, + ..Default::default() + }; + write_summary(&path, &s).unwrap(); + let raw = std::fs::read_to_string(&path).unwrap(); + let back: Summary = serde_json::from_str(&raw).unwrap(); + assert!((back.total_pnl_usd - 123.45).abs() < 0.01); + assert_eq!(back.n_trades, 7); + } + + #[test] + fn write_trades_csv_header_and_one_row() { + let dir = tempfile::tempdir().expect("tmpdir"); + let path = dir.path().join("trades.csv"); + let recs = vec![trade(1_000_000_000, 2_000_000_000, 3, 12345)]; + write_trades_csv(&path, &recs).unwrap(); + let s = std::fs::read_to_string(&path).unwrap(); + assert!(s.contains("entry_ts_ns,exit_ts_ns,side")); + assert!(s.contains("1000000000,2000000000,buy,3")); + assert!(s.contains("123.45")); // $123.45 from fp 12345 + } +} diff --git a/crates/ml-backtesting/src/harness.rs b/crates/ml-backtesting/src/harness.rs index 85e1ba103..26bc6c715 100644 --- a/crates/ml-backtesting/src/harness.rs +++ b/crates/ml-backtesting/src/harness.rs @@ -39,6 +39,9 @@ pub struct BacktestHarness { 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>, } impl BacktestHarness { @@ -70,6 +73,7 @@ impl BacktestHarness { let sim = LobSimCuda::new(cfg.n_parallel, dev)?; + let pnl_curves = (0..cfg.n_parallel).map(|_| Vec::with_capacity(1024)).collect(); Ok(Self { cfg, loader, @@ -77,6 +81,7 @@ impl BacktestHarness { sim, decision_count: 0, event_count: 0, + pnl_curves, }) } @@ -106,6 +111,12 @@ impl BacktestHarness { 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.cfg.max_events > 0 && self.event_count >= self.cfg.max_events { break; @@ -118,6 +129,25 @@ impl BacktestHarness { }) } + /// After `run()` completes, write per-cell artifacts to + /// `/cell_/{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()))?; + for b in 0..self.cfg.n_parallel { + let cell_dir = out_dir.join(format!("cell_{b:04}")); + 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 } diff --git a/crates/ml-backtesting/src/lib.rs b/crates/ml-backtesting/src/lib.rs index 0fd968cdd..06082c88a 100644 --- a/crates/ml-backtesting/src/lib.rs +++ b/crates/ml-backtesting/src/lib.rs @@ -16,6 +16,8 @@ pub use ml_core::MLError; pub mod action_loader; +pub mod aggregate; +pub mod artifacts; pub mod barrier_backtest; pub mod harness; pub mod lob;