Aura/crates/aura-engine/src/report.rs

//! Run summary metrics + the reproducible run manifest (C18 / C12): the
//! `(manifest, metrics)` pair a run produces "from day one". The metrics are a
//! **post-run pure reduction** over a run's recorded streams — a node cannot
//! reduce end-of-run (C8 caps a node at one record per `eval`, with no terminal
//! `eval`), so the World drains its recording sinks after [`Harness::run`](crate::Harness::run)
//! and folds them here. Output is canonical JSON (C14): the schema is tiny,
//! closed, and flat. `to_json` renders via serde (the report types derive it,
//! cycle 0029) — the same encoder the run registry uses, so a record's stdout
//! and on-disk shapes coincide.

use aura_core::{Scalar, ScalarKind, SeriesFold, Timestamp};
use std::collections::HashMap;

/// Summary metrics reduced from a run's recorded streams — the `-> metrics`
/// half of C12's atomic sim unit. Pure function of the recorded streams.
#[derive(Clone, Debug, PartialEq, serde::Serialize, serde::Deserialize)]
pub struct RunMetrics {
    /// Final cumulative pip equity — the last value of the (cumulative)
    /// pip-equity curve. `0.0` if the curve is empty.
    pub total_pips: f64,
    /// Largest peak-to-trough drop on the cumulative pip curve:
    /// `max_t (running_peak(t) - equity(t))`, always `>= 0.0` (`0.0` if the
    /// curve is monotonic non-decreasing or empty).
    pub max_drawdown: f64,
    /// Count of adjacent recorded bias samples whose sign differs (a zero
    /// bias normalizes to sign `0`, so flat is distinct from long/short).
    /// A turnover proxy: it counts long<->short reversals *and* transitions
    /// into/out of flat — the plain sign-change count over the bias series.
    /// The serde alias accepts the pre-rename `exposure_sign_flips` key so legacy
    /// `runs.jsonl` lines still deserialise (C14/C18 back-compat).
    #[serde(alias = "exposure_sign_flips")]
    pub bias_sign_flips: u64,
    /// Optional Stage-1 R metrics block. `None` for a pip-only run (and for legacy
    /// `runs.jsonl` written before this field existed — `serde(default)`); omitted from
    /// the JSON entirely when absent (`skip_serializing_if`), so the pip-only on-disk
    /// shape stays byte-unchanged (C14/C18 back-compat).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub r: Option<RMetrics>,
}

/// R-based signal-quality metrics (Stage-1), reduced from a `PositionManagement` dense
/// record stream by [`summarize_r`]. Account- and instrument-agnostic (pure R). Carries
/// the enriched dispersion/churn fields (SQN, conviction terciles, net-of-cost).
#[derive(Clone, Debug, PartialEq, serde::Serialize, serde::Deserialize)]
pub struct RMetrics {
    pub expectancy_r: f64,   // mean realised R over all trades (equal-weighted; headline)
    pub n_trades: u64,
    pub win_rate: f64,       // fraction with R > 0
    pub avg_win_r: f64,
    pub avg_loss_r: f64,
    pub profit_factor: f64,  // sum(win R) / |sum(loss R)|; 0.0 if no losses or no trades
    pub max_r_drawdown: f64, // worst peak-to-trough on the by-trade cumulative-R curve, >= 0
    pub n_open_at_end: u64,  // positions force-closed at window end (counted, not hidden)
    pub sqn: f64,            // √n · mean_R / sample-stdev_R; n<2 or zero-variance -> 0.0
    /// (mean_R / sample-stdev_R) · √(min(n, SQN_CAP)): the n-normalized SQN
    /// ("SQN score", Van Tharp), turnover-robust vs. the raw `sqn`. n<2 or
    /// zero-variance -> 0.0.
    #[serde(default)]
    pub sqn_normalized: f64,
    pub net_expectancy_r: f64, // mean(R - round_trip_cost / latched_dist) — churn-honest
    pub conviction_terciles_r: [f64; 3], // E[R] by conviction_at_entry tercile (asc); <3 trades -> [0,0,0]
}

// Dense `PositionManagement` record column indices — the lockstep contract with
// aura-std's FIELD_NAMES/RECORD_KINDS. The record crosses the crate boundary as a
// type-erased `Scalar` vector (C4 SoA), so it is read positionally, not by importing
// the producer's types; the layout is shared by convention and `stage1_r_e2e.rs`
// guards that it matches.
mod r_col {
    pub const CLOSED: usize = 0;
    pub const REALIZED_R: usize = 1;
    pub const DIRECTION: usize = 4;
    pub const ENTRY_PRICE: usize = 6;
    pub const STOP_PRICE: usize = 7;
    pub const CONVICTION_AT_ENTRY: usize = 9;
    pub const SIZE: usize = 10;
    pub const OPEN: usize = 11;
    pub const UNREALIZED_R: usize = 12;
}

/// Van Tharp's conventional trade-count cap for the n-normalized SQN ("SQN
/// score"): capping n stops the single-number objective rewarding sheer turnover.
const SQN_CAP: u64 = 100;

