Aura/crates/aura-engine/tests/r_meanrev_e2e.rs

//! 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 r_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:?}");
}