Brummel
cfe7bad0ff
M3 machinery for BLOCKER #138: sink reductions can now fold online instead of buffering every per-cycle row to an unbounded channel. - aura-core: additive `Node::finalize` end-of-stream hook (default no-op) + `SeriesFold` online accumulator (last / max_drawdown / sign_flips). - aura-engine: `Harness::run` calls `finalize` once per node in topo order after the source loop; `summarize` refactored to drive `SeriesFold` (byte-identical), the now-dead `sign0` removed. - aura-std: `GatedRecorder` (emits only gated rows + the final row on finalize) and `SeriesReducer` (folds one f64 series, emits one summary row on finalize), both emitting the `Recorder` channel type — no aura-engine dependency. - Integration tests: folded reductions are byte-for-byte equal to the raw-recorder path; the R-reduction's peak memory is O(trades), independent of cycle count. Ratified two off-plan fixes the implementer made in the tests (both plan defects, not code defects): the GatedRecorder correctly flushes the final non-gated row on finalize, so the retained count is K_TRADES + 1; and the heterogeneous PM record needs kind-correct zero scalars (a blanket f64 zero is rejected by the kind-typed AnyColumn::push). Wiring these reducers into the stage1-r sweep (the actual O(cycles)->O(trades) win) follows next. refs #138
95 lines
3.2 KiB
Rust
95 lines
3.2 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![],
|
|
},
|
|
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)])]);
|
|
}
|
|
}
|