From 2cebb3cca11f9825f4f2006d913fa8856d2531d3 Mon Sep 17 00:00:00 2001
From: kingchenc <kingchencp@gmail.com>
Date: Fri, 22 May 2026 12:22:33 +0200
Subject: [PATCH] C10: reject same-bucket out-of-order ticks
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

push rejected ticks that went backwards across buckets but absorbed any
tick whose timestamp fell inside the open bucket — including one older
than the last tick already absorbed. Such a stale tick silently
overwrote the bar's close with an outdated price.

Track last_ts on OpenBar (set in from_tick, advanced in absorb) and, on
the same-bucket path, reject a tick whose timestamp predates it with
Error::Malformed, leaving the open bar untouched. Ticks that share a
timestamp are still accepted, since several trades can land in the same
millisecond.
---
 crates/wickra-data/src/aggregator.rs | 47 ++++++++++++++++++++++++++--
 1 file changed, 44 insertions(+), 3 deletions(-)

diff --git a/crates/wickra-data/src/aggregator.rs b/crates/wickra-data/src/aggregator.rs
index 321af6da..5b1493f0 100644
--- a/crates/wickra-data/src/aggregator.rs
+++ b/crates/wickra-data/src/aggregator.rs
@@ -83,6 +83,10 @@ pub struct TickAggregator {
 #[derive(Debug, Clone, Copy)]
 struct OpenBar {
     bucket_start: i64,
+    /// Timestamp of the most recently absorbed tick. Used to reject ticks that
+    /// arrive out of order *within* the current bucket — without it an older
+    /// tick would silently overwrite `close` with a stale price.
+    last_ts: i64,
     open: f64,
     high: f64,
     low: f64,
@@ -94,6 +98,7 @@ impl OpenBar {
     fn from_tick(t: Tick, bucket_start: i64) -> Self {
         Self {
             bucket_start,
+            last_ts: t.timestamp,
             open: t.price,
             high: t.price,
             low: t.price,
@@ -111,6 +116,7 @@ impl OpenBar {
         }
         self.close = t.price;
         self.volume += t.volume;
+        self.last_ts = t.timestamp;
     }
 
     fn into_candle(self) -> Candle {
@@ -158,9 +164,10 @@ impl TickAggregator {
     /// per skipped bucket.
     ///
     /// # Errors
-    /// Returns [`Error::Malformed`] if `tick.timestamp` is strictly less than
-    /// the start of the currently open bar (out-of-order ticks are not
-    /// supported), or if gap filling overflows the timestamp range.
+    /// Returns [`Error::Malformed`] if `tick.timestamp` goes backwards — both
+    /// across buckets (older than the open bar's start) and within a bucket
+    /// (older than the last tick absorbed into it) — or if gap filling
+    /// overflows the timestamp range. Ticks sharing a timestamp are accepted.
     pub fn push(&mut self, tick: Tick) -> Result<Vec<Candle>> {
         let bucket = self.timeframe.floor(tick.timestamp);
         if let Some(mut bar) = self.open_bar {
@@ -181,6 +188,16 @@ impl TickAggregator {
                 self.open_bar = Some(OpenBar::from_tick(tick, bucket));
                 return Ok(out);
             }
+            // Same bucket: reject a tick that predates the last one absorbed,
+            // which would otherwise overwrite `close` with a stale price.
+            // Equal timestamps are allowed — several trades can share a
+            // millisecond.
+            if tick.timestamp < bar.last_ts {
+                return Err(Error::Malformed(format!(
+                    "tick timestamp {} predates the last tick {} in the same bucket",
+                    tick.timestamp, bar.last_ts
+                )));
+            }
             bar.absorb(tick);
             self.open_bar = Some(bar);
             return Ok(Vec::new());
@@ -303,6 +320,30 @@ mod tests {
         assert!(matches!(err, Error::Malformed(_)));
     }
 
+    #[test]
+    fn rejects_same_bucket_out_of_order_tick() {
+        let mut agg = TickAggregator::new(Timeframe::new(60).unwrap());
+        agg.push(t(10.0, 50)).unwrap();
+        // ts=10 is still bucket 0 but predates the tick at ts=50 — rejecting
+        // it prevents a stale price silently overwriting `close`.
+        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);
+    }
+
+    #[test]
+    fn accepts_same_bucket_ticks_sharing_a_timestamp() {
+        let mut agg = TickAggregator::new(Timeframe::new(60).unwrap());
+        agg.push(t(10.0, 20)).unwrap();
+        // 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();
+        assert_eq!(bar.high, 12.0);
+        assert_eq!(bar.close, 11.0);
+    }
+
     #[test]
     fn skips_empty_buckets_without_gap_fill() {
         let mut agg = TickAggregator::new(Timeframe::new(60).unwrap());