Aura/crates/aura-composites/tests/risk_executor.rs

//! The `RiskExecutor` composite (Stage-1, #128): a per-symbol risk-based executor over a
//! bias stream — `stop-rule -> Sizer -> position-management`, exposing the dense R-record.
//! Proves the composite bootstraps, runs end-to-end, and folds to the SAME R-outcomes as
//! the hand-wired iter-1 chain, and that R is invariant under the Sizer's `risk_budget`
//! (Stage-1 feed-forward). The Veto is a DOCUMENTED SEAM, not a runtime node (a
//! pass-through identity is exactly what C19/C23 DCE deletes), so it appears nowhere here.
use aura_core::{
    Cell, Ctx, FieldSpec, Firing, Node, NodeSchema, PortSpec, PrimitiveBuilder, Scalar,
    ScalarKind, Timestamp,
};
use aura_composites::{risk_executor, StopRule};
use aura_engine::{summarize_r, GraphBuilder, RunMetrics, VecSource};
use aura_std::{Recorder, PM_FIELD_NAMES, PM_RECORD_KINDS};
use std::sync::mpsc::channel;

// The dense-record columns this fixture reads, named in lockstep with the sibling
// `stage1_r_e2e.rs` (which names `REALIZED_R = 1` the same way) so the cross-crate
// `PM_FIELD_NAMES` layout is never referenced by a bare literal.
const REALIZED_R: usize = 1;
const SIZE: usize = 10;

/// An always-long strategy stand-in: emits a constant `+1` bias once price is present. The
/// strategy is upstream of the RiskExecutor; this is the minimal in-graph producer so the
/// whole chain runs off the single price source (a second bias *source* would k-way-merge
/// into separate cycles and mark stale prices — see harness.rs C4 tie-breaking).
struct ConstLongBias {
    out: [Cell; 1],
}
impl ConstLongBias {
    fn builder() -> PrimitiveBuilder {
        PrimitiveBuilder::new(
            "ConstLongBias",
            NodeSchema {
                inputs: vec![PortSpec { kind: ScalarKind::F64, firing: Firing::Any, name: "price".into() }],
                output: vec![FieldSpec { name: "bias".into(), kind: ScalarKind::F64 }],
                params: vec![],
            },
            |_| Box::new(ConstLongBias { out: [Cell::from_f64(0.0)] }),
        )
    }
}
impl Node for ConstLongBias {
    fn lookbacks(&self) -> Vec<usize> {
        vec![1]
    }
    fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> {
        if ctx.f64_in(0).is_empty() {
            return None;
        }
        self.out[0] = Cell::from_f64(1.0);
        Some(&self.out)
    }
    fn label(&self) -> String {
        "ConstLongBias".into()
    }
}

/// Bootstrap a harness: one price source -> ConstLongBias (the strategy) + RiskExecutor;
/// the RiskExecutor's dense R-record into a Recorder. Returns the drained ledger.
fn run_executor(prices: &[f64], stop_distance: f64, risk_budget: f64) -> Vec<(Timestamp, Vec<Scalar>)> {
    let (tx, rx) = channel();
    let mut g = GraphBuilder::new("risk_harness");
    let price = g.source_role("price", ScalarKind::F64);
    let strat = g.add(ConstLongBias::builder());
    let exec = g.add(risk_executor(StopRule::Fixed(stop_distance), risk_budget));
    let rec = g.add(Recorder::builder(PM_RECORD_KINDS.to_vec(), Firing::Any, tx));
    g.feed(price, [strat.input("price"), exec.input("price")]);
    g.connect(strat.output("bias"), exec.input("bias"));
    for (i, field) in PM_FIELD_NAMES.iter().enumerate() {
        // `input` takes a `&'static str` (names resolve at the authoring boundary, C23);
        // leak the per-column port name so the runtime-built `col[i]` satisfies that bound.
        let col: &'static str = format!("col[{i}]").leak();
        g.connect(exec.output(field), rec.input(col));
    }
    let mut h = g
        .build()
        .expect("risk_harness wires")
        .bootstrap_with_params(vec![])
        .expect("bootstraps");
    let stream: Vec<(Timestamp, Scalar)> = prices
        .iter()
        .enumerate()
        .map(|(i, &p)| (Timestamp(i as i64), Scalar::f64(p)))
        .collect();
    h.run(vec![Box::new(VecSource::new(stream))]);
    rx.try_iter().collect()
}

/// Property: the composite bootstraps, runs, and folds to the documented hand value. A
/// constant long over a monotonically rising price never stops or flips, so the position
/// is open at window end: entry @100 latched on FixedStop(10), last mark @105 -> window-end
/// R = (105-100)/10 = +0.5, the only trade -> expectancy 0.5 (the bootstrapped-composite
/// twin of the iter-1 hand-wired `open_at_window_end_is_folded_into_expectancy_not_dropped`).
#[test]
fn risk_executor_bootstraps_and_folds_to_expected_rmetric() {
    let ledger = run_executor(&[100.0, 102.0, 105.0], 10.0, 1.0);
    let m = summarize_r(&ledger, 0.0);
    assert_eq!(m.n_open_at_end, 1, "the open position must be counted");
    assert_eq!(m.n_trades, 1);
    assert!((m.expectancy_r - 0.5).abs() < 1e-9, "window-end R = (105-100)/10; got {}", m.expectancy_r);
}

