Brummel/Aura
1
0
Fork 0
Code Issues 31 Pull Requests Actions Packages Projects Releases Wiki Activity

test(aura-engine): GER40 session-breakout e2e — the milestone capstone

Browse Source 
Hand-wires all seven new aura-std nodes (Resample, Delay, Gt, Session, EqConst
x2, And, Latch) + SimBroker into a raw-index FlatGraph over one synthetic
Frankfurt session, pinning the whole vocabulary composed end-to-end:
- held = [0,0,1,1,0] over the 5 bar emissions: flat, latched bar3-close (the
  strict breakout landing on session bar 3) through bar4, exit at bar5-close;
- equity = [0,0,0,3,8] pips (SimBroker lagged-exposure integration, pip=1.0);
- a no-entry control (bar3 close == bar2 high) pins the strict-> semantic;
- two disjoint runs byte-identical (C1).
Both spec traps exercised (bar6 rollover closes bar5; bars 1-2 warm Delay[1]).
Phase-1 C9 deliverable; build-step 8 (capstone) of milestone 'Strategy node
vocabulary I'.

closes #91
This commit is contained in:
Brummel
2026-06-17 17:26:54 +02:00
parent 17197fed91
commit 4187f3bbc1
3 changed files with 323 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+2
View File
@@ -73,6 +73,8 @@ version = "0.1.0"
dependencies = [
"aura-core",
"aura-std",
"chrono",
"chrono-tz",
"serde",
"serde_json",
]
crates/aura-engine/Cargo.toml
+6
View File
@@ -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 }
crates/aura-engine/tests/ger40_breakout.rs
+315
View File
@@ -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<Box<dyn Source>> {
let col = |pick: fn(&M1) -> f64| -> Box<dyn Source> {
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<Scalar>)>,
tx_held: mpsc::Sender<(Timestamp, Vec<Scalar>)>,
) -> Harness {
use aura_core::Firing;
// 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(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<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(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<M1> {
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<Scalar>)>) -> Vec<f64> {
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<Scalar>)> = rx_eq.try_iter().collect();
let hd: Vec<(Timestamp, Vec<Scalar>)> = rx_held.try_iter().collect();
(eq, hd)
};
assert_eq!(run_once(), run_once(), "same input → byte-identical recorded output (C1)");
}