Aura/crates/aura-ingest/examples/shared/breakout_real.rs

//! Shared wiring for the real-data GER40 15m session-breakout demonstration.
//!
//! This is the *single* definition of the breakout harness used by BOTH the
//! runnable example (`examples/ger40_breakout_real.rs`) and the gated
//! determinism test (`tests/ger40_breakout_real.rs`), so the 11-node FlatGraph
//! is wired exactly once and never drifts between the two. It is included into
//! each consumer via `#[path = "..."] mod`, not compiled as its own target.
//!
//! The DAG is the VERBATIM breakout graph proven synthetically in
//! `aura-engine/tests/ger40_breakout.rs` — only the sources change: real
//! `M1FieldSource`s over a data-server window replace the synthetic
//! `VecSource`s. Source declaration order (open, high, low, close) is the C4
//! merge tie-break order Resample's `Barrier(0)` group depends on, so it is
//! load-bearing and preserved.
//!
//! ```text
//! open,high,low,close (4 real M1 sources, FIXED order)
//!   -> Resample(15)  ⇒ {open15,high15,low15,close15}
//!   close15 -> Gt.a
//!   high15  -> Delay(1) -> Gt.b           (prevHigh15)
//!   Gt -> breakout:bool                   (close15 > prevHigh15, STRICT)
//!   close15 -> Session(9,0,Berlin,15) ⇒ bars_since_open:i64
//!       bars_since_open -> EqConst(3) -> isBar3
//!       bars_since_open -> EqConst(5) -> isBar5Close
//!   And(breakout, isBar3) -> entry
//!   Latch(set=entry, reset=isBar5Close) -> held:f64
//!   held -> SimBroker(exposure=held, price=close15) ⇒ equity   [pip_size=1.0]
//!   equity -> Recorder(A);  held -> Recorder(B)
//!   bars_since_open -> Recorder(C);  breakout -> Recorder(D)
//! ```
//!
//! The synthetic test taps two channels (equity, held). The real demonstration
//! taps two more — `bars_since_open` (i64) and `breakout` (bool) — so the
//! entry/exit trace can name *why* each session did or did not fire, without
//! reaching into engine internals (it reads only what sinks record, C8/C18).

#![allow(dead_code)] // each consumer (example / test) uses a subset of this API

use std::sync::mpsc;
use std::sync::Arc;

use aura_core::{NodeSchema, Scalar, ScalarKind, Timestamp};
use aura_engine::{
    summarize, Composite, Edge, FlatGraph, GraphBuilder, Harness, RunManifest, RunReport, Source,
    SourceSpec, Target,
};
use aura_std::{And, Delay, EqConst, Gt, Latch, Recorder, Resample, Session, SimBroker};
use chrono::TimeZone;
use chrono_tz::Europe::Berlin;
use data_server::DataServer;

use aura_ingest::{M1Field, M1FieldSource};

// ---------------------------------------------------------------------------
// Node indices in the FlatGraph (one fixed layout, the same nodes 0..=8 as
// `aura-engine/tests/ger40_breakout.rs`, plus two extra trace taps 11/12).
// ---------------------------------------------------------------------------
pub const RESAMPLE: usize = 0;
pub const DELAY: usize = 1;
pub const GT: usize = 2;
pub const SESSION: usize = 3;
pub const EQ3: usize = 4;
pub const EQ5: usize = 5;
pub const AND: usize = 6;
pub const LATCH: usize = 7;
pub const BROKER: usize = 8;
pub const REC_EQUITY: usize = 9; // tap A — SimBroker equity (f64, pips)
pub const REC_HELD: usize = 10; // tap B — Latch exposure (f64, 0.0/1.0)
pub const REC_BARS: usize = 11; // tap C — Session bars_since_open (i64)
pub const REC_BREAKOUT: usize = 12; // tap D — Gt breakout flag (bool)

/// The instrument. GER40 is an index, so pips are index points: `pip_size = 1.0`.
pub const SYMBOL: &str = "GER40";

