feat(aura-std): Resample — M1->coarse OHLC, C2 emit-on-rollover
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
This commit is contained in:
@@ -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;
|
||||
|
||||
@@ -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<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 }]);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user