refactor(stage1-r): the vol stop is a composition, not a fused node

Applies the principle the user named: a node is a primitive only if it is NOT
DAG-expressible from other primitives; a function that needs a missing primitive
gets the primitive added, not fused away.

- Remove the fused VolStop node (it was pure feed-forward arithmetic).
- The volatility stop is now a composition of primitives, vol_stop(length, k):
  k * Sqrt(Ema(Mul(d,d), length)), d = Sub(price, Delay(price,1)) -- a rolling
  EWMA standard deviation. Built with GraphBuilder; proven end-to-end
  (bootstraps + runs + emits k*sigma) in tests/vol_stop_composite.rs.
- Migrate the one VolStop caller (stage1_r_e2e.rs) off it: the R-is-stop-defined
  test now contrasts a tight vs a wide FixedStop (two constant distances still
  fold to different R, the property under test).

Uses the Mul + Sqrt primitives added in the prior commit. FixedStop stays the
only stop-rule primitive (a triggered constant). Full suite + clippy green.

refs #117 #119

crates/aura-engine/tests/stage1_r_e2e.rs

@@ -27,7 +27,7 @@
 
 use aura_core::{AnyColumn, Ctx, Node, Scalar, ScalarKind, Timestamp};
 use aura_engine::{RMetrics, summarize_r};
-use aura_std::{FixedStop, PM_FIELD_NAMES, PM_RECORD_KINDS, PM_WIDTH, PositionManagement, VolStop};
+use aura_std::{FixedStop, PM_FIELD_NAMES, PM_RECORD_KINDS, PM_WIDTH, PositionManagement};
 
 // The indices `report.rs::summarize_r` reads the dense record by. Re-declared here
 // (the consumer's `r_col` module is private to `aura-engine`) so the guard test can