/// SimBroker pip size for GER40. An index quotes in points, so one pip is one
/// point — `1.0`, exactly as the synthetic capstone test uses.
pub const PIP_SIZE: f64 = 1.0;

/// Session config: Frankfurt cash open 09:00 Europe/Berlin, 15m bars. The same
/// `Session::new(9, 0, Europe::Berlin, 15)` the synthetic test pins.
pub const SESSION_HOUR: u32 = 9;
pub const SESSION_MINUTE: u32 = 0;
pub const BAR_MINUTES: i64 = 15;

/// The four recorder channels a built harness drains after a run.
pub struct Taps {
    pub equity: mpsc::Receiver<(Timestamp, Vec<Scalar>)>,
    pub held: mpsc::Receiver<(Timestamp, Vec<Scalar>)>,
    pub bars: mpsc::Receiver<(Timestamp, Vec<Scalar>)>,
    pub breakout: mpsc::Receiver<(Timestamp, Vec<Scalar>)>,
}

/// Hand-wire the breakout DAG into a frozen [`Harness`] with four recording
/// taps. A fresh build per call (fresh nodes + channels) keeps two runs
/// disjoint for the determinism assertion (C1). The wiring of nodes 0..=10 is
/// VERBATIM from `aura-engine/tests/ger40_breakout.rs`; nodes 11/12 add the two
/// trace taps.
pub fn build_harness() -> (Harness, Taps) {
    use aura_core::Firing;

    let (tx_equity, rx_equity) = mpsc::channel();
    let (tx_held, rx_held) = mpsc::channel();
    let (tx_bars, rx_bars) = mpsc::channel();
    let (tx_breakout, rx_breakout) = mpsc::channel();

    // Each node's declared signature (kind/firing per port), parallel to `nodes`.
    let signatures: Vec<NodeSchema> = vec![
        Resample::builder().schema().clone(), // 0
        Delay::builder().schema().clone(),     // 1
        Gt::builder().schema().clone(),        // 2
        Session::builder(SESSION_HOUR, SESSION_MINUTE, Berlin, BAR_MINUTES)
            .schema()
            .clone(), // 3
        EqConst::builder().schema().clone(), // 4
        EqConst::builder().schema().clone(), // 5
        And::builder().schema().clone(),     // 6
        Latch::builder().schema().clone(),   // 7
        SimBroker::builder(PIP_SIZE).schema().clone(), // 8
        Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_equity.clone())
            .schema()
            .clone(), // 9
        Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_held.clone())
            .schema()
            .clone(), // 10
        Recorder::builder(vec![ScalarKind::I64], Firing::Any, tx_bars.clone())
            .schema()
            .clone(), // 11
        Recorder::builder(vec![ScalarKind::Bool], Firing::Any, tx_breakout.clone())
            .schema()
            .clone(), // 12
    ];

    let nodes: Vec<Box<dyn aura_core::Node>> = vec![
        Box::new(Resample::new(15)),                                            // 0
        Box::new(Delay::new(1)),                                                // 1
        Box::new(Gt::new()),                                                    // 2
        Box::new(Session::new(SESSION_HOUR, SESSION_MINUTE, Berlin, BAR_MINUTES)), // 3
        Box::new(EqConst::new(3)),                                              // 4
        Box::new(EqConst::new(5)),                                              // 5
        Box::new(And::new()),                                                   // 6
        Box::new(Latch::new()),                                                 // 7
        Box::new(SimBroker::new(PIP_SIZE)),                                     // 8
        Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_equity)),    // 9
        Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_held)),      // 10
        Box::new(Recorder::new(&[ScalarKind::I64], Firing::Any, tx_bars)),      // 11
        Box::new(Recorder::new(&[ScalarKind::Bool], Firing::Any, tx_breakout)), // 12
    ];

    // Resample output fields: 0=open15, 1=high15, 2=low15, 3=close15.
    const F_HIGH15: usize = 1;
    const F_CLOSE15: usize = 3;

    let edges = vec![
        // close15 -> Gt.a (slot 0)
        Edge { from: RESAMPLE, to: GT, slot: 0, from_field: F_CLOSE15 },
        // high15 -> Delay.series (slot 0) -> Gt.b (slot 1)
        Edge { from: RESAMPLE, to: DELAY, slot: 0, from_field: F_HIGH15 },
        Edge { from: DELAY, to: GT, slot: 1, from_field: 0 },
        // close15 -> Session.trigger (value ignored)
        Edge { from: RESAMPLE, to: SESSION, slot: 0, from_field: F_CLOSE15 },
        // bars_since_open -> EqConst(3) / EqConst(5)
        Edge { from: SESSION, to: EQ3, slot: 0, from_field: 0 },
        Edge { from: SESSION, to: EQ5, slot: 0, from_field: 0 },
        // And(breakout, isBar3) -> entry
        Edge { from: GT, to: AND, slot: 0, from_field: 0 },
        Edge { from: EQ3, to: AND, slot: 1, from_field: 0 },
        // Latch(set=entry, reset=isBar5Close)
        Edge { from: AND, to: LATCH, slot: 0, from_field: 0 },
        Edge { from: EQ5, to: LATCH, slot: 1, from_field: 0 },
        // held -> SimBroker.exposure (slot 0); close15 -> SimBroker.price (slot 1)
        Edge { from: LATCH, to: BROKER, slot: 0, from_field: 0 },
        Edge { from: RESAMPLE, to: BROKER, slot: 1, from_field: F_CLOSE15 },
        // equity -> tap A; held -> tap B
        Edge { from: BROKER, to: REC_EQUITY, slot: 0, from_field: 0 },
        Edge { from: LATCH, to: REC_HELD, slot: 0, from_field: 0 },
        // trace taps: bars_since_open -> tap C; breakout -> tap D
        Edge { from: SESSION, to: REC_BARS, slot: 0, from_field: 0 },
        Edge { from: GT, to: REC_BREAKOUT, slot: 0, from_field: 0 },
    ];

    // Four f64 sources, one per OHLC field, each feeding one Resample slot. The
    // declaration order open(0), high(1), low(2), close(3) is the merge tie-break
    // order for the Barrier(0) group — identical to the synthetic capstone.
    let sources = (0..4)
        .map(|slot| SourceSpec {
            kind: ScalarKind::F64,
            targets: vec![Target { node: RESAMPLE, slot }],
        })
        .collect();

    let h = Harness::bootstrap(FlatGraph { nodes, signatures, sources, edges })
        .expect("valid breakout DAG");
    (h, Taps { equity: rx_equity, held: rx_held, bars: rx_bars, breakout: rx_breakout })
}

