feat(aura-ingest): GER40 breakout as a Composite blueprint (spec 0051)

Author the shipped session-breakout as a reusable Composite blueprint via
GraphBuilder — the World-consumable form (param_space + bootstrap) — alongside
the hand-wired FlatGraph it reproduces. Applies the spec-0051 decisions:
- bar_period_minutes is a construction arg binding BOTH Resample.period_minutes
  and Session's period, locked equal, so a sweep cannot desync them (#96);
- delay.lag is bound out (a structural constant, not a tuning knob — C34);
- the two EqConst targets (named entry_bar / exit_bar) are the ONLY params, so
  param_space() == {entry_bar.target, exit_bar.target} — the genuine tuning
  knobs, which makes walk_forward non-degenerate (#97).

The ger40_breakout_real example now bootstraps from the blueprint (still prints
-45.0 for 2024-09). New gated test ger40_breakout_blueprint.rs proves: (a) the
param_space is exactly the two targets (no lag, no period); (b) C23 — the
blueprint reproduces the FlatGraph bit-identically over real GER40 2024-09;
(c) determinism.

First increment of the milestone; the sweep / walk_forward / compare World-family
demos follow.

refs #94, #96, #97

crates/aura-ingest/examples/ger40_breakout_real.rs

@@ -5,9 +5,11 @@
 //!
 //! This is a *pure composition* of SHIPPED `aura-std` nodes: no project-specific
 //! signal code, so it lives in the engine repo's `examples/` carveout (C9), not
-//! a separate project crate. It reuses the breakout FlatGraph VERBATIM (see
-//! `shared/breakout_real.rs`) — only the sources change: four real
-//! `M1FieldSource`s replace the synthetic `VecSource`s.
+//! a separate project crate. The strategy is authored as a World-consumable
+//! `Composite` blueprint (`ger40_breakout_blueprint`, spec 0051) and bootstrapped
+//! here by name (`entry_bar=3`, `exit_bar=5`); the four real `M1FieldSource`s
+//! feed its four OHLC source roles. The blueprint reproduces the shipped
+//! hand-wired `FlatGraph` EXACTLY (C23), so this prints the same report + trace.
 //!
 //! Run:
 //! ```text
@@ -21,6 +23,7 @@
 use std::sync::Arc;
 
 use chrono::TimeZone;
+use chrono_tz::Europe::Berlin;
 use data_server::{DataServer, DEFAULT_DATA_PATH};
 
 // The breakout wiring is defined once in the shared module and included by both
@@ -53,7 +56,16 @@ fn main() {
         return;
     };
 
-    let (mut h, taps) = build_harness();
+    // Author the breakout as a Composite blueprint and bootstrap it by name with
+    // the two genuine tuning knobs (entry bar 3, exit bar 5). The bar-period (15m),
+    // session open (09:00 Berlin) and delay.lag are structural — bound at
+    // construction, not exposed as params (spec 0051 §2).
+    let (bp, taps) = ger40_breakout_blueprint(BAR_MINUTES, SESSION_HOUR, SESSION_MINUTE, Berlin);
+    let mut h = bp
+        .with("entry_bar.target", aura_core::Scalar::i64(3))
+        .with("exit_bar.target", aura_core::Scalar::i64(5))
+        .bootstrap()
+        .expect("breakout blueprint bootstraps with entry=3, exit=5");
     h.run(sources);
 
     // Drain the per-bar trace first (it consumes the equity/held/bars/breakout

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

@@ -41,7 +41,8 @@ use std::sync::Arc;
 
 use aura_core::{NodeSchema, Scalar, ScalarKind, Timestamp};
 use aura_engine::{
-    summarize, Edge, FlatGraph, Harness, RunManifest, RunReport, Source, SourceSpec, Target,
+    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;
@@ -192,6 +193,119 @@ pub fn build_harness() -> (Harness, Taps) {
     (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 (spec 0051) — 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 spec-0051 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
+    // (§2): 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

crates/aura-ingest/tests/ger40_breakout_blueprint.rs

@@ -0,0 +1,135 @@
+//! Gated integration test: the GER40 session-breakout shipped as a
+//! World-consumable `Composite` blueprint (spec 0051). Proves the blueprint
+//! reproduces the shipped hand-wired `FlatGraph` EXACTLY (C23) and that its
+//! `param_space()` is the two genuine tuning knobs only — no `delay.lag`, no
+//! bar-period (the desync trap, #96).
+//!
+//! Mirrors `tests/ger40_breakout_real.rs` / `real_bars.rs`: the data-gated
+//! checks skip with a note where the local Pepperstone archive is absent, so
+//! `cargo test --workspace` stays green anywhere. The `param_space()` check is
+//! PURE (no data) and runs everywhere.
+//!
+//! The breakout wiring — both the shipped hand-wired `FlatGraph`
+//! (`build_harness()`) and the new blueprint factory
+//! (`ger40_breakout_blueprint()`) — lives once in the shared module
+//! (`shared/breakout_real.rs`) and is reused by the example and the tests.
+
+use std::sync::Arc;
+
+use aura_core::Scalar;
+use chrono_tz::Europe::Berlin;
+use data_server::{DataServer, DEFAULT_DATA_PATH};
+
+#[path = "../examples/shared/breakout_real.rs"]
+mod breakout_real;
+use breakout_real::*;
+
+// September 2024 (inclusive), the same window the example demonstrates and the
+// hand-wired determinism test pins.
+const YEAR: i32 = 2024;
+const MONTH: u32 = 9;
+
+/// (a) PURE — `param_space()` of the blueprint is EXACTLY the two genuine tuning
+/// knobs `entry_bar.target` and `exit_bar.target`, and contains NEITHER
+/// `delay.lag` NOR any bar-period param. This is the #96 resolution made
+/// executable: the bar-period is bound structurally (both Resample and Session
+/// at construction, locked equal) and `delay.lag` is bound out, so neither can
+/// be swept into a desync. No data needed — runs everywhere.
+#[test]
+fn param_space_is_exactly_entry_and_exit_bar_targets() {
+    let (bp, _taps) = ger40_breakout_blueprint(15, 9, 0, Berlin);
+    let space = bp.param_space();
+
+    let names: Vec<&str> = space.iter().map(|p| p.name.as_str()).collect();
+    assert_eq!(
+        names,
+        vec!["entry_bar.target", "exit_bar.target"],
+        "param_space() must be exactly the two EqConst targets (spec 0051 §2)",
+    );
+
+    // Belt-and-braces on the two specific leaks #96 named, independent of order.
+    assert!(
+        !names.iter().any(|n| n.contains("lag")),
+        "delay.lag must be bound out of the space (a structural constant, not a knob)",
+    );
+    assert!(
+        !names.iter().any(|n| n.contains("period")),
+        "no bar-period param may leak (it is a structural construction arg, locked equal)",
+    );
+}
+
+/// (c) determinism (PURE) — two blueprints constructed with the same args expose
+/// the identical `param_space()`; the construction is a deterministic function of
+/// its args. (The data-gated run-level determinism is asserted in
+/// `composite_matches_flatgraph_bit_identical`.)
+#[test]
+fn blueprint_param_space_is_construction_deterministic() {
+    let a = ger40_breakout_blueprint(15, 9, 0, Berlin).0.param_space();
+    let b = ger40_breakout_blueprint(15, 9, 0, Berlin).0.param_space();
+    assert_eq!(a, b, "blueprint param_space() is a deterministic function of its args");
+}
+
+/// Run the blueprint-bootstrapped harness over `[from_ms, to_ms]` and drain the
+/// recorded per-bar `(held, equity)` series. Fresh build per call (fresh nodes +
+/// channels) keeps two runs disjoint for the C1 determinism assertion.
+fn run_blueprint(server: &Arc<DataServer>, from_ms: i64, to_ms: i64) -> Vec<(f64, f64)> {
+    let (bp, taps) = ger40_breakout_blueprint(BAR_MINUTES, SESSION_HOUR, SESSION_MINUTE, Berlin);
+    let mut h = bp
+        .with("entry_bar.target", Scalar::i64(3))
+        .with("exit_bar.target", Scalar::i64(5))
+        .bootstrap()
+        .expect("blueprint bootstraps with entry=3, exit=5");
+    let sources =
+        open_ohlc_sources(server, from_ms, to_ms).expect("GER40 has data in the Sept-2024 window");
+    h.run(sources);
+    let trace = drain_trace(&taps);
+    trace.iter().map(|b| (b.held, b.equity)).collect()
+}
+
+/// Run the shipped hand-wired `FlatGraph` over the same window — the C23
+/// reference series.
+fn run_flatgraph(server: &Arc<DataServer>, from_ms: i64, to_ms: i64) -> Vec<(f64, f64)> {
+    let (mut h, taps) = build_harness();
+    let sources =
+        open_ohlc_sources(server, from_ms, to_ms).expect("GER40 has data in the Sept-2024 window");
+    h.run(sources);
+    let trace = drain_trace(&taps);
+    trace.iter().map(|b| (b.held, b.equity)).collect()
+}
+
+/// (b) C23 fidelity (gated) — over the GER40 2024-09 window, bootstrapping the
+/// blueprint (entry=3, exit=5, bar_period=15) and running it produces a recorded
+/// `(held, equity)` series BIT-IDENTICAL to the existing hand-wired `FlatGraph`
+/// `build_harness()` over the same window. This is the behaviour-preserving
+/// guarantee (C23): the Composite is just the authoring form of the same flat
+/// graph, so the numbers must match exactly — not approximately.
+///
+/// (c) determinism (gated) — the blueprint path is bit-identical on a fresh
+/// disjoint rerun (C1).
+#[test]
+fn composite_matches_flatgraph_bit_identical() {
+    let server = Arc::new(DataServer::new(DEFAULT_DATA_PATH));
+    if !server.has_symbol(SYMBOL) {
+        eprintln!("skip: no local data at {DEFAULT_DATA_PATH} (symbol {SYMBOL} absent)");
+        return; // hermetic elsewhere; exercises the real path where files exist
+    }
+
+    let (from_ms, to_ms) = utc_month_window_ms(YEAR, MONTH);
+
+    let blueprint_series = run_blueprint(&server, from_ms, to_ms);
+    assert!(!blueprint_series.is_empty(), "window resolved to zero bars");
+
+    // C23: the Composite blueprint reproduces the hand-wired FlatGraph EXACTLY.
+    let flatgraph_series = run_flatgraph(&server, from_ms, to_ms);
+    assert_eq!(
+        blueprint_series, flatgraph_series,
+        "blueprint must reproduce the hand-wired FlatGraph bit-identically (C23)",
+    );
+
+    // C1: a fresh disjoint blueprint run over the identical window is bit-identical.
+    let blueprint_rerun = run_blueprint(&server, from_ms, to_ms);
+    assert_eq!(
+        blueprint_series, blueprint_rerun,
+        "blueprint recorded (held, equity) series bit-identical across runs (C1)",
+    );
+}