/// Reduce a `PositionManagement` dense record stream into R-metrics. Pure (C1).
/// The trade ledger is the rows where `closed_this_cycle`; a position still open on the
/// last row is force-closed at its `unrealized_r` (a window-end trade — never silently
/// folded as unrealised MtM). Empty input -> a well-defined all-zero `RMetrics`.
pub fn summarize_r(record: &[(Timestamp, Vec<Scalar>)], round_trip_cost: f64) -> RMetrics {
    // Collect one entry per trade: its realised R, the entry-conviction |bias|, and the
    // latched R-distance (|entry - stop|, the frozen R-denominator) recovered from the
    // record. The ledger is the closed rows; a position still open on the last row is
    // force-closed at its unrealized R (a window-end trade).
    struct Trade {
        r: f64,
        bias_abs: f64,
        latched: f64,
    }
    let mut trades: Vec<Trade> = Vec::new();
    for (_, row) in record {
        if row[r_col::CLOSED].as_bool() {
            trades.push(Trade {
                r: row[r_col::REALIZED_R].as_f64(),
                bias_abs: row[r_col::CONVICTION_AT_ENTRY].as_f64(),
                latched: (row[r_col::ENTRY_PRICE].as_f64() - row[r_col::STOP_PRICE].as_f64()).abs(),
            });
        }
    }
    let mut n_open_at_end = 0u64;
    if let Some((_, last)) = record.last()
        && last[r_col::OPEN].as_bool()
    {
        trades.push(Trade {
            r: last[r_col::UNREALIZED_R].as_f64(),
            bias_abs: last[r_col::CONVICTION_AT_ENTRY].as_f64(),
            latched: (last[r_col::ENTRY_PRICE].as_f64() - last[r_col::STOP_PRICE].as_f64()).abs(),
        });
        n_open_at_end = 1;
    }
    let n = trades.len() as u64;
    if n == 0 {
        return RMetrics {
            expectancy_r: 0.0,
            n_trades: 0,
            win_rate: 0.0,
            avg_win_r: 0.0,
            avg_loss_r: 0.0,
            profit_factor: 0.0,
            max_r_drawdown: 0.0,
            n_open_at_end: 0,
            sqn: 0.0,
            sqn_normalized: 0.0,
            net_expectancy_r: 0.0,
            conviction_terciles_r: [0.0; 3],
        };
    }
    let rs: Vec<f64> = trades.iter().map(|t| t.r).collect();
    let sum: f64 = rs.iter().sum();
    let mean = sum / n as f64;
    let wins: Vec<f64> = rs.iter().copied().filter(|&r| r > 0.0).collect();
    let losses: Vec<f64> = rs.iter().copied().filter(|&r| r <= 0.0).collect();
    let sum_win: f64 = wins.iter().sum();
    let sum_loss: f64 = losses.iter().sum(); // <= 0
    let avg = |v: &[f64]| if v.is_empty() { 0.0 } else { v.iter().sum::<f64>() / v.len() as f64 };
    // by-trade cumulative-R drawdown
    let mut peak = f64::NEG_INFINITY;
    let mut cum = 0.0;
    let mut max_dd = 0.0_f64;
    for &r in &rs {
        cum += r;
        if cum > peak {
            peak = cum;
        }
        let dd = peak - cum;
        if dd > max_dd {
            max_dd = dd;
        }
    }
    // SQN = √n · mean / sample-stdev (raw, Van Tharp). SQN100 = √(min(n,SQN_CAP)) ·
    // mean / sd — the same dispersion ratio with the trade count capped
    // (turnover-robust). n < 2 or zero variance -> 0.0 for both (dispersion undefined).
    let (sqn, sqn_normalized) = if n < 2 {
        (0.0, 0.0)
    } else {
        let var = rs.iter().map(|&r| (r - mean).powi(2)).sum::<f64>() / (n as f64 - 1.0);
        let sd = var.sqrt();
        if sd > 0.0 {
            // raw sqn kept bit-identical to the pre-0067 expression (sqrt(n)*mean/sd,
            // not factored via a shared ratio) so the existing metric is byte-unchanged.
            (
                (n as f64).sqrt() * mean / sd,
                (n.min(SQN_CAP) as f64).sqrt() * mean / sd,
            )
        } else {
            (0.0, 0.0)
        }
    };
    // net-of-cost: subtract one round-trip spread (price units) per trade, expressed in R
    // by dividing by that trade's latched R-distance (a zero distance contributes no cost).
    let net_sum: f64 = trades
        .iter()
        .map(|t| t.r - if t.latched > 0.0 { round_trip_cost / t.latched } else { 0.0 })
        .sum();
    let net_expectancy_r = net_sum / n as f64;
    // conviction terciles: sort by conviction_at_entry ascending, split into three contiguous
    // near-equal-count buckets (floor boundaries i*n/3), E[R] per bucket. < 3 trades -> 0s.
    let conviction_terciles_r = if n < 3 {
        [0.0; 3]
    } else {
        let mut by_conv: Vec<&Trade> = trades.iter().collect();
        by_conv.sort_by(|a, b| a.bias_abs.total_cmp(&b.bias_abs));
        let nn = by_conv.len();
        let mut out = [0.0; 3];
        for (i, slot) in out.iter_mut().enumerate() {
            let lo = i * nn / 3;
            let hi = (i + 1) * nn / 3;
            let bucket = &by_conv[lo..hi];
            *slot = if bucket.is_empty() {
                0.0
            } else {
                bucket.iter().map(|t| t.r).sum::<f64>() / bucket.len() as f64
            };
        }
        out
    };
    RMetrics {
        expectancy_r: mean,
        n_trades: n,
        win_rate: wins.len() as f64 / n as f64,
        avg_win_r: avg(&wins),
        avg_loss_r: avg(&losses),
        profit_factor: if sum_loss < 0.0 { sum_win / (-sum_loss) } else { 0.0 },
        max_r_drawdown: max_dd,
        n_open_at_end,
        sqn,
        sqn_normalized,
        net_expectancy_r,
        conviction_terciles_r,
    }
}

/// Derive the broker-independent position-event table from a `PositionManagement`
/// dense record (read positionally, C7 SoA), as the first difference of the
/// executed book (`deal = target - book - in_flight`; `in_flight = 0` for the
/// instant-fill backtest). Pure (C1); no look-ahead — each event's `event_ts` is
/// its own cycle (C2). `instrument_id` is supplied by the caller (the engine never
/// imports `InstrumentSpec`). A reversal emits `Close` then the opposite open at
/// one `event_ts` (close first); a position still open on the last row emits its
/// open with no synthetic `Close` (the table records actual executed events —
/// unlike `summarize_r`, which force-closes for the R metric).
pub fn derive_position_events(
    record: &[(Timestamp, Vec<Scalar>)],
    instrument_id: i64,
) -> Vec<PositionEvent> {
    // The derive's single-position book (in_flight is structurally 0).
    struct Book {
        position_id: i64,
        volume: f64,
    }
    let mut out: Vec<PositionEvent> = Vec::new();
    let mut book: Option<Book> = None;
    let mut next_id: i64 = 0;
    for (ts, row) in record {
        // 1) close first: the book held into this cycle exited this cycle.
        if row[r_col::CLOSED].as_bool()
            && let Some(b) = book.take()
        {
            out.push(PositionEvent {
                event_ts: *ts,
                action: PositionAction::Close,
                position_id: b.position_id,
                instrument_id,
                volume: b.volume,
            });
        }
        // 2) then open: a position is open at cycle end the book is not tracking.
        if row[r_col::OPEN].as_bool() && book.is_none() {
            let dir = row[r_col::DIRECTION].as_i64();
            let volume = row[r_col::SIZE].as_f64();
            let action = if dir >= 0 { PositionAction::Buy } else { PositionAction::Sell };
            out.push(PositionEvent {
                event_ts: *ts,
                action,
                position_id: next_id,
                instrument_id,
                volume,
            });
            book = Some(Book { position_id: next_id, volume });
            next_id += 1;
        }
    }
    out
}

/// The three position-event actions (C10). Direction IS the action; volume is
/// unsigned. Serde-encoded as its i64 mapping (`Buy=0, Sell=1, Close=2`) so the
/// persisted/columnar form stays C7-scalar and ledger-faithful (`action: i64`).
#[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(into = "i64", try_from = "i64")]
pub enum PositionAction {
    Buy,
    Sell,
    Close,
}

impl From<PositionAction> for i64 {
    fn from(a: PositionAction) -> i64 {
        match a {
            PositionAction::Buy => 0,
            PositionAction::Sell => 1,
            PositionAction::Close => 2,
        }
    }
}

impl TryFrom<i64> for PositionAction {
    type Error = String;
    fn try_from(v: i64) -> Result<Self, Self::Error> {
        match v {
            0 => Ok(PositionAction::Buy),
            1 => Ok(PositionAction::Sell),
            2 => Ok(PositionAction::Close),
            other => Err(format!("invalid PositionAction i64: {other}")),
        }
    }
}

/// One row of C10's derived position-event table — the broker-independent audit
/// view (the first difference of the exposure state). A post-run value type
/// (sibling of [`RunMetrics`]), NOT a per-`eval` node output (C8). Multiple events
/// may share one `event_ts` (a reversal: Close then open, close-before-open). No
/// `open_ts` — a position's open time is its opening event's `event_ts`.
#[derive(Clone, Copy, Debug, PartialEq, serde::Serialize, serde::Deserialize)]
pub struct PositionEvent {
    pub event_ts: Timestamp,
    pub action: PositionAction,
    /// Monotonic, assigned at open. A Close references an existing `position_id`.
    pub position_id: i64,
    pub instrument_id: i64,
    /// Lots, unsigned. A partial close carries its own (smaller) volume.
    pub volume: f64,
}

/// The reproducible run descriptor (C18). **Caller-supplied**: the engine
/// cannot introspect a git commit, an RNG seed, or a broker label — the World
/// that bootstraps and runs the harness fills these in.
#[derive(Clone, Debug, PartialEq, serde::Serialize, serde::Deserialize)]
pub struct RunManifest {
    /// Node/engine identity: the git commit of the frozen artifact (C18 —
    /// commit = identity; the frozen bot *is* a commit).
    pub commit: String,
    /// The bound tuning params as ordered `name -> value` pairs. Each value is a
    /// self-describing [`Scalar`], so the param's kind (an `i64` length vs an
    /// `f64` scale) survives into the record instead of collapsing to `f64`.
    pub params: Vec<(String, Scalar)>,
    /// The data-window: inclusive `(from, to)` epoch-ns bounds (C12).
    pub window: (Timestamp, Timestamp),
    /// The RNG seed (C12 seed-as-input). `0` for a seed-free synthetic run.
    pub seed: u64,
    /// The broker profile label, e.g. `"sim-optimal(pip_size=0.0001)"`.
    pub broker: String,
}

