perf: bit-exact batch fast paths + streaming-first benchmark docs (#202)
## Summary - Dedicated batch fast paths for **EMA, RSI, Bollinger, MACD and ATR** (used by the Python bindings): one allocation filled in a single pass, warmup encoded as `NaN`, no per-element `Option` or input re-validation. Each is **bit-for-bit equal** to replaying `update` — SMA/Bollinger keep the drift-reseed cadence, the EMA-family keep the seed division and `mul_add` recurrences. Adds the `BatchNanExt` extension trait. - **Cross-library benchmark refresh**: `compare_libraries.py` reports the median across timing rounds (`--rounds` / `--streaming-rounds`), gains `--skip-batch` / `--skip-streaming`, and runs every peer through the streaming arena (recompute for batch-only libraries). `wickra-bench` drives the batch fast paths against `kand`. - **README** benchmark section reordered streaming-first (the order-of-magnitude result), with measured TA-Lib/tulipy/pandas-ta numbers in place of the CI-only placeholders. ## Impact - Python batch ~2× faster on EMA/RSI/MACD/ATR; streaming path unchanged. - The `batch == streaming` equivalence stays bit-exact. ## Verification - `cargo fmt` · `cargo clippy --workspace --all-targets --all-features -- -D warnings` (clean) - `cargo test --workspace --all-features` — 3782 unit + 420 doc tests pass - Python `pytest` — streaming-vs-batch, known-values, input-validation, smoke pass ## Notes - Node/WASM bindings keep their existing batch; the fast paths are Python-only for now.
This commit is contained in:
kingchenc
@@ -75,6 +75,67 @@ impl Atr {
|
||||
None
|
||||
}
|
||||
}
|
||||
|
||||
/// Vectorized batch over raw high/low/close columns: one `f64` per bar
|
||||
/// (`NaN` during warmup). The caller guarantees the three slices are equal
|
||||
/// length and finite with valid OHLC ordering (the binding validates once up
|
||||
/// front); ATR only reads high, low and the previous close.
|
||||
///
|
||||
/// For a fresh indicator long enough to seed (`n >= period`) it runs the
|
||||
/// true-range seed once and then the bare Wilder recurrence in a tight loop —
|
||||
/// no per-bar `Candle` construction/validation, no `Option`, identical
|
||||
/// division at the seed and `mul_add` afterwards, so the result is
|
||||
/// *bit-for-bit* equal to replaying `update` over the same candles. Shorter
|
||||
/// or non-fresh inputs defer to an exact `update` replay.
|
||||
pub fn batch_atr(&mut self, high: &[f64], low: &[f64], close: &[f64]) -> Vec<f64> {
|
||||
let p = self.period;
|
||||
let n = high.len();
|
||||
if self.seeded || !self.seed_buf.is_empty() || self.prev_close.is_some() || n < p {
|
||||
let mut out = vec![f64::NAN; n];
|
||||
for i in 0..n {
|
||||
let candle = Candle::new_unchecked(close[i], high[i], low[i], close[i], 0.0, 0);
|
||||
if let Some(v) = self.update(candle) {
|
||||
out[i] = v;
|
||||
}
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
// Warmup `[0, p-1)` is `NaN`; the first ATR is emitted at index `p - 1`.
|
||||
let mut out = vec![f64::NAN; p - 1];
|
||||
out.reserve(n - (p - 1));
|
||||
// Seed: mean of the first `period` true ranges. TR₀ has no previous close.
|
||||
let mut prev_close = close[0];
|
||||
let mut sum_tr = high[0] - low[0];
|
||||
self.seed_buf.push(sum_tr);
|
||||
for i in 1..p {
|
||||
let (h, l) = (high[i], low[i]);
|
||||
let tr = (h - l)
|
||||
.max((h - prev_close).abs())
|
||||
.max((l - prev_close).abs());
|
||||
prev_close = close[i];
|
||||
self.seed_buf.push(tr);
|
||||
sum_tr += tr;
|
||||
}
|
||||
let mut avg = sum_tr / p as f64;
|
||||
out.push(avg);
|
||||
// Steady state: Wilder smoothing, reciprocal hoisted out of the loop.
|
||||
for i in p..n {
|
||||
let (h, l) = (high[i], low[i]);
|
||||
let tr = (h - l)
|
||||
.max((h - prev_close).abs())
|
||||
.max((l - prev_close).abs());
|
||||
prev_close = close[i];
|
||||
avg = avg.mul_add(self.n_minus_1, tr) * self.inv_period;
|
||||
out.push(avg);
|
||||
}
|
||||
|
||||
// Leave state where a full `update` replay would (seeded; seed_buf retained).
|
||||
self.prev_close = Some(prev_close);
|
||||
self.avg = avg;
|
||||
self.seeded = true;
|
||||
out
|
||||
}
|
||||
}
|
||||
|
||||
impl Indicator for Atr {
|
||||
@@ -266,6 +327,81 @@ mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
fn bits_eq(a: &[f64], b: &[f64]) -> bool {
|
||||
a.len() == b.len()
|
||||
&& a.iter()
|
||||
.zip(b)
|
||||
.all(|(x, y)| x == y || (x.is_nan() && y.is_nan()))
|
||||
}
|
||||
|
||||
fn atr_replay(period: usize, high: &[f64], low: &[f64], close: &[f64]) -> Vec<f64> {
|
||||
let mut a = Atr::new(period).unwrap();
|
||||
(0..high.len())
|
||||
.map(|i| {
|
||||
let candle = Candle::new_unchecked(close[i], high[i], low[i], close[i], 0.0, 0);
|
||||
a.update(candle).unwrap_or(f64::NAN)
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// Valid OHLC columns from a wandering base price.
|
||||
fn columns(n: usize) -> (Vec<f64>, Vec<f64>, Vec<f64>) {
|
||||
let base: Vec<f64> = (0..n)
|
||||
.map(|i| (f64::from(u32::try_from(i).unwrap()) * 0.3).sin() * 5.0 + 100.0)
|
||||
.collect();
|
||||
let high = base.iter().map(|b| b + 1.0).collect();
|
||||
let low = base.iter().map(|b| b - 1.0).collect();
|
||||
(high, low, base)
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn batch_atr_fast_path_is_bit_identical() {
|
||||
let (high, low, close) = columns(300);
|
||||
let mut atr = Atr::new(14).unwrap();
|
||||
let got = atr.batch_atr(&high, &low, &close);
|
||||
assert!(bits_eq(&got, &atr_replay(14, &high, &low, &close)));
|
||||
let mut ref_atr = Atr::new(14).unwrap();
|
||||
for i in 0..high.len() {
|
||||
ref_atr.update(Candle::new_unchecked(
|
||||
close[i], high[i], low[i], close[i], 0.0, 0,
|
||||
));
|
||||
}
|
||||
let next = Candle::new_unchecked(101.0, 102.0, 100.0, 101.0, 0.0, 0);
|
||||
assert_eq!(atr.update(next), ref_atr.update(next));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn batch_atr_falls_back_when_not_fresh() {
|
||||
let (high, low, close) = columns(40);
|
||||
let mut atr = Atr::new(14).unwrap();
|
||||
atr.update(Candle::new_unchecked(
|
||||
close[0], high[0], low[0], close[0], 0.0, 0,
|
||||
));
|
||||
let mut ref_atr = Atr::new(14).unwrap();
|
||||
ref_atr.update(Candle::new_unchecked(
|
||||
close[0], high[0], low[0], close[0], 0.0, 0,
|
||||
));
|
||||
let want: Vec<f64> = (0..high.len())
|
||||
.map(|i| {
|
||||
ref_atr
|
||||
.update(Candle::new_unchecked(
|
||||
close[i], high[i], low[i], close[i], 0.0, 0,
|
||||
))
|
||||
.unwrap_or(f64::NAN)
|
||||
})
|
||||
.collect();
|
||||
assert!(bits_eq(&atr.batch_atr(&high, &low, &close), &want));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn batch_atr_sub_period_slice_falls_back() {
|
||||
let (high, low, close) = columns(5);
|
||||
let mut atr = Atr::new(14).unwrap();
|
||||
let got = atr.batch_atr(&high, &low, &close);
|
||||
assert!(bits_eq(&got, &atr_replay(14, &high, &low, &close)));
|
||||
assert!(got.iter().all(|x| x.is_nan()));
|
||||
}
|
||||
|
||||
proptest::proptest! {
|
||||
#![proptest_config(proptest::test_runner::Config::with_cases(48))]
|
||||
#[test]
|
||||
|
||||
@@ -108,6 +108,82 @@ impl BollingerBands {
|
||||
self.multiplier
|
||||
}
|
||||
|
||||
/// Vectorized flat batch for bindings: returns `n * 4` values laid out as
|
||||
/// `[upper, middle, lower, stddev]` per input row, warmup rows all `NaN`.
|
||||
///
|
||||
/// For a fresh, all-finite slice it inlines `update`'s rolling `sum`/`sum_sq`
|
||||
/// and drift-reseed, writing the four band values directly instead of an
|
||||
/// `Option<BollingerOutput>` per element. Same add/subtract order, same reseed
|
||||
/// cadence, same variance/`sqrt` math — so it is *bit-for-bit* equal to
|
||||
/// replaying `update`, including the long-stream drift bound. Any other state,
|
||||
/// or a non-finite element, defers to the exact `update` replay.
|
||||
///
|
||||
/// This is a *separate* entry point from the trait [`batch`](crate::BatchExt::batch),
|
||||
/// which returns `Vec<Option<BollingerOutput>>`; only the bindings, which want
|
||||
/// a flat `f64` buffer, call this.
|
||||
pub fn batch_bands(&mut self, inputs: &[f64]) -> Vec<f64> {
|
||||
let p = self.period;
|
||||
let n = inputs.len();
|
||||
if self.count != 0
|
||||
|| self.updates_since_recompute != 0
|
||||
|| !inputs.iter().all(|x| x.is_finite())
|
||||
{
|
||||
// Slow path: exact replay of `update` into the flat layout.
|
||||
let mut out = vec![f64::NAN; n * 4];
|
||||
for (i, &x) in inputs.iter().enumerate() {
|
||||
if let Some(o) = self.update(x) {
|
||||
out[i * 4] = o.upper;
|
||||
out[i * 4 + 1] = o.middle;
|
||||
out[i * 4 + 2] = o.lower;
|
||||
out[i * 4 + 3] = o.stddev;
|
||||
}
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
let p_f64 = p as f64;
|
||||
let mult = self.multiplier;
|
||||
// Pre-sized output: warmup rows stay NaN, ready rows are written in place
|
||||
// by index — no per-row `push` length/capacity check.
|
||||
let mut out = vec![f64::NAN; n * 4];
|
||||
for (i, &x) in inputs.iter().enumerate() {
|
||||
if self.count == p {
|
||||
let old = self.buf[self.head];
|
||||
self.sum -= old;
|
||||
self.sum_sq -= old * old;
|
||||
self.buf[self.head] = x;
|
||||
self.sum += x;
|
||||
self.sum_sq += x * x;
|
||||
} else {
|
||||
self.buf[self.head] = x;
|
||||
self.sum += x;
|
||||
self.sum_sq += x * x;
|
||||
self.count += 1;
|
||||
}
|
||||
self.head += 1;
|
||||
if self.head == p {
|
||||
self.head = 0;
|
||||
}
|
||||
self.updates_since_recompute += 1;
|
||||
if self.updates_since_recompute >= RECOMPUTE_EVERY * p {
|
||||
let chronological = self.buf[self.head..].iter().chain(&self.buf[..self.head]);
|
||||
self.sum = chronological.clone().copied().sum();
|
||||
self.sum_sq = chronological.map(|&v| v * v).sum();
|
||||
self.updates_since_recompute = 0;
|
||||
}
|
||||
if self.count == p {
|
||||
let mean = self.sum / p_f64;
|
||||
let stddev = (self.sum_sq / p_f64 - mean * mean).max(0.0).sqrt();
|
||||
let band = mult * stddev;
|
||||
out[i * 4] = mean + band;
|
||||
out[i * 4 + 1] = mean;
|
||||
out[i * 4 + 2] = mean - band;
|
||||
out[i * 4 + 3] = stddev;
|
||||
}
|
||||
}
|
||||
out
|
||||
}
|
||||
|
||||
fn current(&self) -> Option<BollingerOutput> {
|
||||
if self.count != self.period {
|
||||
return None;
|
||||
@@ -352,6 +428,79 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
fn bits_eq(a: &[f64], b: &[f64]) -> bool {
|
||||
a.len() == b.len()
|
||||
&& a.iter()
|
||||
.zip(b)
|
||||
.all(|(x, y)| x == y || (x.is_nan() && y.is_nan()))
|
||||
}
|
||||
|
||||
/// Flat `n*4` `[upper, middle, lower, stddev]` replay of `update`.
|
||||
fn bb_replay(period: usize, mult: f64, series: &[f64]) -> Vec<f64> {
|
||||
let mut bb = BollingerBands::new(period, mult).unwrap();
|
||||
let mut out = Vec::with_capacity(series.len() * 4);
|
||||
for &x in series {
|
||||
match bb.update(x) {
|
||||
Some(o) => out.extend_from_slice(&[o.upper, o.middle, o.lower, o.stddev]),
|
||||
None => out.extend_from_slice(&[f64::NAN; 4]),
|
||||
}
|
||||
}
|
||||
out
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn batch_bands_fast_path_is_bit_identical_with_reseed() {
|
||||
// > 16*period inputs so the drift-reseed branch fires inside batch_bands.
|
||||
let series: Vec<f64> = (0..500)
|
||||
.map(|i| (f64::from(i) * 0.2).sin() * 10.0 + 50.0)
|
||||
.collect();
|
||||
let mut bb = BollingerBands::new(20, 2.0).unwrap();
|
||||
let got = bb.batch_bands(&series);
|
||||
assert!(bits_eq(&got, &bb_replay(20, 2.0, &series)));
|
||||
// State continues identically.
|
||||
let mut ref_bb = BollingerBands::new(20, 2.0).unwrap();
|
||||
for &x in &series {
|
||||
ref_bb.update(x);
|
||||
}
|
||||
assert_eq!(bb.update(55.0), ref_bb.update(55.0));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn batch_bands_falls_back_on_non_finite() {
|
||||
let series = [1.0, 2.0, 3.0, f64::NAN, 5.0, 6.0, 7.0];
|
||||
let mut bb = BollingerBands::new(3, 2.0).unwrap();
|
||||
assert!(bits_eq(
|
||||
&bb.batch_bands(&series),
|
||||
&bb_replay(3, 2.0, &series)
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn batch_bands_falls_back_when_not_fresh() {
|
||||
let mut bb = BollingerBands::new(3, 2.0).unwrap();
|
||||
bb.update(99.0);
|
||||
let series = [1.0, 2.0, 3.0, 4.0];
|
||||
let mut ref_bb = BollingerBands::new(3, 2.0).unwrap();
|
||||
ref_bb.update(99.0);
|
||||
let mut want = Vec::new();
|
||||
for &x in &series {
|
||||
match ref_bb.update(x) {
|
||||
Some(o) => want.extend_from_slice(&[o.upper, o.middle, o.lower, o.stddev]),
|
||||
None => want.extend_from_slice(&[f64::NAN; 4]),
|
||||
}
|
||||
}
|
||||
assert!(bits_eq(&bb.batch_bands(&series), &want));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn batch_bands_sub_period_slice_is_all_nan() {
|
||||
let series = [1.0, 2.0, 3.0];
|
||||
let mut bb = BollingerBands::new(10, 2.0).unwrap();
|
||||
let got = bb.batch_bands(&series);
|
||||
assert!(bits_eq(&got, &bb_replay(10, 2.0, &series)));
|
||||
assert!(got.iter().all(|x| x.is_nan()) && got.len() == 12);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn ignores_non_finite_input() {
|
||||
let mut bb = BollingerBands::new(5, 2.0).unwrap();
|
||||
|
||||
@@ -102,6 +102,68 @@ impl Ema {
|
||||
}
|
||||
}
|
||||
|
||||
/// Whether the EMA has seen no input yet (neither seeded nor mid-warmup).
|
||||
/// Lets composite indicators (e.g. MACD) decide if a fast batch path is safe.
|
||||
pub(crate) fn is_fresh(&self) -> bool {
|
||||
!self.seeded && self.warmup_buf.is_empty()
|
||||
}
|
||||
|
||||
/// Force the EMA into its seeded steady state with `current` as the latest
|
||||
/// value. Used by composite fused batch paths (MACD) to leave each sub-EMA
|
||||
/// where a per-tick `update` replay would, so a later `update` continues
|
||||
/// correctly. The post-seed recurrence never re-reads `warmup_buf`, so it is
|
||||
/// left as-is.
|
||||
pub(crate) fn seed_to(&mut self, current: f64) {
|
||||
self.current = current;
|
||||
self.seeded = true;
|
||||
}
|
||||
|
||||
/// Vectorized batch returning one `f64` per input (`NaN` during warmup).
|
||||
///
|
||||
/// Shadows the generic [`BatchNanExt::batch_nan`](crate::BatchNanExt) blanket
|
||||
/// default via inherent-method resolution. For a fresh indicator over an
|
||||
/// all-finite slice it runs the seed (mean of the first `period`) once and
|
||||
/// then the bare `alpha * x + (1 - alpha) * prev` recurrence in a tight loop
|
||||
/// with no per-element `is_finite`/`seeded` branch and no `Option` — yet uses
|
||||
/// the identical `mul_add`, so the result is *bit-for-bit* equal to replaying
|
||||
/// `update`. Any other state, or a non-finite element, defers to the exact
|
||||
/// `update` replay.
|
||||
pub fn batch_nan(&mut self, inputs: &[f64]) -> Vec<f64> {
|
||||
let p = self.period;
|
||||
if self.seeded || !self.warmup_buf.is_empty() || !inputs.iter().all(|x| x.is_finite()) {
|
||||
return inputs
|
||||
.iter()
|
||||
.map(|&x| self.update(x).unwrap_or(f64::NAN))
|
||||
.collect();
|
||||
}
|
||||
|
||||
let n = inputs.len();
|
||||
if n < p {
|
||||
// Not enough to seed; mirror `update` stashing inputs for warmup.
|
||||
self.warmup_buf.extend_from_slice(inputs);
|
||||
return vec![f64::NAN; n];
|
||||
}
|
||||
|
||||
// Warmup `[0, p-1)` is `NaN`; values from the seed on are pushed once each.
|
||||
let mut out = vec![f64::NAN; p - 1];
|
||||
out.reserve(n - (p - 1));
|
||||
let seed = inputs[..p].iter().copied().sum::<f64>() / p as f64;
|
||||
let mut cur = seed;
|
||||
out.push(seed);
|
||||
let (alpha, oma) = (self.alpha, self.one_minus_alpha);
|
||||
for &x in &inputs[p..] {
|
||||
cur = alpha.mul_add(x, oma * cur);
|
||||
out.push(cur);
|
||||
}
|
||||
|
||||
// Leave state exactly where `update` would: seeded on `current`, with the
|
||||
// first `period` inputs retained in `warmup_buf` (never cleared post-seed).
|
||||
self.current = cur;
|
||||
self.seeded = true;
|
||||
self.warmup_buf.extend_from_slice(&inputs[..p]);
|
||||
out
|
||||
}
|
||||
|
||||
/// Internal helper that feeds a value without finiteness validation. The caller
|
||||
/// guarantees `input.is_finite()`. Used by MACD which has already validated.
|
||||
pub(crate) fn step_unchecked(&mut self, input: f64) -> Option<f64> {
|
||||
@@ -288,6 +350,71 @@ mod tests {
|
||||
assert_eq!(ema.update(f64::INFINITY), before);
|
||||
}
|
||||
|
||||
fn bits_eq(a: &[f64], b: &[f64]) -> bool {
|
||||
a.len() == b.len()
|
||||
&& a.iter()
|
||||
.zip(b)
|
||||
.all(|(x, y)| x == y || (x.is_nan() && y.is_nan()))
|
||||
}
|
||||
|
||||
fn ema_replay(period: usize, series: &[f64]) -> Vec<f64> {
|
||||
let mut e = Ema::new(period).unwrap();
|
||||
series
|
||||
.iter()
|
||||
.map(|&x| e.update(x).unwrap_or(f64::NAN))
|
||||
.collect()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn batch_nan_fast_path_is_bit_identical() {
|
||||
let series: Vec<f64> = (0..300)
|
||||
.map(|i| (f64::from(i) * 0.25).cos() * 8.0 + 40.0)
|
||||
.collect();
|
||||
let mut ema = Ema::new(14).unwrap();
|
||||
let got = ema.batch_nan(&series);
|
||||
assert!(bits_eq(&got, &ema_replay(14, &series)));
|
||||
let mut ref_ema = Ema::new(14).unwrap();
|
||||
for &x in &series {
|
||||
ref_ema.update(x);
|
||||
}
|
||||
assert_eq!(ema.update(7.5), ref_ema.update(7.5));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn batch_nan_falls_back_on_non_finite() {
|
||||
let series = [1.0, 2.0, 3.0, f64::INFINITY, 5.0, 6.0, 7.0];
|
||||
let mut ema = Ema::new(3).unwrap();
|
||||
assert!(bits_eq(&ema.batch_nan(&series), &ema_replay(3, &series)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn batch_nan_falls_back_when_warming() {
|
||||
let mut ema = Ema::new(3).unwrap();
|
||||
ema.update(10.0); // mid-warmup: warmup_buf non-empty, not seeded
|
||||
let series = [1.0, 2.0, 3.0, 4.0];
|
||||
let mut ref_ema = Ema::new(3).unwrap();
|
||||
ref_ema.update(10.0);
|
||||
let want: Vec<f64> = series
|
||||
.iter()
|
||||
.map(|&x| ref_ema.update(x).unwrap_or(f64::NAN))
|
||||
.collect();
|
||||
assert!(bits_eq(&ema.batch_nan(&series), &want));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn batch_nan_sub_period_slice_stays_unseeded() {
|
||||
let series = [1.0, 2.0];
|
||||
let mut ema = Ema::new(5).unwrap();
|
||||
let got = ema.batch_nan(&series);
|
||||
assert!(got.iter().all(|x| x.is_nan()) && got.len() == 2);
|
||||
assert!(!ema.is_ready());
|
||||
// Warmup state was stashed: feeding the rest seeds exactly as a full stream.
|
||||
assert!(bits_eq(
|
||||
&[ema.update(3.0).unwrap_or(f64::NAN)],
|
||||
&[ema_replay(5, &[1.0, 2.0, 3.0])[2]]
|
||||
));
|
||||
}
|
||||
|
||||
proptest::proptest! {
|
||||
#![proptest_config(proptest::test_runner::Config::with_cases(48))]
|
||||
#[test]
|
||||
|
||||
@@ -86,6 +86,116 @@ impl MacdIndicator {
|
||||
pub const fn value(&self) -> Option<MacdOutput> {
|
||||
self.last
|
||||
}
|
||||
|
||||
/// Vectorized flat batch for bindings: `n * 3` values laid out as
|
||||
/// `[macd, signal, histogram]` per input row, warmup rows all `NaN`.
|
||||
///
|
||||
/// For a fresh, all-finite slice long enough for a full output it runs the
|
||||
/// fast EMA, slow EMA and signal EMA as three recurrences fused into a single
|
||||
/// pass with one allocation — no `Option` per tick, no per-EMA intermediate
|
||||
/// buffers, identical SMA-mean seeds (division) and `mul_add` recurrences. The
|
||||
/// result is *bit-for-bit* equal to replaying `update`. Anything else (not
|
||||
/// fresh, non-finite, or too short to emit) defers to the exact `update`
|
||||
/// replay.
|
||||
///
|
||||
/// Separate from the trait [`batch`](crate::BatchExt::batch), which stays a
|
||||
/// bit-identical `update` replay; only the bindings call this.
|
||||
pub fn batch_macd(&mut self, inputs: &[f64]) -> Vec<f64> {
|
||||
let n = inputs.len();
|
||||
let (fp, sp, gp) = (self.fast_period, self.slow_period, self.signal_period);
|
||||
// First full output needs the slow EMA seeded (index sp-1) plus gp signal
|
||||
// values: index sp + gp - 2. Below that, or non-fresh/non-finite, replay.
|
||||
if self.last.is_some()
|
||||
|| !self.fast.is_fresh()
|
||||
|| !self.slow.is_fresh()
|
||||
|| !self.signal_ema.is_fresh()
|
||||
|| n < sp + gp - 1
|
||||
|| !inputs.iter().all(|x| x.is_finite())
|
||||
{
|
||||
let mut out = vec![f64::NAN; n * 3];
|
||||
for (i, &x) in inputs.iter().enumerate() {
|
||||
if let Some(o) = self.update(x) {
|
||||
out[i * 3] = o.macd;
|
||||
out[i * 3 + 1] = o.signal;
|
||||
out[i * 3 + 2] = o.histogram;
|
||||
}
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
// Pre-sized output: warmup rows stay NaN, full-output rows are written in
|
||||
// place by index — no per-row `push` length/capacity check.
|
||||
let mut out = vec![f64::NAN; n * 3];
|
||||
let (fa, fo) = (self.fast.alpha(), 1.0 - self.fast.alpha());
|
||||
let (sa, so) = (self.slow.alpha(), 1.0 - self.slow.alpha());
|
||||
let (ga, go) = (self.signal_ema.alpha(), 1.0 - self.signal_ema.alpha());
|
||||
let (fp_f, sp_f, gp_f) = (fp as f64, sp as f64, gp as f64);
|
||||
|
||||
let (mut fast_val, mut slow_val, mut sig) = (0.0_f64, 0.0_f64, 0.0_f64);
|
||||
let (mut fsum, mut ssum, mut gsum) = (0.0_f64, 0.0_f64, 0.0_f64);
|
||||
let mut sig_count = 0usize; // signal-EMA seed progress (raw MACD values seen)
|
||||
let mut sig_seeded = false;
|
||||
let mut last = MacdOutput {
|
||||
macd: 0.0,
|
||||
signal: 0.0,
|
||||
histogram: 0.0,
|
||||
};
|
||||
|
||||
for (i, &x) in inputs.iter().enumerate() {
|
||||
// Fast EMA: SMA-seeded at index fp-1, then recurrence.
|
||||
if i < fp {
|
||||
fsum += x;
|
||||
if i == fp - 1 {
|
||||
fast_val = fsum / fp_f;
|
||||
}
|
||||
} else {
|
||||
fast_val = fa.mul_add(x, fo * fast_val);
|
||||
}
|
||||
// Slow EMA: SMA-seeded at index sp-1, then recurrence.
|
||||
if i < sp {
|
||||
ssum += x;
|
||||
if i == sp - 1 {
|
||||
slow_val = ssum / sp_f;
|
||||
}
|
||||
} else {
|
||||
slow_val = sa.mul_add(x, so * slow_val);
|
||||
}
|
||||
if i + 1 < sp {
|
||||
continue; // slow EMA not seeded yet → no raw MACD line
|
||||
}
|
||||
let macd = fast_val - slow_val;
|
||||
// Signal EMA over the MACD line: SMA-seeded over its first gp values.
|
||||
let signal = if sig_seeded {
|
||||
sig = ga.mul_add(macd, go * sig);
|
||||
sig
|
||||
} else {
|
||||
gsum += macd;
|
||||
sig_count += 1;
|
||||
if sig_count < gp {
|
||||
continue; // signal EMA still seeding → no full output
|
||||
}
|
||||
sig = gsum / gp_f;
|
||||
sig_seeded = true;
|
||||
sig
|
||||
};
|
||||
let histogram = macd - signal;
|
||||
out[i * 3] = macd;
|
||||
out[i * 3 + 1] = signal;
|
||||
out[i * 3 + 2] = histogram;
|
||||
last = MacdOutput {
|
||||
macd,
|
||||
signal,
|
||||
histogram,
|
||||
};
|
||||
}
|
||||
|
||||
// Leave every sub-EMA and `last` where a full `update` replay would.
|
||||
self.fast.seed_to(fast_val);
|
||||
self.slow.seed_to(slow_val);
|
||||
self.signal_ema.seed_to(sig);
|
||||
self.last = Some(last);
|
||||
out
|
||||
}
|
||||
}
|
||||
|
||||
impl Indicator for MacdIndicator {
|
||||
@@ -256,6 +366,79 @@ mod tests {
|
||||
assert_eq!(macd.update(1.0), None);
|
||||
}
|
||||
|
||||
fn bits_eq(a: &[f64], b: &[f64]) -> bool {
|
||||
a.len() == b.len()
|
||||
&& a.iter()
|
||||
.zip(b)
|
||||
.all(|(x, y)| x == y || (x.is_nan() && y.is_nan()))
|
||||
}
|
||||
|
||||
/// Flat `n*3` `[macd, signal, histogram]` replay of `update`.
|
||||
fn macd_replay(series: &[f64]) -> Vec<f64> {
|
||||
let mut m = MacdIndicator::classic();
|
||||
let mut out = Vec::with_capacity(series.len() * 3);
|
||||
for &x in series {
|
||||
match m.update(x) {
|
||||
Some(o) => out.extend_from_slice(&[o.macd, o.signal, o.histogram]),
|
||||
None => out.extend_from_slice(&[f64::NAN; 3]),
|
||||
}
|
||||
}
|
||||
out
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn batch_macd_fast_path_is_bit_identical() {
|
||||
let series: Vec<f64> = (0..300)
|
||||
.map(|i| (f64::from(i) * 0.4).cos() * 10.0 + 100.0)
|
||||
.collect();
|
||||
let mut macd = MacdIndicator::classic();
|
||||
let got = macd.batch_macd(&series);
|
||||
assert!(bits_eq(&got, &macd_replay(&series)));
|
||||
// Sub-EMA + last state left where the replay would: continued update agrees.
|
||||
let mut ref_macd = MacdIndicator::classic();
|
||||
for &x in &series {
|
||||
ref_macd.update(x);
|
||||
}
|
||||
let (a, b) = (macd.update(101.0), ref_macd.update(101.0));
|
||||
assert_eq!(a.is_some(), b.is_some());
|
||||
assert_relative_eq!(a.unwrap().macd, b.unwrap().macd, epsilon = 1e-12);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn batch_macd_falls_back_on_non_finite() {
|
||||
let mut series: Vec<f64> = (0..60).map(|i| f64::from(i) + 100.0).collect();
|
||||
series[40] = f64::NAN;
|
||||
let mut macd = MacdIndicator::classic();
|
||||
assert!(bits_eq(&macd.batch_macd(&series), &macd_replay(&series)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn batch_macd_falls_back_when_not_fresh() {
|
||||
let series: Vec<f64> = (0..60).map(|i| f64::from(i) + 100.0).collect();
|
||||
let mut macd = MacdIndicator::classic();
|
||||
macd.update(50.0);
|
||||
let mut ref_macd = MacdIndicator::classic();
|
||||
ref_macd.update(50.0);
|
||||
let mut want = Vec::new();
|
||||
for &x in &series {
|
||||
match ref_macd.update(x) {
|
||||
Some(o) => want.extend_from_slice(&[o.macd, o.signal, o.histogram]),
|
||||
None => want.extend_from_slice(&[f64::NAN; 3]),
|
||||
}
|
||||
}
|
||||
assert!(bits_eq(&macd.batch_macd(&series), &want));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn batch_macd_too_short_for_output_falls_back() {
|
||||
// n < slow + signal - 1 (= 34): no full output, routed to the replay.
|
||||
let series: Vec<f64> = (0..20).map(|i| f64::from(i) + 100.0).collect();
|
||||
let mut macd = MacdIndicator::classic();
|
||||
let got = macd.batch_macd(&series);
|
||||
assert!(bits_eq(&got, &macd_replay(&series)));
|
||||
assert!(got.iter().all(|x| x.is_nan()));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn ignores_non_finite_input() {
|
||||
let mut macd = MacdIndicator::classic();
|
||||
|
||||
@@ -81,6 +81,76 @@ impl Rsi {
|
||||
self.last_value
|
||||
}
|
||||
|
||||
/// Vectorized batch returning one `f64` per input (`NaN` during warmup).
|
||||
///
|
||||
/// Shadows the generic [`BatchNanExt::batch_nan`](crate::BatchNanExt) blanket
|
||||
/// default. RSI is a recursive (IIR) filter — Wilder smoothing — so it cannot
|
||||
/// be SIMD-vectorized any more than the C peers manage; the win is purely in
|
||||
/// stripping per-tick overhead. For a fresh indicator over an all-finite slice
|
||||
/// long enough to seed (`n > period`) it runs the seed once and then the bare
|
||||
/// smoothing recurrence in a tight loop with no per-tick `is_finite`/`has_prev`/
|
||||
/// `avgs_seeded` branch and no `Option`, using the identical division at the
|
||||
/// seed and `mul_add`/`rsi_from_avgs` afterwards — so it is *bit-for-bit* equal
|
||||
/// to replaying `update`. Shorter or non-fresh/non-finite inputs defer to the
|
||||
/// exact `update` replay.
|
||||
pub fn batch_nan(&mut self, inputs: &[f64]) -> Vec<f64> {
|
||||
let p = self.period;
|
||||
let n = inputs.len();
|
||||
if self.has_prev
|
||||
|| self.avgs_seeded
|
||||
|| !self.seed_buf_gains.is_empty()
|
||||
|| n <= p
|
||||
|| !inputs.iter().all(|x| x.is_finite())
|
||||
{
|
||||
return inputs
|
||||
.iter()
|
||||
.map(|&x| self.update(x).unwrap_or(f64::NAN))
|
||||
.collect();
|
||||
}
|
||||
|
||||
// Warmup `[0, p)` is `NaN`; outputs from index `p` on are pushed once each.
|
||||
let mut out = vec![f64::NAN; p];
|
||||
out.reserve(n - p);
|
||||
// Seed from the first `period` diffs (inputs[1..=p]); index 0 only sets the
|
||||
// baseline. Retain the seed gains/losses exactly as `update` leaves them.
|
||||
let mut prev = inputs[0];
|
||||
let (mut sum_gain, mut sum_loss) = (0.0_f64, 0.0_f64);
|
||||
for &x in &inputs[1..=p] {
|
||||
let diff = x - prev;
|
||||
prev = x;
|
||||
let gain = if diff > 0.0 { diff } else { 0.0 };
|
||||
let loss = if diff < 0.0 { -diff } else { 0.0 };
|
||||
self.seed_buf_gains.push(gain);
|
||||
self.seed_buf_losses.push(loss);
|
||||
sum_gain += gain;
|
||||
sum_loss += loss;
|
||||
}
|
||||
let p_f64 = p as f64;
|
||||
let mut ag = sum_gain / p_f64;
|
||||
let mut al = sum_loss / p_f64;
|
||||
out.push(Self::rsi_from_avgs(ag, al));
|
||||
|
||||
// Steady state: Wilder smoothing, reciprocal hoisted, one `rsi_from_avgs`.
|
||||
for &x in &inputs[p + 1..] {
|
||||
let diff = x - prev;
|
||||
prev = x;
|
||||
let gain = if diff > 0.0 { diff } else { 0.0 };
|
||||
let loss = if diff < 0.0 { -diff } else { 0.0 };
|
||||
ag = ag.mul_add(self.n_minus_1, gain) * self.inv_period;
|
||||
al = al.mul_add(self.n_minus_1, loss) * self.inv_period;
|
||||
out.push(Self::rsi_from_avgs(ag, al));
|
||||
}
|
||||
|
||||
// Leave state where a full `update` replay would.
|
||||
self.prev_close = prev;
|
||||
self.has_prev = true;
|
||||
self.avg_gain = ag;
|
||||
self.avg_loss = al;
|
||||
self.avgs_seeded = true;
|
||||
self.last_value = Some(out[n - 1]);
|
||||
out
|
||||
}
|
||||
|
||||
fn rsi_from_avgs(avg_gain: f64, avg_loss: f64) -> f64 {
|
||||
// Algebraically `100 - 100/(1 + ag/al)` collapses to `100·ag/(ag+al)`,
|
||||
// which needs a single division instead of two and removes the separate
|
||||
@@ -376,6 +446,65 @@ mod tests {
|
||||
assert_eq!(rsi.value(), before);
|
||||
}
|
||||
|
||||
fn bits_eq(a: &[f64], b: &[f64]) -> bool {
|
||||
a.len() == b.len()
|
||||
&& a.iter()
|
||||
.zip(b)
|
||||
.all(|(x, y)| x == y || (x.is_nan() && y.is_nan()))
|
||||
}
|
||||
|
||||
fn rsi_replay(period: usize, series: &[f64]) -> Vec<f64> {
|
||||
let mut r = Rsi::new(period).unwrap();
|
||||
series
|
||||
.iter()
|
||||
.map(|&x| r.update(x).unwrap_or(f64::NAN))
|
||||
.collect()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn batch_nan_fast_path_is_bit_identical() {
|
||||
let series: Vec<f64> = (0..300)
|
||||
.map(|i| (f64::from(i) * 0.3).sin() * 5.0 + f64::from(i) * 0.1 + 100.0)
|
||||
.collect();
|
||||
let mut rsi = Rsi::new(14).unwrap();
|
||||
let got = rsi.batch_nan(&series);
|
||||
assert!(bits_eq(&got, &rsi_replay(14, &series)));
|
||||
let mut ref_rsi = Rsi::new(14).unwrap();
|
||||
for &x in &series {
|
||||
ref_rsi.update(x);
|
||||
}
|
||||
assert_eq!(rsi.update(123.0), ref_rsi.update(123.0));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn batch_nan_falls_back_on_non_finite() {
|
||||
let series = [10.0, 11.0, 9.0, f64::NAN, 12.0, 13.0, 8.0];
|
||||
let mut rsi = Rsi::new(3).unwrap();
|
||||
assert!(bits_eq(&rsi.batch_nan(&series), &rsi_replay(3, &series)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn batch_nan_falls_back_when_not_fresh() {
|
||||
let mut rsi = Rsi::new(3).unwrap();
|
||||
rsi.update(50.0);
|
||||
let series = [51.0, 49.0, 52.0, 53.0, 50.0];
|
||||
let mut ref_rsi = Rsi::new(3).unwrap();
|
||||
ref_rsi.update(50.0);
|
||||
let want: Vec<f64> = series
|
||||
.iter()
|
||||
.map(|&x| ref_rsi.update(x).unwrap_or(f64::NAN))
|
||||
.collect();
|
||||
assert!(bits_eq(&rsi.batch_nan(&series), &want));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn batch_nan_too_short_to_seed_falls_back() {
|
||||
// n <= period: routed to the exact replay (cannot seed yet).
|
||||
let series = [10.0, 11.0, 12.0];
|
||||
let mut rsi = Rsi::new(3).unwrap();
|
||||
assert!(bits_eq(&rsi.batch_nan(&series), &rsi_replay(3, &series)));
|
||||
}
|
||||
|
||||
proptest::proptest! {
|
||||
#![proptest_config(proptest::test_runner::Config::with_cases(48))]
|
||||
#[test]
|
||||
|
||||
@@ -86,6 +86,62 @@ impl Sma {
|
||||
None
|
||||
}
|
||||
}
|
||||
|
||||
/// Vectorized batch returning one `f64` per input (`NaN` during warmup).
|
||||
///
|
||||
/// Shadows the generic [`BatchNanExt::batch_nan`](crate::BatchNanExt) blanket
|
||||
/// default via inherent-method resolution. For a fresh, all-finite slice it
|
||||
/// inlines `update`'s rolling sum and drift-reseed, writing the mean as a bare
|
||||
/// `f64` (warmup → `NaN`) instead of allocating an `Option<f64>` per element
|
||||
/// and walking the result a second time. Same add/subtract order, same reseed
|
||||
/// cadence, same `sum / period` division — so it is *bit-for-bit* equal to
|
||||
/// replaying `update`, including the long-stream drift bound. Any other state,
|
||||
/// or a non-finite element, defers to the exact `update` replay.
|
||||
pub fn batch_nan(&mut self, inputs: &[f64]) -> Vec<f64> {
|
||||
let p = self.period;
|
||||
if self.count != 0
|
||||
|| self.updates_since_recompute != 0
|
||||
|| !inputs.iter().all(|x| x.is_finite())
|
||||
{
|
||||
return inputs
|
||||
.iter()
|
||||
.map(|&x| self.update(x).unwrap_or(f64::NAN))
|
||||
.collect();
|
||||
}
|
||||
|
||||
let p_f64 = p as f64;
|
||||
let mut out = Vec::with_capacity(inputs.len());
|
||||
for &x in inputs {
|
||||
if self.count == p {
|
||||
self.sum -= self.buf[self.head];
|
||||
self.buf[self.head] = x;
|
||||
self.sum += x;
|
||||
} else {
|
||||
self.buf[self.head] = x;
|
||||
self.sum += x;
|
||||
self.count += 1;
|
||||
}
|
||||
self.head += 1;
|
||||
if self.head == p {
|
||||
self.head = 0;
|
||||
}
|
||||
self.updates_since_recompute += 1;
|
||||
if self.updates_since_recompute >= RECOMPUTE_EVERY * p {
|
||||
self.sum = self.buf[self.head..]
|
||||
.iter()
|
||||
.chain(&self.buf[..self.head])
|
||||
.copied()
|
||||
.sum();
|
||||
self.updates_since_recompute = 0;
|
||||
}
|
||||
out.push(if self.count == p {
|
||||
self.sum / p_f64
|
||||
} else {
|
||||
f64::NAN
|
||||
});
|
||||
}
|
||||
out
|
||||
}
|
||||
}
|
||||
|
||||
impl Indicator for Sma {
|
||||
@@ -246,6 +302,69 @@ mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
/// NaN-aware bit-equality for the `f64`-with-NaN-warmup batch outputs.
|
||||
fn bits_eq(a: &[f64], b: &[f64]) -> bool {
|
||||
a.len() == b.len()
|
||||
&& a.iter()
|
||||
.zip(b)
|
||||
.all(|(x, y)| x == y || (x.is_nan() && y.is_nan()))
|
||||
}
|
||||
|
||||
fn sma_replay(period: usize, series: &[f64]) -> Vec<f64> {
|
||||
let mut s = Sma::new(period).unwrap();
|
||||
series
|
||||
.iter()
|
||||
.map(|&x| s.update(x).unwrap_or(f64::NAN))
|
||||
.collect()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn batch_nan_fast_path_is_bit_identical_with_reseed() {
|
||||
// > 16*period inputs so the drift-reseed branch fires inside batch_nan.
|
||||
let series: Vec<f64> = (0..500)
|
||||
.map(|i| (f64::from(i) * 0.2).sin() * 10.0 + 50.0)
|
||||
.collect();
|
||||
let mut sma = Sma::new(14).unwrap();
|
||||
let got = sma.batch_nan(&series);
|
||||
assert!(bits_eq(&got, &sma_replay(14, &series)));
|
||||
// State left where the replay would: continued updates agree.
|
||||
let mut ref_sma = Sma::new(14).unwrap();
|
||||
for &x in &series {
|
||||
ref_sma.update(x);
|
||||
}
|
||||
assert_eq!(sma.update(42.0), ref_sma.update(42.0));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn batch_nan_falls_back_on_non_finite() {
|
||||
let series = [1.0, 2.0, f64::NAN, 4.0, 5.0, 6.0];
|
||||
let mut sma = Sma::new(3).unwrap();
|
||||
assert!(bits_eq(&sma.batch_nan(&series), &sma_replay(3, &series)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn batch_nan_falls_back_when_not_fresh() {
|
||||
let mut sma = Sma::new(3).unwrap();
|
||||
sma.update(99.0);
|
||||
let series = [1.0, 2.0, 3.0, 4.0];
|
||||
let mut ref_sma = Sma::new(3).unwrap();
|
||||
ref_sma.update(99.0);
|
||||
let want: Vec<f64> = series
|
||||
.iter()
|
||||
.map(|&x| ref_sma.update(x).unwrap_or(f64::NAN))
|
||||
.collect();
|
||||
assert!(bits_eq(&sma.batch_nan(&series), &want));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn batch_nan_sub_period_slice_is_all_nan() {
|
||||
let series = [1.0, 2.0, 3.0];
|
||||
let mut sma = Sma::new(10).unwrap();
|
||||
let got = sma.batch_nan(&series);
|
||||
assert!(bits_eq(&got, &sma_replay(10, &series)));
|
||||
assert!(got.iter().all(|x| x.is_nan()));
|
||||
}
|
||||
|
||||
proptest::proptest! {
|
||||
#![proptest_config(proptest::test_runner::Config::with_cases(64))]
|
||||
#[test]
|
||||
|
||||
Reference in New Issue
Block a user