Aura/crates/aura-strategy/src/vol_tf_stop.rs

//! `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.

/// Nanoseconds per minute — the bucket-id divisor's one scale factor.
const NS_PER_MINUTE: i64 = 60 * 1_000_000_000;

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 * NS_PER_MINUTE,
            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 },
                ],
                doc: "volatility- and timeframe-scaled protective stop rule",
            },
            |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().snap_to_nearest_minute().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 / NS_PER_MINUTE,
            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 * NS_PER_MINUTE)))
                    .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)");
    }

    /// Property (#280): bucket membership follows a price sample's NOMINAL minute,
    /// not the raw provider stamp. Provider M1 stamps carry sub-second jitter
    /// around minute boundaries; a sample whose nominal minute is a bucket boundary
    /// but stamped a fraction of a second early must open the new bucket —
    /// completing the previous one — rather than folding back into it. Mis-bucketing
    /// the boundary sample suppresses the rollover, so the stop emission over the
    /// true bucket closes never fires.
    #[test]
    fn buckets_a_boundary_sample_by_its_nominal_minute_despite_subsecond_jitter() {
        // 5m buckets: boundaries at 0, 5min, 10min (multiples of period_ns).
        let period_ns = 5 * NS_PER_MINUTE; // 300_000_000_000
        let mut n = VolTfStop::new(5, 1, 2.0);
        let mut col = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)];
        let mut feed = |ns: i64, price: f64| {
            col[0].push(Scalar::f64(price)).unwrap();
            n.eval(Ctx::new(&col, Timestamp(ns))).map(|c| c[0].f64())
        };

        // Bucket 0 opens with a clean stamp; its running close is 100.0.
        assert_eq!(feed(0, 100.0), None);
        // The next sample's NOMINAL minute is 05:00 (the bucket-0/bucket-1
        // boundary) but the provider stamped it 600ms early at 04:59:59.400. By its
        // nominal minute it opens bucket 1 (close 105.0), rolling over bucket 0 and
        // seeding prev_close = 100.0 (first completed bucket: no delta yet).
        assert_eq!(feed(period_ns - 600_000_000, 105.0), None);
        // Nominal 10:00 rolls over bucket 1: Δ = 105 − 100, stop = k·|Δ| = 2·5 = 10.
        assert_eq!(
            feed(2 * period_ns, 108.0),
            Some(10.0),
            "the jitter-early boundary sample must roll over bucket 0 so this \
             rollover emits k·|Δ| over the TRUE bucket closes (100→105)"
        );
    }

    /// 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}");
    }
}