/// A run's full structured result: the descriptor plus the metrics it
/// reproduces. The durable run record of C18 ("stores manifests + metrics,
/// re-derives full results on demand").
#[derive(Clone, Debug, PartialEq, serde::Serialize, serde::Deserialize)]
pub struct RunReport {
    pub manifest: RunManifest,
    pub metrics: RunMetrics,
}

impl RunReport {
    /// Render the canonical, machine-readable JSON (C14) via serde — the same
    /// encoder the run registry uses on disk, so a record's stdout shape and its
    /// `runs.jsonl` shape are byte-identical. `params` is an array of
    /// `[name, value]` pairs where `value` is a self-describing tagged scalar
    /// (serde's externally-tagged enum: `{"I64": 10}` for a length, `{"F64": 2.5}`
    /// for a scale). Consumers parse the tagged object, never a bare number.
    pub fn to_json(&self) -> String {
        serde_json::to_string(self).expect("a finite RunReport always serializes")
    }
}

/// Reduce a run's recorded pip-equity + exposure streams into summary metrics.
/// Pure — identical inputs yield identical metrics (C1/C12). Timestamps are
/// carried in the input to match exactly what a sink records; the reduction
/// itself is value-only (it does not read the timestamps).
pub fn summarize(
    equity: &[(Timestamp, f64)],
    exposure: &[(Timestamp, f64)],
) -> RunMetrics {
    // total_pips + max_drawdown fold over equity; sign_flips folds over exposure.
    // SeriesFold is the single arithmetic source of truth (shared with the
    // SeriesReducer sink), so this is byte-identical to the prior inline loops.
    let mut eqf = SeriesFold::new();
    for &(_, v) in equity {
        eqf.fold(v);
    }
    let mut exf = SeriesFold::new();
    for &(_, v) in exposure {
        exf.fold(v);
    }
    RunMetrics {
        total_pips: eqf.last,
        max_drawdown: eqf.max_drawdown,
        bias_sign_flips: exf.sign_flips,
        r: None,
    }
}

/// Bridge a recording sink's recorded `(ts, row)` stream to [`summarize`]:
/// extract one `f64` field of each row into `(ts, f64)` samples. Panics if a
/// row has no such field or the field is not an `f64` scalar — a wiring bug (a
/// sink's declared kinds are fixed at bootstrap, so a correctly-wired
/// equity/exposure sink always yields `f64` at field 0), surfaced like the
/// engine's other "checked at wiring" contract violations rather than silently
/// dropped.
pub fn f64_field(rows: &[(Timestamp, Vec<Scalar>)], field: usize) -> Vec<(Timestamp, f64)> {
    rows.iter()
        .map(|(ts, row)| {
            let Some(&scalar) = row.get(field) else {
                panic!("f64_field: row has no field {field} (row width {})", row.len());
            };
            if scalar.kind() != ScalarKind::F64 {
                panic!("f64_field: field {field} is not an f64 scalar: {scalar:?}");
            }
            (*ts, scalar.as_f64())
        })
        .collect()
}

/// One spine row joined with each side stream's row recorded at the same
/// timestamp. `sides` is parallel to the `sides` argument of [`join_on_ts`]; an
/// entry is `None` where that side did not fire at this spine timestamp.
#[derive(Clone, Debug, PartialEq)]
pub struct JoinedRow {
    pub ts: Timestamp,
    pub spine: Vec<Scalar>,
    pub sides: Vec<Option<Vec<Scalar>>>,
}

/// Join recording-sink tap streams on their recorded timestamp (C8/C18: a post-run
/// reduction over recorded sink output; C3: NOT an in-graph join).
///
/// `spine` defines the row set — exactly one [`JoinedRow`] per spine entry, in
/// spine order. Each side stream is looked up by timestamp: `Some(row)` where it
/// fired at that timestamp, `None` where it did not. The helper does not interpret
/// a row's columns (it returns each whole); the caller maps `None` to whatever
/// default its column means.
///
/// Precondition (C1): each stream has at most one row per timestamp — a sink fires
/// at most once per cycle and cycles have unique timestamps. A duplicate timestamp
/// within one stream resolves last-write-wins. A side row whose timestamp is absent
/// from the spine is dropped (the spine defines the rows).
pub fn join_on_ts(
    spine: &[(Timestamp, Vec<Scalar>)],
    sides: &[&[(Timestamp, Vec<Scalar>)]],
) -> Vec<JoinedRow> {
    let side_maps: Vec<HashMap<i64, &Vec<Scalar>>> = sides
        .iter()
        .map(|s| s.iter().map(|(t, row)| (t.0, row)).collect())
        .collect();

    spine
        .iter()
        .map(|(ts, row)| JoinedRow {
            ts: *ts,
            spine: row.clone(),
            sides: side_maps.iter().map(|m| m.get(&ts.0).map(|r| (*r).clone())).collect(),
        })
        .collect()
}

/// One drained recorder tap in columnar (SoA) form: parallel arrays, one per
/// recorded column, plus the shared recorded-timestamp axis. Chart-ready (a column
/// is a series of numbers) and kind-tagged (C7) — values are coerced to f64 for
/// plotting; the base type survives in `kinds`. Pure: identical rows always encode
/// to the same value (C1).
#[derive(Clone, Debug, PartialEq, serde::Serialize, serde::Deserialize)]
pub struct ColumnarTrace {
    pub tap: String,
    pub kinds: Vec<String>,
    pub ts: Vec<i64>,
    pub columns: Vec<Vec<f64>>,
}

impl ColumnarTrace {
    /// Transpose a drained tap's `(ts, row)` pairs into columns. An empty `rows`
    /// yields empty `ts` and one empty column per kind. Each cell is coerced to f64:
    /// f64 as-is, i64 as f64, bool 1.0/0.0, timestamp epoch as f64. Panics if any
    /// row's width disagrees with `kinds.len()` — a wiring bug (a sink's column
    /// count is fixed at bootstrap), surfaced as a named panic like
    /// [`f64_field`] rather than a bare index-out-of-bounds (wider) or silent
    /// ragged columns (narrower).
    pub fn from_rows(tap: &str, kinds: &[ScalarKind], rows: &[(Timestamp, Vec<Scalar>)]) -> Self {
        let ts: Vec<i64> = rows.iter().map(|(t, _)| t.0).collect();
        let mut columns: Vec<Vec<f64>> = vec![Vec::with_capacity(rows.len()); kinds.len()];
        for (_, row) in rows {
            if row.len() != kinds.len() {
                panic!(
                    "from_rows: row width {} disagrees with kinds.len() {} (tap {tap:?})",
                    row.len(),
                    kinds.len(),
                );
            }
            for (c, scalar) in row.iter().enumerate() {
                columns[c].push(scalar_to_f64(*scalar));
            }
        }
        ColumnarTrace {
            tap: tap.to_string(),
            kinds: kinds.iter().map(kind_tag).collect(),
            ts,
            columns,
        }
    }

    /// Inverse for the serve/align path: rebuild `(ts, row)` pairs with each cell as
    /// `Scalar::f64(columns[c][r])` — uniformly f64 (the on-disk store is f64
    /// columns; the base type survives only in `kinds`). A true value round-trip for
    /// f64 taps; keeps the serve-side `as_f64` projection total.
    pub fn to_rows(&self) -> Vec<(Timestamp, Vec<Scalar>)> {
        self.ts
            .iter()
            .enumerate()
            .map(|(r, &t)| {
                let row = self.columns.iter().map(|col| Scalar::f64(col[r])).collect();
                (Timestamp(t), row)
            })
            .collect()
    }
}