/// Author the GER40 session-breakout as a World-consumable [`Composite`]
/// blueprint — the canonical shippable form of the strategy, of which
/// [`build_harness`]'s hand-wired `FlatGraph` is the compiled substrate (C11/C23).
///
/// This is the SAME 13-node breakout `build_harness` wires by raw index, authored
/// here fluently by name via [`GraphBuilder`] (canonical template
/// `aura-engine/src/test_fixtures.rs`). Being a `Composite` it exposes
/// `param_space()` / the by-name binder / `bootstrap_with_cells`, so `sweep` /
/// `walk_forward` / compare consume it with NO re-authoring (the #94 friction).
///
/// The structural decisions are baked here, not left to the caller:
/// - `bar_period_minutes` binds BOTH the period-bearing nodes at construction —
///   `Resample`'s `period_minutes` param AND `Session`'s baked period — so the two
///   are LOCKED EQUAL and a sweep cannot desync them to 0.0 pips (#96 / C34: a
///   15m and a 30m breakout are *different strategies*, not one tuned).
/// - `delay.lag` is bound out to its structural constant `1` (the "previous bar's
///   high" register is not a tuning knob — the exact C34 deform-vs-tune case).
/// - The two `EqConst` nodes are `.named("entry_bar")` / `.named("exit_bar")` with
///   their `target` param LEFT UNBOUND — so `param_space()` is EXACTLY
///   `{ entry_bar.target, exit_bar.target }`, the two genuine tuning knobs.
///
/// The four OHLC fields are root **source roles** (open, high, low, close) in the
/// fixed C4 merge order, fanning into Resample's four `Barrier(0)` slots. The
/// SimBroker equity is exposed as an output field (`equity`, for the later World
/// demos) AND tapped into a recorder, alongside the held/bars/breakout taps — so
/// the standalone example/test drains the same trace `build_harness` does today.
///
/// Returns the blueprint plus the four recording [`Taps`]. A fresh build per call
/// (fresh nodes + channels) keeps two bootstraps disjoint for the C1 determinism
/// assertion, exactly as `build_harness` does.
pub fn ger40_breakout_blueprint(
    bar_period_minutes: i64,
    open_hour: u32,
    open_minute: u32,
    tz: chrono_tz::Tz,
) -> (Composite, Taps) {
    use aura_core::Firing;

    let (tx_equity, rx_equity) = mpsc::channel();
    let (tx_held, rx_held) = mpsc::channel();
    let (tx_bars, rx_bars) = mpsc::channel();
    let (tx_breakout, rx_breakout) = mpsc::channel();

    let mut g = GraphBuilder::new("ger40_breakout");

    // Strategy nodes. The two period-bearing nodes are bound EQUAL by construction:
    // Resample's `period_minutes` param and Session's baked period both take
    // `bar_period_minutes`, so no sweep can desync them. `delay.lag` is bound out.
    let resample = g.add(
        Resample::builder().bind("period_minutes", Scalar::i64(bar_period_minutes)),
    );
    let delay = g.add(Delay::builder().bind("lag", Scalar::i64(1)));
    let gt = g.add(Gt::builder());
    let session = g.add(Session::builder(open_hour, open_minute, tz, bar_period_minutes));
    // The ONLY two params: the EqConst targets, named so the space reads
    // `entry_bar.target` / `exit_bar.target`. Their `target` is left UNBOUND.
    let entry_bar = g.add(EqConst::builder().named("entry_bar"));
    let exit_bar = g.add(EqConst::builder().named("exit_bar"));
    let and = g.add(And::builder());
    let latch = g.add(Latch::builder());
    let broker = g.add(SimBroker::builder(PIP_SIZE));

    // Recording sinks: equity (A), held (B), bars_since_open (C), breakout (D).
    let rec_equity = g.add(Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_equity));
    let rec_held = g.add(Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_held));
    let rec_bars = g.add(Recorder::builder(vec![ScalarKind::I64], Firing::Any, tx_bars));
    let rec_breakout =
        g.add(Recorder::builder(vec![ScalarKind::Bool], Firing::Any, tx_breakout));

    // Four OHLC source roles (open, high, low, close) in the fixed C4 merge order,
    // each fanning into the matching Resample Barrier(0) slot (port names match
    // the field names: "open"/"high"/"low"/"close").
    let open = g.source_role("open", ScalarKind::F64);
    let high = g.source_role("high", ScalarKind::F64);
    let low = g.source_role("low", ScalarKind::F64);
    let close = g.source_role("close", ScalarKind::F64);
    g.feed(open, [resample.input("open")]);
    g.feed(high, [resample.input("high")]);
    g.feed(low, [resample.input("low")]);
    g.feed(close, [resample.input("close")]);

    // close15 -> Gt.a; high15 -> Delay -> Gt.b (prevHigh15, STRICT comparator).
    g.connect(resample.output("close"), gt.input("a"));
    g.connect(resample.output("high"), delay.input("series"));
    g.connect(delay.output("value"), gt.input("b"));
    // close15 -> Session.trigger (value ignored; fires once per completed bar).
    g.connect(resample.output("close"), session.input("trigger"));
    // bars_since_open -> EqConst(entry) / EqConst(exit).
    g.connect(session.output("bars_since_open"), entry_bar.input("value"));
    g.connect(session.output("bars_since_open"), exit_bar.input("value"));
    // And(breakout, isEntryBar) -> entry.
    g.connect(gt.output("value"), and.input("a"));
    g.connect(entry_bar.output("value"), and.input("b"));
    // Latch(set=entry, reset=isExitBar) -> held.
    g.connect(and.output("value"), latch.input("set"));
    g.connect(exit_bar.output("value"), latch.input("reset"));
    // held -> SimBroker.exposure; close15 -> SimBroker.price.
    g.connect(latch.output("value"), broker.input("exposure"));
    g.connect(resample.output("close"), broker.input("price"));
    // equity -> tap A; held -> tap B; bars -> tap C; breakout -> tap D.
    g.connect(broker.output("equity"), rec_equity.input("col[0]"));
    g.connect(latch.output("value"), rec_held.input("col[0]"));
    g.connect(session.output("bars_since_open"), rec_bars.input("col[0]"));
    g.connect(gt.output("value"), rec_breakout.input("col[0]"));

    // Expose the SimBroker equity at the boundary (for the later World demos);
    // the recorder taps still drain the standalone trace.
    g.expose(broker.output("equity"), "equity");

    let bp = g.build().expect("breakout blueprint resolves by name");
    (bp, Taps { equity: rx_equity, held: rx_held, bars: rx_bars, breakout: rx_breakout })
}

