test(0072): mean-reversion signal-composition e2e (characterization pin)

Hand-wired EWMA Bollinger-band fade subgraph (price -> Ema mean + dev -> Mul
square -> Ema var -> Sqrt sigma -> LinComb k-band -> Add/Sub bands -> Gt/Gt ->
Latch/Latch -> Sub bias). Pins the fade direction (above mean+k*sigma -> short,
below -> long), the latch hold, and flat-series -> no fade. Green on arrival
(composes only existing aura-std nodes; no new node), so it is a characterization
pin rather than a RED->GREEN unit.

refs #137

crates/aura-engine/tests/stage1_meanrev_e2e.rs

@@ -0,0 +1,88 @@
+//! Mean-reversion signal composition: price -> {Ema mean, Sub dev} -> Mul sq ->
+//! Ema var -> Sqrt sigma -> LinComb(k) band -> {Add upper, Sub lower} ->
+//! {Gt, Gt} -> {Latch, Latch} -> Sub = bias in {-1,0,+1}. Tests the FADE
+//! direction (price above mean+k*sigma -> short -1; below mean-k*sigma -> long
+//! +1), the latched hold, and that no fade fires on a flat series. Built
+//! straight from aura-std nodes (no CLI dependency); mirrors stage1_breakout_e2e.
+
+use aura_core::{Scalar, Timestamp};
+use aura_engine::{GraphBuilder, Harness, Source, VecSource};
+use aura_std::{Add, Ema, Gt, Latch, LinComb, Mul, Recorder, Sqrt, Sub};
+use std::sync::mpsc;
+
+// Feed `closes` into the Bollinger-fade signal subgraph (window n, band width k)
+// and tap the exposure (bias) through a Recorder. Returns the emitted bias values
+// in cycle order.
+fn run_meanrev_bias(closes: &[f64], n: i64, k: f64) -> Vec<f64> {
+    let (tx, rx) = mpsc::channel();
+    let mut g = GraphBuilder::new("meanrev_sig");
+    let mean = g.add(Ema::builder().bind("length", Scalar::i64(n)));
+    let dev = g.add(Sub::builder()); // price - mean
+    let sq = g.add(Mul::builder()); // dev * dev
+    let var = g.add(Ema::builder().bind("length", Scalar::i64(n))); // EWMA variance
+    let sigma = g.add(Sqrt::builder());
+    let band = g.add(LinComb::builder(1).bind("weights[0]", Scalar::f64(k))); // k*sigma
+    let upper = g.add(Add::builder()); // mean + k*sigma
+    let lower = g.add(Sub::builder()); // mean - k*sigma
+    let gt_hi = g.add(Gt::builder()); // price > upper
+    let gt_lo = g.add(Gt::builder()); // lower > price
+    let short_latch = g.add(Latch::builder());
+    let long_latch = g.add(Latch::builder());
+    let bias = g.add(Sub::builder()); // long_latch - short_latch
+    let rec = g.add(Recorder::builder(vec![aura_core::ScalarKind::F64], aura_core::Firing::Any, tx));
+    let price = g.source_role("price", aura_core::ScalarKind::F64);
+    g.feed(price, [mean.input("series"), dev.input("lhs"), gt_hi.input("a"), gt_lo.input("b")]);
+    g.connect(mean.output("value"), dev.input("rhs"));
+    g.connect(dev.output("value"), sq.input("lhs"));
+    g.connect(dev.output("value"), sq.input("rhs"));
+    g.connect(sq.output("value"), var.input("series"));
+    g.connect(var.output("value"), sigma.input("value"));
+    g.connect(sigma.output("value"), band.input("term[0]"));
+    g.connect(mean.output("value"), upper.input("lhs"));
+    g.connect(band.output("value"), upper.input("rhs"));
+    g.connect(mean.output("value"), lower.input("lhs"));
+    g.connect(band.output("value"), lower.input("rhs"));
+    g.connect(upper.output("value"), gt_hi.input("b"));
+    g.connect(lower.output("value"), gt_lo.input("a"));
+    g.connect(gt_hi.output("value"), short_latch.input("set"));
+    g.connect(gt_lo.output("value"), short_latch.input("reset"));
+    g.connect(gt_lo.output("value"), long_latch.input("set"));
+    g.connect(gt_hi.output("value"), long_latch.input("reset"));
+    g.connect(long_latch.output("value"), bias.input("lhs"));
+    g.connect(short_latch.output("value"), bias.input("rhs"));
+    g.connect(bias.output("value"), rec.input("col[0]"));
+    let flat = g.build().expect("meanrev signal wiring resolves").compile_with_params(&[]).expect("compiles");
+    let mut h = Harness::bootstrap(flat).expect("bootstraps");
+    let prices: Vec<(Timestamp, Scalar)> =
+        closes.iter().enumerate().map(|(i, &c)| (Timestamp(i as i64), Scalar::f64(c))).collect();
+    let src: Vec<Box<dyn Source>> = vec![Box::new(VecSource::new(prices))];
+    h.run(src);
+    rx.try_iter().map(|(_, row): (Timestamp, Vec<Scalar>)| row[0].as_f64()).collect()
+}
+
+#[test]
+fn meanrev_fades_against_the_move_and_holds_the_latch() {
+    // k = 0 makes the band collapse to the mean (upper = lower = mean), so the
+    // fade fires on ANY deviation from the lagging EWMA mean -- isolating the
+    // direction + latch from the sigma threshold (the k>0 band is exercised by
+    // the real-data CLI screen). n = 3 -> alpha = 0.5, so the mean lags the level
+    // clearly. A long calm, then a sustained jump UP, then a sustained drop DOWN.
+    //   calm (price == mean -> no break) | up (price > mean -> SHORT) | down (price < mean -> LONG)
+    let closes = [100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 130.0, 130.0, 130.0, 70.0, 70.0, 70.0];
+    let bias = run_meanrev_bias(&closes, 3, 0.0);
+    assert!(!bias.is_empty(), "the signal must emit once warmed up");
+    assert_eq!(*bias.first().unwrap(), 0.0, "calm bars (price == mean) must not fade; got {bias:?}");
+    let first_short = bias.iter().position(|&b| b == -1.0).expect("an up-move must fade SHORT (-1)");
+    let first_long = bias.iter().position(|&b| b == 1.0).expect("a down-move must fade LONG (+1)");
+    assert!(first_short < first_long, "short (up-fade) must precede long (down-fade); got {bias:?}");
+    assert_eq!(*bias.last().unwrap(), 1.0, "the down-fade long must hold to the end; got {bias:?}");
+}
+
+#[test]
+fn meanrev_flat_series_never_fades() {
+    // A perfectly flat series: dev == 0, sigma == 0, band == mean, so price is
+    // never strictly beyond the band -> no break ever -> bias pinned at 0.
+    let bias = run_meanrev_bias(&[100.0; 10], 3, 2.0);
+    assert!(!bias.is_empty(), "the signal must emit once warmed up");
+    assert!(bias.iter().all(|&b| b == 0.0), "a flat series must never fade; got {bias:?}");
+}