Aura/crates/aura-std/src/resample.rs

//! `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).
//! 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<Acc>,
    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<usize> {
        // 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<AnyColumn> {
        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<Vec<f64>> {
        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 }]);
    }
}