/// The four-kind tag string for a column (the on-disk `kinds` form; `ScalarKind`
/// derives no serde, so tags are plain strings).
fn kind_tag(kind: &ScalarKind) -> String {
    match kind {
        ScalarKind::F64 => "F64",
        ScalarKind::I64 => "I64",
        ScalarKind::Bool => "Bool",
        ScalarKind::Timestamp => "Timestamp",
    }
    .to_string()
}

/// Coerce a recorded scalar to the f64 a chart plots: f64 as-is, i64 as f64,
/// bool 1.0/0.0, timestamp epoch as f64.
fn scalar_to_f64(s: Scalar) -> f64 {
    match s.kind() {
        ScalarKind::F64 => s.as_f64(),
        ScalarKind::I64 => s.as_i64() as f64,
        ScalarKind::Bool => {
            if s.as_bool() {
                1.0
            } else {
                0.0
            }
        }
        ScalarKind::Timestamp => s.as_ts().0 as f64,
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{Edge, FlatGraph, Harness, SourceSpec, Target, VecSource};
    use aura_core::{Firing, NodeSchema, PortSpec, ScalarKind};
    use aura_std::{Bias, Recorder, SimBroker, Sma, Sub};
    use std::sync::mpsc;

    #[test]
    fn position_action_round_trips_through_i64() {
        for a in [PositionAction::Buy, PositionAction::Sell, PositionAction::Close] {
            let n: i64 = a.into();
            assert_eq!(PositionAction::try_from(n), Ok(a));
        }
        assert_eq!(i64::from(PositionAction::Buy), 0);
        assert_eq!(i64::from(PositionAction::Sell), 1);
        assert_eq!(i64::from(PositionAction::Close), 2);
    }

    #[test]
    fn position_action_rejects_out_of_range_i64() {
        assert!(PositionAction::try_from(3).is_err());
        assert!(PositionAction::try_from(-1).is_err());
    }

    #[test]
    fn position_event_serde_round_trips_with_bare_int_action() {
        let ev = PositionEvent {
            event_ts: Timestamp(42),
            action: PositionAction::Sell,
            position_id: 7,
            instrument_id: 3,
            volume: 0.5,
        };
        let json = serde_json::to_string(&ev).expect("serialize");
        // action encodes as a bare integer (C7 scalar shape), not a tagged enum
        assert!(json.contains("\"action\":1"), "action not bare-int encoded: {json}");
        let back: PositionEvent = serde_json::from_str(&json).expect("deserialize");
        assert_eq!(back, ev);
    }

    #[test]
    fn position_events_may_share_one_event_ts_on_reversal() {
        // a stop-and-reverse: Close then open at the SAME event_ts (close-before-open).
        let ts = Timestamp(100);
        let close = PositionEvent {
            event_ts: ts, action: PositionAction::Close,
            position_id: 1, instrument_id: 3, volume: 0.5,
        };
        let open = PositionEvent {
            event_ts: ts, action: PositionAction::Sell,
            position_id: 2, instrument_id: 3, volume: 0.5,
        };
        let table = [close, open];
        assert_eq!(table[0].event_ts, table[1].event_ts);
        assert_eq!(table[0].action, PositionAction::Close);
        assert_eq!(table[1].action, PositionAction::Sell);
    }

    /// The declared signature of a `Recorder` over one f64 column (the sink shape
    /// the two-sink harness uses).
    fn f64_recorder_sig() -> NodeSchema {
        NodeSchema {
            inputs: vec![PortSpec { kind: ScalarKind::F64, firing: Firing::Any, name: "in".into() }],
            output: vec![],
            params: vec![],
        }
    }

    /// Build an f64 source stream from (timestamp, value) points (mirrors the
    /// harness.rs test helper; the e2e test needs its own copy — the harness
    /// test module's is private to that module).
    fn f64_stream(points: &[(i64, f64)]) -> Vec<(Timestamp, Scalar)> {
        points.iter().map(|&(t, v)| (Timestamp(t), Scalar::f64(v))).collect()
    }

    /// Bootstrap the cycle-0007 signal-quality harness with TWO sinks: one on
    /// the SimBroker equity output (node 4 -> node 5) and one on the Bias
    /// output (node 3 -> node 6). Returns the harness plus the two receivers.
    #[allow(clippy::type_complexity)]
    fn build_two_sink_harness() -> (
        Harness,
        mpsc::Receiver<(Timestamp, Vec<Scalar>)>,
        mpsc::Receiver<(Timestamp, Vec<Scalar>)>,
    ) {
        let (tx_eq, rx_eq) = mpsc::channel();
        let (tx_ex, rx_ex) = mpsc::channel();
        let h = Harness::bootstrap(FlatGraph {
            nodes: vec![
                Box::new(Sma::new(2)),                                       // 0
                Box::new(Sma::new(4)),                                       // 1
                Box::new(Sub::new()),                                        // 2
                Box::new(Bias::new(0.5)),                                    // 3
                Box::new(SimBroker::new(0.0001)),                            // 4
                Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_eq)), // 5 equity sink
                Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_ex)), // 6 exposure sink
            ],
            signatures: vec![
                Sma::builder().schema().clone(),
                Sma::builder().schema().clone(),
                Sub::builder().schema().clone(),
                Bias::builder().schema().clone(),
                SimBroker::builder(0.0001).schema().clone(),
                f64_recorder_sig(),
                f64_recorder_sig(),
            ],
            sources: vec![SourceSpec {
                kind: ScalarKind::F64,
                targets: vec![
                    Target { node: 0, slot: 0 },
                    Target { node: 1, slot: 0 },
                    Target { node: 4, slot: 1 }, // price into the broker
                ],
            }],
            edges: vec![
                Edge { from: 0, to: 2, slot: 0, from_field: 0 },
                Edge { from: 1, to: 2, slot: 1, from_field: 0 },
                Edge { from: 2, to: 3, slot: 0, from_field: 0 },
                Edge { from: 3, to: 4, slot: 0, from_field: 0 },
                Edge { from: 4, to: 5, slot: 0, from_field: 0 }, // equity -> sink 5
                Edge { from: 3, to: 6, slot: 0, from_field: 0 }, // exposure -> sink 6
            ],
        })
        .expect("valid signal-quality DAG");
        (h, rx_eq, rx_ex)
    }

    fn run_once() -> RunReport {
        let (mut h, rx_eq, rx_ex) = build_two_sink_harness();
        h.run(vec![Box::new(VecSource::new(f64_stream(&[
            (1, 1.0000),
            (2, 1.0010),
            (3, 1.0025),
            (4, 1.0020),
            (5, 1.0040),
        ])))]);
        let eq_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_eq.try_iter().collect();
        let ex_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_ex.try_iter().collect();
        let equity = f64_field(&eq_rows, 0);
        let exposure = f64_field(&ex_rows, 0);
        let metrics = summarize(&equity, &exposure);
        RunReport {
            manifest: RunManifest {
                commit: "test-commit".to_string(),
                params: vec![
                    ("sma_fast".to_string(), Scalar::i64(2)),
                    ("sma_slow".to_string(), Scalar::i64(4)),
                    ("bias_scale".to_string(), Scalar::f64(0.5)),
                ],
                window: (Timestamp(1), Timestamp(5)),
                seed: 0,
                broker: "sim-optimal(pip_size=0.0001)".to_string(),
            },
            metrics,
        }
    }

    #[test]
    fn report_is_deterministic_end_to_end() {
        let r1 = run_once();
        let r2 = run_once();
        // a run actually emitted metrics over a non-empty pip curve
        assert!(r1.metrics.total_pips.is_finite());
        // same manifest -> same metrics (C1/C12): two runs are bit-identical
        assert_eq!(r1.metrics, r2.metrics);
        assert_eq!(r1.to_json(), r2.to_json());
    }

    // Build a minimal dense record row: only the columns summarize_r reads matter.
    // The four written slots reference `r_col::*` (not bare literals) so the helper
    // tracks the same constants it exists to test — a literal would mask drift in them.
    fn r_row(closed: bool, realized: f64, open: bool, unreal: f64) -> (Timestamp, Vec<Scalar>) {
        let mut v = vec![Scalar::f64(0.0); r_col::UNREALIZED_R + 1];
        v[r_col::CLOSED] = Scalar::bool(closed);
        v[r_col::REALIZED_R] = Scalar::f64(realized);
        v[r_col::OPEN] = Scalar::bool(open);
        v[r_col::UNREALIZED_R] = Scalar::f64(unreal);
        (Timestamp(0), v)
    }

    // A fuller closed-trade dense row: also sets entry_price (6), stop_price (7) and
    // conviction_at_entry (9) — the geometry summarize_r recovers latched_dist and
    // conviction from. Width up to UNREALIZED_R+1 (summarize_r never reads col 13).
    fn r_row_full(realized: f64, entry: f64, stop: f64, bias_abs: f64) -> (Timestamp, Vec<Scalar>) {
        let mut v = vec![Scalar::f64(0.0); r_col::UNREALIZED_R + 1];
        v[r_col::CLOSED] = Scalar::bool(true);
        v[r_col::REALIZED_R] = Scalar::f64(realized);
        v[r_col::ENTRY_PRICE] = Scalar::f64(entry);
        v[r_col::STOP_PRICE] = Scalar::f64(stop);
        v[r_col::CONVICTION_AT_ENTRY] = Scalar::f64(bias_abs);
        v[r_col::OPEN] = Scalar::bool(false);
        (Timestamp(0), v)
    }
    #[test]
    fn summarize_r_is_zero_on_empty() {
        let m = summarize_r(&[], 0.0);
        assert_eq!(m.n_trades, 0);
        assert_eq!(m.expectancy_r, 0.0);
        assert_eq!(m.max_r_drawdown, 0.0);
        assert_eq!(m.sqn, 0.0);
        assert_eq!(m.sqn_normalized, 0.0);
        assert_eq!(m.net_expectancy_r, 0.0);
        assert_eq!(m.conviction_terciles_r, [0.0; 3]);
    }
    #[test]
    fn summarize_r_expectancy_winrate_profit_factor() {
        // three closed trades: +2, -1, +1 (no open at end). E[R] = 2/3.
        let rec = vec![
            r_row(true, 2.0, true, 0.0),
            r_row(false, 0.0, true, 0.0),
            r_row(true, -1.0, true, 0.0),
            r_row(true, 1.0, false, 0.0), // last row not open
        ];
        let m = summarize_r(&rec, 0.0);
        assert_eq!(m.n_trades, 3);
        assert!((m.expectancy_r - (2.0 / 3.0)).abs() < 1e-9);
        assert!((m.win_rate - (2.0 / 3.0)).abs() < 1e-9);
        assert!((m.profit_factor - 3.0).abs() < 1e-9); // (2+1)/1
        assert_eq!(m.n_open_at_end, 0);
    }
    #[test]
    fn summarize_r_force_closes_open_position_at_window_end() {
        // one closed +1, then last row open with unrealized -0.5 -> a window-end trade.
        let rec = vec![r_row(true, 1.0, true, 0.0), r_row(false, 0.0, true, -0.5)];
        let m = summarize_r(&rec, 0.0);
        assert_eq!(m.n_trades, 2);
        assert_eq!(m.n_open_at_end, 1);
        assert!((m.expectancy_r - 0.25).abs() < 1e-9); // (1 + -0.5)/2
    }
    #[test]
    fn summarize_r_max_drawdown_on_by_trade_curve() {
        // cum: +3, +1 (dd 2), +4 -> max dd = 2.
        let rec = vec![r_row(true, 3.0, false, 0.0), r_row(true, -2.0, false, 0.0), r_row(true, 3.0, false, 0.0)];
        assert!((summarize_r(&rec, 0.0).max_r_drawdown - 2.0).abs() < 1e-9);
    }

    #[test]
    fn summarize_r_sqn_is_sqrt_n_mean_over_stdev() {
        // R = [1, 1, 1, -1]: mean 0.5; sample var = ((0.5)^2*3 + (1.5)^2)/3 = 1.0; sd = 1;
        // sqn = sqrt(4)*0.5/1 = 1.0.
        let rec = vec![
            r_row(true, 1.0, false, 0.0),
            r_row(true, 1.0, false, 0.0),
            r_row(true, 1.0, false, 0.0),
            r_row(true, -1.0, false, 0.0),
        ];
        let m = summarize_r(&rec, 0.0);
        assert!((m.sqn - 1.0).abs() < 1e-9, "sqn = sqrt(n)*mean/sd; got {}", m.sqn);
    }

    #[test]
    fn summarize_r_sqn_zero_when_under_two_trades_or_zero_variance() {
        // n < 2 -> 0
        assert_eq!(summarize_r(&[r_row(true, 2.0, false, 0.0)], 0.0).sqn, 0.0);
        // identical R -> zero variance -> 0
        let flat = vec![r_row(true, 1.0, false, 0.0), r_row(true, 1.0, false, 0.0), r_row(true, 1.0, false, 0.0)];
        assert_eq!(summarize_r(&flat, 0.0).sqn, 0.0);
    }

    #[test]
    fn summarize_r_sqn_normalized_caps_trade_count_at_100() {
        // 144 closed trades alternating R = 1.0 / 2.0 (nonzero variance, sd > 0).
        // raw sqn = √144·q, SQN100 = √(min(144,100))·q = √100·q for the SAME ratio
        // q = mean/sd, so sqn / sqn_normalized = √(144/100) = 1.2, independent of q.
        let rec: Vec<_> = (0..144)
            .map(|i| r_row(true, if i % 2 == 0 { 1.0 } else { 2.0 }, false, 0.0))
            .collect();
        let m = summarize_r(&rec, 0.0);
        assert_eq!(m.n_trades, 144);
        assert!(m.sqn_normalized > 0.0, "sqn_normalized nonzero; got {}", m.sqn_normalized);
        assert!(m.sqn_normalized < m.sqn, "capped SQN100 < raw sqn for n > 100");
        assert!(
            (m.sqn / m.sqn_normalized - 1.2).abs() < 1e-9,
            "sqn / sqn_normalized = √(144/100) = 1.2; got {}",
            m.sqn / m.sqn_normalized
        );
    }

    #[test]
    fn summarize_r_sqn_normalized_equals_raw_sqn_below_cap() {
        // n = 4 <= 100 -> cap inactive -> SQN100 == raw sqn (same ledger as
        // summarize_r_sqn_is_sqrt_n_mean_over_stdev, where sqn = 1.0).
        let rec = vec![
            r_row(true, 1.0, false, 0.0),
            r_row(true, 1.0, false, 0.0),
            r_row(true, 1.0, false, 0.0),
            r_row(true, -1.0, false, 0.0),
        ];
        let m = summarize_r(&rec, 0.0);
        assert!((m.sqn_normalized - m.sqn).abs() < 1e-12, "below cap, SQN100 == sqn");
        assert!((m.sqn_normalized - 1.0).abs() < 1e-9, "sqn_normalized = 1.0 here");
    }

    #[test]
    fn summarize_r_conviction_terciles_order_by_bias() {
        // Calibrated: |bias| correlates with R. Sorted ascending by |bias|, the three
        // terciles (2 trades each) are [-2,-1]->-1.5, [0,0]->0, [+1,+2]->+1.5.
        let calibrated = vec![
            r_row_full(-2.0, 100.0, 99.0, 0.1),
            r_row_full(-1.0, 100.0, 99.0, 0.2),
            r_row_full(0.0, 100.0, 99.0, 0.5),
            r_row_full(0.0, 100.0, 99.0, 0.6),
            r_row_full(1.0, 100.0, 99.0, 0.9),
            r_row_full(2.0, 100.0, 99.0, 1.0),
        ];
        let m = summarize_r(&calibrated, 0.0);
        assert!((m.conviction_terciles_r[0] - (-1.5)).abs() < 1e-9, "low tercile: {:?}", m.conviction_terciles_r);
        assert!((m.conviction_terciles_r[2] - 1.5).abs() < 1e-9, "high tercile: {:?}", m.conviction_terciles_r);
        assert!(m.conviction_terciles_r[2] > m.conviction_terciles_r[0], "high conviction must out-earn low when calibrated");

        // Anti-calibrated: high |bias| earns LESS. The ordering must INVERT — proving the
        // metric reads |bias|, not trade order. Same R multiset, |bias| reversed.
        let anti = vec![
            r_row_full(2.0, 100.0, 99.0, 0.1),
            r_row_full(1.0, 100.0, 99.0, 0.2),
            r_row_full(0.0, 100.0, 99.0, 0.5),
            r_row_full(0.0, 100.0, 99.0, 0.6),
            r_row_full(-1.0, 100.0, 99.0, 0.9),
            r_row_full(-2.0, 100.0, 99.0, 1.0),
        ];
        let a = summarize_r(&anti, 0.0);
        assert!(a.conviction_terciles_r[2] < a.conviction_terciles_r[0], "anti-calibrated must invert: {:?}", a.conviction_terciles_r);
    }

    #[test]
    fn summarize_r_conviction_terciles_zero_under_three_trades() {
        let rec = vec![r_row_full(2.0, 100.0, 99.0, 0.5), r_row_full(-1.0, 100.0, 99.0, 0.5)];
        assert_eq!(summarize_r(&rec, 0.0).conviction_terciles_r, [0.0; 3]);
    }

    #[test]
    fn summarize_r_net_of_cost_subtracts_round_trip_per_trade() {
        // one trade: R=+2, entry 100, stop 90 -> latched 10. round_trip_cost 1.0 (price
        // units) -> cost in R = 1/10 = 0.1. gross E[R]=2.0, net = 1.9.
        let rec = vec![r_row_full(2.0, 100.0, 90.0, 0.5)];
        let gross = summarize_r(&rec, 0.0);
        let net = summarize_r(&rec, 1.0);
        assert!((gross.expectancy_r - 2.0).abs() < 1e-9);
        assert!((gross.net_expectancy_r - 2.0).abs() < 1e-9, "zero cost -> net == gross");
        assert!((net.expectancy_r - 2.0).abs() < 1e-9, "cost never changes gross expectancy");
        assert!((net.net_expectancy_r - 1.9).abs() < 1e-9, "net = 2 - 1/10; got {}", net.net_expectancy_r);
    }

    // One PositionManagement dense record row for the derive tests: only the columns
    // derive_position_events reads (closed, direction, size, open) are set; the rest
    // default to 0. Distinct per-row timestamps (the derive keys events on the row's
    // own cycle, not the entry_ts column).
    fn pm_row(ts: i64, closed: bool, dir: i64, size: f64, open: bool) -> (Timestamp, Vec<Scalar>) {
        let mut v = vec![Scalar::f64(0.0); r_col::UNREALIZED_R + 1];
        v[r_col::CLOSED] = Scalar::bool(closed);
        v[r_col::DIRECTION] = Scalar::i64(dir);
        v[r_col::SIZE] = Scalar::f64(size);
        v[r_col::OPEN] = Scalar::bool(open);
        (Timestamp(ts), v)
    }

    #[test]
    fn derive_reversal_emits_close_then_opposite_open_at_one_ts() {
        let rec = vec![
            pm_row(10, false, 1, 2.0, true),   // open long
            pm_row(20, true, -1, 3.0, true),   // reversal: close long, reopen short
            pm_row(30, true, -1, 3.0, false),  // close short, flat
        ];
        let ev = derive_position_events(&rec, 42);
        assert_eq!(ev.len(), 4);
        assert_eq!(ev[0].action, PositionAction::Buy);
        assert_eq!(ev[0].event_ts, Timestamp(10));
        assert_eq!(ev[0].position_id, 0);
        assert_eq!(ev[0].volume, 2.0);
        assert_eq!(ev[0].instrument_id, 42);
        assert_eq!(ev[1].action, PositionAction::Close);
        assert_eq!(ev[1].position_id, 0);
        assert_eq!(ev[1].event_ts, Timestamp(20));
        assert_eq!(ev[1].volume, 2.0); // close sizes the actual (old) book, not the new leg
        assert_eq!(ev[2].action, PositionAction::Sell);
        assert_eq!(ev[2].position_id, 1);
        assert_eq!(ev[2].event_ts, Timestamp(20)); // same instant as the close
        assert_eq!(ev[2].volume, 3.0);
        assert_eq!(ev[3].action, PositionAction::Close);
        assert_eq!(ev[3].position_id, 1);
        assert_eq!(ev[3].event_ts, Timestamp(30));
    }

    #[test]
    fn derive_normal_open_hold_close_lifecycle() {
        let rec = vec![
            pm_row(1, false, 1, 1.0, true),  // open long
            pm_row(2, false, 1, 1.0, true),  // hold (no event)
            pm_row(3, true, 1, 1.0, false),  // close to flat
        ];
        let ev = derive_position_events(&rec, 9);
        assert_eq!(ev.len(), 2);
        assert_eq!(ev[0].action, PositionAction::Buy);
        assert_eq!(ev[0].event_ts, Timestamp(1));
        assert_eq!(ev[1].action, PositionAction::Close);
        assert_eq!(ev[1].event_ts, Timestamp(3));
        assert_eq!(ev[1].position_id, 0);
    }

    #[test]
    fn derive_short_entry_emits_sell() {
        let ev = derive_position_events(&[pm_row(5, false, -1, 1.0, true)], 0);
        assert_eq!(ev.len(), 1);
        assert_eq!(ev[0].action, PositionAction::Sell);
        assert_eq!(ev[0].position_id, 0);
    }

    #[test]
    fn derive_window_end_open_has_no_synthetic_close() {
        // opened and never closed in-window -> open event only, no Close.
        let rec = vec![
            pm_row(1, false, 1, 1.0, true),
            pm_row(2, false, 1, 1.0, true), // still open on the last row
        ];
        let ev = derive_position_events(&rec, 0);
        assert_eq!(ev.len(), 1);
        assert_eq!(ev[0].action, PositionAction::Buy);
    }

    #[test]
    fn derive_empty_record_is_empty_table() {
        let rec: Vec<(Timestamp, Vec<Scalar>)> = vec![];
        assert!(derive_position_events(&rec, 0).is_empty());
    }

    #[test]
    fn derive_position_ids_are_monotonic_across_trades() {
        let rec = vec![
            pm_row(1, false, 1, 1.0, true),   // open id0
            pm_row(2, true, 1, 1.0, false),   // close id0
            pm_row(3, false, -1, 1.0, true),  // open id1 (short)
            pm_row(4, true, -1, 1.0, false),  // close id1
        ];
        let ev = derive_position_events(&rec, 0);
        assert_eq!(ev.len(), 4);
        assert_eq!(ev[0].position_id, 0);
        assert_eq!(ev[1].position_id, 0);
        assert_eq!(ev[2].position_id, 1);
        assert_eq!(ev[3].position_id, 1);
    }

    fn samples(values: &[f64]) -> Vec<(Timestamp, f64)> {
        values
            .iter()
            .enumerate()
            .map(|(i, &v)| (Timestamp(i as i64 + 1), v))
            .collect()
    }

    #[test]
    fn summarize_total_pips_is_last_cumulative_value() {
        let equity = samples(&[0.0, 5.0, 4.0, 12.0]);
        let m = summarize(&equity, &[]);
        assert_eq!(m.total_pips, 12.0);
    }

    #[test]
    fn summarize_is_zero_on_empty_streams() {
        let m = summarize(&[], &[]);
        assert_eq!(m.total_pips, 0.0);
        assert_eq!(m.max_drawdown, 0.0);
        assert_eq!(m.bias_sign_flips, 0);
    }

    #[test]
    fn summarize_max_drawdown_is_worst_peak_to_trough() {
        // peak 10 then trough 5 (drop 5), recovers to 8; worst drop is 5,
        // not the final drop (10 -> 8 = 2).
        let equity = samples(&[0.0, 10.0, 5.0, 8.0]);
        let m = summarize(&equity, &[]);
        assert_eq!(m.max_drawdown, 5.0);
    }

    #[test]
    fn summarize_max_drawdown_zero_on_monotonic_curve() {
        let equity = samples(&[0.0, 1.0, 2.0, 3.0]);
        let m = summarize(&equity, &[]);
        assert_eq!(m.max_drawdown, 0.0);
    }

    #[test]
    fn summarize_sign_flips_counts_signum_changes() {
        // signum series: + + - 0 -  ->  flips at +->-, -->0, 0->- = 3.
        let exposure = samples(&[0.5, 0.5, -0.5, 0.0, -0.5]);
        let m = summarize(&[], &exposure);
        assert_eq!(m.bias_sign_flips, 3);
    }

    #[test]
    fn summarize_sign_flips_zero_on_constant_sign() {
        let exposure = samples(&[0.2, 0.5, 1.0, 0.7]);
        let m = summarize(&[], &exposure);
        assert_eq!(m.bias_sign_flips, 0);
    }

    #[test]
    fn f64_field_projects_the_named_field() {
        let rows = vec![
            (Timestamp(1), vec![Scalar::f64(1.5), Scalar::i64(9)]),
            (Timestamp(2), vec![Scalar::f64(2.5), Scalar::i64(8)]),
        ];
        assert_eq!(
            f64_field(&rows, 0),
            vec![(Timestamp(1), 1.5), (Timestamp(2), 2.5)],
        );
    }

    #[test]
    #[should_panic(expected = "not an f64 scalar")]
    fn f64_field_panics_on_kind_mismatch() {
        let rows = vec![(Timestamp(1), vec![Scalar::i64(7)])];
        let _ = f64_field(&rows, 0);
    }

    #[test]
    fn to_json_renders_the_canonical_form() {
        let report = RunReport {
            manifest: RunManifest {
                commit: "abc123".to_string(),
                params: vec![
                    ("sma_fast".to_string(), Scalar::i64(2)),
                    ("sma_slow".to_string(), Scalar::i64(4)),
                    ("bias_scale".to_string(), Scalar::f64(1.0)),
                ],
                window: (Timestamp(1), Timestamp(6)),
                seed: 0,
                broker: "sim-optimal(pip_size=1.0)".to_string(),
            },
            metrics: RunMetrics {
                total_pips: 12.0,
                max_drawdown: 1.0,
                bias_sign_flips: 1,
                r: None,
            },
        };
        assert_eq!(
            report.to_json(),
            r#"{"manifest":{"commit":"abc123","params":[["sma_fast",{"I64":2}],["sma_slow",{"I64":4}],["bias_scale",{"F64":1.0}]],"window":[1,6],"seed":0,"broker":"sim-optimal(pip_size=1.0)"},"metrics":{"total_pips":12.0,"max_drawdown":1.0,"bias_sign_flips":1}}"#,
        );
    }

    #[test]
    fn to_json_equals_serde_disk_shape() {
        // the same RunReport value the canonical-form test builds.
        let report = RunReport {
            manifest: RunManifest {
                commit: "abc123".to_string(),
                params: vec![
                    ("sma_fast".to_string(), Scalar::i64(2)),
                    ("sma_slow".to_string(), Scalar::i64(4)),
                    ("bias_scale".to_string(), Scalar::f64(1.0)),
                ],
                window: (Timestamp(1), Timestamp(6)),
                seed: 0,
                broker: "sim-optimal(pip_size=1.0)".to_string(),
            },
            metrics: RunMetrics { total_pips: 12.0, max_drawdown: 1.0, bias_sign_flips: 1, r: None },
        };
        // stdout (to_json) and disk (serde_json::to_string) are now the same bytes.
        assert_eq!(report.to_json(), serde_json::to_string(&report).unwrap());
    }

    #[test]
    fn runreport_serde_round_trips() {
        let report = RunReport {
            manifest: RunManifest {
                commit: "abc123".to_string(),
                params: vec![
                    ("sma_fast".to_string(), Scalar::i64(2)),
                    ("sma_slow".to_string(), Scalar::i64(4)),
                    ("bias_scale".to_string(), Scalar::f64(1.0)),
                ],
                window: (Timestamp(1), Timestamp(6)),
                seed: 0,
                broker: "sim-optimal(pip_size=1.0)".to_string(),
            },
            metrics: RunMetrics { total_pips: 12.0, max_drawdown: 1.0, bias_sign_flips: 1, r: None },
        };
        let json = serde_json::to_string(&report).expect("serialize RunReport");
        // window is a 2-element [from, to] array (Timestamp newtype is transparent)
        assert!(json.contains("\"window\":[1,6]"), "window shape: {json}");
        let back: RunReport = serde_json::from_str(&json).expect("deserialize RunReport");
        assert_eq!(back, report);
    }

    #[test]
    fn runmetrics_with_r_block_round_trips() {
        let m = RunMetrics {
            total_pips: 3.0,
            max_drawdown: 1.0,
            bias_sign_flips: 2,
            r: Some(RMetrics {
                expectancy_r: 0.5,
                n_trades: 4,
                win_rate: 0.5,
                avg_win_r: 1.5,
                avg_loss_r: -0.5,
                profit_factor: 3.0,
                max_r_drawdown: 0.5,
                n_open_at_end: 1,
                sqn: 1.0,
                sqn_normalized: 1.0,
                net_expectancy_r: 0.4,
                conviction_terciles_r: [-0.5, 0.5, 1.5],
            }),
        };
        let json = serde_json::to_string(&m).expect("serialize");
        assert!(json.contains("\"r\":{"), "r block present when Some: {json}");
        let back: RunMetrics = serde_json::from_str(&json).expect("deserialize");
        assert_eq!(back, m);
    }

    #[test]
    fn rmetrics_deserializes_without_sqn_normalized_field_to_zero() {
        // C18 back-compat: a pre-0067 `r:` block (serialized before sqn_normalized
        // existed) lacks the field; serde(default) must fill it with 0.0.
        let legacy = r#"{"expectancy_r":1.0,"n_trades":4,"win_rate":0.5,"avg_win_r":1.5,"avg_loss_r":-0.5,"profit_factor":3.0,"max_r_drawdown":0.5,"n_open_at_end":0,"sqn":1.0,"net_expectancy_r":0.4,"conviction_terciles_r":[0.0,0.0,0.0]}"#;
        let m: RMetrics = serde_json::from_str(legacy).expect("legacy RMetrics deserializes");
        assert_eq!(m.sqn_normalized, 0.0);
        assert_eq!(m.sqn, 1.0);
    }

    #[test]
    fn legacy_exposure_sign_flips_key_reads_via_alias_and_new_output_uses_bias() {
        // a legacy runs.jsonl metrics line carries the OLD key — the serde alias must accept it.
        let legacy = r#"{"total_pips":12.0,"max_drawdown":1.0,"exposure_sign_flips":3}"#;
        let m: RunMetrics = serde_json::from_str(legacy).expect("legacy exposure_sign_flips deserialises");
        assert_eq!(m.bias_sign_flips, 3);
        // new output serialises the NEW key, and the old key is gone from output.
        let json = serde_json::to_string(&m).unwrap();
        assert!(json.contains("\"bias_sign_flips\":3"), "new output uses bias_sign_flips: {json}");
        assert!(!json.contains("exposure_sign_flips"), "old key absent from new output: {json}");
    }

    #[test]
    fn legacy_runmetrics_without_r_field_deserialises_to_none() {
        // a pre-`r` runs.jsonl line: no `r` key at all.
        let legacy = r#"{"total_pips":12.0,"max_drawdown":1.0,"exposure_sign_flips":1}"#;
        let m: RunMetrics = serde_json::from_str(legacy).expect("legacy line still deserialises");
        assert_eq!(m.r, None);
        assert_eq!(m.total_pips, 12.0);
    }

    #[test]
    fn runmetrics_none_r_is_omitted_from_json() {
        let m = RunMetrics { total_pips: 1.0, max_drawdown: 0.0, bias_sign_flips: 0, r: None };
        let json = serde_json::to_string(&m).expect("serialize");
        assert!(!json.contains("\"r\""), "a None r must be omitted from the JSON: {json}");
    }

    #[test]
    fn join_on_ts_aligns_streams_of_different_cardinality() {
        // spine fires every bar; side A is one row shorter (no ts 10, like cold
        // Delay(1) on the first bar); side B fires on a subset (only ts 20, 40,
        // like a Session filter before the open).
        let spine = vec![
            (Timestamp(10), vec![Scalar::f64(1.0)]),
            (Timestamp(20), vec![Scalar::f64(2.0)]),
            (Timestamp(30), vec![Scalar::f64(3.0)]),
            (Timestamp(40), vec![Scalar::f64(4.0)]),
        ];
        let side_a = vec![
            (Timestamp(20), vec![Scalar::bool(true)]),
            (Timestamp(30), vec![Scalar::bool(false)]),
            (Timestamp(40), vec![Scalar::bool(true)]),
        ];
        let side_b = vec![
            (Timestamp(20), vec![Scalar::i64(0)]),
            (Timestamp(40), vec![Scalar::i64(2)]),
        ];

        let joined = join_on_ts(&spine, &[&side_a, &side_b]);

        // one row per spine entry, in spine order
        assert_eq!(joined.len(), 4);
        assert_eq!(
            joined.iter().map(|j| j.ts).collect::<Vec<_>>(),
            vec![Timestamp(10), Timestamp(20), Timestamp(30), Timestamp(40)]
        );

        // ts 10: spine present, both sides absent (the zip-by-index misalignment case)
        assert_eq!(joined[0].spine, vec![Scalar::f64(1.0)]);
        assert_eq!(joined[0].sides[0], None);
        assert_eq!(joined[0].sides[1], None);

        // ts 20: both sides present and aligned to THIS ts
        assert_eq!(joined[1].sides[0], Some(vec![Scalar::bool(true)]));
        assert_eq!(joined[1].sides[1], Some(vec![Scalar::i64(0)]));

        // ts 30: side A present, side B absent
        assert_eq!(joined[2].sides[0], Some(vec![Scalar::bool(false)]));
        assert_eq!(joined[2].sides[1], None);

        // ts 40: both present
        assert_eq!(joined[3].sides[0], Some(vec![Scalar::bool(true)]));
        assert_eq!(joined[3].sides[1], Some(vec![Scalar::i64(2)]));
    }

    #[test]
    fn join_on_ts_drops_side_rows_absent_from_spine() {
        // a side row whose ts is not in the spine is dropped — the spine defines
        // the row set.
        let spine = vec![(Timestamp(10), vec![Scalar::f64(1.0)])];
        let side = vec![
            (Timestamp(10), vec![Scalar::i64(7)]),
            (Timestamp(99), vec![Scalar::i64(8)]), // ts 99 absent from spine -> dropped
        ];

        let joined = join_on_ts(&spine, &[&side]);

        assert_eq!(joined.len(), 1);
        assert_eq!(joined[0].ts, Timestamp(10));
        assert_eq!(joined[0].sides[0], Some(vec![Scalar::i64(7)]));
    }

    #[test]
    fn columnar_trace_round_trips_f64_rows() {
        let rows = vec![
            (Timestamp(2), vec![Scalar::f64(10.0)]),
            (Timestamp(3), vec![Scalar::f64(20.0)]),
            (Timestamp(4), vec![Scalar::f64(30.0)]),
        ];
        let ct = ColumnarTrace::from_rows("equity", &[ScalarKind::F64], &rows);
        assert_eq!(ct.tap, "equity");
        assert_eq!(ct.kinds, vec!["F64".to_string()]);
        assert_eq!(ct.ts, vec![2, 3, 4]);
        assert_eq!(ct.columns, vec![vec![10.0, 20.0, 30.0]]);
        // to_rows is the inverse for f64 taps
        assert_eq!(ct.to_rows(), rows);
    }

    #[test]
    fn columnar_trace_empty_rows_yields_empty_columns_sized_to_kinds() {
        let ct = ColumnarTrace::from_rows("exposure", &[ScalarKind::F64], &[]);
        assert_eq!(ct.ts, Vec::<i64>::new());
        assert_eq!(ct.columns, vec![Vec::<f64>::new()]);
        assert!(ct.to_rows().is_empty());
    }

    #[test]
    fn columnar_trace_coerces_non_f64_kinds_to_f64() {
        let rows = vec![
            (Timestamp(1), vec![Scalar::i64(7), Scalar::bool(true)]),
            (Timestamp(2), vec![Scalar::i64(9), Scalar::bool(false)]),
        ];
        let ct = ColumnarTrace::from_rows("mix", &[ScalarKind::I64, ScalarKind::Bool], &rows);
        assert_eq!(ct.kinds, vec!["I64".to_string(), "Bool".to_string()]);
        assert_eq!(ct.columns, vec![vec![7.0, 9.0], vec![1.0, 0.0]]);
    }

    /// The fourth coercion arm: a Timestamp-kind column tags as "Timestamp" and
    /// its epoch-ns survives the f64 projection (`scalar_to_f64` reads `as_ts().0`),
    /// closing the kind for which `from_rows` would otherwise be untested.
    #[test]
    fn columnar_trace_coerces_timestamp_kind_to_epoch_f64() {
        let rows = vec![
            (Timestamp(1), vec![Scalar::ts(Timestamp(1_700_000_000_000_000_000))]),
            (Timestamp(2), vec![Scalar::ts(Timestamp(1_700_000_000_000_000_060))]),
        ];
        let ct = ColumnarTrace::from_rows("clock", &[ScalarKind::Timestamp], &rows);
        assert_eq!(ct.kinds, vec!["Timestamp".to_string()]);
        assert_eq!(
            ct.columns,
            vec![vec![1_700_000_000_000_000_000.0, 1_700_000_000_000_000_060.0]],
        );
    }

    /// A row wider than the declared kinds is a wiring bug (a sink's column count
    /// is fixed at bootstrap), surfaced as a named panic like `f64_field` — not a
    /// bare index-out-of-bounds.
    #[test]
    #[should_panic(expected = "row width 2 disagrees with kinds.len() 1")]
    fn from_rows_panics_on_row_wider_than_kinds() {
        let rows = vec![(Timestamp(1), vec![Scalar::f64(1.0), Scalar::f64(2.0)])];
        let _ = ColumnarTrace::from_rows("wide", &[ScalarKind::F64], &rows);
    }

    /// Symmetric to the wide case: a row narrower than the declared kinds is the
    /// same wiring-bug class and carries the same named panic.
    #[test]
    #[should_panic(expected = "row width 1 disagrees with kinds.len() 2")]
    fn from_rows_panics_on_row_narrower_than_kinds() {
        let rows = vec![(Timestamp(1), vec![Scalar::f64(1.0)])];
        let _ = ColumnarTrace::from_rows("narrow", &[ScalarKind::F64, ScalarKind::F64], &rows);
    }
}