diff --git a/crates/aura-core/src/lib.rs b/crates/aura-core/src/lib.rs index 2be6bf5..b68e8aa 100644 --- a/crates/aura-core/src/lib.rs +++ b/crates/aura-core/src/lib.rs @@ -35,6 +35,7 @@ mod ctx; mod error; mod node; mod scalar; +mod series_fold; pub use any::AnyColumn; pub use cell::Cell; @@ -45,3 +46,4 @@ pub use node::{ zip_params, FieldSpec, Firing, Node, NodeSchema, ParamSpec, PortSpec, PrimitiveBuilder, }; pub use scalar::{Scalar, ScalarKind, Timestamp}; +pub use series_fold::SeriesFold; diff --git a/crates/aura-core/src/node.rs b/crates/aura-core/src/node.rs index d972cb2..438c2ef 100644 --- a/crates/aura-core/src/node.rs +++ b/crates/aura-core/src/node.rs @@ -280,6 +280,12 @@ pub trait Node { fn label(&self) -> String { "node".to_string() } + + /// End-of-stream hook: `Harness::run` calls it once per node, in topological + /// order, after the source loop drains. A folding sink overrides it to flush + /// its accumulated summary; the default is a no-op, so existing nodes are + /// unaffected. Takes `&mut self` like `eval`, so `Node` stays object-safe. + fn finalize(&mut self) {} } #[cfg(test)] @@ -315,6 +321,14 @@ mod tests { assert_eq!(Bare.label(), "node"); } + #[test] + fn default_finalize_is_a_callable_noop() { + // a node that does not override finalize uses the default no-op; calling + // it compiles and does nothing (additive, non-breaking for existing nodes). + let mut b = Bare; + b.finalize(); + } + #[test] fn primitive_builder_label_is_the_bare_type() { // param-generic: the label is just the node type, no knob suffix — the diff --git a/crates/aura-core/src/series_fold.rs b/crates/aura-core/src/series_fold.rs new file mode 100644 index 0000000..43d8feb --- /dev/null +++ b/crates/aura-core/src/series_fold.rs @@ -0,0 +1,97 @@ +//! `SeriesFold` — the online accumulator behind `summarize`'s three pip +//! statistics (`last`, `max_drawdown`, `sign_flips`). One arithmetic source of +//! truth, foldable a value at a time, so a slice driver (`summarize`) and a +//! per-`eval` streaming sink (`SeriesReducer`) compute byte-identical results. + +/// Three-way sign: `-1.0` / `0.0` / `+1.0`. A zero maps to `0.0` so flat is +/// distinct from long/short in the flip count (unlike `f64::signum`). +fn sign0(v: f64) -> f64 { + if v > 0.0 { + 1.0 + } else if v < 0.0 { + -1.0 + } else { + 0.0 + } +} + +/// Online fold of a single `f64` series into the pip summary statistics. The +/// operations and their order mirror `summarize`'s loops exactly, so a +/// slice-driven and a per-value-driven fold agree bit-for-bit (IEEE-754 +/// non-associativity respected). +#[derive(Clone, Debug)] +pub struct SeriesFold { + /// the last value folded (`total_pips` for an equity series); `0.0` if none. + pub last: f64, + peak: f64, + /// running `max(peak - value)`, always `>= 0.0`. + pub max_drawdown: f64, + prev_sign: Option, + /// count of adjacent normalized-sign changes. + pub sign_flips: u64, +} + +impl Default for SeriesFold { + fn default() -> Self { + Self { last: 0.0, peak: f64::NEG_INFINITY, max_drawdown: 0.0, prev_sign: None, sign_flips: 0 } + } +} + +impl SeriesFold { + pub fn new() -> Self { + Self::default() + } + + /// Fold one value into the running statistics. + pub fn fold(&mut self, v: f64) { + self.last = v; + if v > self.peak { + self.peak = v; + } + let dd = self.peak - v; + if dd > self.max_drawdown { + self.max_drawdown = dd; + } + let s = sign0(v); + if let Some(p) = self.prev_sign + && s != p + { + self.sign_flips += 1; + } + self.prev_sign = Some(s); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn series_fold_empty_is_zero() { + let f = SeriesFold::new(); + assert_eq!(f.last, 0.0); + assert_eq!(f.max_drawdown, 0.0); + assert_eq!(f.sign_flips, 0); + } + + #[test] + fn series_fold_tracks_last_and_drawdown() { + let mut f = SeriesFold::new(); + for v in [10.0, 12.0, 8.0, 9.0] { + f.fold(v); + } + assert_eq!(f.last, 9.0); + assert_eq!(f.max_drawdown, 4.0); // peak 12 - trough 8 + assert_eq!(f.sign_flips, 0); // all positive + } + + #[test] + fn series_fold_counts_sign_flips_zero_distinct() { + let mut f = SeriesFold::new(); + for v in [1.0, -1.0, 0.0, 2.0] { + f.fold(v); + } + // signs + - 0 + -> flips at -, 0, + = 3 + assert_eq!(f.sign_flips, 3); + } +} diff --git a/crates/aura-engine/src/harness.rs b/crates/aura-engine/src/harness.rs index 5b44651..c26f8ec 100644 --- a/crates/aura-engine/src/harness.rs +++ b/crates/aura-engine/src/harness.rs @@ -441,6 +441,13 @@ impl Harness { } } } + + // end-of-stream: flush every node once, in topological order, so folding + // sinks (C8) emit their accumulated summary. This runs AFTER the + // deterministic event loop, so it adds no within-sim concurrency (C1). + for &nidx in topo.iter() { + nodes[nidx].node.finalize(); + } } } @@ -2265,4 +2272,49 @@ mod tests { }; assert_eq!(build(), build()); } + + struct FinalizeProbe { + id: usize, + tx: mpsc::Sender, + } + impl Node for FinalizeProbe { + fn lookbacks(&self) -> Vec { + vec![1] + } + fn eval(&mut self, _ctx: Ctx<'_>) -> Option<&[Cell]> { + None + } + fn finalize(&mut self) { + let _ = self.tx.send(self.id); + } + } + + #[test] + fn run_finalizes_every_node_once_after_the_stream() { + let (tx, rx) = mpsc::channel(); + // two sink probes, both fed by the single source; no inter-node edges. + let mut h = boot( + vec![ + Box::new(FinalizeProbe { id: 0, tx: tx.clone() }), + Box::new(FinalizeProbe { id: 1, tx }), + ], + vec![ + recorder_sig(&[ScalarKind::F64], Firing::Any), + recorder_sig(&[ScalarKind::F64], Firing::Any), + ], + vec![SourceSpec { + kind: ScalarKind::F64, + targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }], + }], + vec![], + ) + .expect("valid"); + // nothing finalized before the run. + assert!(rx.try_recv().is_err()); + h.run(vec![Box::new(VecSource::new(f64_stream(&[(1, 1.0), (2, 2.0)])))]); + // each node finalized exactly once, after the stream drained. + let mut fired: Vec = rx.try_iter().collect(); + fired.sort_unstable(); + assert_eq!(fired, vec![0, 1]); + } } diff --git a/crates/aura-engine/src/report.rs b/crates/aura-engine/src/report.rs index 0cf761e..ab193e5 100644 --- a/crates/aura-engine/src/report.rs +++ b/crates/aura-engine/src/report.rs @@ -8,7 +8,7 @@ //! cycle 0029) — the same encoder the run registry uses, so a record's stdout //! and on-disk shapes coincide. -use aura_core::{Scalar, ScalarKind, Timestamp}; +use aura_core::{Scalar, ScalarKind, SeriesFold, Timestamp}; use std::collections::HashMap; /// Summary metrics reduced from a run's recorded streams — the `-> metrics` @@ -370,48 +370,22 @@ pub fn summarize( equity: &[(Timestamp, f64)], exposure: &[(Timestamp, f64)], ) -> RunMetrics { - // total pips: the last cumulative equity value (0.0 if empty). - let total_pips = equity.last().map(|&(_, v)| v).unwrap_or(0.0); - - // max drawdown: the largest running-peak-minus-value, always >= 0.0. - let mut peak = f64::NEG_INFINITY; - let mut max_drawdown = 0.0_f64; + // total_pips + max_drawdown fold over equity; sign_flips folds over exposure. + // SeriesFold is the single arithmetic source of truth (shared with the + // SeriesReducer sink), so this is byte-identical to the prior inline loops. + let mut eqf = SeriesFold::new(); for &(_, v) in equity { - if v > peak { - peak = v; - } - let dd = peak - v; - if dd > max_drawdown { - max_drawdown = dd; - } + eqf.fold(v); } - - // bias sign-flips: adjacent samples whose normalized sign differs. - let mut bias_sign_flips = 0u64; - let mut prev: Option = None; + let mut exf = SeriesFold::new(); for &(_, v) in exposure { - let s = sign0(v); - if let Some(p) = prev - && s != p - { - bias_sign_flips += 1; - } - prev = Some(s); + exf.fold(v); } - - RunMetrics { total_pips, max_drawdown, bias_sign_flips, r: None } -} - -/// Three-way sign: `-1.0` / `0.0` / `+1.0`. Unlike `f64::signum` (which returns -/// `+1.0` for `+0.0`), a zero exposure maps to `0.0` so flat is distinct from -/// long/short in the sign-flip count. -fn sign0(v: f64) -> f64 { - if v > 0.0 { - 1.0 - } else if v < 0.0 { - -1.0 - } else { - 0.0 + RunMetrics { + total_pips: eqf.last, + max_drawdown: eqf.max_drawdown, + bias_sign_flips: exf.sign_flips, + r: None, } } diff --git a/crates/aura-engine/tests/streaming_reduction_equivalence.rs b/crates/aura-engine/tests/streaming_reduction_equivalence.rs new file mode 100644 index 0000000..e8d4a04 --- /dev/null +++ b/crates/aura-engine/tests/streaming_reduction_equivalence.rs @@ -0,0 +1,110 @@ +//! Folded sink reductions are byte-for-byte equal to the raw-recorder path. +//! Drives the folding sinks node-by-node (the `stage1_r_e2e.rs` direct-node +//! style) over a synthetic dense R-record + equity/exposure series, and asserts +//! `summarize_r` over `GatedRecorder`'s emitted rows equals `summarize_r` over +//! the full record, and a `SeriesReducer` summary equals `summarize`'s fields. + +use aura_core::{AnyColumn, Ctx, Node, Scalar, ScalarKind, Timestamp}; +use aura_engine::{summarize, summarize_r}; +use aura_std::{GatedRecorder, SeriesReducer, PM_RECORD_KINDS, PM_WIDTH}; +use std::sync::mpsc; + +const CLOSED: usize = 0; +const REALIZED_R: usize = 1; +const OPEN: usize = 11; +const UNREALIZED_R: usize = 12; +const ENTRY_PRICE: usize = 6; +const STOP_PRICE: usize = 7; + +/// The kind-correct zero for a dense-record column. The PM record is +/// heterogeneous (`PM_RECORD_KINDS`: Bool/I64/Timestamp slots beside the f64s), +/// so each unset slot must default to a scalar of its column's kind — a blanket +/// f64 zero would be rejected by the kind-typed `AnyColumn::push`. +fn zero_of(kind: ScalarKind) -> Scalar { + match kind { + ScalarKind::F64 => Scalar::f64(0.0), + ScalarKind::I64 => Scalar::i64(0), + ScalarKind::Bool => Scalar::bool(false), + ScalarKind::Timestamp => Scalar::ts(Timestamp(0)), + } +} + +/// One dense R-record row. `closed`/`open` set the gate + the open-at-end flag; +/// the priced fields make `summarize_r`'s R arithmetic well-defined. +fn pm_row(closed: bool, realized: f64, open: bool, entry: f64, stop: f64) -> Vec { + let mut v: Vec = PM_RECORD_KINDS.iter().map(|&k| zero_of(k)).collect(); + debug_assert_eq!(v.len(), PM_WIDTH); + v[CLOSED] = Scalar::bool(closed); + v[REALIZED_R] = Scalar::f64(realized); + v[OPEN] = Scalar::bool(open); + v[UNREALIZED_R] = Scalar::f64(if open { realized } else { 0.0 }); + v[ENTRY_PRICE] = Scalar::f64(entry); + v[STOP_PRICE] = Scalar::f64(stop); + v +} + +#[test] +fn gated_recorder_then_summarize_r_equals_raw_summarize_r() { + // a realistic mix: holds, closed wins/losses, a hold tail, then an open-at-end row. + let full: Vec<(Timestamp, Vec)> = [ + pm_row(false, 0.0, false, 0.0, 0.0), + pm_row(true, 1.5, false, 100.0, 95.0), + pm_row(false, 0.0, false, 0.0, 0.0), + pm_row(true, -1.0, false, 100.0, 95.0), + pm_row(false, 0.0, false, 0.0, 0.0), + pm_row(false, 0.7, true, 100.0, 95.0), // open at end + ] + .into_iter() + .enumerate() + .map(|(i, r)| (Timestamp(i as i64 + 1), r)) + .collect(); + + // RAW: summarize_r over the whole dense record. + let raw = summarize_r(&full, 0.0); + + // FOLDED: drive GatedRecorder (gate = CLOSED), collect its emitted rows, fold. + let (tx, rx) = mpsc::channel(); + let mut g = GatedRecorder::new(&PM_RECORD_KINDS, CLOSED, tx); + let mut cols: Vec = + PM_RECORD_KINDS.iter().map(|&k| AnyColumn::with_capacity(k, 1)).collect(); + for (ts, row) in &full { + for (i, s) in row.iter().enumerate() { + cols[i].push(*s).unwrap(); + } + assert_eq!(g.eval(Ctx::new(&cols, *ts)), None); + } + g.finalize(); + let folded_rows: Vec<(Timestamp, Vec)> = rx.try_iter().collect(); + let folded = summarize_r(&folded_rows, 0.0); + + assert_eq!(folded, raw, "GatedRecorder+summarize_r must equal raw summarize_r"); +} + +#[test] +fn series_reducer_equals_summarize_pip_fields() { + let equity: Vec<(Timestamp, f64)> = + [10.0, 12.0, 8.0, 9.0, 7.0].iter().enumerate().map(|(i, &v)| (Timestamp(i as i64), v)).collect(); + let exposure: Vec<(Timestamp, f64)> = + [1.0, 1.0, -1.0, 0.0, 1.0].iter().enumerate().map(|(i, &v)| (Timestamp(i as i64), v)).collect(); + + let raw = summarize(&equity, &exposure); + + // FOLDED: one SeriesReducer per series; read pip fields from their summaries. + let fold_series = |series: &[(Timestamp, f64)]| -> Vec { + let (tx, rx) = mpsc::channel(); + let mut r = SeriesReducer::new(tx); + let mut col = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)]; + for &(ts, v) in series { + col[0].push(Scalar::f64(v)).unwrap(); + r.eval(Ctx::new(&col, ts)); + } + r.finalize(); + rx.try_iter().next().expect("one summary row").1 + }; + let eqs = fold_series(&equity); + let exs = fold_series(&exposure); + + assert_eq!(eqs[0].as_f64(), raw.total_pips, "total_pips"); + assert_eq!(eqs[1].as_f64(), raw.max_drawdown, "max_drawdown"); + assert_eq!(exs[2].as_i64() as u64, raw.bias_sign_flips, "bias_sign_flips"); +} diff --git a/crates/aura-engine/tests/streaming_reduction_memory.rs b/crates/aura-engine/tests/streaming_reduction_memory.rs new file mode 100644 index 0000000..880a657 --- /dev/null +++ b/crates/aura-engine/tests/streaming_reduction_memory.rs @@ -0,0 +1,121 @@ +//! BLOCKER #138 capstone: the R-reduction's peak retained memory is +//! O(trades)+O(1), independent of cycle count. Drives `GatedRecorder` (gate = +//! CLOSED) through its real `mpsc` channel with a FIXED small number of closed +//! trades but a VARIABLE large number of hold cycles, generated LAZILY (one row +//! at a time, never all at once), and measures peak live bytes with a counting +//! global allocator. A retain-everything sink's peak would scale with the cycle +//! count; the folding sink's does not. Sole test in this binary, so the global +//! peak counter is not polluted by other tests running in parallel. + +use std::alloc::{GlobalAlloc, Layout, System}; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::mpsc; + +use aura_core::{AnyColumn, Ctx, Node, Scalar, ScalarKind, Timestamp}; +use aura_std::{GatedRecorder, PM_RECORD_KINDS, PM_WIDTH}; + +struct CountingAlloc; +static LIVE: AtomicUsize = AtomicUsize::new(0); +static PEAK: AtomicUsize = AtomicUsize::new(0); + +unsafe impl GlobalAlloc for CountingAlloc { + unsafe fn alloc(&self, layout: Layout) -> *mut u8 { + let p = unsafe { System.alloc(layout) }; + if !p.is_null() { + let now = LIVE.fetch_add(layout.size(), Ordering::Relaxed) + layout.size(); + PEAK.fetch_max(now, Ordering::Relaxed); + } + p + } + unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) { + LIVE.fetch_sub(layout.size(), Ordering::Relaxed); + unsafe { System.dealloc(ptr, layout) } + } +} + +#[global_allocator] +static A: CountingAlloc = CountingAlloc; + +fn reset_peak() { + PEAK.store(LIVE.load(Ordering::Relaxed), Ordering::Relaxed); +} +fn peak() -> usize { + PEAK.load(Ordering::Relaxed) +} + +const CLOSED: usize = 0; +const REALIZED_R: usize = 1; +const K_TRADES: usize = 8; + +/// Peak live bytes to drive `GatedRecorder` over `K_TRADES` closed trades, each +/// followed by `holds` hold cycles, sourced lazily (one row built at a time), +/// then fold the emitted rows. With a folding sink the peak is O(trades). +fn peak_bytes(holds: usize) -> usize { + reset_peak(); + let (tx, rx) = mpsc::channel(); + let mut g = GatedRecorder::new(&PM_RECORD_KINDS, CLOSED, tx); + let mut cols: Vec = + PM_RECORD_KINDS.iter().map(|&k| AnyColumn::with_capacity(k, 1)).collect(); + let mut ts = 0i64; + let push = |g: &mut GatedRecorder, cols: &mut Vec, ts: &mut i64, closed: bool, r: f64| { + // kind-correct zeros: the PM record is heterogeneous (PM_RECORD_KINDS has + // Bool/I64/Timestamp slots), so each unset slot defaults to its column's + // kind — a blanket f64 zero would be rejected by the kind-typed push. + let mut row: Vec = PM_RECORD_KINDS + .iter() + .map(|&k| match k { + ScalarKind::F64 => Scalar::f64(0.0), + ScalarKind::I64 => Scalar::i64(0), + ScalarKind::Bool => Scalar::bool(false), + ScalarKind::Timestamp => Scalar::ts(Timestamp(0)), + }) + .collect(); + debug_assert_eq!(row.len(), PM_WIDTH); + row[CLOSED] = Scalar::bool(closed); + row[REALIZED_R] = Scalar::f64(r); + for (i, s) in row.iter().enumerate() { + cols[i].push(*s).unwrap(); + } + g.eval(Ctx::new(cols, Timestamp(*ts))); + *ts += 1; + }; + for k in 0..K_TRADES { + let r = if k % 2 == 0 { 1.0 } else { -0.5 }; + push(&mut g, &mut cols, &mut ts, true, r); + for _ in 0..holds { + push(&mut g, &mut cols, &mut ts, false, 0.0); + } + } + g.finalize(); + let rows: Vec<(Timestamp, Vec)> = rx.try_iter().collect(); + // the K_TRADES gated (closed) rows, plus the one genuine final row the + // GatedRecorder flushes on finalize (the last cycle is a hold, so it was not + // gated) — O(trades)+O(1), independent of `holds`, which is the whole point. + assert_eq!(rows.len(), K_TRADES + 1, "only the closed trades + the final row are retained"); + let p = peak(); + drop(rows); + p +} + +#[test] +fn r_reduction_peak_memory_is_independent_of_cycle_count() { + const SMALL_HOLDS: usize = 64; // 8 * (1+64) = 520 cycles + const LARGE_HOLDS: usize = 50_000; // 8 * (1+50000) = 400_008 cycles + + let small = peak_bytes(SMALL_HOLDS); + let large = peak_bytes(LARGE_HOLDS); + + // cycle count grew ~770x; an O(trades)+O(1) sink holds its peak essentially + // flat. Generous 4x headroom so only genuine O(cycles) retention trips it. + assert!( + large <= small.saturating_mul(4), + "R-reduction peak scales with cycle count (O(cycles) retention): \ + small {} cycles -> {} bytes; large {} cycles -> {} bytes ({:.0}x). \ + The folding sink must retain O(trades)+O(1).", + K_TRADES * (1 + SMALL_HOLDS), + small, + K_TRADES * (1 + LARGE_HOLDS), + large, + large as f64 / small.max(1) as f64, + ); +} diff --git a/crates/aura-std/src/gated_recorder.rs b/crates/aura-std/src/gated_recorder.rs new file mode 100644 index 0000000..943a799 --- /dev/null +++ b/crates/aura-std/src/gated_recorder.rs @@ -0,0 +1,146 @@ +//! `GatedRecorder` — a `Recorder` (C8 sink) that emits a recorded row only on +//! cycles where a designated `bool` gate column is true, plus the genuine final +//! row once on `finalize`. Fed a dense per-decision record gated on its +//! "closed-this-cycle" column, it yields exactly the rows a trade-ledger +//! reduction reads — the closed rows and the last row — in O(trades) memory +//! instead of O(cycles). C7-pure: it holds only owned state + the channel. + +use aura_core::{Cell, Ctx, Firing, Node, NodeSchema, PortSpec, PrimitiveBuilder, Scalar, ScalarKind, Timestamp}; +use std::sync::mpsc::Sender; + +/// A recording sink that emits only rows whose `gate_col` (a `bool` column) is +/// true, plus the final row once on `finalize` (deduped against the last gated +/// emit). Warm-up behaves like `Recorder`: `None` until every column is warm, +/// recording nothing. +pub struct GatedRecorder { + kinds: Vec, + gate_col: usize, + last: Option<(Timestamp, Vec)>, + last_emitted: bool, + tx: Sender<(Timestamp, Vec)>, +} + +impl GatedRecorder { + pub fn new(kinds: &[ScalarKind], gate_col: usize, tx: Sender<(Timestamp, Vec)>) -> Self { + Self { kinds: kinds.to_vec(), gate_col, last: None, last_emitted: false, tx } + } + + /// Param-generic recipe. `kinds` / `gate_col` / `tx` are non-param construction + /// args (captured); the node declares no params. One input port per kind. + pub fn builder( + kinds: Vec, + gate_col: usize, + firing: Firing, + tx: Sender<(Timestamp, Vec)>, + ) -> PrimitiveBuilder { + let inputs = kinds + .iter() + .enumerate() + .map(|(i, &kind)| PortSpec { kind, firing, name: format!("col[{i}]") }) + .collect(); + let build_kinds = kinds.clone(); + PrimitiveBuilder::new( + "GatedRecorder", + NodeSchema { inputs, output: vec![], params: vec![] }, // sink: empty output (C8) + move |_| Box::new(GatedRecorder::new(&build_kinds, gate_col, tx.clone())), + ) + } +} + +impl Node for GatedRecorder { + fn lookbacks(&self) -> Vec { + vec![1; self.kinds.len()] + } + + fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> { + let mut row = Vec::with_capacity(self.kinds.len()); + for (i, &kind) in self.kinds.iter().enumerate() { + let scalar = match kind { + ScalarKind::F64 => Scalar::f64(ctx.f64_in(i).get(0)?), + ScalarKind::I64 => Scalar::i64(ctx.i64_in(i).get(0)?), + ScalarKind::Bool => Scalar::bool(ctx.bool_in(i).get(0)?), + ScalarKind::Timestamp => Scalar::ts(ctx.ts_in(i).get(0)?), + }; + row.push(scalar); + } + let gated = row[self.gate_col].as_bool(); + if gated { + let _ = self.tx.send((ctx.now(), row.clone())); + } + self.last = Some((ctx.now(), row)); + self.last_emitted = gated; + None + } + + fn finalize(&mut self) { + // ensure a downstream `summarize_r` sees the true final row (e.g. an + // open-at-end position) without double-counting a final row already + // emitted as a closed trade. + if !self.last_emitted + && let Some(row) = self.last.take() + { + let _ = self.tx.send(row); + } + } + + fn label(&self) -> String { + "GatedRecorder".to_string() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use aura_core::AnyColumn; + use std::sync::mpsc; + + fn drive(seq: &[(bool, f64)]) -> Vec<(Timestamp, Vec)> { + let (tx, rx) = mpsc::channel(); + let mut g = GatedRecorder::new(&[ScalarKind::Bool, ScalarKind::F64], 0, tx); + let mut inputs = vec![ + AnyColumn::with_capacity(ScalarKind::Bool, 1), + AnyColumn::with_capacity(ScalarKind::F64, 1), + ]; + for (t, &(gate, payload)) in seq.iter().enumerate() { + inputs[0].push(Scalar::bool(gate)).unwrap(); + inputs[1].push(Scalar::f64(payload)).unwrap(); + assert_eq!(g.eval(Ctx::new(&inputs, Timestamp(t as i64 + 1))), None); + } + g.finalize(); + rx.try_iter().collect() + } + + #[test] + fn emits_gated_rows_and_flushes_a_nongated_final_row() { + // hold, closed, hold(final). emits the closed row, then flushes the final hold. + let out = drive(&[(false, 1.0), (true, 2.0), (false, 3.0)]); + assert_eq!( + out, + vec![ + (Timestamp(2), vec![Scalar::bool(true), Scalar::f64(2.0)]), + (Timestamp(3), vec![Scalar::bool(false), Scalar::f64(3.0)]), + ] + ); + } + + #[test] + fn does_not_double_emit_a_gated_final_row() { + // final row IS gated -> emitted once by the gate; finalize must not re-emit. + let out = drive(&[(false, 1.0), (true, 2.0)]); + assert_eq!(out, vec![(Timestamp(2), vec![Scalar::bool(true), Scalar::f64(2.0)])]); + } + + #[test] + fn cold_warmup_records_nothing() { + let (tx, rx) = mpsc::channel(); + let mut g = GatedRecorder::new(&[ScalarKind::Bool, ScalarKind::F64], 0, tx); + let inputs = vec![ + AnyColumn::with_capacity(ScalarKind::Bool, 1), + AnyColumn::with_capacity(ScalarKind::F64, 1), + ]; + // both columns cold -> None, nothing recorded, finalize flushes nothing. + assert_eq!(g.eval(Ctx::new(&inputs, Timestamp(1))), None); + g.finalize(); + assert!(rx.try_recv().is_err()); + } +} diff --git a/crates/aura-std/src/lib.rs b/crates/aura-std/src/lib.rs index 36af28c..fc641f8 100644 --- a/crates/aura-std/src/lib.rs +++ b/crates/aura-std/src/lib.rs @@ -21,6 +21,7 @@ mod bias; mod delay; mod ema; mod eqconst; +mod gated_recorder; mod gt; mod latch; mod lincomb; @@ -29,6 +30,7 @@ mod mul; mod position_management; mod recorder; mod resample; +mod series_reducer; mod session; mod sim_broker; mod sizer; @@ -42,6 +44,7 @@ pub use bias::Bias; pub use delay::Delay; pub use ema::Ema; pub use eqconst::EqConst; +pub use gated_recorder::GatedRecorder; pub use gt::Gt; pub use latch::Latch; pub use lincomb::LinComb; @@ -53,6 +56,7 @@ pub use position_management::{ }; pub use recorder::Recorder; pub use resample::Resample; +pub use series_reducer::SeriesReducer; pub use session::Session; pub use sim_broker::SimBroker; pub use sizer::Sizer; diff --git a/crates/aura-std/src/series_reducer.rs b/crates/aura-std/src/series_reducer.rs new file mode 100644 index 0000000..83099a8 --- /dev/null +++ b/crates/aura-std/src/series_reducer.rs @@ -0,0 +1,94 @@ +//! `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)>, +} + +impl SeriesReducer { + pub fn new(tx: Sender<(Timestamp, Vec)>) -> Self { + Self { fold: SeriesFold::new(), last_ts: Timestamp(0), tx } + } + + pub fn builder(firing: Firing, tx: Sender<(Timestamp, Vec)>) -> 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 { + 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)> = 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)> = rx.try_iter().collect(); + assert_eq!(out, vec![(Timestamp(0), vec![Scalar::f64(0.0), Scalar::f64(0.0), Scalar::i64(0)])]); + } +}