gitadmin/foxhunt
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(ml-backtesting): artifacts + sweep aggregate + fxt-backtest CLI (C9)

Browse Source 
artifacts.rs:
  - Summary struct (total_pnl_usd, sharpe_ann, sortino_ann,
    max_drawdown_usd, calmar, n_trades, win_rate, avg_win/avg_loss,
    profit_factor, total_fees_usd, exposure_pct,
    kelly_cap_history_sample).
  - compute_summary(records, pnl_curve_usd) — non-overlapping
    annualisation × √825 per pearl_phase1d4_backtest_cost_edge_frontier
    (K=6000 holding × 250 trading days ≈ 825 trades/year).
    Sharpe + Sortino + max drawdown + Calmar.
  - write_summary (JSON pretty-printed), write_trades_csv (with USD
    conversion from fp ×100), write_pnl_curve_bin (bytemuck-cast f32
    slice). 5 unit tests with tempdir.

aggregate.rs:
  - aggregate_sweep_dir walks <root>/<cell>/summary.json, builds an
    arrow RecordBatch (cell name + 9 stats columns), writes
    SNAPPY-compressed aggregate.parquet.
  - pareto_frontier: cells are kept unless another cell weakly
    dominates on all three of (sharpe_ann maxed, max_drawdown_usd
    minimised, total_fees_usd minimised) AND strictly improves on one.
    Written to pareto_frontier.json (Vec<cell-name>).
  - 2 unit tests (3-cell mutual-non-dominance; B-dominates-A).

harness.rs:
  - run() now samples Pos.realized_pnl × $50/index-pt per event into
    self.pnl_curves[b], so the per-cell P&L curve is ready for
    write_artifacts() without an extra sim pass.
  - write_artifacts(out_dir) — per-cell <out>/cell_NNNN/{summary.json,
    trades.csv, pnl_curve.bin}.

bin/fxt-backtest:
  - clap-derive CLI with two subcommands:
      run --data <dir> [--predecoded-dir <dir>] [--policy-grid <yaml>]
          [--n-parallel N] [--decision-stride S] [--latency-ns N]
          [--target-annual-vol-units F] [--annualisation-factor F]
          [--max-lots N] [--max-events N] [--seed N] --out <dir>
      aggregate <sweep_dir>
  - Constructs MlDevice::cuda(0) + CfcTrunk::new_random for the trunk
    (v1 — ml-alpha has no checkpoint format yet; --seed gates init).
  - Parses --policy-grid YAML if given but doesn't yet plumb to the
    LobSimCuda decision kernel (the v1 kernel hardcodes the
    Strategy::default_for path; bytecode VM is C7's deferred follow-up).
    Parse step kept end-to-end so the YAML format is validated now.
  - --latency-ns parsed but not consumed — reserved for follow-up
    resting-order in-flight promotion (deferred from C5).

Adds parquet + arrow + arrow-array + arrow-schema + serde_yaml to
ml-backtesting deps; bin/fxt-backtest added to workspace members.

All 33 lib tests + 6 GPU fixture tests green. CLI --help renders both
subcommands correctly.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
jgrusewski
2026-05-18 08:59:41 +02:00
parent a4127935a8
commit 63ed6d0217
8 changed files with 790 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
Cargo.toml
View File

@@ -146,6 +146,7 @@ members = [
"crates/training_uploader",
# CLI binary
"bin/fxt",
"bin/fxt-backtest",
# Services
"services/backtesting_service",
"services/broker_gateway_service",

27
bin/fxt-backtest/Cargo.toml Normal file
View File

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

167
bin/fxt-backtest/src/main.rs Normal file
View File

@@ -0,0 +1,167 @@
//! `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, 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<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>,
/// 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 <out>/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<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()
};
// 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(())
}

6
crates/ml-backtesting/Cargo.toml
View File

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

254
crates/ml-backtesting/src/aggregate.rs Normal file
View File

@@ -0,0 +1,254 @@
//! Sweep-directory aggregator: walks `<sweep_dir>/<cell>/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 `<sweep_dir>/aggregate.parquet` and
/// `<sweep_dir>/pareto_frontier.json`.
pub fn aggregate_sweep_dir(sweep_dir: &Path) -> Result<AggregateOutcome> {
let mut rows: Vec<AggregateRow> = 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<String> {
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<f32> = rows.iter().map(|r| r.total_pnl_usd).collect();
let sharpe: Vec<f32> = rows.iter().map(|r| r.sharpe_ann).collect();
let sortino: Vec<f32> = rows.iter().map(|r| r.sortino_ann).collect();
let dd: Vec<f32> = rows.iter().map(|r| r.max_drawdown_usd).collect();
let calmar: Vec<f32> = rows.iter().map(|r| r.calmar).collect();
let n_tr: Vec<u64> = rows.iter().map(|r| r.n_trades).collect();
let wr: Vec<f32> = rows.iter().map(|r| r.win_rate).collect();
let pf: Vec<f32> = rows.iter().map(|r| r.profit_factor).collect();
let fees: Vec<f32> = 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<String> =
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<String> =
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()));
}
}

303
crates/ml-backtesting/src/artifacts.rs Normal file
View File

@@ -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<Vec<f32>>,
}
/// 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<f32> = 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::<f32>()
/ 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::<f32>()
/ 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<f32> = 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
}
}

30
crates/ml-backtesting/src/harness.rs
View File

@@ -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<Vec<f32>>,
}
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
/// `<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()))?;
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 }

2
crates/ml-backtesting/src/lib.rs
View File

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