//! Breakout signal composition: close -> Delay(1) -> {RollingMax,RollingMin}(N) ->
//! {Gt,Gt} -> {Latch,Latch} -> Sub = bias in {-1,0,+1}. Tests the C2 causality guard
//! (the channel excludes the current bar via Delay(1)) and the held ±1 direction,
//! built straight from aura-std nodes (no CLI dependency).

use aura_core::{Scalar, Timestamp};
use aura_engine::{GraphBuilder, Harness, Source, VecSource};
use aura_std::{Delay, Gt, Latch, Recorder, RollingMax, RollingMin, Sub};
use std::sync::mpsc;

// A minimal single-f64 source role "price" fed into the breakout signal subgraph,
// tapping the Sub (bias) output through a Recorder. channel N = 3.
fn run_breakout_bias(closes: &[f64]) -> Vec<(Timestamp, f64)> {
    let (tx, rx) = mpsc::channel();
    let mut g = GraphBuilder::new("breakout_sig");
    let delay = g.add(Delay::builder().bind("lag", Scalar::i64(1)));
    let mx = g.add(RollingMax::builder().bind("length", Scalar::i64(3)));
    let mn = g.add(RollingMin::builder().bind("length", Scalar::i64(3)));
    let gt_up = g.add(Gt::builder());
    let gt_down = g.add(Gt::builder());
    let up_latch = g.add(Latch::builder());
    let down_latch = g.add(Latch::builder());
    let bias = g.add(Sub::builder());
    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, [delay.input("series"), gt_up.input("a"), gt_down.input("b")]);
    g.connect(delay.output("value"), mx.input("series"));
    g.connect(delay.output("value"), mn.input("series"));
    g.connect(mx.output("value"), gt_up.input("b")); // close > max(prior N)
    g.connect(mn.output("value"), gt_down.input("a")); // min(prior N) > close
    g.connect(gt_up.output("value"), up_latch.input("set"));
    g.connect(gt_down.output("value"), up_latch.input("reset"));
    g.connect(gt_down.output("value"), down_latch.input("set"));
    g.connect(gt_up.output("value"), down_latch.input("reset"));
    g.connect(up_latch.output("value"), bias.input("lhs"));
    g.connect(down_latch.output("value"), bias.input("rhs"));
    g.connect(bias.output("value"), rec.input("col[0]"));
    let flat = g.build().expect("breakout 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(|(ts, row): (Timestamp, Vec<Scalar>)| (ts, row[0].as_f64())).collect()
}

#[test]
fn breakout_channel_excludes_the_current_bar_c2() {
    // C2 (no look-ahead), contrastive form: on a strictly monotonically rising series
    // every warmed bar is a fresh all-time high. The channel is max(close[t-N..t-1])
    // (Delay(1) excludes the current bar), so close[t] > that prior max on EVERY warmed
    // bar -> a continuous UP-break -> bias pinned at +1.
    //
    // If the channel WRONGLY included the current bar (the look-ahead bug), the test
    // would be `close[t] > max(close[t-N..t])`, and a value is never strictly greater
    // than the max of a window that contains itself -> the up-break could NEVER fire ->
    // bias would never reach +1. So a sustained +1 here is only reachable when the
    // current bar is structurally excluded: this asserts the C2 guard directly.
    let closes = [100.0, 101.0, 102.0, 103.0, 104.0, 105.0];
    let vals: Vec<f64> = run_breakout_bias(&closes).iter().map(|(_, v)| *v).collect();
    assert!(
        !vals.is_empty() && vals.iter().all(|&v| v == 1.0),
        "a strictly rising series must up-break every warmed bar (+1); got {vals:?}"
    );
}

#[test]
fn breakout_is_causal_and_holds_pm1_direction() {
    // closes: an up-break at bar 3, quiet hold, a down-break at bar 6.
    //   idx:    0    1    2    3    4    5   6    7
    let closes = [100.0, 101.0, 99.0, 102.0, 100.0, 100.0, 95.0, 100.0];
    // bar3: max(close[0..2])=101, close[3]=102 > 101 -> UP-break. (C2: a window INCLUDING
    //       bar3 would be max=102, and 102>102 strict is false -> no break; the break
    //       firing here proves the current bar is excluded.)
    // bars4,5: no break -> bias HOLDS +1.
    // bar6: min(close[3..5])=100, close[6]=95 < 100 -> DOWN-break -> bias flips to -1.
    // bar7: no break -> HOLDS -1.
    let got = run_breakout_bias(&closes);
    let vals: Vec<f64> = got.iter().map(|(_, v)| *v).collect();
    // bias is emitted only once the channel warms up (Delay(1)+RollingMax(3) -> first at bar 3).
    assert_eq!(vals, vec![1.0, 1.0, 1.0, -1.0, -1.0], "expected +1 held then -1 held; got {vals:?}");
}