feat(std): VolTfStop — the timescale-matched volatility stop primitive

k · sqrt(EMA(delta^2, length)) over completed period_minutes buckets —
the vol regime on a coarser clock, as a fused primitive in the Resample
mold: the in-progress bucket lives in node state, the finished bucket's
close is differenced against the previous one on rollover, and the stop
distance is emitted ONCE per completed bucket (C2: a partial bucket
never exists downstream; Firing::Any consumers hold the last value).
EMA follows Ema's exact convention (SMA seed over the first length
deltas, alpha = 2/(length+1)), pinned by a constant-|delta|
correspondence test against the per-cycle vol regime. Not rostered in
the std vocabulary (stop primitives are risk-executor internals).

First slice of #262; the regime variants, executor arm, and manifest
round-trip follow.

refs #262

crates/aura-std/src/lib.rs

@@ -54,6 +54,7 @@ mod stop_rule;
 mod sub;
 mod vocabulary;
 mod vol_slippage_cost;
+mod vol_tf_stop;
 pub use abs::Abs;
 pub use add::Add;
 pub use and::And;
@@ -97,3 +98,4 @@ pub use stop_rule::FixedStop;
 pub use sub::Sub;
 pub use vocabulary::{std_vocabulary, std_vocabulary_types};
 pub use vol_slippage_cost::VolSlippageCost;
+pub use vol_tf_stop::VolTfStop;

crates/aura-std/src/stop_rule.rs

@@ -2,13 +2,16 @@
 //! 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 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.
+//! `FixedStop` and `VolTfStop` (`vol_tf_stop.rs`) are the fused stop-rule
+//! PRIMITIVES: `FixedStop` is 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); `VolTfStop` folds
+//! `k · √EMA(Δ², length)` over completed timescale-matched buckets. The
+//! per-cycle volatility stop remains 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,

crates/aura-std/src/vol_tf_stop.rs

