From f0471ba824bb6cea10adfb0d1f1ead651dc9b4bf Mon Sep 17 00:00:00 2001 From: kingchenc Date: Fri, 22 May 2026 12:25:56 +0200 Subject: [PATCH] C11: validate volume when finalising aggregated candles MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit OpenBar::into_candle and RolledBar::into_candle built their result with Candle::new_unchecked, skipping the finiteness check. volume is summed across every absorbed tick/candle, so a long or large run can drift it to +inf — and an inf-volume candle would silently poison every downstream indicator. Switch both to Candle::new, which validates volume finiteness, and return Result. The OHLC fields are finite and correctly ordered by construction, so the only invariant Candle::new can reject here is a non-finite volume. push propagates the error with `?`; both flush methods now return Result> and resample_all pulls the result through. --- crates/wickra-data/src/aggregator.rs | 45 +++++++++++++++++++++------- crates/wickra-data/src/resample.rs | 43 +++++++++++++++++++++----- 2 files changed, 71 insertions(+), 17 deletions(-) diff --git a/crates/wickra-data/src/aggregator.rs b/crates/wickra-data/src/aggregator.rs index 5b1493f0..0580a78a 100644 --- a/crates/wickra-data/src/aggregator.rs +++ b/crates/wickra-data/src/aggregator.rs @@ -119,8 +119,17 @@ impl OpenBar { self.last_ts = t.timestamp; } - fn into_candle(self) -> Candle { - Candle::new_unchecked( + /// Finalise the bar into a validated [`Candle`]. + /// + /// # Errors + /// Returns [`Error::Core`] if the accumulated `volume` is no longer finite. + /// `volume` is summed across every absorbed tick, so an astronomically + /// long or large run can drift it to `inf`; emitting such a candle would + /// silently poison every downstream indicator, so it is surfaced instead. + /// The OHLC fields are finite and correctly ordered by construction, so + /// `Candle::new` only ever rejects this bar for a non-finite volume. + fn into_candle(self) -> Result { + Candle::new( self.open, self.high, self.low, @@ -128,6 +137,7 @@ impl OpenBar { self.volume, self.bucket_start, ) + .map_err(Error::from) } } @@ -179,7 +189,7 @@ impl TickAggregator { } if bucket > bar.bucket_start { // Close the previous bar and start a new one with this tick. - let closed = bar.into_candle(); + let closed = bar.into_candle()?; let mut out = Vec::with_capacity(1); out.push(closed); if self.fill_gaps { @@ -229,8 +239,12 @@ impl TickAggregator { /// Drain the currently open bar (if any) and return it. Useful at the end of /// a backtest or when shutting down a live aggregator. - pub fn flush(&mut self) -> Option { - self.open_bar.take().map(OpenBar::into_candle) + /// + /// # Errors + /// Returns an error if the open bar's accumulated volume is non-finite + /// (see [`OpenBar::into_candle`]). + pub fn flush(&mut self) -> Result> { + self.open_bar.take().map(OpenBar::into_candle).transpose() } /// Configured timeframe. @@ -284,7 +298,7 @@ mod tests { assert!(agg.push(t(12.0, 15)).unwrap().is_empty()); assert!(agg.push(t(8.0, 30)).unwrap().is_empty()); assert!(agg.push(t(11.0, 50)).unwrap().is_empty()); - let bar = agg.flush().expect("open bar"); + let bar = agg.flush().unwrap().expect("open bar"); assert_eq!(bar.open, 10.0); assert_eq!(bar.high, 12.0); assert_eq!(bar.low, 8.0); @@ -307,7 +321,7 @@ mod tests { assert_eq!(closed.close, 12.0); // The new tick at ts=60 opens the next bar. - let still_open = agg.flush().unwrap(); + let still_open = agg.flush().unwrap().unwrap(); assert_eq!(still_open.open, 15.0); assert_eq!(still_open.timestamp, 60); } @@ -329,7 +343,7 @@ mod tests { let err = agg.push(t(99.0, 10)).unwrap_err(); assert!(matches!(err, Error::Malformed(_))); // The open bar is untouched: close is still the ts=50 price. - assert_eq!(agg.flush().unwrap().close, 10.0); + assert_eq!(agg.flush().unwrap().unwrap().close, 10.0); } #[test] @@ -339,11 +353,22 @@ mod tests { // Two trades in the same millisecond are legitimate. agg.push(t(12.0, 20)).unwrap(); agg.push(t(11.0, 20)).unwrap(); - let bar = agg.flush().unwrap(); + let bar = agg.flush().unwrap().unwrap(); assert_eq!(bar.high, 12.0); assert_eq!(bar.close, 11.0); } + #[test] + fn flushes_a_non_finite_volume_as_an_error() { + let mut agg = TickAggregator::new(Timeframe::new(60).unwrap()); + // Two near-max volumes sum to +inf — the closed candle would carry a + // non-finite volume that poisons every downstream indicator. + agg.push(Tick::new(10.0, f64::MAX, 0).unwrap()).unwrap(); + agg.push(Tick::new(10.0, f64::MAX, 1).unwrap()).unwrap(); + let err = agg.flush().unwrap_err(); + assert!(matches!(err, Error::Core(_))); + } + #[test] fn skips_empty_buckets_without_gap_fill() { let mut agg = TickAggregator::new(Timeframe::new(60).unwrap()); @@ -379,7 +404,7 @@ mod tests { } // The tick at ts=200 opens bucket 180. - assert_eq!(agg.flush().unwrap().timestamp, 180); + assert_eq!(agg.flush().unwrap().unwrap().timestamp, 180); } #[test] diff --git a/crates/wickra-data/src/resample.rs b/crates/wickra-data/src/resample.rs index 9f097a89..5c131c31 100644 --- a/crates/wickra-data/src/resample.rs +++ b/crates/wickra-data/src/resample.rs @@ -49,8 +49,16 @@ impl RolledBar { self.volume += c.volume; } - fn into_candle(self) -> Candle { - Candle::new_unchecked( + /// Finalise the rolled bar into a validated [`Candle`]. + /// + /// # Errors + /// Returns [`Error::Core`] if the accumulated `volume` is no longer finite. + /// `volume` is summed across every absorbed candle, so a long or large run + /// can drift it to `inf`; emitting such a candle would silently poison + /// every downstream indicator, so it is surfaced instead. The OHLC fields + /// are finite and correctly ordered by construction. + fn into_candle(self) -> Result { + Candle::new( self.open, self.high, self.low, @@ -58,6 +66,7 @@ impl RolledBar { self.volume, self.bucket_start, ) + .map_err(Error::from) } } @@ -86,7 +95,7 @@ impl Resampler { Ok(None) } Some(bar) if bucket > bar.bucket_start => { - let closed = bar.into_candle(); + let closed = bar.into_candle()?; self.open = Some(RolledBar::from_candle(candle, bucket)); Ok(Some(closed)) } @@ -102,8 +111,12 @@ impl Resampler { } /// Flush the currently open coarser bar, if any. - pub fn flush(&mut self) -> Option { - self.open.take().map(RolledBar::into_candle) + /// + /// # Errors + /// Returns an error if the open bar's accumulated volume is non-finite + /// (see [`RolledBar::into_candle`]). + pub fn flush(&mut self) -> Result> { + self.open.take().map(RolledBar::into_candle).transpose() } } @@ -121,7 +134,7 @@ where out.push(closed); } } - if let Some(last) = r.flush() { + if let Some(last) = r.flush()? { out.push(last); } Ok(out) @@ -174,8 +187,24 @@ mod tests { let mut r = Resampler::new(Timeframe::new(5).unwrap()); assert!(r.push(c(0, 10.0, 11.0, 9.0, 10.5, 1.0)).unwrap().is_none()); assert!(r.push(c(3, 10.5, 12.0, 10.0, 11.0, 1.0)).unwrap().is_none()); - let bar = r.flush().unwrap(); + let bar = r.flush().unwrap().unwrap(); assert_eq!(bar.high, 12.0); assert_eq!(bar.low, 9.0); } + + #[test] + fn flushes_a_non_finite_volume_as_an_error() { + let mut r = Resampler::new(Timeframe::new(5).unwrap()); + // Two near-max volumes in the same bucket sum to +inf. + assert!(r + .push(c(0, 10.0, 11.0, 9.0, 10.5, f64::MAX)) + .unwrap() + .is_none()); + assert!(r + .push(c(1, 10.0, 11.0, 9.0, 10.5, f64::MAX)) + .unwrap() + .is_none()); + let err = r.flush().unwrap_err(); + assert!(matches!(err, Error::Core(_))); + } }