@@ -119,26 +119,26 @@ fn long_path(prices: &[f64]) -> Vec<(f64, f64)> {
 /// distances. With a no-gap stop hit at exactly the stop level the fold returns
 /// exactly -1R *whatever the distance*, so to observe the stop-defines-R property
 /// end-to-end the price must overshoot the stop: then a wider FixedStop (distance
-/// 10) realises a shallower R-loss than a tight VolStop. A run that ignored the
-/// stop distance (e.g. measured R in raw pips) would collapse these to one number.
+/// 10) realises a shallower R-loss than a tight FixedStop (distance 1). A run that
+/// ignored the stop distance (e.g. measured R in raw pips) would collapse these.
 #[test]
 fn r_is_stop_defined_two_stops_fold_to_different_expectancy() {
     // Open long ~100, then a sharp drop that overshoots both stops.
     // FixedStop(10): stop @90, fill @85 -> R = (85-100)/10 = -1.5.
     let mut fixed = FixedStop::new(10.0);
     let wide = run_chain(&mut fixed, &long_path(&[100.0, 100.0, 85.0]));
-    // VolStop(1, 1.0): length-1 VolStop warms after one return; entry on the cycle
-    // after warm-up. returns: 100->101 (=1), warm; entry @101 with dist=1, stop @100;
-    // drop to 96. A unit (1.0) far tighter than the wide FixedStop unit (10.0).
-    let mut vol = VolStop::new(1, 1.0);
-    let vol_m = run_chain(&mut vol, &long_path(&[100.0, 101.0, 96.0]));
-    assert!(wide.n_trades >= 1 && vol_m.n_trades >= 1);
-    // R is the (exit-entry)/distance ratio: the tight VolStop unit (1.0) yields a far
-    // deeper R-loss than the wide FixedStop unit (10.0) for a comparable price drop.
+    // A TIGHT FixedStop (distance 1.0): entry @100, stop @99; drop to 96 overshoots,
+    // fill @96 -> R = (96-100)/1 = -4. A unit (1.0) far tighter than the wide unit (10.0),
+    // so for a comparable drop it folds to a ~10x deeper R-loss — R is stop-defined.
+    let mut tight = FixedStop::new(1.0);
+    let tight_m = run_chain(&mut tight, &long_path(&[100.0, 100.0, 96.0]));
+    assert!(wide.n_trades >= 1 && tight_m.n_trades >= 1);
+    // R is the (exit-entry)/distance ratio: the tight unit (1.0) yields a far deeper
+    // R-loss than the wide unit (10.0) for a comparable price drop.
     assert!(
-        vol_m.expectancy_r < wide.expectancy_r,
-        "stop choice must change folded R: vol={:?} wide={:?}",
-        vol_m.expectancy_r,
+        tight_m.expectancy_r < wide.expectancy_r,
+        "stop choice must change folded R: tight={:?} wide={:?}",
+        tight_m.expectancy_r,
         wide.expectancy_r,
     );
 }

crates/aura-engine/tests/vol_stop_composite.rs

@@ -0,0 +1,65 @@
+//! The volatility stop as a COMPOSITION of primitives (the post-correction design):
+//! `stop_distance = k · Sqrt(Ema(Mul(Δ,Δ), length))`, `Δ = Sub(price, Delay(price,1))`
+//! — a rolling EWMA standard deviation. This proves the decomposition wires through
+//! `GraphBuilder`, bootstraps, and computes the right number end-to-end — the principle
+//! that a node expressible as a DAG of primitives is a composition, not a fused node.
+use aura_core::{Firing, Scalar, ScalarKind, Timestamp};
+use aura_engine::{Composite, GraphBuilder, VecSource};
+use aura_std::{Delay, Ema, LinComb, Mul, Recorder, Sqrt, Sub};
+use std::sync::mpsc::channel;
+
+/// Build the volatility-stop composite: `price -> k · rolling-EWMA-stddev`. The
+/// `length`, `k`, and the `lag=1` register are bound at construction (no free params).
+fn vol_stop(length: i64, k: f64) -> Composite {
+    let mut g = GraphBuilder::new("vol_stop");
+    let price = g.input_role("price");
+    let delay = g.add(Delay::builder().bind("lag", Scalar::i64(1))); // z^-1: prev price
+    let sub = g.add(Sub::builder()); // Δ = price - prev
+    let sq = g.add(Mul::builder()); // Δ·Δ = Δ²
+    let ema = g.add(Ema::builder().bind("length", Scalar::i64(length))); // EWMA variance
+    let sqrt = g.add(Sqrt::builder()); // -> σ (price units)
+    let scale = g.add(LinComb::builder(1).bind("weights[0]", Scalar::f64(k))); // k·σ
+    g.feed(price, [delay.input("series"), sub.input("lhs")]);
+    g.connect(delay.output("value"), sub.input("rhs"));
+    g.connect(sub.output("value"), sq.input("lhs"));
+    g.connect(sub.output("value"), sq.input("rhs")); // square: feed Δ to both legs
+    g.connect(sq.output("value"), ema.input("series"));
+    g.connect(ema.output("value"), sqrt.input("value"));
+    g.connect(sqrt.output("value"), scale.input("term[0]"));
+    g.expose(scale.output("value"), "stop_distance");
+    g.build().expect("vol_stop composite wires")
+}
+
+/// An alternating ±1 price path makes `Δ² ≡ 1`, so the EWMA variance is `1`, `σ = 1`,
+/// and `stop_distance = k·σ = k`. With `k = 2.0` the warmed-up output is exactly `2.0`.
+#[test]
+fn vol_stop_emits_k_times_rolling_stddev() {
+    let (tx, rx) = channel();
+    let mut g = GraphBuilder::new("vol_stop_harness");
+    let price = g.source_role("price", ScalarKind::F64);
+    let vs = g.add(vol_stop(/*length*/ 3, /*k*/ 2.0));
+    let rec = g.add(Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx));
+    g.feed(price, [vs.input("price")]);
+    g.connect(vs.output("stop_distance"), rec.input("col[0]"));
+    let mut h = g
+        .build()
+        .expect("harness wires")
+        .bootstrap_with_params(vec![])
+        .expect("bootstraps");
+
+    let prices = [100.0, 101.0, 100.0, 101.0, 100.0, 101.0, 100.0, 101.0, 100.0, 101.0];
+    let stream: Vec<(Timestamp, Scalar)> = prices
+        .iter()
+        .enumerate()
+        .map(|(i, &p)| (Timestamp(i as i64), Scalar::f64(p)))
+        .collect();
+    h.run(vec![Box::new(VecSource::new(stream))]);
+
+    let rows: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
+    let last = rows.last().expect("vol_stop produced output after warm-up");
+    assert!(
+        (last.1[0].as_f64() - 2.0).abs() < 1e-9,
+        "expected stop_distance = k·σ = 2.0, got {}",
+        last.1[0].as_f64()
+    );
+}