@@ -0,0 +1,214 @@
+//! `VolTfStop` — timescale-matched volatility stop (#262):
+//! `k · √EMA(Δ², length)` over completed `period_minutes` buckets — the vol
+//! regime on a coarser clock. A fused primitive in the `Resample` mold: the
+//! in-progress bucket lives in node state; on rollover the finished bucket's
+//! close is differenced against the previous bucket's close, Δ² folds into an
+//! `Ema`-convention EMA (SMA seed over the first `length` deltas,
+//! `alpha = 2/(length+1)`), and the stop distance is emitted ONCE (C2: a
+//! partial bucket never exists downstream). Mid-bucket the node emits
+//! nothing, so the `Firing::Any` consumers (`Sizer`, `PositionManagement`)
+//! hold the last completed bucket's value; the R-latch samples at entry.
+
+use aura_core::{
+    Cell, Ctx, FieldSpec, Firing, Node, NodeSchema, ParamSpec, PortSpec, PrimitiveBuilder,
+    ScalarKind,
+};
+
+pub struct VolTfStop {
+    period_ns: i64,
+    k: f64,
+    // EMA(Δ²) over completed buckets — `Ema`'s exact convention.
+    alpha: f64,
+    length: usize,
+    seeded: bool,
+    warmup_sum: f64,
+    count: usize,
+    ema: f64,
+    // Bucket accumulator (`Resample`'s mold): current id, its running close,
+    // and the previous COMPLETED bucket's close.
+    bucket: Option<i64>,
+    close_in_bucket: f64,
+    prev_close: Option<f64>,
+    out: [Cell; 1],
+}
+
+impl VolTfStop {
+    pub fn new(period_minutes: i64, length: i64, k: f64) -> Self {
+        assert!(period_minutes >= 1, "VolTfStop period_minutes must be >= 1");
+        assert!(length >= 1, "VolTfStop length must be >= 1");
+        assert!(k > 0.0, "VolTfStop k must be > 0");
+        Self {
+            period_ns: period_minutes * 60 * 1_000_000_000,
+            k,
+            alpha: 2.0 / (length as f64 + 1.0),
+            length: length as usize,
+            seeded: false,
+            warmup_sum: 0.0,
+            count: 0,
+            ema: 0.0,
+            bucket: None,
+            close_in_bucket: 0.0,
+            prev_close: None,
+            out: [Cell::from_f64(0.0)],
+        }
+    }
+
+    /// The param-generic recipe (#262): a graph builder adds this fused
+    /// primitive the same way it adds any other node — all three knobs are
+    /// structural (baked into the accumulator/EMA state at construction), so
+    /// `risk_executor`'s `VolTf` arm binds all three via `.bind(..)` (the
+    /// `FixedStop::builder().bind("distance", ..)` idiom, chained thrice).
+    pub fn builder() -> PrimitiveBuilder {
+        PrimitiveBuilder::new(
+            "VolTfStop",
+            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: "period_minutes".into(), kind: ScalarKind::I64 },
+                    ParamSpec { name: "length".into(), kind: ScalarKind::I64 },
+                    ParamSpec { name: "k".into(), kind: ScalarKind::F64 },
+                ],
+            },
+            |p| Box::new(VolTfStop::new(p[0].i64(), p[1].i64(), p[2].f64())),
+        )
+    }
+}
+
+impl Node for VolTfStop {
+    // The in-progress bucket lives in node state, not the input columns.
+    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; // fire with price (warm-up filter)
+        }
+        let price = w[0];
+        let bucket = ctx.now().0 / self.period_ns;
+        match self.bucket {
+            // First ever sample: open the accumulator, nothing complete yet.
+            None => {
+                self.bucket = Some(bucket);
+                self.close_in_bucket = price;
+                None
+            }
+            // Same bucket: the running close advances (last close wins).
+            Some(b) if bucket == b => {
+                self.close_in_bucket = price;
+                None
+            }
+            // Rollover: the finished bucket's close is final — difference it
+            // against the previous completed close, fold, emit once.
+            Some(_) => {
+                let finished_close = self.close_in_bucket;
+                self.bucket = Some(bucket);
+                self.close_in_bucket = price;
+                let Some(prev) = self.prev_close.replace(finished_close) else {
+                    return None; // first completed bucket: no delta yet
+                };
+                let d = finished_close - prev;
+                let d2 = d * d;
+                if !self.seeded {
+                    // SMA-seed over the first `length` deltas (Ema's convention).
+                    self.warmup_sum += d2;
+                    self.count += 1;
+                    if self.count < self.length {
+                        return None;
+                    }
+                    self.ema = self.warmup_sum / self.length as f64;
+                    self.seeded = true;
+                } else {
+                    // O(1) recurrence, exactly Ema's.
+                    self.ema += self.alpha * (d2 - self.ema);
+                }
+                self.out[0] = Cell::from_f64(self.k * self.ema.sqrt());
+                Some(&self.out)
+            }
+        }
+    }
+
+    fn label(&self) -> String {
+        format!(
+            "VolTfStop({}m,{},{})",
+            self.period_ns / (60 * 1_000_000_000),
+            self.length,
+            self.k
+        )
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use aura_core::{AnyColumn, Scalar, ScalarKind, Timestamp};
+
+    // Drive the node like the engine does: one eval per (minute, price) sample,
+    // mirroring the `feed`/`step` eval-harness idiom of `stop_rule.rs` /
+    // `resample.rs` — a capacity-1 ring column whose push overwrites the single
+    // slot, re-presenting the newest sample each cycle.
+    fn drive(node: &mut VolTfStop, samples: &[(i64, f64)]) -> Vec<Option<f64>> {
+        let mut col = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)];
+        samples
+            .iter()
+            .map(|&(ts_min, price)| {
+                col[0].push(Scalar::f64(price)).unwrap();
+                node.eval(Ctx::new(&col, Timestamp(ts_min * 60 * 1_000_000_000)))
+                    .map(|c| c[0].f64())
+            })
+            .collect()
+    }
+
+    /// Property (#262): mid-bucket cycles emit NOTHING — the stop exists only
+    /// at bucket rollover (C2: a partial bucket never reaches downstream).
+    #[test]
+    fn emits_only_on_bucket_rollover() {
+        let mut n = VolTfStop::new(60, 1, 2.0);
+        // Minutes 0..59 are one bucket; minute 60 rolls it over; 61..119 the
+        // next; 120 rolls again (first delta -> first emission).
+        let out = drive(
+            &mut n,
+            &[(0, 100.0), (30, 101.0), (59, 102.0), (60, 103.0), (90, 104.0), (120, 105.0)],
+        );
+        assert_eq!(out[0], None, "first sample opens the accumulator");
+        assert_eq!(out[1], None, "mid-bucket");
+        assert_eq!(out[2], None, "mid-bucket");
+        assert_eq!(out[3], None, "first rollover: a close exists, no delta yet");
+        assert_eq!(out[4], None, "mid-bucket");
+        assert!(out[5].is_some(), "second rollover: first delta -> first stop");
+    }
+
+    /// Property (#262): the emitted value is k · √EMA(Δ², length) over the
+    /// BUCKET closes. length=1 (alpha=1, the Ema identity) makes each stop
+    /// exactly k·|Δ| of the last completed bucket pair.
+    #[test]
+    fn stop_is_k_times_root_ema_of_bucket_close_deltas() {
+        let mut n = VolTfStop::new(60, 1, 2.0);
+        // Bucket closes: 102 (bucket 0), 104 (bucket 1), 109 (bucket 2).
+        let out = drive(
+            &mut n,
+            &[(0, 100.0), (59, 102.0), (61, 103.0), (119, 104.0), (121, 108.0), (179, 109.0), (181, 110.0)],
+        );
+        // Rollover at 61 (close 102, no delta), 121 (Δ=2 -> 2·|2|=4), 181 (Δ=5 -> 2·|5|=10).
+        assert_eq!(out[2], None, "first completed bucket seeds prev_close only");
+        assert_eq!(out[4], Some(4.0), "k·|Δ| with length=1: 2·(104−102)");
+        assert_eq!(out[6], Some(10.0), "2·(109−104)");
+    }
+
+    /// Correspondence pin (#262 acceptance): on strictly 1-minute-spaced
+    /// input with CONSTANT |Δ| per minute, vol_tf{1, length, k} converges to
+    /// exactly what the composite vol stop computes — k·|Δ| — the constant
+    /// input makes the one-cycle emission lag immaterial.
+    #[test]
+    fn one_minute_periods_match_the_vol_regime_on_constant_deltas() {
+        let mut n = VolTfStop::new(1, 3, 2.0);
+        // Alternating ±1 prices: every minute-delta has |Δ| = 1, Δ² = 1.
+        let samples: Vec<(i64, f64)> =
+            (0..12).map(|m| (m, if m % 2 == 0 { 100.0 } else { 101.0 })).collect();
+        let out = drive(&mut n, &samples);
+        let last = out.last().copied().flatten().expect("steady state reached");
+        assert!((last - 2.0).abs() < 1e-12, "k·√EMA(1) = 2.0, got {last}");
+    }
+}