/// Open the four real OHLC `M1FieldSource`s over `[from_ms, to_ms]` (inclusive
/// Unix-ms) in the FIXED order open, high, low, close — the C4 merge tie-break
/// order Resample's `Barrier(0)` group expects. Returns `None` if any field has
/// no file overlapping the window (so the caller can skip cleanly).
///
/// This is exactly the multi-field OHLC open that `M1FieldSource` makes the
/// caller spell out four times by hand — see the friction note in the report.
pub fn open_ohlc_sources(
    server: &Arc<DataServer>,
    from_ms: i64,
    to_ms: i64,
) -> Option<Vec<Box<dyn Source>>> {
    let open = M1FieldSource::open(server, SYMBOL, Some(from_ms), Some(to_ms), M1Field::Open)?;
    let high = M1FieldSource::open(server, SYMBOL, Some(from_ms), Some(to_ms), M1Field::High)?;
    let low = M1FieldSource::open(server, SYMBOL, Some(from_ms), Some(to_ms), M1Field::Low)?;
    let close = M1FieldSource::open(server, SYMBOL, Some(from_ms), Some(to_ms), M1Field::Close)?;
    let srcs: Vec<Box<dyn Source>> =
        vec![Box::new(open), Box::new(high), Box::new(low), Box::new(close)];
    Some(srcs)
}

