From 4f88d675434f661fa5b9b2c65a1acad4d8b77498 Mon Sep 17 00:00:00 2001 From: Brummel Date: Thu, 25 Jun 2026 13:33:21 +0200 Subject: [PATCH] =?UTF-8?q?audit(0071):=20cycle=20close=20=E2=80=94=20brea?= =?UTF-8?q?kout=20candidate=20refuted;=20drop=20ephemera?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Donchian channel-breakout Stage-1 R candidate is refuted: no cross-index generalization, violent IS->OOS sign flips at long channels, no significant positive cell (full verdict recorded on #137). The breakout machinery (RollingMax / RollingMin nodes, --strategy stage1-breakout) is permanent and kept; only the per-cycle spec/plan are dropped per the ephemeral-artifact lifecycle convention. refs #137 --- docs/plans/0071-stage1-breakout.md | 899 ----------------------------- docs/specs/0071-stage1-breakout.md | 362 ------------ 2 files changed, 1261 deletions(-) delete mode 100644 docs/plans/0071-stage1-breakout.md delete mode 100644 docs/specs/0071-stage1-breakout.md diff --git a/docs/plans/0071-stage1-breakout.md b/docs/plans/0071-stage1-breakout.md deleted file mode 100644 index 73f54ff..0000000 --- a/docs/plans/0071-stage1-breakout.md +++ /dev/null @@ -1,899 +0,0 @@ -# Stage-1 Donchian Breakout Candidate — Implementation Plan - -> **Parent spec:** `docs/specs/0071-stage1-breakout.md` -> -> **For agentic workers:** REQUIRED SUB-SKILL: use the `implement` skill to run -> this plan. Steps use `- [ ]` checkboxes for tracking. - -**Goal:** Add a channel-breakout signal candidate (`--strategy stage1-breakout`) -that screens under the same Stage-1 R yardstick as `stage1-r`, swapping only the -signal leg (SMA-cross → Donchian channel direction). - -**Architecture:** Two new aura-std nodes (`RollingMax`/`RollingMin`, sliding-window -extremum via a monotonic deque, mirroring the `Sma`/`Delay` node shape). A new -`stage1_breakout_graph` mirrors `stage1_r_graph` but builds the signal leg `close → -Delay(1) → {RollingMax,RollingMin}(N) → {Gt,Gt} → {Latch,Latch} → Sub = bias ∈ -{−1,0,+1}`, feeding the identical bias seam; the vol-stop defines R unchanged. A -`stage1_breakout_sweep_family` iterates the channel × stop grid manually with -fully-bound graphs (sidestepping parameter-ganging, since one `channel` drives two -nodes). A `Strategy::Stage1Breakout` variant + `--channel` flag wire it to the CLI. - -**Tech Stack:** aura-std (node primitives), aura-core (Node trait), aura-cli -(strategy graph + sweep family + arg parse), aura-engine (GraphBuilder, Harness, -SweepFamily/SweepPoint, summarize/summarize_r). - ---- - -**Files this plan creates or modifies:** - -- Create: `crates/aura-std/src/rolling_max.rs` — sliding-window max node (descending monotonic deque) -- Create: `crates/aura-std/src/rolling_min.rs` — sliding-window min node (ascending monotonic deque) -- Modify: `crates/aura-std/src/lib.rs:18-66` — register the two modules + re-exports -- Create: `crates/aura-engine/tests/stage1_breakout_e2e.rs` — C2-causality + ±1-latch composition test -- Modify: `crates/aura-cli/src/main.rs` — imports (Gt/Latch/RollingMax/RollingMin, SweepPoint); `stage1_breakout_graph`; `stage1_breakout_sweep_family`; `Strategy::Stage1Breakout` (enum 1286, parse arm 1336, dispatch arm 1422); `--channel` flag (parse 1345) + `Stage1RGrid.channel` field (1123-1140) -- Test: `crates/aura-cli/tests/cli_run.rs` — `aura sweep --strategy stage1-breakout` seam: r block present + folded-no-trace == raw-trace - ---- - -### Task 1: `RollingMax` node - -**Files:** -- Create: `crates/aura-std/src/rolling_max.rs` -- Modify: `crates/aura-std/src/lib.rs:18-66` - -- [ ] **Step 1: Write the node file with its failing tests** - -Create `crates/aura-std/src/rolling_max.rs`: - -```rust -//! `RollingMax` — maximum over the last `length` values of one f64 input. -//! -//! Sliding-window maximum maintained by a **descending monotonic deque** (the -//! textbook O(1)-amortized sliding-window-max): each cycle pops from the back every -//! element <= the new sample (they can never again be the window max), pushes the new -//! one, and evicts the front once its stream index leaves the window. The front is -//! always the current window max. A per-cycle O(N) re-scan would be too slow at the -//! large `N` (channel lengths in the thousands of bars) the breakout screen sweeps. -//! -//! Like `Sma`/`Delay`, the window lives in node state, so `eval` reads only the newest -//! sample and `lookbacks()` is `1`. Warm-up is **skip-emit**: `None` until `length` -//! samples have passed (the window is not yet full). The deque is pre-sized to -//! `length` so the hot path is allocation-free (C7). Per the "operator is topology" -//! convention (see `gt.rs`), `RollingMax` and `RollingMin` are two node types, not one -//! node with a max/min param. - -use aura_core::{ - Cell, Ctx, FieldSpec, Firing, Node, NodeSchema, ParamSpec, PortSpec, PrimitiveBuilder, - ScalarKind, -}; -use std::collections::VecDeque; - -/// Maximum over the last `length` values of one f64 input, maintained by a descending -/// monotonic deque (front = current window max). `None` during warm-up. -pub struct RollingMax { - length: usize, - // descending monotonic deque of (value, stream_index); front is the window max. - // node-owned, bounded by `length` (C7), pre-sized so push_back does not allocate. - deque: VecDeque<(f64, u64)>, - seen: u64, // stream position of the next sample (drives front eviction) - count: usize, // samples seen so far, capped at `length` — the warm-up gate - out: [Cell; 1], -} - -impl RollingMax { - /// Build a rolling max of window `length` (must be >= 1; mirror `Sma::new`). - pub fn new(length: usize) -> Self { - assert!(length >= 1, "RollingMax length must be >= 1"); - Self { - length, - deque: VecDeque::with_capacity(length), - seen: 0, - count: 0, - out: [Cell::from_f64(0.0)], - } - } - - /// The param-generic recipe for a blueprint primitive: declares `length` and builds - /// through `RollingMax::new`. - pub fn builder() -> PrimitiveBuilder { - PrimitiveBuilder::new( - "RollingMax", - NodeSchema { - inputs: vec![PortSpec { kind: ScalarKind::F64, firing: Firing::Any, name: "series".into() }], - output: vec![FieldSpec { name: "value".into(), kind: ScalarKind::F64 }], - params: vec![ParamSpec { name: "length".into(), kind: ScalarKind::I64 }], - }, - |p| Box::new(RollingMax::new(p[0].i64() as usize)), - ) - } -} - -impl Node for RollingMax { - // The window lives in node state, so only the newest sample is read each cycle. - fn lookbacks(&self) -> Vec { - vec![1] - } - - fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> { - let w = ctx.f64_in(0); - if w.is_empty() { - return None; // no sample yet - } - let x = w[0]; // index 0 = newest (financial indexing) - let idx = self.seen; - self.seen += 1; - // descending invariant: a back element <= x can never again be the max — drop it. - while let Some(&(v, _)) = self.deque.back() { - if v <= x { - self.deque.pop_back(); - } else { - break; - } - } - self.deque.push_back((x, idx)); - // evict the front once it leaves the window [idx-length+1, idx] (no subtraction: - // front index `i` is out when `i + length <= idx`). - while let Some(&(_, i)) = self.deque.front() { - if i + self.length as u64 <= idx { - self.deque.pop_front(); - } else { - break; - } - } - if self.count < self.length { - self.count += 1; - } - if self.count < self.length { - return None; // not yet warmed up - } - self.out[0] = Cell::from_f64(self.deque.front().expect("deque non-empty after push").0); - Some(&self.out) - } - - fn label(&self) -> String { - format!("RollingMax({})", self.length) - } -} - -#[cfg(test)] -mod tests { - use super::*; - use aura_core::{AnyColumn, Scalar, Timestamp}; - - fn drive(node: &mut RollingMax, feed: &[f64]) -> Vec> { - let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)]; - let mut out = Vec::new(); - for &v in feed { - inputs[0].push(Scalar::f64(v)).unwrap(); - out.push(node.eval(Ctx::new(&inputs, Timestamp(0))).map(|r| r[0].f64())); - } - out - } - - #[test] - fn rolling_max_warms_up_then_tracks_the_window_max() { - // max of [1,3,2], [3,2,5], [2,5,4] once warmed up; silent for the first two. - let mut node = RollingMax::new(3); - let got = drive(&mut node, &[1.0, 3.0, 2.0, 5.0, 4.0]); - assert_eq!(got, vec![None, None, Some(3.0), Some(5.0), Some(5.0)]); - } - - #[test] - fn rolling_max_evicts_the_expiring_window_maximum() { - // the front-eviction path: the early big value must leave the window. length 2. - // windows: [9,1]->9, [1,2]->2, [2,3]->3 (the 9 has expired, not stuck as max). - let mut node = RollingMax::new(2); - let got = drive(&mut node, &[9.0, 1.0, 2.0, 3.0]); - assert_eq!(got, vec![None, Some(9.0), Some(2.0), Some(3.0)]); - } - - #[test] - fn rolling_max_length_one_is_identity() { - let mut node = RollingMax::new(1); - assert_eq!(drive(&mut node, &[7.0, 9.0, 4.0]), vec![Some(7.0), Some(9.0), Some(4.0)]); - } - - #[test] - fn deque_matches_naive_window_max() { - // the monotonic deque must equal a naive O(N) per-window max at every warmed - // cycle, across a long deterministic series (mirrors Sma's re-sum cross-check). - let length = 37; - let mut node = RollingMax::new(length); - let series: Vec = - (0..3_000).map(|i| ((i * 1103515245 + 12345) % 1000) as f64 * 0.5 - 250.0).collect(); - let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)]; - for (t, &x) in series.iter().enumerate() { - inputs[0].push(Scalar::f64(x)).unwrap(); - let got = node.eval(Ctx::new(&inputs, Timestamp(0))).map(|r| r[0].f64()); - if t + 1 >= length { - let want = series[t + 1 - length..=t].iter().cloned().fold(f64::NEG_INFINITY, f64::max); - assert_eq!(got, Some(want), "at t={t}"); - } else { - assert_eq!(got, None, "silent until warmed up at t={t}"); - } - } - } - - #[test] - fn lookback_is_one_window_lives_in_node_state() { - assert_eq!(RollingMax::new(20).lookbacks(), vec![1]); - } - - #[test] - fn length_must_be_at_least_one() { - let r = std::panic::catch_unwind(|| RollingMax::new(0)); - assert!(r.is_err(), "RollingMax::new(0) must panic"); - } - - #[test] - fn label_carries_the_window() { - assert_eq!(RollingMax::new(20).label(), "RollingMax(20)"); - } -} -``` - -- [ ] **Step 2: Register the module + re-export in `lib.rs`** - -In `crates/aura-std/src/lib.rs`, add `mod rolling_max;` in the alphabetical `mod` -block (between `mod resample;` and `mod series_reducer;`), and `pub use -rolling_max::RollingMax;` in the `pub use` block (near `pub use sma::Sma;`). - -- [ ] **Step 3: Run the tests to verify they pass** - -Run: `cargo test -p aura-std rolling_max` -Expected: PASS — `rolling_max_warms_up_then_tracks_the_window_max`, -`rolling_max_evicts_the_expiring_window_maximum`, `rolling_max_length_one_is_identity`, -`deque_matches_naive_window_max`, `lookback_is_one_window_lives_in_node_state`, -`length_must_be_at_least_one`, `label_carries_the_window` all pass (7 tests). - -(The module must compile first; if Step 1's tests are written before Step 2's -registration, `cargo test` cannot see the file — Step 2 is part of making them run.) - ---- - -### Task 2: `RollingMin` node - -**Files:** -- Create: `crates/aura-std/src/rolling_min.rs` -- Modify: `crates/aura-std/src/lib.rs:18-66` - -- [ ] **Step 1: Write the node file with its failing tests** - -Create `crates/aura-std/src/rolling_min.rs` — the mirror of `RollingMax`: an -**ascending** monotonic deque (pop the back while `v >= x`; front = current window -min), warm-up skip-emit, one `length` I64 param. - -```rust -//! `RollingMin` — minimum over the last `length` values of one f64 input. -//! -//! The mirror of `RollingMax`: an **ascending** monotonic deque (front = window min). -//! Each cycle pops from the back every element >= the new sample, pushes the new one, -//! and evicts the front once it leaves the window. See `rolling_max.rs` for the shared -//! rationale (O(1) amortized, window in node state, warm-up skip-emit, C7). - -use aura_core::{ - Cell, Ctx, FieldSpec, Firing, Node, NodeSchema, ParamSpec, PortSpec, PrimitiveBuilder, - ScalarKind, -}; -use std::collections::VecDeque; - -/// Minimum over the last `length` values of one f64 input, maintained by an ascending -/// monotonic deque (front = current window min). `None` during warm-up. -pub struct RollingMin { - length: usize, - deque: VecDeque<(f64, u64)>, // ascending; front = window min - seen: u64, - count: usize, - out: [Cell; 1], -} - -impl RollingMin { - /// Build a rolling min of window `length` (must be >= 1). - pub fn new(length: usize) -> Self { - assert!(length >= 1, "RollingMin length must be >= 1"); - Self { - length, - deque: VecDeque::with_capacity(length), - seen: 0, - count: 0, - out: [Cell::from_f64(0.0)], - } - } - - /// The param-generic recipe for a blueprint primitive: declares `length` and builds - /// through `RollingMin::new`. - pub fn builder() -> PrimitiveBuilder { - PrimitiveBuilder::new( - "RollingMin", - NodeSchema { - inputs: vec![PortSpec { kind: ScalarKind::F64, firing: Firing::Any, name: "series".into() }], - output: vec![FieldSpec { name: "value".into(), kind: ScalarKind::F64 }], - params: vec![ParamSpec { name: "length".into(), kind: ScalarKind::I64 }], - }, - |p| Box::new(RollingMin::new(p[0].i64() as usize)), - ) - } -} - -impl Node for RollingMin { - fn lookbacks(&self) -> Vec { - vec![1] - } - - fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> { - let w = ctx.f64_in(0); - if w.is_empty() { - return None; - } - let x = w[0]; - let idx = self.seen; - self.seen += 1; - // ascending invariant: a back element >= x can never again be the min — drop it. - while let Some(&(v, _)) = self.deque.back() { - if v >= x { - self.deque.pop_back(); - } else { - break; - } - } - self.deque.push_back((x, idx)); - while let Some(&(_, i)) = self.deque.front() { - if i + self.length as u64 <= idx { - self.deque.pop_front(); - } else { - break; - } - } - if self.count < self.length { - self.count += 1; - } - if self.count < self.length { - return None; - } - self.out[0] = Cell::from_f64(self.deque.front().expect("deque non-empty after push").0); - Some(&self.out) - } - - fn label(&self) -> String { - format!("RollingMin({})", self.length) - } -} - -#[cfg(test)] -mod tests { - use super::*; - use aura_core::{AnyColumn, Scalar, Timestamp}; - - fn drive(node: &mut RollingMin, feed: &[f64]) -> Vec> { - let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)]; - let mut out = Vec::new(); - for &v in feed { - inputs[0].push(Scalar::f64(v)).unwrap(); - out.push(node.eval(Ctx::new(&inputs, Timestamp(0))).map(|r| r[0].f64())); - } - out - } - - #[test] - fn rolling_min_warms_up_then_tracks_the_window_min() { - // min of [9,7,8], [7,8,5], [8,5,6]; silent for the first two. - let mut node = RollingMin::new(3); - let got = drive(&mut node, &[9.0, 7.0, 8.0, 5.0, 6.0]); - assert_eq!(got, vec![None, None, Some(7.0), Some(5.0), Some(5.0)]); - } - - #[test] - fn rolling_min_evicts_the_expiring_window_minimum() { - // length 2; the early small value must leave the window: [1,9]->1, [9,8]->8, [8,7]->7. - let mut node = RollingMin::new(2); - let got = drive(&mut node, &[1.0, 9.0, 8.0, 7.0]); - assert_eq!(got, vec![None, Some(1.0), Some(8.0), Some(7.0)]); - } - - #[test] - fn rolling_min_length_one_is_identity() { - let mut node = RollingMin::new(1); - assert_eq!(drive(&mut node, &[7.0, 9.0, 4.0]), vec![Some(7.0), Some(9.0), Some(4.0)]); - } - - #[test] - fn deque_matches_naive_window_min() { - let length = 37; - let mut node = RollingMin::new(length); - let series: Vec = - (0..3_000).map(|i| ((i * 1103515245 + 12345) % 1000) as f64 * 0.5 - 250.0).collect(); - let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)]; - for (t, &x) in series.iter().enumerate() { - inputs[0].push(Scalar::f64(x)).unwrap(); - let got = node.eval(Ctx::new(&inputs, Timestamp(0))).map(|r| r[0].f64()); - if t + 1 >= length { - let want = series[t + 1 - length..=t].iter().cloned().fold(f64::INFINITY, f64::min); - assert_eq!(got, Some(want), "at t={t}"); - } else { - assert_eq!(got, None, "silent until warmed up at t={t}"); - } - } - } - - #[test] - fn lookback_is_one_window_lives_in_node_state() { - assert_eq!(RollingMin::new(20).lookbacks(), vec![1]); - } - - #[test] - fn length_must_be_at_least_one() { - let r = std::panic::catch_unwind(|| RollingMin::new(0)); - assert!(r.is_err(), "RollingMin::new(0) must panic"); - } - - #[test] - fn label_carries_the_window() { - assert_eq!(RollingMin::new(20).label(), "RollingMin(20)"); - } -} -``` - -- [ ] **Step 2: Register the module + re-export in `lib.rs`** - -In `crates/aura-std/src/lib.rs`, add `mod rolling_min;` (after `mod rolling_max;`) -and `pub use rolling_min::RollingMin;` (after `pub use rolling_max::RollingMax;`). - -- [ ] **Step 3: Run the tests to verify they pass** - -Run: `cargo test -p aura-std rolling_min` -Expected: PASS — 7 tests (`rolling_min_warms_up_then_tracks_the_window_min`, -`rolling_min_evicts_the_expiring_window_minimum`, `rolling_min_length_one_is_identity`, -`deque_matches_naive_window_min`, `lookback_is_one_window_lives_in_node_state`, -`length_must_be_at_least_one`, `label_carries_the_window`). - ---- - -### Task 3: Breakout signal composition test (C2 causality + ±1 latch) - -Validates the breakout *composition* — `close → Delay(1) → {RollingMax,RollingMin}(N) -→ {Gt,Gt} → {Latch,Latch} → Sub` — directly via `GraphBuilder`/`Harness`, using the -new nodes (Tasks 1-2) + existing `Delay`/`Gt`/`Latch`/`Sub`. Independent of the CLI -binary (which cannot be imported by an integration test). Pins the two load-bearing -acceptance properties: C2 (the channel excludes the current bar) and the held ±1 -direction. New file name avoids the existing unrelated `ger40_breakout.rs`. - -**Files:** -- Create: `crates/aura-engine/tests/stage1_breakout_e2e.rs` - -- [ ] **Step 1: Write the failing composition test** - -Create `crates/aura-engine/tests/stage1_breakout_e2e.rs`: - -```rust -//! Breakout signal composition: close -> Delay(1) -> {RollingMax,RollingMin}(N) -> -//! {Gt,Gt} -> {Latch,Latch} -> Sub = bias in {-1,0,+1}. Tests the C2 causality guard -//! (the channel excludes the current bar via Delay(1)) and the held ±1 direction, -//! built straight from aura-std nodes (no CLI dependency). - -use aura_core::{Scalar, Timestamp}; -use aura_engine::{GraphBuilder, Harness, Source, VecSource}; -use aura_std::{Delay, Gt, Latch, Recorder, RollingMax, RollingMin, Sub}; -use std::sync::mpsc; - -// A minimal single-f64 source role "price" fed into the breakout signal subgraph, -// tapping the Sub (bias) output through a Recorder. channel N = 3. -fn run_breakout_bias(closes: &[f64]) -> Vec<(Timestamp, f64)> { - let (tx, rx) = mpsc::channel(); - let mut g = GraphBuilder::new("breakout_sig"); - let delay = g.add(Delay::builder().bind("lag", Scalar::i64(1))); - let mx = g.add(RollingMax::builder().bind("length", Scalar::i64(3))); - let mn = g.add(RollingMin::builder().bind("length", Scalar::i64(3))); - let gt_up = g.add(Gt::builder()); - let gt_down = g.add(Gt::builder()); - let up_latch = g.add(Latch::builder()); - let down_latch = g.add(Latch::builder()); - let bias = g.add(Sub::builder()); - let rec = g.add(Recorder::builder(vec![aura_core::ScalarKind::F64], aura_core::Firing::Any, tx)); - let price = g.source_role("price", aura_core::ScalarKind::F64); - g.feed(price, [delay.input("series"), gt_up.input("a"), gt_down.input("b")]); - g.connect(delay.output("value"), mx.input("series")); - g.connect(delay.output("value"), mn.input("series")); - g.connect(mx.output("value"), gt_up.input("b")); // close > max(prior N) - g.connect(mn.output("value"), gt_down.input("a")); // min(prior N) > close - g.connect(gt_up.output("value"), up_latch.input("set")); - g.connect(gt_down.output("value"), up_latch.input("reset")); - g.connect(gt_down.output("value"), down_latch.input("set")); - g.connect(gt_up.output("value"), down_latch.input("reset")); - g.connect(up_latch.output("value"), bias.input("lhs")); - g.connect(down_latch.output("value"), bias.input("rhs")); - g.connect(bias.output("value"), rec.input("col[0]")); - let flat = g.build().expect("breakout signal wiring resolves").compile_with_params(&[]).expect("compiles"); - let mut h = Harness::bootstrap(flat).expect("bootstraps"); - let prices: Vec<(Timestamp, f64)> = closes.iter().enumerate().map(|(i, &c)| (Timestamp(i as i64), c)).collect(); - let src: Vec> = vec![Box::new(VecSource::new(prices))]; - h.run(src); - rx.try_iter().map(|(ts, row): (Timestamp, Vec)| (ts, row[0].as_f64())).collect() -} - -#[test] -fn breakout_is_causal_and_holds_pm1_direction() { - // closes: an up-break at bar 3, quiet hold, a down-break at bar 6. - // idx: 0 1 2 3 4 5 6 7 - let closes = [100.0, 101.0, 99.0, 102.0, 100.0, 100.0, 95.0, 100.0]; - // bar3: max(close[0..2])=101, close[3]=102 > 101 -> UP-break. (C2: a window INCLUDING - // bar3 would be max=102, and 102>102 strict is false -> no break; the break - // firing here proves the current bar is excluded.) - // bars4,5: no break -> bias HOLDS +1. - // bar6: min(close[3..5])=100, close[6]=95 < 100 -> DOWN-break -> bias flips to -1. - // bar7: no break -> HOLDS -1. - let got = run_breakout_bias(&closes); - let vals: Vec = got.iter().map(|(_, v)| *v).collect(); - // bias is emitted only once the channel warms up (Delay(1)+RollingMax(3) -> first at bar 3). - assert_eq!(vals, vec![1.0, 1.0, 1.0, -1.0, -1.0], "expected +1 held then -1 held; got {vals:?}"); -} -``` - -- [ ] **Step 2: Run the test to verify it passes** - -Run: `cargo test -p aura-engine --test stage1_breakout_e2e` -Expected: PASS — `breakout_is_causal_and_holds_pm1_direction` (1 test). The -`GraphBuilder`/`source_role`/`feed`/`connect`/`build` + `compile_with_params(&[])` + -`Harness::bootstrap` + `h.run(Vec>)` idiom is the one -`stage1_r_graph` / `run_stage1_r` use in `crates/aura-cli/src/main.rs` (NOT the raw -`FlatGraph`/`Edge`/`Target` wiring in `ger40_breakout.rs`); if a name differs, match -main.rs — the asserted bias sequence `[+1,+1,+1,-1,-1]` is the fixed contract. - ---- - -### Task 4: CLI breakout strategy (`stage1_breakout_graph`, family, `--strategy`/`--channel`) - -All main.rs changes land together: the `Strategy` variant, its parse + dispatch arms -(the dispatch `match` is exhaustive — rustc forces the arm, so the family it calls -must exist in the same task), the graph builder, the family, and the `--channel` flag -+ grid field. Ends with a green workspace build + the CLI seam test. - -**Files:** -- Modify: `crates/aura-cli/src/main.rs` (imports; enum 1286; parse 1336; dispatch 1422; grid 1123-1140 + 1345; new fns) -- Test: `crates/aura-cli/tests/cli_run.rs` - -- [ ] **Step 1: Extend imports** - -In `crates/aura-cli/src/main.rs`, add `Delay`, `Gt`, `Latch`, `RollingMax`, -`RollingMin` to the `use aura_std::{...}` block (~lines 30-33; `Sub` is already there, -but `Delay`/`Gt`/`Latch` are **not** — stage1_r_graph never used them), and add -`SweepPoint` to the `use aura_engine::{...}` block (~lines 19-23; `SweepFamily` is -already there, `SweepPoint` is not). - -- [ ] **Step 2: Add the `channel` grid field** - -In `crates/aura-cli/src/main.rs:1123-1140`, add a `channel: Vec` field to -`struct Stage1RGrid` and `channel: vec![1920]` to its `impl Default`: - -```rust -#[derive(Clone, Debug, PartialEq)] -struct Stage1RGrid { - fast: Vec, - slow: Vec, - stop_length: Vec, - stop_k: Vec, - channel: Vec, -} - -impl Default for Stage1RGrid { - fn default() -> Self { - Self { - fast: vec![2, 3], - slow: vec![6, 12], - stop_length: vec![STAGE1_R_STOP_LENGTH], - stop_k: vec![STAGE1_R_STOP_K], - channel: vec![1920], - } - } -} -``` - -(The two defaults tests at `main.rs:3113` and `:3154` compare against -`Stage1RGrid::default()` / a `.clone()`, not a struct literal, so they stay green with -the new field. The `stage1-r` family never reads `channel`, so its grid + goldens are -untouched.) - -- [ ] **Step 3: Add the `--channel` parse arm** - -In `crates/aura-cli/src/main.rs:1345-1348`, add to the flag match (alongside -`--fast`/`--slow`/`--stop-length`/`--stop-k`): - -```rust - "--channel" => grid.channel = parse_csv_list(value).map_err(|()| usage())?, -``` - -- [ ] **Step 4: Add the `Strategy::Stage1Breakout` variant + parse + dispatch arms** - -In `crates/aura-cli/src/main.rs`: - -`enum Strategy` (1286-1290): -```rust -enum Strategy { - SmaCross, - Momentum, - Stage1R, - Stage1Breakout, -} -``` - -parse arm (after the `"stage1-r"` arm at 1339): -```rust - "stage1-breakout" => Strategy::Stage1Breakout, -``` - -dispatch arm (after the `Strategy::Stage1R` arm at 1425): -```rust - Strategy::Stage1Breakout => stage1_breakout_sweep_family(persist.then_some(name), &data, grid), -``` - -- [ ] **Step 5: Add `stage1_breakout_graph`** - -In `crates/aura-cli/src/main.rs`, near `stage1_r_graph`, add (the broker / executor / -taps / reduce branch are copied verbatim from `stage1_r_graph` 1968-2026; only the -signal leg and the bound vol-stop differ): - -```rust -/// The stage1-r harness with its signal leg swapped for a Donchian channel breakout: -/// `close -> Delay(1) -> {RollingMax,RollingMin}(channel) -> {Gt,Gt} -> {Latch,Latch} -/// -> Sub = bias in {-1,0,+1}`. The one Delay(1) on close feeds both rolling nodes, so -/// each channel covers `close[t-N..t-1]` (the C2 guard: the current bar is excluded). -/// `channel = None` leaves the lengths open (unused today; the family binds them); the -/// stop is bound (defines R), identical downstream to stage1_r_graph. -fn stage1_breakout_graph( - tx_eq: mpsc::Sender<(Timestamp, Vec)>, - tx_ex: mpsc::Sender<(Timestamp, Vec)>, - tx_r: mpsc::Sender<(Timestamp, Vec)>, - tx_req: mpsc::Sender<(Timestamp, Vec)>, - channel: Option, - stop_length: i64, - stop_k: f64, - reduce: bool, -) -> Composite { - let mut g = GraphBuilder::new("stage1_breakout"); - // Donchian breakout signal leg. - let delay = g.add(Delay::builder().bind("lag", Scalar::i64(1))); - let mut mx_b = RollingMax::builder().named("channel_hi"); - let mut mn_b = RollingMin::builder().named("channel_lo"); - if let Some(n) = channel { - mx_b = mx_b.bind("length", Scalar::i64(n)); - mn_b = mn_b.bind("length", Scalar::i64(n)); - } - let mx = g.add(mx_b); - let mn = g.add(mn_b); - let gt_up = g.add(Gt::builder()); - let gt_down = g.add(Gt::builder()); - let up_latch = g.add(Latch::builder()); - let down_latch = g.add(Latch::builder()); - let exposure = g.add(Sub::builder()); // up_latch - down_latch -> bias in {-1,0,+1} - // pip branch (verbatim from stage1_r_graph). - let broker = g.add(SimBroker::builder(SYNTHETIC_PIP_SIZE)); - let gate_col = PM_FIELD_NAMES - .iter() - .position(|&n| n == "closed_this_cycle") - .expect("PM record has a closed_this_cycle column"); - let eq = if reduce { - g.add(SeriesReducer::builder(Firing::Any, tx_eq)) - } else { - g.add(Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_eq)) - }; - let ex = if reduce { - g.add(SeriesReducer::builder(Firing::Any, tx_ex)) - } else { - g.add(Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_ex)) - }; - let exec = g.add(risk_executor(StopRule::Vol { length: stop_length, k: stop_k }, 1.0)); - let rrec = if reduce { - g.add(GatedRecorder::builder(PM_RECORD_KINDS.to_vec(), gate_col, Firing::Any, tx_r)) - } else { - g.add(Recorder::builder(PM_RECORD_KINDS.to_vec(), Firing::Any, tx_r)) - }; - let price = g.source_role("price", ScalarKind::F64); - g.feed( - price, - [ - delay.input("series"), - gt_up.input("a"), - gt_down.input("b"), - broker.input("price"), - exec.input("price"), - ], - ); - g.connect(delay.output("value"), mx.input("series")); - g.connect(delay.output("value"), mn.input("series")); - g.connect(mx.output("value"), gt_up.input("b")); - g.connect(mn.output("value"), gt_down.input("a")); - g.connect(gt_up.output("value"), up_latch.input("set")); - g.connect(gt_down.output("value"), up_latch.input("reset")); - g.connect(gt_down.output("value"), down_latch.input("set")); - g.connect(gt_up.output("value"), down_latch.input("reset")); - g.connect(up_latch.output("value"), exposure.input("lhs")); - g.connect(down_latch.output("value"), exposure.input("rhs")); - g.connect(exposure.output("value"), broker.input("exposure")); - g.connect(exposure.output("value"), ex.input("col[0]")); - g.connect(exposure.output("value"), exec.input("bias")); - g.connect(broker.output("equity"), eq.input("col[0]")); - for (i, field) in PM_FIELD_NAMES.iter().enumerate() { - g.connect(exec.output(field), rrec.input(COL_PORTS[i].as_str())); - } - if !reduce { - let r_equity = g.add( - LinComb::builder(2) - .bind("weights[0]", Scalar::f64(1.0)) - .bind("weights[1]", Scalar::f64(1.0)), - ); - let req = g.add(Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_req)); - g.connect(exec.output("cum_realized_r"), r_equity.input("term[0]")); - g.connect(exec.output("unrealized_r"), r_equity.input("term[1]")); - g.connect(r_equity.output("value"), req.input("col[0]")); - } - g.build().expect("stage1_breakout wiring resolves") -} -``` - -- [ ] **Step 6: Add `stage1_breakout_sweep_family`** - -In `crates/aura-cli/src/main.rs`, near `stage1_r_sweep_family`, add (the metrics -reduce/trace branch is copied verbatim from `stage1_r_sweep_family` 1232-1258): - -```rust -/// `aura sweep --strategy stage1-breakout`: sweep the breakout harness over a channel × -/// stop grid. One `channel` length drives BOTH rolling nodes (parameter-ganging, #61), -/// so the family iterates the cartesian product MANUALLY with a fully-bound graph per -/// point (compile_with_params(&[]) + Harness::bootstrap, like run_stage1_r) rather than -/// the open .axis/bootstrap_with_cells path. Each member folds the dense R-record via -/// summarize_r, so the family is rankable by sqn / expectancy_r / ... (parity with -/// stage1-r). With --trace, raw recorders persist the per-cycle streams. -fn stage1_breakout_sweep_family(trace: Option<&str>, data: &DataSource, grid: &Stage1RGrid) -> SweepFamily { - let pip = data.pip_size(); - let window = data.full_window(); - let mut varying: HashSet = HashSet::new(); - if grid.channel.len() > 1 { - varying.insert("channel".to_string()); - } - if grid.stop_length.len() > 1 { - varying.insert("stop_length".to_string()); - } - if grid.stop_k.len() > 1 { - varying.insert("stop_k".to_string()); - } - let mut points = Vec::new(); - for &c in &grid.channel { - for &sl in &grid.stop_length { - for &sk in &grid.stop_k { - let (tx_eq, rx_eq) = mpsc::channel(); - let (tx_ex, rx_ex) = mpsc::channel(); - let (tx_r, rx_r) = mpsc::channel(); - let (tx_req, rx_req) = mpsc::channel(); - let reduce = trace.is_none(); - let flat = stage1_breakout_graph(tx_eq, tx_ex, tx_r, tx_req, Some(c), sl, sk, reduce) - .compile_with_params(&[]) - .expect("valid stage1-breakout blueprint"); - let mut h = Harness::bootstrap(flat).expect("valid stage1-breakout harness"); - h.run(data.run_sources()); - let named: Vec<(String, Scalar)> = vec![ - ("channel".to_string(), Scalar::i64(c)), - ("stop_length".to_string(), Scalar::i64(sl)), - ("stop_k".to_string(), Scalar::f64(sk)), - ]; - let key = member_key(&named, &varying); - let mut manifest = sim_optimal_manifest(named, window, 0, pip); - manifest.broker = stage1_r_broker_label(pip); - let metrics = if reduce { - let r_rows: Vec<(Timestamp, Vec)> = rx_r.try_iter().collect(); - let (total_pips, max_drawdown) = rx_eq - .try_iter() - .next() - .map(|(_, row)| (row[0].as_f64(), row[1].as_f64())) - .unwrap_or((0.0, 0.0)); - let bias_sign_flips = - rx_ex.try_iter().next().map(|(_, row)| row[2].as_i64() as u64).unwrap_or(0); - let mut m = RunMetrics { total_pips, max_drawdown, bias_sign_flips, r: None }; - m.r = Some(summarize_r(&r_rows, 0.0)); - m - } else { - let eq_rows: Vec<(Timestamp, Vec)> = rx_eq.try_iter().collect(); - let ex_rows: Vec<(Timestamp, Vec)> = rx_ex.try_iter().collect(); - let r_rows: Vec<(Timestamp, Vec)> = rx_r.try_iter().collect(); - let req_rows: Vec<(Timestamp, Vec)> = rx_req.try_iter().collect(); - if let Some(name) = trace { - persist_traces_r(&format!("{name}/{key}"), &manifest, &eq_rows, &ex_rows, &req_rows); - } - let mut m = summarize(&f64_field(&eq_rows, 0), &f64_field(&ex_rows, 0)); - m.r = Some(summarize_r(&r_rows, 0.0)); - m - }; - points.push(SweepPoint { params: vec![], report: RunReport { manifest, metrics } }); - } - } - } - SweepFamily { space: vec![], points } -} -``` - -- [ ] **Step 7: Run the workspace build + suite to verify green** - -Run: `cargo build --workspace` -Expected: clean build, 0 errors (the exhaustive dispatch `match` now has its -`Stage1Breakout` arm; the new fns compile). - -Run: `cargo test --workspace` -Expected: PASS — all existing suites still green (stage1-r / sma / momentum goldens -byte-identical; the breakout adds a new path, touches no existing graph), plus the new -node + composition tests from Tasks 1-3. - -- [ ] **Step 8: Write the CLI seam test** - -In `crates/aura-cli/tests/cli_run.rs`, add (mirrors -`sweep_strategy_stage1_r_folded_no_trace_metrics_equal_raw_trace_metrics` at `:2000`; -reuses the in-file `BIN`, `temp_cwd`, `metrics_object`, `Command` already used there): - -```rust -/// The breakout strategy reaches the CLI seam: `aura sweep --strategy stage1-breakout` -/// emits an R-bearing member, and the folded no-trace path equals the raw --trace path -/// byte-for-byte (parity with the stage1-r fold-vs-raw guard). channel 3 warms up on -/// the synthetic stream; one channel value × the single default stop = one member. -#[test] -fn sweep_strategy_stage1_breakout_folded_no_trace_metrics_equal_raw_trace_metrics() { - let cwd = temp_cwd("sweep-stage1-breakout-fold-vs-raw"); - - let folded = Command::new(BIN) - .args(["sweep", "--strategy", "stage1-breakout", "--channel", "3"]) - .current_dir(&cwd) - .output() - .expect("spawn folded (no-trace) stage1-breakout sweep"); - assert!( - folded.status.success(), - "folded no-trace breakout sweep exit: {:?}; stderr: {}", - folded.status, - String::from_utf8_lossy(&folded.stderr) - ); - let folded_out = String::from_utf8(folded.stdout).expect("utf-8"); - - let raw = Command::new(BIN) - .args(["sweep", "--strategy", "stage1-breakout", "--channel", "3", "--trace", "b1"]) - .current_dir(&cwd) - .output() - .expect("spawn raw (--trace) stage1-breakout sweep"); - assert!( - raw.status.success(), - "raw --trace breakout sweep exit: {:?}; stderr: {}", - raw.status, - String::from_utf8_lossy(&raw.stderr) - ); - let raw_out = String::from_utf8(raw.stdout).expect("utf-8"); - - let folded_lines: Vec<&str> = folded_out.lines().collect(); - let raw_lines: Vec<&str> = raw_out.lines().collect(); - assert_eq!(folded_lines.len(), 1, "folded breakout sweep must print 1 member: {folded_out:?}"); - assert_eq!( - raw_lines.len(), - folded_lines.len(), - "member count must match: folded {} vs raw {}", - folded_lines.len(), - raw_lines.len() - ); - - for (i, (f, r)) in folded_lines.iter().zip(raw_lines.iter()).enumerate() { - let fm = metrics_object(f); - let rm = metrics_object(r); - assert!(fm.contains("\"r\":{"), "folded breakout member {i} must carry an r block: {fm}"); - assert_eq!( - fm, rm, - "folded vs raw breakout metrics diverge at member {i}\n folded: {fm}\n raw: {rm}" - ); - } - let _ = std::fs::remove_dir_all(&cwd); -} -``` - -- [ ] **Step 9: Run the seam test + final gates** - -Run: `cargo test -p aura-cli --test cli_run stage1_breakout` -Expected: PASS — the new `stage1-breakout` seam test(s). - -Run: `cargo test --workspace` -Expected: PASS (whole workspace). - -Run: `cargo clippy --workspace --all-targets -- -D warnings` -Expected: clean, no warnings. - ---- diff --git a/docs/specs/0071-stage1-breakout.md b/docs/specs/0071-stage1-breakout.md deleted file mode 100644 index 4498bcb..0000000 --- a/docs/specs/0071-stage1-breakout.md +++ /dev/null @@ -1,362 +0,0 @@ -# Stage-1 Donchian Breakout Candidate — Design Spec - -**Date:** 2026-06-25 -**Status:** Draft — awaiting user spec review -**Authors:** orchestrator + Claude - -> Reference issue: Brummel/Aura #137 (edge-research run log). The node-design -> forks F1–F5 are decided and recorded as a #137 comment; this spec produces them, -> it does not re-derive them. /boss autonomous within the active "Edge research" -> milestone. - -## Goal - -Add a **channel-breakout** signal candidate to the Stage-1 R research surface, so -the cross-symbol + temporal-OOS screen can be run on a *structurally different* -trend mechanic than the (now-refuted) SMA-cross momentum candidate. The -deliverable is a runnable - -``` -aura sweep --strategy stage1-breakout --real GER40 [--from ] [--to ] \ - --channel 480,960,1920,3840,7680 --stop-length 1920 --stop-k 2.0 -``` - -that emits the **same R-metrics block** (`expectancy_r`, `sqn`, `sqn_normalized`, -…) per grid member as `--strategy stage1-r`, so the breakout is screened under the -identical R yardstick and the only thing that differs from the momentum candidate -is the **signal leg**. - -Frictionless Stage-1 R (no costs). The breakout direction is the only new signal; -the vol-stop defines R exactly as in `stage1-r` (F3), giving a clean A/B. - -## Architecture - -The breakout is **the existing `stage1_r_graph` with its signal leg swapped**. In -`stage1_r_graph` the leg is `price → {fast SMA, slow SMA} → Sub(spread) → -Bias(0.5) → bias`. The breakout leg replaces that with a Donchian channel -direction: - -``` -price(close) ─┬───────────────────────────────────→ Gt_up.a , Gt_down.b (raw close[t]) - └→ Delay(1) ─┬→ RollingMax(N) → Gt_up.b (max close[t-N..t-1]) - └→ RollingMin(N) → Gt_down.a (min close[t-N..t-1]) - -Gt_up = (close[t] > max(close[t-N..t-1])) → up_break : bool (STRICT) -Gt_down = (min(close[t-N..t-1]) > close[t]) → down_break: bool (STRICT) - -up_latch = Latch(set = up_break, reset = down_break) → f64 {0,1} -down_latch = Latch(set = down_break, reset = up_break) → f64 {0,1} -bias = Sub(up_latch, down_latch) → f64 {-1, 0, +1} -``` - -`bias` then feeds `broker.exposure`, the exposure tap, and `exec.bias` — the -**identical seam** the SMA leg fed (`exposure.output("bias")`). Everything -downstream of the signal — `SimBroker`, the `RiskExecutor(vol_stop)`, the R-record, -the pip/eq/ex taps, the `reduce`-mode folding sinks — is **unchanged and shared** -with `stage1_r_graph`. - -**Why this composition (F2).** A breakout is a discrete regime flip held at the -extreme: `up_latch − down_latch` is `+1` after the last break was up, `−1` after -the last break was down, `0` before the first break. It is built from two existing -`Latch` nodes (set/reset, reset-dominant) plus a `Sub` — no new latch type is -needed. The two break legs cannot pulse on the same bar (a close cannot be both -strictly above the prior-N max and strictly below the prior-N min), so the -reset-dominant tie rule never fires. After a vol-stop exit while the latch still -holds `+1`, the executor re-enters in the same direction — desirable -trend-following behaviour, and the same executor composition the SMA path already -validated. - -**Why one `Delay(1)` on close, not two on the extrema (C2 — no look-ahead).** The -channel must exclude the current bar: `close[t]` is tested against -`max/min(close[t-N..t-1])`. Feeding one `Delay(1)` of `close` into both rolling -nodes makes each rolling output at cycle `t` cover `close[t-N..t-1]` (its newest -input is `close[t-1]`), so the comparison never sees `close[t]` on both sides. -Omitting the `Delay(1)` would let the rolling window include `close[t]`, making -`close[t] > max(window)` essentially never true — a degenerate (not look-ahead) -signal. The `Delay(1)` is the load-bearing causality guard and gets an explicit -test. - -**Why manual grid iteration, not the `.axis` param-space (F4 + ganging).** The -single `channel` length `N` drives **two** nodes (`RollingMax` and `RollingMin`). -Driving two slots from one swept value is parameter-ganging (#61), not yet -first-classed. So `stage1_breakout_sweep_family` iterates the grid **explicitly** -— for each `(channel, stop_length, stop_k)` point it builds a **fully-bound** -`stage1_breakout_graph` (both rolling nodes bound to the same `N`), -`compile_with_params(&[])` + `Harness::bootstrap`, runs, summarizes — exactly the -fully-bound path `run_stage1_r` already uses. This sidesteps the `.axis`/ganging -machinery entirely; the graph has no open slots. - -## Concrete code shapes - -### The user-facing program (the criterion's evidence) - -The researcher runs (and this is what the cycle delivers): - -```bash -# breakout screen, full history, GER40, channel grid, stop matched to slow signal -aura sweep --strategy stage1-breakout --real GER40 \ - --channel 480,960,1920,3840,7680 --stop-length 1920 --stop-k 2.0 \ - --name ger40-breakout - -# temporal OOS split (same as the momentum screen): IS then OOS -aura sweep --strategy stage1-breakout --real GER40 --to 1609459200000 \ - --channel 1920 --stop-length 1920 --stop-k 2.0 --name ger40-bo-is -aura sweep --strategy stage1-breakout --real GER40 --from 1609459200000 \ - --channel 1920 --stop-length 1920 --stop-k 2.0 --name ger40-bo-oos - -# rank by the R yardstick (same as stage1-r) -aura runs family rank sqn_normalized -``` - -Each member prints one JSON line whose `metrics.r` block is the same `RMetrics` -shape `stage1-r` emits (`expectancy_r`, `sqn`, `sqn_normalized`, `win_rate`, -`profit_factor`, `n_trades`, …), so the breakout is directly comparable and -rankable. - -### New node `RollingMax` (and its sibling `RollingMin`) — builder shape - -Mirrors `Sma`/`Delay` (node-owned window, `lookbacks()==[1]`, warm-up skip-emit, -`PrimitiveBuilder` with one `length` I64 param). The window statistic is the -**sliding maximum** maintained by a **monotonic (descending) deque** for O(1) -amortized updates — a per-cycle O(N) re-scan would be too slow at `N`≈7680 over -millions of bars. `RollingMin` is the mirror (ascending deque). Per the "operator -is topology" convention (see `gt.rs` docs) these are **two node types**, not one -node with a max/min param. - -```rust -// crates/aura-std/src/rolling_max.rs (rolling_min.rs is the mirror) -pub struct RollingMax { - length: usize, - // monotonic descending deque of (value, stream_index); front = current window max. - // node-owned, sized/bounded by `length` (C7, no per-cycle alloc on the hot path). - deque: VecDeque<(f64, u64)>, - seen: u64, // stream position, to evict the front when it exits the window - count: usize, // warm-up gate, silent until `length` (like Sma) - out: [Cell; 1], -} - -impl RollingMax { - pub fn new(length: usize) -> Self { /* assert length >= 1, mirror Sma::new */ } - pub fn builder() -> PrimitiveBuilder { - PrimitiveBuilder::new( - "RollingMax", - NodeSchema { - inputs: vec![PortSpec { kind: F64, firing: Any, name: "series".into() }], - output: vec![FieldSpec { name: "value".into(), kind: F64 }], - params: vec![ParamSpec { name: "length".into(), kind: I64 }], - }, - |p| Box::new(RollingMax::new(p[0].i64() as usize)), - ) - } -} - -impl Node for RollingMax { - fn lookbacks(&self) -> Vec { vec![1] } // window lives in node state - fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> { - let w = ctx.f64_in(0); - if w.is_empty() { return None; } - let x = w[0]; // newest (financial indexing) - // pop-back while back-value <= x (descending invariant); push (x, seen); - // pop-front while front index <= seen - length (evict out-of-window); - // bump seen/count; None until count >= length; else emit deque.front().value - // ... full body is the planner's - } -} -``` - -### `stage1_breakout_graph` — the swapped leg (vs `stage1_r_graph`) - -Same signature shape and the **same** broker / executor / taps / reduce branch as -`stage1_r_graph` (lines ~1944-2028), with the signal leg replaced. `channel: -Option` replaces `fast_len`/`slow_len`; when `Some(n)` it is bound to BOTH -rolling nodes. - -```rust -// before (stage1_r_graph signal leg) -let fast = g.add(Sma::builder().named("fast") /* +bind length */); -let slow = g.add(Sma::builder().named("slow") /* +bind length */); -let spread = g.add(Sub::builder()); -let exposure = g.add(Bias::builder().named("bias").bind("scale", Scalar::f64(0.5))); -g.feed(price, [fast.input("series"), slow.input("series"), broker.input("price"), exec.input("price")]); -g.connect(fast.output("value"), spread.input("lhs")); -g.connect(slow.output("value"), spread.input("rhs")); -g.connect(spread.output("value"), exposure.input("signal")); -let bias_out = exposure.output("bias"); - -// after (stage1_breakout_graph signal leg) — channel bound to BOTH rolling nodes -let delay = g.add(Delay::builder().bind("lag", Scalar::i64(1))); -let mut mx_b = RollingMax::builder().named("channel_hi"); -let mut mn_b = RollingMin::builder().named("channel_lo"); -if let Some(n) = channel { mx_b = mx_b.bind("length", Scalar::i64(n)); - mn_b = mn_b.bind("length", Scalar::i64(n)); } -let mx = g.add(mx_b); -let mn = g.add(mn_b); -let gt_up = g.add(Gt::builder()); // close[t] > max(prior N) -let gt_down = g.add(Gt::builder()); // min(prior N) > close[t] -let up_latch = g.add(Latch::builder()); -let down_latch = g.add(Latch::builder()); -let bias = g.add(Sub::builder()); // up_latch - down_latch -> {-1,0,+1} -g.feed(price, [delay.input("series"), gt_up.input("a"), gt_down.input("b"), - broker.input("price"), exec.input("price")]); -g.connect(delay.output("value"), mx.input("series")); -g.connect(delay.output("value"), mn.input("series")); -g.connect(mx.output("value"), gt_up.input("b")); -g.connect(mn.output("value"), gt_down.input("a")); -g.connect(gt_up.output("value"), up_latch.input("set")); -g.connect(gt_down.output("value"), up_latch.input("reset")); -g.connect(gt_down.output("value"), down_latch.input("set")); -g.connect(gt_up.output("value"), down_latch.input("reset")); -g.connect(up_latch.output("value"), bias.input("lhs")); -g.connect(down_latch.output("value"), bias.input("rhs")); -let bias_out = bias.output("value"); -// ... bias_out feeds broker.exposure, ex tap, exec.bias — VERBATIM from stage1_r_graph -``` - -### `--strategy` / `Strategy` enum / dispatch (before → after) - -```rust -// enum (main.rs ~1286) -enum Strategy { SmaCross, Momentum, Stage1R, Stage1Breakout } // + variant - -// parse (main.rs ~1336) -"stage1-r" => Strategy::Stage1R, -"stage1-breakout" => Strategy::Stage1Breakout, // + arm - -// dispatch (main.rs ~1422) -Strategy::Stage1R => stage1_r_sweep_family(persist.then_some(name), &data, grid), -Strategy::Stage1Breakout => stage1_breakout_sweep_family(persist.then_some(name), &data, grid), -``` - -### `--channel` grid flag - -A new `--channel ` flag (parsed exactly like `--fast`/`--slow` via -`parse_csv_list`) supplies the breakout channel-length axis. The breakout reuses -`--stop-length` / `--stop-k` for the vol-stop (same as `stage1-r`); `--fast` / -`--slow` are not used by the breakout strategy. The grid representation carries a -`channel: Vec` defaulting to `vec![1920]` (a single sensible value, so a -no-flag `aura sweep --strategy stage1-breakout` runs one member; the screen always -passes `--channel` explicitly). Lowest-friction representation: extend the existing -`Stage1RGrid` with a `channel` field, keeping one parser + one grid struct; the -planner may instead introduce a dedicated `Stage1BreakoutGrid` — either resolves -the same behaviour. The breakout family reads `channel` + `stop_length` + -`stop_k`; the `stage1-r` family ignores `channel` (its default does not perturb -the existing stage1-r grid, so stage1-r goldens are untouched). - -### `stage1_breakout_sweep_family` — manual cartesian iteration - -```rust -fn stage1_breakout_sweep_family(trace: Option<&str>, data: &DataSource, grid: &Stage1RGrid) - -> SweepFamily -{ - let pip = data.pip_size(); - let window = data.full_window(); - let varying = /* which of {channel, stop_length, stop_k} has > 1 value */; - let mut points = Vec::new(); - for &c in &grid.channel { - for &sl in &grid.stop_length { - for &sk in &grid.stop_k { - let reduce = trace.is_none(); - let (tx_eq, rx_eq) = mpsc::channel(); /* ex, r, req similarly */ - // fully-bound graph: channel = Some(c), stop bound to Vol{ length: sl, k: sk } - let flat = stage1_breakout_graph(tx_eq, tx_ex, tx_r, tx_req, Some(c), sl, sk, reduce) - .compile_with_params(&[]).expect("bound breakout blueprint"); - let mut h = Harness::bootstrap(flat).expect("valid breakout harness"); - h.run(data.run_sources()); - // reduce vs trace branch: VERBATIM the stage1_r_sweep_family logic - // (folded GatedRecorder/SeriesReducer rows -> RunMetrics + summarize_r, - // or raw recorders + persist_traces_r + summarize on --trace) - let manifest = /* sim_optimal_manifest with channel/stop_length/stop_k, broker label */; - points.push(/* RunReport { manifest, metrics } as a SweepFamily point */); - } - } - } - SweepFamily { points } -} -``` - -(`stage1_breakout_graph` takes the stop knobs as bound values, so it uses -`stop_open = false` internally and binds `StopRule::Vol { length: sl, k: sk }` — -the `risk_executor(..)` path, not `risk_executor_vol_open`.) - -## Components - -| Component | File | Status | -|---|---|---| -| `RollingMax` (sliding-window max, monotonic deque) | `crates/aura-std/src/rolling_max.rs` | NEW | -| `RollingMin` (sliding-window min, mirror) | `crates/aura-std/src/rolling_min.rs` | NEW | -| re-export both | `crates/aura-std/src/lib.rs` | MODIFY | -| `stage1_breakout_graph` | `crates/aura-cli/src/main.rs` | NEW | -| `stage1_breakout_sweep_family` | `crates/aura-cli/src/main.rs` | NEW | -| `Strategy::Stage1Breakout` + parse + dispatch | `crates/aura-cli/src/main.rs` | MODIFY | -| `--channel` flag + grid `channel` field | `crates/aura-cli/src/main.rs` | MODIFY | -| reused verbatim: `Delay`, `Gt`, `Latch`, `Sub`, `SimBroker`, `risk_executor`, the taps, the reduce-mode folding sinks | (existing) | REUSE | - -## Data flow - -Per cycle (after warm-up `N+1` bars): `close → Delay(1) → {RollingMax, RollingMin} -→ {Gt_up, Gt_down} → {up_latch, down_latch} → Sub = bias ∈ {−1,0,+1}`. `bias` → -`SimBroker.exposure` (pip equity tap), exposure tap, `RiskExecutor.bias`. The -`RiskExecutor` applies the vol-stop (defines R), emits the dense R-record → folded -by `GatedRecorder` (reduce) or raw `Recorder` (trace) → `summarize_r` → the `r` -block. Pip eq/ex taps fold via `SeriesReducer`/`summarize` exactly as today. -Identical to the `stage1-r` data flow from `bias` onward. - -## Error handling - -- `RollingMax::new` / `RollingMin::new` assert `length >= 1` (mirror `Sma`/`Delay`). -- A malformed `--channel 480,x` rejects to the subcommand `usage()` (via - `parse_csv_list`, exactly like `--fast`). -- Warm-up: each new node emits `None` until its window is full; `Gt` emits `None` - until both legs present; `Latch` emits `None` while both legs cold; so `bias` is - `None` (flat) until the first break after warm-up — no fabricated signal. -- An unknown `--strategy` value still rejects to `usage()` (existing path). - -## Testing strategy - -RED-first. New tests, all additive; **every existing golden** (stage1-r / sma / -momentum; single run + sweep + `--trace`) stays byte-identical (the breakout adds -a new strategy path and two new nodes; it touches no existing graph). - -1. **`RollingMax` / `RollingMin` correctness + warm-up** (unit, per node): - warms up silent for `length-1` samples, then emits the sliding max/min; mirror - `sma_warms_up_then_tracks_the_window_mean`. Include a length-1 identity case. -2. **Monotonic-deque == naive reference** (unit): drive a long deterministic noisy - series through the node and a reference O(N) per-window `max`/`min`; assert - equality at every warmed cycle (the deque must not drift from the true window - extremum). Mirror `incremental_matches_full_resum_within_tolerance`. -3. **C2 causality — the channel excludes the current bar** (integration, the - load-bearing test): on a crafted close series with a single new all-time high - at bar `t`, assert the up-break fires **at `t`** (close[t] > max(close[t-N..t-1])) - and that a graph wired WITHOUT the `Delay(1)` would NOT fire on that bar — i.e. - pin that the comparison is against `max(close[t-N..t-1])`, never a window that - includes `close[t]`. -4. **Breakout direction latch ±1 hold** (integration): a series with an up-break - then quiet bars then a down-break yields `bias` = `+1` held across the quiet - bars, flipping to `−1` on the down-break, `0` before the first break. -5. **CLI seam** (cli_run-style): `aura sweep --strategy stage1-breakout --real - --channel ` produces a member whose `metrics.r` block is present - and has the `RMetrics` fields (parity with the stage1-r sweep seam test); and - that folded-no-trace metrics equal raw-`--trace` metrics byte-for-byte (mirror - the 0070 equivalence test). -6. **Existing goldens unchanged**: `cargo test --workspace` green; the stage1-r / - sma / momentum report goldens byte-identical. - -Final gates: `cargo test --workspace` and `cargo clippy --workspace ---all-targets -- -D warnings`. - -## Acceptance criteria - -- `aura sweep --strategy stage1-breakout --real GER40 --channel - [--from][--to] [--stop-length][--stop-k]` runs and prints one R-bearing - `RunReport` JSON line per channel (× stop) grid point, rankable via `aura runs - family rank `. -- The breakout signal is causal (C2): `close[t]` is compared only against - `max/min(close[t-N..t-1])` — proven by test 3. -- Invariants preserved: C1 (deterministic synchronous run; manual iteration builds - disjoint fully-bound harnesses), C2 (Delay(1) channel exclusion), C7 (node-owned - deque/ring state, no `Rc`/`RefCell`), C8 (each new node one output, emits nothing - when cold). -- All pre-existing goldens byte-identical; workspace tests + clippy green. -- The deliverable enables the next research step: a cross-symbol (GER40 + FRA40) - and temporal-OOS breakout screen under the identical R yardstick used for the - refuted momentum candidate.