crates/aura-std/src/lib.rs

@@ -56,5 +56,5 @@ pub use session::Session;
 pub use sim_broker::SimBroker;
 pub use sma::Sma;
 pub use sqrt::Sqrt;
-pub use stop_rule::{FixedStop, VolStop};
+pub use stop_rule::FixedStop;
 pub use sub::Sub;

crates/aura-std/src/stop_rule.rs

@@ -1,17 +1,24 @@
 //! Stop-rule nodes — emit a protective-stop *distance* (price units, ≥ 0,
 //! direction-agnostic). The stop DEFINES the risk unit R (1R = the loss if stopped);
 //! position-management latches the entry-cycle distance as the frozen R-denominator.
-//! `FixedStop` is a constant distance (test fixture / structural-axis sibling);
-//! `VolStop` is a close-to-close volatility stop `k * EMA_length(|price - prev_price|)`
-//! (true-range ATR is deferred — it needs OHLC). One fused node each (no Abs/Mul
-//! primitive exists, so abs+delta+EMA fuse inside VolStop).
+//!
+//! `FixedStop` is the only stop-rule PRIMITIVE: a constant distance gated on its price
+//! input (a source-less `Const` has no firing trigger in the push model, so the
+//! triggered-constant shape is the honest primitive). The volatility stop is NOT a
+//! primitive — it is a COMPOSITION of primitives, `k · Sqrt(Ema(Mul(Δ,Δ), length))`
+//! with `Δ = Sub(price, Delay(price,1))` (a rolling EWMA standard deviation), built with
+//! `GraphBuilder` — see `crates/aura-engine/tests/vol_stop_composite.rs`. True-range ATR
+//! (a richer stop) is deferred — it needs OHLC.
 use aura_core::{
     Cell, Ctx, FieldSpec, Firing, Node, NodeSchema, ParamSpec, PortSpec, PrimitiveBuilder,
     ScalarKind,
 };
 