/// Build the inclusive Unix-ms window for the whole calendar month `(year,
/// month)` in UTC: `[first instant of the 1st, last ms of the last day]`. UTC
/// keeps the window boundary trivial to reason about; the Session node still
/// indexes bars in Berlin wall-clock inside the graph.
pub fn utc_month_window_ms(year: i32, month: u32) -> (i64, i64) {
    let from = chrono::Utc
        .with_ymd_and_hms(year, month, 1, 0, 0, 0)
        .unwrap()
        .timestamp_millis();
    // first instant of the next month, minus one ms => last ms of this month.
    let (ny, nm) = if month == 12 { (year + 1, 1) } else { (year, month + 1) };
    let next = chrono::Utc
        .with_ymd_and_hms(ny, nm, 1, 0, 0, 0)
        .unwrap()
        .timestamp_millis();
    (from, next - 1)
}

/// One bar's recorded trace row, fused across the four taps by **timestamp**.
///
/// The four sinks do NOT all fire on the same cardinality: `equity` and `held`
/// fire once per completed Resample bar (downstream of Latch/SimBroker), but the
/// `breakout` (Gt) tap is one bar shorter — `Gt` depends on `Delay(1)` of
/// high15, which is cold on the very first bar and emits nothing — and the
/// `bars_since_open` (Session) tap filters bars that precede a session open. So
/// they cannot be fused by positional index; they are joined on the recorded
/// timestamp. `equity`/`held` are the spine (one row per bar); the other two are
/// looked up by ts, defaulting where they did not fire this bar.
pub struct BarTrace {
    pub ts: Timestamp,
    pub equity: f64,
    pub held: f64,
    /// Session bar index, or `-1` where Session did not emit this bar (outside a
    /// tracked session).
    pub bars_since_open: i64,
    /// The strict-breakout flag, or `false` where Gt did not emit (the cold
    /// first bar, before Delay(1) is warm).
    pub breakout: bool,
}

