claude
a32dc38d18
C29 compile/unit seam, task 3 of the self-description plan: every aura-std NodeSchema literal threads its one-line doc. Four texts were corrected against the actual eval/finalize semantics rather than taken from the plan table verbatim: Delay is a lag-N register (not one-step), GatedRecorder flushes an ungated final row at finalize, Latch is a level-sensitive set/reset register (captures no input), SeriesReducer emits its single summary row at finalize (not per cycle). Gate: cargo build -p aura-std --lib clean; full std test run follows at the all-crates gate once strategy/backtest/engine thread their sites (the earlier per-crate test gate was unsatisfiable in isolation -- std's dev-deps pull crates whose sites belong to later tasks). refs #316
96 lines
3.3 KiB
Rust
96 lines
3.3 KiB
Rust
//! `SeriesReducer` — a folding sink (C8) over one `f64` column. It folds each
|
|
//! warm value into a `SeriesFold` and, on `finalize`, emits a single summary row
|
|
//! `[last, max_drawdown, sign_flips]`, so a downstream reduction reads O(1) per
|
|
//! member instead of the whole per-cycle series. C7-pure: owned fold + channel.
|
|
|
|
use aura_core::{Cell, Ctx, Firing, Node, NodeSchema, PortSpec, PrimitiveBuilder, Scalar, ScalarKind, SeriesFold, Timestamp};
|
|
use std::sync::mpsc::Sender;
|
|
|
|
/// A single-`f64`-column folding sink. Emits exactly one row on `finalize`:
|
|
/// `[Scalar::f64(last), Scalar::f64(max_drawdown), Scalar::i64(sign_flips)]`.
|
|
pub struct SeriesReducer {
|
|
fold: SeriesFold,
|
|
last_ts: Timestamp,
|
|
tx: Sender<(Timestamp, Vec<Scalar>)>,
|
|
}
|
|
|
|
impl SeriesReducer {
|
|
pub fn new(tx: Sender<(Timestamp, Vec<Scalar>)>) -> Self {
|
|
Self { fold: SeriesFold::new(), last_ts: Timestamp(0), tx }
|
|
}
|
|
|
|
pub fn builder(firing: Firing, tx: Sender<(Timestamp, Vec<Scalar>)>) -> PrimitiveBuilder {
|
|
PrimitiveBuilder::new(
|
|
"SeriesReducer",
|
|
NodeSchema {
|
|
inputs: vec![PortSpec { kind: ScalarKind::F64, firing, name: "col[0]".to_string() }],
|
|
output: vec![], // sink: empty output (C8)
|
|
params: vec![],
|
|
doc: "folds an input series into a single summary row emitted at finalize",
|
|
},
|
|
move |_| Box::new(SeriesReducer::new(tx.clone())),
|
|
)
|
|
}
|
|
}
|
|
|
|
impl Node for SeriesReducer {
|
|
fn lookbacks(&self) -> Vec<usize> {
|
|
vec![1]
|
|
}
|
|
|
|
fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> {
|
|
let v = ctx.f64_in(0).get(0)?; // None during warm-up
|
|
self.fold.fold(v);
|
|
self.last_ts = ctx.now();
|
|
None
|
|
}
|
|
|
|
fn finalize(&mut self) {
|
|
let row = vec![
|
|
Scalar::f64(self.fold.last),
|
|
Scalar::f64(self.fold.max_drawdown),
|
|
Scalar::i64(self.fold.sign_flips as i64),
|
|
];
|
|
let _ = self.tx.send((self.last_ts, row));
|
|
}
|
|
|
|
fn label(&self) -> String {
|
|
"SeriesReducer".to_string()
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use aura_core::AnyColumn;
|
|
use std::sync::mpsc;
|
|
|
|
#[test]
|
|
fn folds_and_emits_one_summary_row_on_finalize() {
|
|
let (tx, rx) = mpsc::channel();
|
|
let mut r = SeriesReducer::new(tx);
|
|
let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)];
|
|
for (t, v) in [(1_i64, 10.0_f64), (2, 12.0), (3, 8.0)] {
|
|
inputs[0].push(Scalar::f64(v)).unwrap();
|
|
assert_eq!(r.eval(Ctx::new(&inputs, Timestamp(t))), None);
|
|
}
|
|
// nothing emitted per-eval.
|
|
assert!(rx.try_recv().is_err());
|
|
r.finalize();
|
|
let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
|
|
assert_eq!(out.len(), 1);
|
|
assert_eq!(out[0].1[0], Scalar::f64(8.0)); // last
|
|
assert_eq!(out[0].1[1], Scalar::f64(4.0)); // max_drawdown 12 - 8
|
|
assert_eq!(out[0].1[2], Scalar::i64(0)); // sign_flips (all positive)
|
|
}
|
|
|
|
#[test]
|
|
fn empty_finalize_emits_zero_summary() {
|
|
let (tx, rx) = mpsc::channel();
|
|
let mut r = SeriesReducer::new(tx);
|
|
r.finalize();
|
|
let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
|
|
assert_eq!(out, vec![(Timestamp(0), vec![Scalar::f64(0.0), Scalar::f64(0.0), Scalar::i64(0)])]);
|
|
}
|
|
}
|