From d482f02d04f1eb751fb8ee7ce2abf88dd228b77d Mon Sep 17 00:00:00 2001 From: Brummel Date: Thu, 25 Jun 2026 12:31:27 +0200 Subject: [PATCH] plan: 0071 stage1 breakout candidate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implementation plan for the Donchian breakout candidate (parent spec docs/specs/0071-stage1-breakout.md). Four dependency-ordered tasks: RollingMax and RollingMin (monotonic-deque sliding extrema, mirroring the Sma/Delay node shape), a C2-causality + ±1-latch composition test, and the CLI breakout strategy (stage1_breakout_graph + manual-grid sweep family + --strategy/--channel wiring). refs #137 --- docs/plans/0071-stage1-breakout.md | 899 +++++++++++++++++++++++++++++ 1 file changed, 899 insertions(+) create mode 100644 docs/plans/0071-stage1-breakout.md diff --git a/docs/plans/0071-stage1-breakout.md b/docs/plans/0071-stage1-breakout.md new file mode 100644 index 0000000..73f54ff --- /dev/null +++ b/docs/plans/0071-stage1-breakout.md @@ -0,0 +1,899 @@ +# 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. + +---