/// Drain the four taps into a per-bar trace, joining on the recorded timestamp
/// (the sinks fire on different cardinalities — see [`BarTrace`]). Reads only
/// recorded sink output (C8/C18) — never engine internals.
pub fn drain_trace(taps: &Taps) -> Vec<BarTrace> {
    use std::collections::HashMap;

    let equity: Vec<(Timestamp, Vec<Scalar>)> = taps.equity.try_iter().collect();
    let held: Vec<(Timestamp, Vec<Scalar>)> = taps.held.try_iter().collect();
    let bars: Vec<(Timestamp, Vec<Scalar>)> = taps.bars.try_iter().collect();
    let breakout: Vec<(Timestamp, Vec<Scalar>)> = taps.breakout.try_iter().collect();

    // held shares the equity spine (same node-firing cadence); align by index is
    // safe between those two, but we still fuse the side taps by ts.
    let held_by_ts: HashMap<i64, f64> =
        held.iter().map(|(t, r)| (t.0, as_f64(&r[0]))).collect();
    let bars_by_ts: HashMap<i64, i64> =
        bars.iter().map(|(t, r)| (t.0, as_i64(&r[0]))).collect();
    let breakout_by_ts: HashMap<i64, bool> =
        breakout.iter().map(|(t, r)| (t.0, as_bool(&r[0]))).collect();

    equity
        .iter()
        .map(|(ts, row)| BarTrace {
            ts: *ts,
            equity: as_f64(&row[0]),
            held: held_by_ts.get(&ts.0).copied().unwrap_or(0.0),
            bars_since_open: bars_by_ts.get(&ts.0).copied().unwrap_or(-1),
            breakout: breakout_by_ts.get(&ts.0).copied().unwrap_or(false),
        })
        .collect()
}

/// Fold an already-drained per-bar trace into a [`RunReport`] over the real
/// window, exactly as `tests/real_bars.rs` does for the SMA sample. The trace is
/// the single drain of the recording channels (`drain_trace`); this reduction is
/// pure over it, so the example and the test share one report definition. The
/// window is the requested `[from_ms, to_ms]` normalized to epoch-ns.
pub fn report_from_trace(trace: &[BarTrace], from_ms: i64, to_ms: i64) -> RunReport {
    let equity: Vec<(Timestamp, f64)> = trace.iter().map(|b| (b.ts, b.equity)).collect();
    let exposure: Vec<(Timestamp, f64)> = trace.iter().map(|b| (b.ts, b.held)).collect();
    let metrics = summarize(&equity, &exposure);

    RunReport {
        manifest: RunManifest {
            commit: "ger40-breakout-real".to_string(),
            params: vec![
                ("session_hour".to_string(), Scalar::i64(SESSION_HOUR as i64)),
                ("session_minute".to_string(), Scalar::i64(SESSION_MINUTE as i64)),
                ("bar_minutes".to_string(), Scalar::i64(BAR_MINUTES)),
                ("entry_bar".to_string(), Scalar::i64(3)),
                ("exit_bar".to_string(), Scalar::i64(5)),
            ],
            window: (
                aura_ingest::unix_ms_to_epoch_ns(from_ms),
                aura_ingest::unix_ms_to_epoch_ns(to_ms),
            ),
            seed: 0,
            broker: "sim-optimal(pip_size=1)".to_string(),
        },
        metrics,
    }
}

fn as_f64(s: &Scalar) -> f64 {
    match s {
        Scalar::F64(v) => *v,
        other => panic!("expected f64, got {other:?}"),
    }
}

fn as_i64(s: &Scalar) -> i64 {
    match s {
        Scalar::I64(v) => *v,
        other => panic!("expected i64, got {other:?}"),
    }
}

fn as_bool(s: &Scalar) -> bool {
    match s {
        Scalar::Bool(v) => *v,
        other => panic!("expected bool, got {other:?}"),
    }
}