From 0e1eee61bd8b21c2ce5e77afb097ec5cf06ec714 Mon Sep 17 00:00:00 2001 From: Brummel Date: Wed, 17 Jun 2026 17:07:51 +0200 Subject: [PATCH] =?UTF-8?q?feat(aura-std):=20Resample=20=E2=80=94=20M1->co?= =?UTF-8?q?arse=20OHLC,=20C2=20emit-on-rollover?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Aggregates a fine 4-field OHLC stream into period_minutes buckets (open=first, high=max, low=min, close=last), emitting a completed bar ONLY on bucket rollover (ctx.now() crossing into the next bucket) — C2: a bar is actionable only once complete, partials are never emitted and the last partial bar is dropped (no EOF flush). Four Barrier(0) f64 inputs, a 4-field f64 record output (the first multi-field-output node in aura-std, mirroring the Ohlcv engine fixture). The bar carries no timestamp; the engine stamps the close-instant. Build-step 6 of milestone 'Strategy node vocabulary I'. closes #89 --- crates/aura-std/src/lib.rs | 2 + crates/aura-std/src/resample.rs | 244 ++++++++++++++++++++++++++++++++ 2 files changed, 246 insertions(+) create mode 100644 crates/aura-std/src/resample.rs diff --git a/crates/aura-std/src/lib.rs b/crates/aura-std/src/lib.rs index d52b667..1ab03b7 100644 --- a/crates/aura-std/src/lib.rs +++ b/crates/aura-std/src/lib.rs @@ -25,6 +25,7 @@ mod gt; mod latch; mod lincomb; mod recorder; +mod resample; mod sim_broker; mod sma; mod sub; @@ -38,6 +39,7 @@ pub use gt::Gt; pub use latch::Latch; pub use lincomb::LinComb; pub use recorder::Recorder; +pub use resample::Resample; pub use sim_broker::SimBroker; pub use sma::Sma; pub use sub::Sub; diff --git a/crates/aura-std/src/resample.rs b/crates/aura-std/src/resample.rs new file mode 100644 index 0000000..7d7e460 --- /dev/null +++ b/crates/aura-std/src/resample.rs @@ -0,0 +1,244 @@ +//! `Resample` — aggregates a fine OHLC stream (e.g. M1) into coarser OHLC bars +//! (e.g. 15-minute), emitting a **completed** bar ONLY on rollover (C2: a bar is +//! actionable only once complete — never a partial bar). +//! +//! **Schema.** FOUR `f64` inputs `open, high, low, close`, each +//! `Firing::Barrier(0)`: they are co-fresh, fed by four separate M1 sources at +//! the same M1 timestamp, so `Barrier(0)` makes the node fire exactly once when +//! all four carry that timestamp (the `Ohlcv` test fixture in `harness.rs` is the +//! precedent — the first multi-field-output + Barrier group in the engine). +//! Output is a 4-field record `open, high, low, close` (each `f64`). One param +//! `period_minutes: i64`. +//! +//! **Accumulator over the current bucket.** `open` = first M1 open of the window, +//! `high` = running max of M1 highs, `low` = running min of M1 lows, `close` = +//! last M1 close. +//! +//! **Emit-on-rollover via `ctx.now()`.** With +//! `period_ns = period_minutes * 60 * 1_000_000_000`, the bucket of an M1 instant +//! is `ctx.now().0 / period_ns` (integer division on the i64 epoch-ns). Per fired +//! eval (all 4 inputs present, Barrier-gated): +//! - **first ever sample:** start the accumulator, return `None`. +//! - **same bucket:** fold the M1 into the accumulator, return `None`. +//! - **rollover (`bucket > current_bucket`):** the accumulator is a COMPLETE bar +//! → emit `[open, high, low, close]`; THEN restart the accumulator from this +//! M1 and adopt the new bucket; return `Some(the completed bar)`. +//! +//! **Timestamp (C4).** The completed bar carries no timestamp of its own — the +//! engine stamps the emission cycle's `ctx.now()` (the new bucket's first M1 +//! instant, which is the close instant of the bar just emitted, spec 0050 §4.1). +//! The node only returns the 4 OHLC values. +//! +//! **Partial last bar is DROPPED.** There is no end-of-stream flush: a bucket with +//! no following rollover tick never emits. This is C2 again — an incomplete bar is +//! not actionable. + +use aura_core::{ + Cell, Ctx, FieldSpec, Firing, Node, NodeSchema, ParamSpec, PortSpec, PrimitiveBuilder, + ScalarKind, +}; + +/// Open/high/low/close accumulator for the current bucket, plus the bucket index +/// it belongs to. Held across cycles in node state (like `SimBroker`/`Latch`). +struct Acc { + bucket: i64, + open: f64, + high: f64, + low: f64, + close: f64, +} + +/// Resamples a fine 4-field OHLC stream into coarser OHLC bars, emitting a +/// completed bar only on bucket rollover (C2). The first multi-field-output node +/// in `aura-std`. +pub struct Resample { + period_ns: i64, + acc: Option, + out: [Cell; 4], +} + +impl Resample { + /// Build a `Resample` aggregating into `period_minutes`-wide buckets + /// (must be >= 1). + pub fn new(period_minutes: i64) -> Self { + assert!(period_minutes >= 1, "Resample period_minutes must be >= 1"); + Self { + period_ns: period_minutes * 60 * 1_000_000_000, + acc: None, + out: [Cell::from_f64(0.0); 4], + } + } + + /// The param-generic recipe for a blueprint primitive: declares + /// `period_minutes` and builds through `Resample::new`. + pub fn builder() -> PrimitiveBuilder { + PrimitiveBuilder::new( + "Resample", + NodeSchema { + inputs: vec![ + PortSpec { kind: ScalarKind::F64, firing: Firing::Barrier(0), name: "open".into() }, + PortSpec { kind: ScalarKind::F64, firing: Firing::Barrier(0), name: "high".into() }, + PortSpec { kind: ScalarKind::F64, firing: Firing::Barrier(0), name: "low".into() }, + PortSpec { kind: ScalarKind::F64, firing: Firing::Barrier(0), name: "close".into() }, + ], + output: vec![ + FieldSpec { name: "open".into(), kind: ScalarKind::F64 }, + FieldSpec { name: "high".into(), kind: ScalarKind::F64 }, + FieldSpec { name: "low".into(), kind: ScalarKind::F64 }, + FieldSpec { name: "close".into(), kind: ScalarKind::F64 }, + ], + params: vec![ParamSpec { name: "period_minutes".into(), kind: ScalarKind::I64 }], + }, + |p| Box::new(Resample::new(p[0].i64())), + ) + } +} + +impl Node for Resample { + fn lookbacks(&self) -> Vec { + // each of the four inputs is read at depth 1 (newest M1 sample only); the + // window-in-progress lives in node state, not in the input columns. + vec![1, 1, 1, 1] + } + + fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> { + // Barrier(0) guarantees all four arrive co-fresh; read the newest M1. + let (o, h, l, c) = (ctx.f64_in(0)[0], ctx.f64_in(1)[0], ctx.f64_in(2)[0], ctx.f64_in(3)[0]); + let bucket = ctx.now().0 / self.period_ns; + + match self.acc.take() { + // first ever sample: start the accumulator, nothing complete yet. + None => { + self.acc = Some(Acc { bucket, open: o, high: h, low: l, close: c }); + None + } + // same bucket: fold this M1 into the in-progress bar (still partial). + Some(mut acc) if bucket == acc.bucket => { + acc.high = acc.high.max(h); + acc.low = acc.low.min(l); + acc.close = c; // last close wins + self.acc = Some(acc); + None + } + // rollover: the accumulated bar is COMPLETE -> emit it, then restart + // the accumulator from this M1 in the new bucket (C2 emit-on-rollover). + Some(acc) => { + self.out = [ + Cell::from_f64(acc.open), + Cell::from_f64(acc.high), + Cell::from_f64(acc.low), + Cell::from_f64(acc.close), + ]; + self.acc = Some(Acc { bucket, open: o, high: h, low: l, close: c }); + Some(&self.out) + } + } + } + + fn label(&self) -> String { + format!("Resample({}m)", self.period_ns / (60 * 1_000_000_000)) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use aura_core::{AnyColumn, Scalar, Timestamp}; + + /// Four depth-1 f64 input columns, as bootstrap sizes them from `lookbacks` + /// for the Barrier(0) OHLC group. + fn ohlc_inputs() -> Vec { + vec![ + AnyColumn::with_capacity(ScalarKind::F64, 1), + AnyColumn::with_capacity(ScalarKind::F64, 1), + AnyColumn::with_capacity(ScalarKind::F64, 1), + AnyColumn::with_capacity(ScalarKind::F64, 1), + ] + } + + /// Drive one fired eval: populate all four co-fresh inputs (Barrier-gated, so + /// in the real graph they always arrive together) and step the node at the M1 + /// instant `ns`. Each input is a capacity-1 ring, so the push overwrites the + /// single slot — `f64_in(i)[0]` reads exactly this M1 (mirrors the engine + /// re-presenting the newest sample each cycle, as `sma.rs`'s test does). + fn step(node: &mut Resample, inputs: &mut [AnyColumn], ns: i64, o: f64, h: f64, l: f64, c: f64) -> Option> { + for (col, v) in inputs.iter_mut().zip([o, h, l, c]) { + col.push(Scalar::f64(v)).unwrap(); + } + node.eval(Ctx::new(inputs, Timestamp(ns))) + .map(|r| r.iter().map(|c| c.f64()).collect()) + } + + #[test] + fn resample_emits_completed_bar_only_on_rollover_open_first_high_max_low_min_close_last() { + // THE HEADLINE PROPERTY: a coarse bar is emitted ONLY when the cursor + // rolls into the next bucket, and it carries open=first M1 open, + // high=max M1 high, low=min M1 low, close=last M1 close of the WINDOW — + // C2 (a bar is actionable only once complete; partials are never emitted). + let period_minutes = 15; + let period_ns = period_minutes * 60 * 1_000_000_000; // 900_000_000_000 + assert_eq!(period_ns, 900_000_000_000); + + let mut node = Resample::new(period_minutes); + let mut inputs = ohlc_inputs(); + + // --- bucket 0: three M1 ticks, each accumulating, NONE emitted --- + // t=0 : first sample of bucket 0 -> starts the accumulator. + assert_eq!(step(&mut node, &mut inputs, 0, 100.0, 101.0, 99.0, 100.5), None); + // t=60e9: same bucket -> high climbs to 103, low to 100, close to 102. + assert_eq!(step(&mut node, &mut inputs, 60_000_000_000, 100.5, 103.0, 100.0, 102.0), None); + // t=120e9: same bucket -> high stays 103, low drops to 99? no: low=101 > 99 + // so running low stays 99; close becomes 101.5 (last). + assert_eq!(step(&mut node, &mut inputs, 120_000_000_000, 102.0, 102.5, 101.0, 101.5), None); + + // --- first tick of bucket 1 (t=900e9): ROLLOVER -> emit the COMPLETED + // bucket-0 bar: open=100.0 (first), high=103.0 (max), low=99.0 (min), + // close=101.5 (last). THE load-bearing assertion. --- + let emitted = step(&mut node, &mut inputs, 900_000_000_000, 101.5, 104.0, 101.0, 103.0); + assert_eq!(emitted, Some(vec![100.0, 103.0, 99.0, 101.5])); + } + + #[test] + fn resample_drops_the_partial_last_bar_and_resets_the_accumulator() { + // PARTIAL-DROP + clean reset: after bucket 0 rolls over (emitting), the + // accumulator restarts from bucket 1's first M1. With no further rollover, + // bucket 1's bar is incomplete and is NEVER emitted (no EOF flush). Feeding + // a bucket-2 tick THEN emits bucket 1 — proving open=bucket-1's first open + // (101.5), i.e. the accumulator reset cleanly rather than carrying bucket 0. + let mut node = Resample::new(15); + let mut inputs = ohlc_inputs(); + + // bucket 0: two ticks -> open=100.0 (first), high=103.0 (max 101,103), + // low=99.0 (min 99,100), close=102.0 (last close). + assert_eq!(step(&mut node, &mut inputs, 0, 100.0, 101.0, 99.0, 100.5), None); + assert_eq!(step(&mut node, &mut inputs, 60_000_000_000, 100.5, 103.0, 100.0, 102.0), None); + // bucket 1's first tick rolls over bucket 0 (emits) and starts bucket 1. + assert_eq!( + step(&mut node, &mut inputs, 900_000_000_000, 101.5, 104.0, 101.0, 103.0), + Some(vec![100.0, 103.0, 99.0, 102.0]) + ); + // another bucket-1 tick: still accumulating bucket 1, NO emission. + assert_eq!(step(&mut node, &mut inputs, 960_000_000_000, 103.0, 105.0, 102.5, 104.5), None); + // bucket 2's first tick rolls over bucket 1: open=101.5 (bucket 1's FIRST), + // high=105.0 (max of 104,105), low=101.0 (min of 101,102.5), close=104.5. + assert_eq!( + step(&mut node, &mut inputs, 1_800_000_000_000, 104.5, 106.0, 104.0, 105.5), + Some(vec![101.5, 105.0, 101.0, 104.5]) + ); + // ...and bucket 2 (the new partial) is NOT flushed: no further emission. + } + + #[test] + fn output_is_a_four_field_ohlc_record_barrier_gated() { + // schema shape: four Barrier(0) f64 inputs named o/h/l/c, four f64 output + // fields named o/h/l/c, one i64 param. The first multi-field output node. + let s = Resample::builder().schema().clone(); + let in_names: Vec<&str> = s.inputs.iter().map(|p| p.name.as_str()).collect(); + assert_eq!(in_names, ["open", "high", "low", "close"]); + assert!(s.inputs.iter().all(|p| p.kind == ScalarKind::F64 && p.firing == Firing::Barrier(0))); + let out_names: Vec<&str> = s.output.iter().map(|f| f.name.as_str()).collect(); + assert_eq!(out_names, ["open", "high", "low", "close"]); + assert!(s.output.iter().all(|f| f.kind == ScalarKind::F64)); + assert_eq!(s.params, vec![ParamSpec { name: "period_minutes".into(), kind: ScalarKind::I64 }]); + } +}