foxhunt/crates/ml-backtesting/src/aggregate.rs

//! 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;
use serde::{Deserialize, Serialize};
use std::collections::BTreeMap;

/// X17: Tiered verdict for the deployability validation.
#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "kebab-case")]
pub enum VerdictTier {
    PassRobust,
    PassNominal,
    FailInconclusive,
    Fail,
    FailDegenerate,
}

#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct AnchorSpec {
    pub name: String,
    pub cost_tick: f32,
    pub latency_ms: u32,
}

impl AnchorSpec {
    pub fn realistic() -> Self {
        Self { name: "realistic".to_string(), cost_tick: 1.0, latency_ms: 200 }
    }
    pub fn stress() -> Self {
        Self { name: "stress".to_string(), cost_tick: 1.5, latency_ms: 400 }
    }
}

#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct AnchorReport {
    pub anchor: AnchorSpec,
    pub sharpe_per_window: Vec<f32>,
    pub sortino_per_window: Vec<f32>,
    pub max_dd_pct_per_window: Vec<f32>,
    pub profit_factor_per_window: Vec<f32>,
    pub median_sharpe: f32,
    pub median_sortino: f32,
    pub median_max_dd_pct: f32,
    pub median_profit_factor: f32,
    pub gate_sharpe: bool,
    pub gate_max_dd: bool,
    pub pass: bool,
    pub status: String,
}

#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct DeployabilityVerdict {
    pub verdict: VerdictTier,
    pub realistic: AnchorReport,
    pub stress: AnchorReport,
    pub threshold: f32,
    pub windows: Vec<String>,
    pub training_sha: String,
    pub spec_sha: String,
    pub timestamp_utc: String,
}

/// X17: tiered verdict classifier. Spec §3.5.
pub fn classify_verdict(realistic: &AnchorReport, stress: &AnchorReport) -> VerdictTier {
    if realistic.status != "ok" {
        return VerdictTier::FailDegenerate;
    }
    if realistic.pass && stress.status == "ok" && stress.pass {
        return VerdictTier::PassRobust;
    }
    if realistic.pass {
        return VerdictTier::PassNominal;
    }
    let sharpe_grey = realistic.median_sharpe >= 0.8 && realistic.median_sharpe < 1.0;
    let max_dd_grey = realistic.median_max_dd_pct >= 0.20 && realistic.median_max_dd_pct < 0.25;
    if sharpe_grey || max_dd_grey {
        return VerdictTier::FailInconclusive;
    }
    VerdictTier::Fail
}

fn median(xs: &[f32]) -> f32 {
    let mut v = xs.to_vec();
    v.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));
    let n = v.len();
    if n == 0 { return f32::NAN; }
    if n % 2 == 1 { v[n / 2] } else { 0.5 * (v[n / 2 - 1] + v[n / 2]) }
}

/// Parse a cell directory name like
/// `cell_0042_cost1.00_lat200_th0.75_W1` into its (cost_tick, latency_ms,
/// threshold, window) tuple. Returns None on non-matching names so
/// callers can skip them.
fn parse_cell_name(name: &str) -> Option<(f32, u32, f32, String)> {
    let parts: Vec<&str> = name.split('_').collect();
    if parts.len() < 5 || parts[0] != "cell" { return None; }
    let cost: f32 = parts.iter().find(|p| p.starts_with("cost"))?[4..].parse().ok()?;
    let lat: u32  = parts.iter().find(|p| p.starts_with("lat"))?[3..].parse().ok()?;
    let th: f32   = parts.iter().find(|p| p.starts_with("th"))?[2..].parse().ok()?;
    let win = parts.iter().rev().find(|p| p.starts_with('W'))?.to_string();
    Some((cost, lat, th, win))
}