/// Property: R is invariant under the Sizer's `risk_budget` (Stage-1 feed-forward). The
/// same price path at two budgets yields a bit-identical realised-R ledger, while the
/// `size` column scales with the budget — proving size flows through the Sizer into PM yet
/// never touches R.
#[test]
fn risk_executor_r_invariant_under_risk_budget() {
    let path = [100.0, 104.0, 108.0, 110.0, 102.0, 96.0, 94.0];
    let a = run_executor(&path, 5.0, 1.0);
    let b = run_executor(&path, 5.0, 8.0);
    assert_eq!(a.len(), b.len());
    assert!(!a.is_empty());
    // index realized_r and size by the producer's dense-record layout (named consts above).
    let realized =
        |rows: &[(Timestamp, Vec<Scalar>)]| rows.iter().map(|(_, r)| r[REALIZED_R].as_f64()).collect::<Vec<_>>();
    let size =
        |rows: &[(Timestamp, Vec<Scalar>)]| rows.iter().map(|(_, r)| r[SIZE].as_f64()).collect::<Vec<_>>();
    assert_eq!(realized(&a), realized(&b), "realized_r must be invariant under risk_budget");
    // size scaled 8x: at least one cycle has a nonzero size that is exactly 8x a's (same
    // stop distance per cycle, budget 1 -> 8).
    let (sa, sb) = (size(&a), size(&b));
    assert!(
        sa.iter().zip(&sb).any(|(x, y)| *x > 0.0 && (*y - 8.0 * *x).abs() < 1e-9),
        "risk_budget must scale size 8x: a={sa:?} b={sb:?}"
    );
}

/// Property: **a real folded `RMetrics` survives the `RunMetrics.r` serde round-trip
/// byte-for-byte (the Stage-1 on-disk back-compat contract), driven from an actual run —
/// not a hand-built literal.** A bootstrapped RiskExecutor run is folded by `summarize_r`
/// and the result is attached as `RunMetrics.r = Some(..)`; serializing then
/// deserializing must reproduce an equal value, and the `r` key must be present. The
/// `report.rs` unit test asserts this on a hand-written `RMetrics`; here the value is
/// whatever the live fold produced, so a future `RMetrics` field that the fold sets but
/// serde forgets to thread would round-trip-diverge here (the literal test cannot see it).
/// The sibling pip-only-`None` path (omitted from JSON) is the inverse, covered in report.rs.
#[test]
fn folded_rmetrics_survives_runmetrics_serde_round_trip() {
    let ledger = run_executor(&[100.0, 104.0, 108.0, 110.0, 102.0, 96.0, 94.0], 5.0, 1.0);
    let folded = summarize_r(&ledger, 0.0);
    assert!(folded.n_trades >= 1, "the run must produce at least one trade to fold");
    let m = RunMetrics { total_pips: 0.0, max_drawdown: 0.0, bias_sign_flips: 0, r: Some(folded) };
    let json = serde_json::to_string(&m).expect("serialize a run's RunMetrics with an r block");
    assert!(json.contains("\"r\":{"), "the folded r block must be present in the JSON: {json}");
    let back: RunMetrics = serde_json::from_str(&json).expect("deserialize round-trips");
    assert_eq!(back, m, "a live-folded RMetrics must round-trip byte-for-byte");
}

/// Property: the RiskExecutor embeds the `vol_stop` composite (the new `StopRule::Vol`
/// arm) and folds to a finite RMetric — the volatility-defined default the `stage1-r`
/// harness uses. A short k·σ stop over a rising-then-falling path opens a trade and
/// (stop or window-end) closes at least one, so the fold is non-empty and sane.
#[test]
fn risk_executor_vol_stop_arm_bootstraps_and_folds() {
    let (tx, rx) = channel();
    let mut g = GraphBuilder::new("vol_harness");
    let price = g.source_role("price", ScalarKind::F64);
    let strat = g.add(ConstLongBias::builder());
    let exec = g.add(risk_executor(StopRule::Vol { length: 3, k: 2.0 }, 1.0));
    let rec = g.add(Recorder::builder(PM_RECORD_KINDS.to_vec(), Firing::Any, tx));
    g.feed(price, [strat.input("price"), exec.input("price")]);
    g.connect(strat.output("bias"), exec.input("bias"));
    for (i, field) in PM_FIELD_NAMES.iter().enumerate() {
        let col: &'static str = format!("col[{i}]").leak();
        g.connect(exec.output(field), rec.input(col));
    }
    let mut h = g
        .build()
        .expect("vol_harness wires")
        .bootstrap_with_params(vec![])
        .expect("bootstraps");
    let path = [100.0, 101.0, 100.0, 101.0, 103.0, 106.0, 104.0, 99.0, 95.0, 96.0];
    let stream: Vec<(Timestamp, Scalar)> =
        path.iter().enumerate().map(|(i, &p)| (Timestamp(i as i64), Scalar::f64(p))).collect();
    h.run(vec![Box::new(VecSource::new(stream))]);
    let ledger: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
    let m = summarize_r(&ledger, 0.0);
    assert!(m.n_trades >= 1, "the vol-stop executor must fold at least one trade");
    assert!(m.sqn.is_finite(), "SQN must be finite, got {}", m.sqn);
}