diff --git a/crates/aura-std/src/lib.rs b/crates/aura-std/src/lib.rs index 8429824..46c540c 100644 --- a/crates/aura-std/src/lib.rs +++ b/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; diff --git a/crates/aura-std/src/stop_rule.rs b/crates/aura-std/src/stop_rule.rs index d07de74..6c6c723 100644 --- a/crates/aura-std/src/stop_rule.rs +++ b/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, diff --git a/crates/aura-std/src/vol_tf_stop.rs b/crates/aura-std/src/vol_tf_stop.rs new file mode 100644 index 0000000..65fe825 --- /dev/null +++ b/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, + close_in_bucket: f64, + prev_close: Option, + 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 { + 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> { + 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}"); + } +}