-/// Constant stop distance. `distance` must be > 0.
-pub struct FixedStop { distance: f64, out: [Cell; 1] }
+/// Constant stop distance, gated on a price input. `distance` must be > 0.
+pub struct FixedStop {
+    distance: f64,
+    out: [Cell; 1],
+}
 impl FixedStop {
     pub fn new(distance: f64) -> Self {
         assert!(distance > 0.0, "FixedStop distance must be > 0");
@@ -30,74 +37,19 @@ impl FixedStop {
     }
 }
 impl Node for FixedStop {
-    fn lookbacks(&self) -> Vec<usize> { vec![1] }
+    fn lookbacks(&self) -> Vec<usize> {
+        vec![1]
+    }
     fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> {
-        if ctx.f64_in(0).is_empty() { return None; } // fire with price (warm-up filter)
+        if ctx.f64_in(0).is_empty() {
+            return None; // fire with price (warm-up filter)
+        }
         self.out[0] = Cell::from_f64(self.distance);
         Some(&self.out)
     }
-    fn label(&self) -> String { format!("FixedStop({})", self.distance) }
-}
-
-/// Volatility stop distance: `k * EMA_length(|price - prev_price|)`. EMA is the
-/// standard `alpha = 2/(length+1)` recursion, but its WARM-UP DIFFERS from the
-/// crate's [`crate::Ema`]: that node seeds with the SMA of its first `length` samples
-/// (ta-lib convention), whereas `VolStop` seeds with the FIRST abs-return and runs the
-/// recurrence from there (a first-value seed). The divergence is deliberate — the
-/// abs-return stream needs a prior price before any sample exists, so a uniform
-/// `length`-sample SMA seed would push warm-up an extra cycle out and complicate the
-/// frozen-R latch; the first-value seed keeps the stop available as early as the data
-/// allows. Output is `None` until `length` abs-returns have been seen (warm-up).
-/// `prev_price` updated AFTER use (intra-node z⁻¹, C2-clean). `length >= 1`, `k > 0`.
-pub struct VolStop {
-    length: usize,
-    k: f64,
-    prev_price: Option<f64>,
-    ema: f64,
-    count: usize,
-    out: [Cell; 1],
-}
-impl VolStop {
-    pub fn new(length: usize, k: f64) -> Self {
-        assert!(length >= 1, "VolStop length must be >= 1");
-        assert!(k > 0.0, "VolStop k must be > 0");
-        Self { length, k, prev_price: None, ema: 0.0, count: 0, out: [Cell::from_f64(0.0)] }
+    fn label(&self) -> String {
+        format!("FixedStop({})", self.distance)
     }
-    pub fn builder() -> PrimitiveBuilder {
-        PrimitiveBuilder::new(
-            "VolStop",
-            NodeSchema {
-                inputs: vec![PortSpec { kind: ScalarKind::F64, firing: Firing::Any, name: "price".into() }],
-                output: vec![FieldSpec { name: "stop_distance".into(), kind: ScalarKind::F64 }],
-                params: vec![
-                    ParamSpec { name: "length".into(), kind: ScalarKind::I64 },
-                    ParamSpec { name: "k".into(), kind: ScalarKind::F64 },
-                ],
-            },
-            |p| Box::new(VolStop::new(p[0].i64() as usize, p[1].f64())),
-        )
-    }
-}
-impl Node for VolStop {
-    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 price = w[0];
-        let Some(pp) = self.prev_price else {
-            self.prev_price = Some(price);
-            return None; // need a prior price to form the first abs-return
-        };
-        let d = (price - pp).abs();
-        let alpha = 2.0 / (self.length as f64 + 1.0);
-        if self.count == 0 { self.ema = d; } else { self.ema += alpha * (d - self.ema); }
-        self.count += 1;
-        self.prev_price = Some(price); // update AFTER use (C2)
-        if self.count < self.length { return None; } // warm-up
-        self.out[0] = Cell::from_f64(self.k * self.ema);
-        Some(&self.out)
-    }
-    fn label(&self) -> String { format!("VolStop({},{})", self.length, self.k) }
 }
 
 #[cfg(test)]
@@ -118,21 +70,4 @@ mod tests {
         let mut s = FixedStop::new(2.5);
         assert_eq!(feed(&mut s, &[100.0, 110.0, 90.0]), vec![Some(2.5), Some(2.5), Some(2.5)]);
     }
-    #[test]
-    fn vol_stop_is_none_until_warm() {
-        // length 3: needs a prior price (cycle 1 -> None) then 3 abs-returns.
-        let mut s = VolStop::new(3, 2.0);
-        let got = feed(&mut s, &[100.0, 101.0, 102.0, 103.0]);
-        assert_eq!(got[0], None); // no prior price
-        assert_eq!(got[1], None); // 1 return
-        assert_eq!(got[2], None); // 2 returns
-        assert!(got[3].is_some()); // 3 returns -> warm
-    }
-    #[test]
-    fn vol_stop_tracks_k_times_ema_abs_return() {
-        // constant abs-return of 1.0 -> EMA = 1.0 -> distance = k*1.0 = 2.0.
-        let mut s = VolStop::new(2, 2.0);
-        let got = feed(&mut s, &[100.0, 101.0, 102.0, 103.0]);
-        assert_eq!(got.last().unwrap().unwrap(), 2.0);
-    }
 }