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
98 lines
2.7 KiB
Rust
98 lines
2.7 KiB
Rust
//! `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<f64>,
|
|
/// 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);
|
|
}
|
|
}
|