fn build_anchor_report(
    sweep_dir: &Path,
    anchor: AnchorSpec,
    threshold: f32,
    windows: &[&str],
) -> Result<AnchorReport> {
    let mut by_window: BTreeMap<String, Summary> = BTreeMap::new();
    for entry in std::fs::read_dir(sweep_dir)
        .with_context(|| format!("read_dir {}", sweep_dir.display()))?
    {
        let path = entry?.path();
        let name = path.file_name().and_then(|n| n.to_str()).unwrap_or("").to_string();
        let (cell_cost, cell_lat, cell_th, cell_win) = match parse_cell_name(&name) {
            Some(t) => t,
            None => continue,
        };
        if (cell_cost - anchor.cost_tick).abs() > 1e-6 { continue; }
        if cell_lat != anchor.latency_ms { continue; }
        if (cell_th - threshold).abs() > 1e-6 { continue; }
        if !windows.iter().any(|w| **w == cell_win) { continue; }
        let summary_path = path.join("summary.json");
        let s: Summary = serde_json::from_reader(
            std::fs::File::open(&summary_path)
                .with_context(|| format!("open {}", summary_path.display()))?,
        )?;
        by_window.insert(cell_win.clone(), s);
    }
    let mut sharpe = Vec::new();
    let mut sortino = Vec::new();
    let mut max_dd = Vec::new();
    let mut pf = Vec::new();
    let mut status = String::from("ok");
    for w in windows {
        let s = by_window.get(*w).ok_or_else(||
            anyhow::anyhow!("missing cell for anchor={} window={}", anchor.name, w)
        )?;
        if s.n_trades == 0 || !s.sharpe_ann.is_finite() || !s.sortino_ann.is_finite() {
            status = format!("fail-degenerate: {w} trades={} sharpe={}", s.n_trades, s.sharpe_ann);
        }
        sharpe.push(s.sharpe_ann);
        sortino.push(s.sortino_ann);
        max_dd.push(s.max_drawdown_pct);
        pf.push(s.profit_factor);
    }
    let median_sharpe = median(&sharpe);
    let median_sortino = median(&sortino);
    let median_max_dd_pct = median(&max_dd);
    let median_profit_factor = median(&pf);
    let gate_sharpe = median_sharpe > 1.0;
    let gate_max_dd = median_max_dd_pct < 0.20;
    let pass = gate_sharpe && gate_max_dd && status == "ok";
    Ok(AnchorReport {
        anchor,
        sharpe_per_window: sharpe,
        sortino_per_window: sortino,
        max_dd_pct_per_window: max_dd,
        profit_factor_per_window: pf,
        median_sharpe, median_sortino, median_max_dd_pct, median_profit_factor,
        gate_sharpe, gate_max_dd, pass,
        status,
    })
}

/// X17: emit the full deployability verdict by aggregating per-cell
/// summary.json files at both anchors against the supplied threshold +
/// walk-forward window IDs. Spec §3.5.
pub fn emit_deployability_verdict(
    sweep_dir: &Path,
    threshold: f32,
    windows: &[&str],
    training_sha: &str,
    spec_sha: &str,
) -> Result<DeployabilityVerdict> {
    let realistic = build_anchor_report(sweep_dir, AnchorSpec::realistic(), threshold, windows)?;
    let stress = build_anchor_report(sweep_dir, AnchorSpec::stress(), threshold, windows)?;
    let verdict = classify_verdict(&realistic, &stress);
    Ok(DeployabilityVerdict {
        verdict,
        realistic,
        stress,
        threshold,
        windows: windows.iter().map(|s| s.to_string()).collect(),
        training_sha: training_sha.to_string(),
        spec_sha: spec_sha.to_string(),
        timestamp_utc: chrono::Utc::now().to_rfc3339(),
    })
}

/// 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();
    // P6 dual-layout support: legacy fan-out writes <sweep-dir>/<cell>/summary.json
    // (one cell per Argo task). P6 batched flow writes
    // <sweep-dir>/<cell>/sim_<variant>/summary.json (one Argo task, N variants
    // inside one harness). The aggregator walks both layouts: any directory
    // containing summary.json is a cell; cells one level deeper count too.
    fn collect(rows: &mut Vec<AggregateRow>, dir: &Path, cell_label_prefix: &str) -> Result<()> {
        for entry in std::fs::read_dir(dir)
            .with_context(|| format!("read sweep dir {}", 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() {
                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 leaf = p
                    .file_name()
                    .map(|n| n.to_string_lossy().into_owned())
                    .unwrap_or_else(|| p.display().to_string());
                let cell = if cell_label_prefix.is_empty() {
                    leaf
                } else {
                    format!("{cell_label_prefix}/{leaf}")
                };
                rows.push(AggregateRow::from_summary(cell, &s));
            } else {
                // No summary.json at this level — descend one more (P6 batched
                // layout: <cell>/sim_<variant>/summary.json). Skip further
                // recursion to keep the search bounded.
                let leaf = p
                    .file_name()
                    .map(|n| n.to_string_lossy().into_owned())
                    .unwrap_or_else(|| p.display().to_string());
                for sub_entry in std::fs::read_dir(&p)
                    .with_context(|| format!("read cell dir {}", p.display()))?
                {
                    let sub_entry = sub_entry?;
                    let sp = sub_entry.path();
                    if !sp.is_dir() { continue; }
                    let sub_summary = sp.join("summary.json");
                    if !sub_summary.exists() { continue; }
                    let s: Summary = serde_json::from_reader(
                        std::fs::File::open(&sub_summary)
                            .with_context(|| format!("open {}", sub_summary.display()))?,
                    )
                    .with_context(|| format!("parse {}", sub_summary.display()))?;
                    let sub_leaf = sp
                        .file_name()
                        .map(|n| n.to_string_lossy().into_owned())
                        .unwrap_or_else(|| sp.display().to_string());
                    rows.push(AggregateRow::from_summary(format!("{leaf}/{sub_leaf}"), &s));
                }
            }
        }
        Ok(())
    }
    collect(&mut rows, sweep_dir, "")?;
    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()));
    }
}