diff --git a/Cargo.lock b/Cargo.lock index 8194135..7cf094f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -73,6 +73,8 @@ version = "0.1.0" dependencies = [ "aura-core", "aura-std", + "chrono", + "chrono-tz", "serde", "serde_json", ] diff --git a/crates/aura-engine/Cargo.toml b/crates/aura-engine/Cargo.toml index 4fcc1d9..10e7e5b 100644 --- a/crates/aura-engine/Cargo.toml +++ b/crates/aura-engine/Cargo.toml @@ -17,3 +17,9 @@ serde_json = { workspace = true } [dev-dependencies] aura-std = { path = "../aura-std" } +# chrono / chrono-tz build the Berlin-local-wall-clock epoch-ns timestamps the +# GER40 session-breakout E2E fixture (tests/ger40_breakout.rs) feeds the engine, +# exactly as `Session`'s own unit test does. Pinned to aura-std's versions (the +# same DST-aware wall-clock math), test-only — never on the engine's hot path. +chrono = { version = "0.4", default-features = false, features = ["clock"] } +chrono-tz = { version = "0.10", default-features = false } diff --git a/crates/aura-engine/tests/ger40_breakout.rs b/crates/aura-engine/tests/ger40_breakout.rs new file mode 100644 index 0000000..84240e3 --- /dev/null +++ b/crates/aura-engine/tests/ger40_breakout.rs @@ -0,0 +1,315 @@ +//! End-to-end verification of the GER40 / DAX 15-minute session-breakout +//! strategy (spec 0050 §6, build-step 8 — the milestone capstone), hand-wired +//! from the seven new `aura-std` nodes into a raw-index [`FlatGraph`] and run +//! over one synthetic Frankfurt session. +//! +//! This is the Phase-1 C9 deliverable: no composite / strategy-builder API is in +//! scope yet, so the DAG is wired by hand the way `harness.rs`'s own fixtures are +//! (`Edge`/`Target` raw indices, `Recorder` taps). It is a **verification** test — +//! the nodes already exist and are green in isolation, so once wired correctly it +//! must PASS; a failure would mean a real node/seam integration bug, not a missing +//! feature. +//! +//! The DAG (4 synthetic M1 OHLC sources → Resample[15m] → the breakout logic → +//! Latch → SimBroker → two Recorder taps): +//! +//! ```text +//! open,high,low,close (4 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 (tap A); held -> Recorder (tap B) +//! ``` + +use std::sync::mpsc; + +use aura_core::{NodeSchema, Scalar, ScalarKind, Timestamp}; +use aura_engine::{Edge, FlatGraph, Harness, Source, SourceSpec, Target, VecSource}; +use aura_std::{And, Delay, EqConst, Gt, Latch, Recorder, Resample, Session, SimBroker}; +use chrono::TimeZone; +use chrono_tz::Europe::Berlin; + +// --------------------------------------------------------------------------- +// Node indices in the FlatGraph (one fixed layout, referenced by both wiring +// and assertions). +// --------------------------------------------------------------------------- +const RESAMPLE: usize = 0; +const DELAY: usize = 1; +const GT: usize = 2; +const SESSION: usize = 3; +const EQ3: usize = 4; +const EQ5: usize = 5; +const AND: usize = 6; +const LATCH: usize = 7; +const BROKER: usize = 8; +const REC_EQUITY: usize = 9; // tap A +const REC_HELD: usize = 10; // tap B + +/// Build the epoch-ns UTC instant for a Berlin **local wall-clock** minute on +/// 2024-07-01 (CEST, summer, UTC+2) — exactly as `Session`'s own unit test does, +/// so the bucket boundaries and `bars_since_open` are self-consistent with the +/// node. Seconds are always 0 (M1 ticks land on whole minutes). +fn berlin(hour: u32, minute: u32) -> Timestamp { + let ns = Berlin + .with_ymd_and_hms(2024, 7, 1, hour, minute, 0) + .unwrap() + .timestamp_nanos_opt() + .unwrap(); + Timestamp(ns) +} + +/// One M1 tick: the same timestamp pushed into all four OHLC sources, with that +/// tick's (open, high, low, close). Resample folds open=first, high=max, +/// low=min, close=last over a 15m bucket. +struct M1 { + ts: Timestamp, + o: f64, + h: f64, + l: f64, + c: f64, +} + +fn m1(hour: u32, minute: u32, o: f64, h: f64, l: f64, c: f64) -> M1 { + M1 { ts: berlin(hour, minute), o, h, l, c } +} + +/// Split a list of M1 ticks into the four per-field [`VecSource`]s the four OHLC +/// roles consume, declared in the FIXED order open, high, low, close. The k-way +/// merge tie-breaks same-timestamp pushes by source index, so this order makes +/// Resample's `Barrier(0)` group complete deterministically each M1 instant. +fn ohlc_sources(ticks: &[M1]) -> Vec> { + let col = |pick: fn(&M1) -> f64| -> Box { + Box::new(VecSource::new( + ticks.iter().map(|m| (m.ts, Scalar::f64(pick(m)))).collect(), + )) + }; + vec![col(|m| m.o), col(|m| m.h), col(|m| m.l), col(|m| m.c)] // open, high, low, close +} + +/// Hand-wire the whole strategy DAG into a frozen [`Harness`], tapping equity +/// (`tap A`) and `held` (`tap B`) into the two supplied channels. A fresh build +/// per call (fresh nodes + channels) keeps two runs disjoint for the determinism +/// assertion (C1). +fn build_harness( + tx_equity: mpsc::Sender<(Timestamp, Vec)>, + tx_held: mpsc::Sender<(Timestamp, Vec)>, +) -> Harness { + use aura_core::Firing; + + // 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(9, 0, Berlin, 15).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(1.0).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 + ]; + + 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(9, 0, Berlin, 15)), // 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(1.0)), // 8 + Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_equity)), // 9 + Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_held)), // 10 + ]; + + // 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 }, + ]; + + // 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. + let sources = (0..4) + .map(|slot| SourceSpec { + kind: ScalarKind::F64, + targets: vec![Target { node: RESAMPLE, slot }], + }) + .collect(); + + Harness::bootstrap(FlatGraph { nodes, signatures, sources, edges }).expect("valid strategy DAG") +} + +/// The synthetic M1 ticks realising one Frankfurt session whose 15m bars are: +/// bar1 high=100; bar2 high=110, close=108; bar3 close=112 (> bar2 high → the +/// strict breakout, and bars_since_open==3 → ENTRY); bar4 close=115; bar5 +/// close=120; then a single 10:15 rollover tick (bar6) that CLOSES bar5 (else the +/// partial last bar is dropped and exposure leaks to EOF — the spec's trap (a)). +/// bars 1 and 2 precede the bar3 entry so `Delay[1]` is warm (trap (b)). +fn entry_session_ticks() -> Vec { + vec![ + // bar1 09:00-09:15: high=100, close=100 (single tick). + m1(9, 0, 100.0, 100.0, 100.0, 100.0), + // bar2 09:15-09:30: high=110 (first tick), close=108 (second tick). + m1(9, 15, 100.0, 110.0, 100.0, 110.0), + m1(9, 20, 110.0, 110.0, 108.0, 108.0), + // bar3 09:30-09:45: close=112 (> bar2 high 110 → breakout; bar3 → entry). + m1(9, 30, 108.0, 112.0, 108.0, 112.0), + // bar4 09:45-10:00: close=115. + m1(9, 45, 112.0, 115.0, 112.0, 115.0), + // bar5 10:00-10:15: close=120. + m1(10, 0, 115.0, 120.0, 115.0, 120.0), + // bar6 rollover tick at 10:15: ANY value — closes bar5 (trap (a)). + m1(10, 15, 120.0, 120.0, 120.0, 120.0), + ] +} + +/// Drain a recorder channel into the single-column f64 values it captured, in +/// arrival order (the run is deterministic, so the order is stable). +fn drain_f64(rx: &mpsc::Receiver<(Timestamp, Vec)>) -> Vec { + rx.try_iter() + .map(|(_, row)| match row[0] { + Scalar::F64(v) => v, + other => panic!("expected an f64 column, got {other:?}"), + }) + .collect() +} + +/// Property: the GER40 session-breakout strategy, wired end-to-end, latches a +/// long position exactly from the bar3 close (the strict breakout that lands on +/// the third session bar) through the bar5 close, then exits — and the +/// sim-optimal broker integrates `held·Δclose15` into the equity curve over +/// exactly that window. This pins the *whole* node vocabulary composed: Resample +/// emit-on-rollover, Delay[1] = prevHigh15, strict-`>` breakout, Session +/// close-instant indexing, the EqConst i64→bool gates, And, the Latch SR +/// register (bool→f64), and SimBroker's lagged-exposure integration, against one +/// synthetic session. +#[test] +fn ger40_session_breakout_holds_exposure_bar3_to_bar5() { + let (tx_eq, rx_eq) = mpsc::channel(); + let (tx_held, rx_held) = mpsc::channel(); + let mut h = build_harness(tx_eq, tx_held); + h.run(ohlc_sources(&entry_session_ticks())); + + // Tap B — the Latch f64, recorded once per completed bar (Resample emits 5 + // bars: bar1..bar5 at 09:15/09:30/09:45/10:00/10:15). held is 0.0 before the + // bar3 close, 1.0 from the bar3 close through the bar5 close, then 0.0 at the + // bar5 close (reset-dominant exit on isBar5Close). + let held = drain_f64(&rx_held); + assert_eq!( + held, + vec![0.0, 0.0, 1.0, 1.0, 0.0], + "held: flat through bar2-close, latched 1.0 from bar3-close through bar4-close, \ + 0.0 at the bar5-close exit" + ); + + // Tap A — the SimBroker equity (pip_size = 1.0). The broker integrates the + // exposure HELD INTO each cycle (decided at t-1) times Δclose15, derived + // directly from sim_broker.rs (`cum += prev_exposure * (price - prev_price)`): + // close15 stream: 100, 108, 112, 115, 120 + // prev_exposure : (0), 0, 0, 1, 1 (held set on the PRIOR bar) + // step PnL : - 0 0 1*(115-112)=3 1*(120-115)=5 + // cum equity : 0.0 0.0 0.0 3.0 8.0 + // Flat 0.0 through the bar3 close, accruing only while held=1.0 (bar3→bar5), + // final equity 8.0 pips. + let equity = drain_f64(&rx_eq); + assert_eq!( + equity, + vec![0.0, 0.0, 0.0, 3.0, 8.0], + "equity: flat 0.0 until the position is held, then +3 then +8 over the bar3→bar5 hold" + ); +} + +/// Property: the breakout is **strictly** greater-than, pinned end-to-end. When +/// bar3 closes *exactly equal* to bar2's high (110 == 110), `Gt` emits `false`, +/// so `entry` never fires, the Latch stays flat, and equity is 0.0 for the whole +/// session. A regression to `>=` would silently enter here; this is the +/// no-entry control that pins the strict-`>` semantic through the full graph. +#[test] +fn no_entry_when_bar3_close_equals_bar2_high_strict_greater() { + // Identical to the entry session, except bar3 closes at 110 == bar2.high(110), + // so the strict `>` is false and no position is ever taken. + let ticks = vec![ + // bar1 09:00-09:15: high=100. + m1(9, 0, 100.0, 100.0, 100.0, 100.0), + // bar2 09:15-09:30: high=110. + m1(9, 15, 100.0, 110.0, 100.0, 110.0), + m1(9, 20, 110.0, 110.0, 108.0, 108.0), + // bar3 09:30-09:45: close=110 == bar2 high → NOT a strict breakout. + m1(9, 30, 108.0, 110.0, 108.0, 110.0), + // bar4 / bar5 continue upward (irrelevant — no entry ever latched). + m1(9, 45, 110.0, 115.0, 110.0, 115.0), + m1(10, 0, 115.0, 120.0, 115.0, 120.0), + // bar6 rollover tick: closes bar5. + m1(10, 15, 120.0, 120.0, 120.0, 120.0), + ]; + + let (tx_eq, rx_eq) = mpsc::channel(); + let (tx_held, rx_held) = mpsc::channel(); + let mut h = build_harness(tx_eq, tx_held); + h.run(ohlc_sources(&ticks)); + + let held = drain_f64(&rx_held); + assert_eq!( + held, + vec![0.0, 0.0, 0.0, 0.0, 0.0], + "strict `>`: close == prev high is no breakout, so held never latches" + ); + let equity = drain_f64(&rx_eq); + assert_eq!( + equity, + vec![0.0, 0.0, 0.0, 0.0, 0.0], + "no position is ever taken, so equity is flat 0.0 across the session" + ); +} + +/// Property: the whole hand-wired harness is deterministic (C1) — two disjoint +/// runs over byte-identical input produce byte-identical recorded output. Mirrors +/// `harness.rs::fan_out_join_dag_runs_deterministically` on the full strategy DAG. +#[test] +fn ger40_breakout_run_is_deterministic() { + let run_once = || { + let (tx_eq, rx_eq) = mpsc::channel(); + let (tx_held, rx_held) = mpsc::channel(); + let mut h = build_harness(tx_eq, tx_held); + h.run(ohlc_sources(&entry_session_ticks())); + let eq: Vec<(Timestamp, Vec)> = rx_eq.try_iter().collect(); + let hd: Vec<(Timestamp, Vec)> = rx_held.try_iter().collect(); + (eq, hd) + }; + assert_eq!(run_once(), run_once(), "same input → byte-identical recorded output (C1)"); +}