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

audit(0071): cycle close — breakout candidate refuted; drop ephemera

Browse Source 
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
This commit is contained in:
Brummel
2026-06-25 13:33:21 +02:00
parent d845c509d3
commit 4f88d67543
2 changed files with 0 additions and 1261 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/plans/0071-stage1-breakout.md
-899
View File
@@ -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<usize> {
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<Option<f64>> {
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<f64> =
(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<usize> {
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<Option<f64>> {
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<f64> =
(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<Box<dyn Source>> = vec![Box::new(VecSource::new(prices))];
h.run(src);
rx.try_iter().map(|(ts, row): (Timestamp, Vec<Scalar>)| (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<f64> = 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<Box<dyn Source>>)` 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<i64>` field to
`struct Stage1RGrid` and `channel: vec![1920]` to its `impl Default`:
```rust
#[derive(Clone, Debug, PartialEq)]
struct Stage1RGrid {
fast: Vec<i64>,
slow: Vec<i64>,
stop_length: Vec<i64>,
stop_k: Vec<f64>,
channel: Vec<i64>,
}
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<Scalar>)>,
tx_ex: mpsc::Sender<(Timestamp, Vec<Scalar>)>,
tx_r: mpsc::Sender<(Timestamp, Vec<Scalar>)>,
tx_req: mpsc::Sender<(Timestamp, Vec<Scalar>)>,
channel: Option<i64>,
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<String> = 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<Scalar>)> = 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<Scalar>)> = rx_eq.try_iter().collect();
let ex_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_ex.try_iter().collect();
let r_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_r.try_iter().collect();
let req_rows: Vec<(Timestamp, Vec<Scalar>)> = 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.
---
docs/specs/0071-stage1-breakout.md
-362
View File
@@ -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 <ms>] [--to <ms>] \
--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 <id> 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<usize> { 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<i64>` 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 <csv>` 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<i64>` 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
<fixture> --channel <n>` 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 <csv>
[--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 <id> rank <r-metric>`.
- 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.