diff --git a/Cargo.lock b/Cargo.lock index 7cf094f..2d4b43b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -86,6 +86,8 @@ dependencies = [ "aura-core", "aura-engine", "aura-std", + "chrono", + "chrono-tz", "data-server", ] diff --git a/crates/aura-ingest/Cargo.toml b/crates/aura-ingest/Cargo.toml index 3ec5626..40eb7f9 100644 --- a/crates/aura-ingest/Cargo.toml +++ b/crates/aura-ingest/Cargo.toml @@ -18,3 +18,10 @@ aura-engine = { path = "../aura-engine" } [dev-dependencies] # the integration tests bootstrap a sample harness + fold a RunReport aura-std = { path = "../aura-std" } +# chrono / chrono-tz build the Berlin-local-wall-clock window bounds + the +# Session node's timezone for the real GER40 session-breakout example/test +# (examples/ger40_breakout_real.rs, tests/ger40_breakout_real.rs). Pinned to +# aura-std's / aura-engine's versions (the same DST-aware wall-clock math), +# test-only — never on the ingestion crate's normal path. +chrono = { version = "0.4", default-features = false, features = ["clock"] } +chrono-tz = { version = "0.10", default-features = false } diff --git a/crates/aura-ingest/examples/ger40_breakout_real.rs b/crates/aura-ingest/examples/ger40_breakout_real.rs new file mode 100644 index 0000000..e2a4326 --- /dev/null +++ b/crates/aura-ingest/examples/ger40_breakout_real.rs @@ -0,0 +1,127 @@ +//! Runnable demonstration: the GER40 15m session-breakout strategy — the +//! Phase-1 `aura-std` node vocabulary, proven synthetically in +//! `aura-engine/tests/ger40_breakout.rs` — driven over REAL GER40 M1 bars from +//! the local data-server archive (one calendar month). +//! +//! 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. +//! +//! Run: +//! ```text +//! cargo run -p aura-ingest --example ger40_breakout_real +//! ``` +//! Prints the [`RunReport`] metrics (total pips / drawdown / sign-flips) and a +//! compact entry/exit trace — each 0→1 entry and 1→0 exit with its timestamp — +//! so the strategy's behaviour over the ~20 sessions in the month is visibly +//! inspectable. Skips cleanly (no panic) where the archive is absent. + +use std::sync::Arc; + +use chrono::TimeZone; +use data_server::{DataServer, DEFAULT_DATA_PATH}; + +// The breakout wiring is defined once in the shared module and included by both +// this example and the gated determinism test — never duplicated. +#[path = "shared/breakout_real.rs"] +mod breakout_real; +use breakout_real::*; + +// --- The window: September 2024 (inclusive), trivially changeable. ----------- +const YEAR: i32 = 2024; +const MONTH: u32 = 9; + +fn main() { + let server = Arc::new(DataServer::new(DEFAULT_DATA_PATH)); + if !server.has_symbol(SYMBOL) { + println!( + "skip: no local data at {DEFAULT_DATA_PATH} (symbol {SYMBOL} absent) — \ + nothing to demonstrate here." + ); + return; + } + + let (from_ms, to_ms) = utc_month_window_ms(YEAR, MONTH); + + let Some(sources) = open_ohlc_sources(&server, from_ms, to_ms) else { + println!( + "skip: {SYMBOL} has no M1 file overlapping {YEAR}-{MONTH:02} — \ + nothing to demonstrate here." + ); + return; + }; + + let (mut h, taps) = build_harness(); + h.run(sources); + + // Drain the per-bar trace first (it consumes the equity/held/bars/breakout + // channels), then fold the report from that same trace so the mpsc channels + // are drained exactly once (shared with the determinism test). + let trace = drain_trace(&taps); + let report = report_from_trace(&trace, from_ms, to_ms); + let metrics = &report.metrics; + + println!("=== GER40 15m session-breakout — REAL bars ==="); + println!( + "symbol={SYMBOL} window={YEAR}-{MONTH:02} (UTC, inclusive) \ + bars={} pip_size={PIP_SIZE}", + trace.len() + ); + println!("session: {SESSION_HOUR:02}:{SESSION_MINUTE:02} Europe/Berlin, {BAR_MINUTES}m bars; entry bar 3, exit bar 5"); + println!(); + println!("--- metrics (sim-optimal broker, pips = index points) ---"); + println!(" total_pips = {:.1}", metrics.total_pips); + println!(" max_drawdown = {:.1}", metrics.max_drawdown); + println!(" exposure_sign_flips = {}", metrics.exposure_sign_flips); + println!(); + + // --- 2. The compact entry/exit trace. ----------------------------------- + // Each 0→1 is an entry (a strict breakout on session bar 3), each 1→0 is the + // bar-5 exit. Printing only the transitions keeps the ~20 sessions readable. + println!("--- entry / exit transitions (held 0↔1) ---"); + let mut prev_held = 0.0_f64; + let mut entries = 0u32; + let mut exits = 0u32; + let mut entry_equity = 0.0_f64; + for b in &trace { + if b.held == 1.0 && prev_held == 0.0 { + entries += 1; + entry_equity = b.equity; + println!( + " ENTER {} bar={} breakout={} equity={:+.1}", + fmt_ts(b.ts), + b.bars_since_open, + b.breakout, + b.equity + ); + } else if b.held == 0.0 && prev_held == 1.0 { + exits += 1; + println!( + " EXIT {} bar={} equity={:+.1} (session pnl {:+.1})", + fmt_ts(b.ts), + b.bars_since_open, + b.equity, + b.equity - entry_equity + ); + } + prev_held = b.held; + } + if prev_held == 1.0 { + println!(" (note: still held at end of window — a partial last session)"); + } + println!(); + println!("sessions entered = {entries}, exited = {exits}"); +} + +/// Render an epoch-ns [`Timestamp`](aura_core::Timestamp) as a UTC wall-clock +/// `YYYY-MM-DD HH:MM` for the trace. Display only — never feeds the graph. +fn fmt_ts(ts: aura_core::Timestamp) -> String { + let secs = ts.0.div_euclid(1_000_000_000); + let nanos = ts.0.rem_euclid(1_000_000_000) as u32; + match chrono::Utc.timestamp_opt(secs, nanos).single() { + Some(dt) => dt.format("%Y-%m-%d %H:%M UTC").to_string(), + None => format!("ts={}", ts.0), + } +} diff --git a/crates/aura-ingest/examples/shared/breakout_real.rs b/crates/aura-ingest/examples/shared/breakout_real.rs new file mode 100644 index 0000000..d068a7c --- /dev/null +++ b/crates/aura-ingest/examples/shared/breakout_real.rs @@ -0,0 +1,338 @@ +//! 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, Edge, FlatGraph, 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)>, + pub held: mpsc::Receiver<(Timestamp, Vec)>, + pub bars: mpsc::Receiver<(Timestamp, Vec)>, + pub breakout: mpsc::Receiver<(Timestamp, Vec)>, +} + +/// 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 = 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> = 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 }) +} + +/// 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, + from_ms: i64, + to_ms: i64, +) -> Option>> { + 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> = + 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 { + use std::collections::HashMap; + + let equity: Vec<(Timestamp, Vec)> = taps.equity.try_iter().collect(); + let held: Vec<(Timestamp, Vec)> = taps.held.try_iter().collect(); + let bars: Vec<(Timestamp, Vec)> = taps.bars.try_iter().collect(); + let breakout: Vec<(Timestamp, Vec)> = 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 = + held.iter().map(|(t, r)| (t.0, as_f64(&r[0]))).collect(); + let bars_by_ts: HashMap = + bars.iter().map(|(t, r)| (t.0, as_i64(&r[0]))).collect(); + let breakout_by_ts: HashMap = + 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:?}"), + } +} diff --git a/crates/aura-ingest/tests/ger40_breakout_real.rs b/crates/aura-ingest/tests/ger40_breakout_real.rs new file mode 100644 index 0000000..daf3209 --- /dev/null +++ b/crates/aura-ingest/tests/ger40_breakout_real.rs @@ -0,0 +1,88 @@ +//! Gated integration test: the GER40 15m session-breakout strategy — the +//! Phase-1 `aura-std` node vocabulary wired VERBATIM from +//! `aura-engine/tests/ger40_breakout.rs` — driven over REAL GER40 M1 bars from +//! the local data-server archive (September 2024). Mirrors `real_bars.rs`: +//! skips with a note where the local Pepperstone archive is absent, so +//! `cargo test --workspace` stays green anywhere; exercises the real ingestion +//! path where the files exist. +//! +//! The breakout FlatGraph is built once in the shared module (`shared/ +//! breakout_real.rs`) and reused by both this test and the runnable example, so +//! the 11-node wiring is never duplicated. + +use std::sync::Arc; + +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. +const YEAR: i32 = 2024; +const MONTH: u32 = 9; + +/// Build the real breakout harness, run it over `[from_ms, to_ms]`, and return +/// the folded `RunReport` plus the drained per-bar `(held, equity)` series. A +/// fresh build per call keeps the two determinism runs disjoint (C1). +fn run_real( + server: &Arc, + from_ms: i64, + to_ms: i64, +) -> (aura_engine::RunReport, Vec<(f64, f64)>) { + let sources = + open_ohlc_sources(server, from_ms, to_ms).expect("GER40 has data in the Sept-2024 window"); + let (mut h, taps) = build_harness(); + h.run(sources); + + // Drain the trace once; fold the report from the same trace via the shared + // helper so the mpsc channels are consumed exactly once and the report shape + // matches the example's byte-for-byte. + let trace = drain_trace(&taps); + let report = report_from_trace(&trace, from_ms, to_ms); + let series: Vec<(f64, f64)> = trace.iter().map(|b| (b.held, b.equity)).collect(); + (report, series) +} + +/// Property: the real-bar GER40 breakout harness runs end-to-end over a real +/// data-server month and is **deterministic** (C1) — two disjoint runs over the +/// byte-identical archive window produce bit-identical reports AND bit-identical +/// recorded series. Also pins two observable invariants of the composed +/// vocabulary on real data: the metrics are finite, and the Latch output +/// (`held`) is *exactly* 0.0 or 1.0 every bar (a binary exposure register, never +/// a fractional or NaN value). This protects the whole node composition — +/// Resample / Delay / Gt / Session / EqConst / And / Latch / SimBroker — against +/// any regression that would only surface on real (irregular, gap-laden) M1 +/// data, which the synthetic single-session capstone cannot reach. +#[test] +fn ger40_breakout_real_bars_run_is_deterministic() { + 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 (r1, s1) = run_real(&server, from_ms, to_ms); + assert!(!s1.is_empty(), "window resolved to zero bars"); + + // Metrics are finite — a backtest actually ran over real bars. + assert!(r1.metrics.total_pips.is_finite(), "total_pips finite"); + assert!(r1.metrics.max_drawdown.is_finite(), "max_drawdown finite"); + + // Latch output is a binary exposure register: every recorded `held` is + // exactly 0.0 or 1.0 — never fractional, never NaN. + for (held, _) in &s1 { + assert!( + *held == 0.0 || *held == 1.0, + "held must be exactly 0.0 or 1.0 (Latch register), got {held}" + ); + } + + // C1 determinism: a fresh disjoint run over the identical window is + // bit-identical in both the folded report and the raw recorded series. + let (r2, s2) = run_real(&server, from_ms, to_ms); + assert_eq!(r1.to_json(), r2.to_json(), "report bit-identical across runs (C1)"); + assert_eq!(s1, s2, "recorded (held, equity) series bit-identical across runs (C1)"); +}