13.0.0: introduce MillisecondPeriod and pass this into TimeRule and AlignedTimeRule to support sample periods smaller than a second. Different logic in both time rules to avoid drift of trigger timestamp

Author: MathisWellmann

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "trade_aggregation"
-version = "12.0.3"
+version = "13.0.0"
 authors = ["MathisWellmann <[email protected]>"]
 edition = "2021"
 license-file = "LICENSE"

img/relative_price_candles_plot.png

@@ -1,4 +1,4 @@
-use crate::{aggregation_rules::TimestampResolution, AggregationRule, ModularCandle, TakerTrade};
+use crate::{AggregationRule, MillisecondPeriod, ModularCandle, TakerTrade, TimestampResolution};
 
 /// The classic time based aggregation rule,
 /// creating a new candle every n seconds.  The time trigger is aligned such that
@@ -13,7 +13,7 @@ pub struct AlignedTimeRule {
     // The period for the candle in seconds
     // constants can be used nicely here from constants.rs
     // e.g.: M1 -> 1 minute candles
-    period_s: i64,
+    period_in_units_from_trade: i64,
 }
 
 impl AlignedTimeRule {
@@ -24,17 +24,20 @@ impl AlignedTimeRule {
     /// period_s: How many seconds a candle will contain
     /// ts_res: The resolution each Trade timestamp will have
     ///
-    pub fn new(period_s: i64, ts_res: TimestampResolution) -> Self {
-        let ts_multiplier = match ts_res {
-            TimestampResolution::Second => 1,
-            TimestampResolution::Millisecond => 1_000,
-            TimestampResolution::Microsecond => 1_000_000,
-            TimestampResolution::Nanosecond => 1_000_000_000,
+    pub fn new(
+        period_ms: MillisecondPeriod,
+        trade_timestamp_resolution: TimestampResolution,
+    ) -> Self {
+        use TimestampResolution::*;
+        let ts_multiplier = match trade_timestamp_resolution {
+            Millisecond => 1,
+            Microsecond => 1_000,
+            Nanosecond => 1_000_000,
         };
 
         Self {
             reference_timestamp: 0,
-            period_s: period_s * ts_multiplier,
+            period_in_units_from_trade: period_ms.get() as i64 * ts_multiplier,
         }
     }
 
@@ -43,7 +46,7 @@ impl AlignedTimeRule {
     /// each period.
     #[must_use]
     pub fn aligned_timestamp(&self, timestamp: i64) -> i64 {
-        timestamp - (timestamp % self.period_s)
+        timestamp - (timestamp % self.period_in_units_from_trade)
     }
 }
 
@@ -58,9 +61,14 @@ where
             return false;
         }
 
-        let should_trigger = trade.timestamp() - self.reference_timestamp >= self.period_s;
+        let should_trigger =
+            trade.timestamp() - self.reference_timestamp >= self.period_in_units_from_trade;
         if should_trigger {
-            self.reference_timestamp = self.aligned_timestamp(trade.timestamp());
+            // Advance the trigger timestamp to the next period.
+            // If the period is too small, then the trade timestamps will out-pace and always trigger.
+            // Alternatively doing `self.reference_timestamp = self.aligned_timestamp(trade.timestamp())` will cause drift
+            // and not produce enough samples.
+            self.reference_timestamp = self.reference_timestamp + self.period_in_units_from_trade
         }
 
         should_trigger

src/aggregation_rules/aligned_time_rule.rs

@@ -1,57 +1,40 @@
-use crate::{AggregationRule, ModularCandle, TakerTrade};
-
-/// The resolution of the "TakerTrade" timestamps
-#[derive(Debug, Clone, Copy)]
-pub enum TimestampResolution {
-    /// The timestamp of the TakerTrade is measured in seconds
-    Second,
-
-    /// The timestamp of the TakerTrade is measured in milliseconds
-    Millisecond,
-
-    /// The timestamp of the TakerTrade is measured in microseconds
-    Microsecond,
-
-    /// The timestamp of the TakerTrade is measured in nanoseconds
-    Nanosecond,
-}
+use crate::{AggregationRule, MillisecondPeriod, ModularCandle, TakerTrade, TimestampResolution};
 
 /// The classic time based aggregation rule,
 /// creating a new candle every n seconds
 #[derive(Debug, Clone)]
 pub struct TimeRule {
-    /// If true, the reference timestamp needs to be reset
-    init: bool,
-
     // The timestamp this rule uses as a reference
+    // in the unit of the incoming trades.
     reference_timestamp: i64,
 
-    // The period for the candle in seconds
-    // constants can be used nicely here from constants.rs
-    // e.g.: M1 -> 1 minute candles
-    period_s: i64,
+    // The period for the candle in the timestamp resolution of the candle as provided in the constructor.
+    period_in_units_from_trade: i64,
 }
 
 impl TimeRule {
     /// Create a new instance of the time rule,
     /// with a given candle period in seconds
     ///
     /// # Arguments:
-    /// period_s: How many seconds a candle will contain
-    /// ts_res: The resolution each Trade timestamp will have
+    /// `period_ms`: How many milliseconds a candle will contain.
+    /// `trade_timestamp_resolution`: The resolution each Trade timestamp will have
     ///
-    pub fn new(period_s: i64, ts_res: TimestampResolution) -> Self {
-        let ts_multiplier = match ts_res {
-            TimestampResolution::Second => 1,
-            TimestampResolution::Millisecond => 1_000,
-            TimestampResolution::Microsecond => 1_000_000,
-            TimestampResolution::Nanosecond => 1_000_000_000,
+    pub fn new(
+        period_ms: MillisecondPeriod,
+        trade_timestamp_resolution: TimestampResolution,
+    ) -> Self {
+        use TimestampResolution::*;
+        // Given the timestamp resolution of the trades, a certain multiplier is required to compute the number of units for the sample period.
+        let ts_multiplier = match trade_timestamp_resolution {
+            Millisecond => 1,
+            Microsecond => 1_000,
+            Nanosecond => 1_000_000,
         };
 
         Self {
-            init: true,
             reference_timestamp: 0,
-            period_s: period_s * ts_multiplier,
+            period_in_units_from_trade: period_ms.get() as i64 * ts_multiplier,
         }
     }
 }
@@ -62,13 +45,17 @@ where
     T: TakerTrade,
 {
     fn should_trigger(&mut self, trade: &T, _candle: &C) -> bool {
-        if self.init {
+        if self.reference_timestamp == 0 {
             self.reference_timestamp = trade.timestamp();
-            self.init = false;
         }
-        let should_trigger = trade.timestamp() - self.reference_timestamp > self.period_s;
+        let should_trigger =
+            trade.timestamp() - self.reference_timestamp > self.period_in_units_from_trade;
         if should_trigger {
-            self.init = true;
+            // Advance the trigger timestamp to the next period.
+            // If the period is too small, then the trade timestamps will out-pace and always trigger.
+            // Alternatively doing `self.reference_timestamp = trade.timestamp()` will cause drift
+            // and not produce enough samples.
+            self.reference_timestamp = self.reference_timestamp + self.period_in_units_from_trade;
         }
 
         should_trigger
@@ -107,14 +94,14 @@ mod tests {
             false,
         );
         let candles = aggregate_all_trades(&trades, &mut aggregator);
-        assert_eq!(candles.len(), 395);
+        assert_eq!(candles.len(), 396);
 
         let mut aggregator = GenericAggregator::<OhlcCandle, TimeRule, Trade>::new(
             TimeRule::new(M5, TimestampResolution::Millisecond),
             false,
         );
         let candles = aggregate_all_trades(&trades, &mut aggregator);
-        assert_eq!(candles.len(), 1180);
+        assert_eq!(candles.len(), 1190);
 
         let mut aggregator = GenericAggregator::<OhlcCandle, TimeRule, Trade>::new(
             TimeRule::new(H1, TimestampResolution::Millisecond),
@@ -130,14 +117,6 @@ mod tests {
         let trades_ms = load_trades_from_csv("data/Bitmex_XBTUSD_1M.csv").unwrap();
 
         // we can therefore transform them into seconds, microseconds and nanoseconds respectively
-        let trades_s: Vec<Trade> = trades_ms
-            .iter()
-            .map(|v| Trade {
-                timestamp: v.timestamp / 1000,
-                price: v.price,
-                size: v.size,
-            })
-            .collect();
         let trades_micros: Vec<Trade> = trades_ms
             .iter()
             .map(|v| Trade {
@@ -155,33 +134,26 @@ mod tests {
             })
             .collect();
 
-        // And make sure they produce the same number of candles regardless of timestamp resolution
-        let mut aggregator = GenericAggregator::<OhlcCandle, TimeRule, Trade>::new(
-            TimeRule::new(M15, TimestampResolution::Second),
-            false,
-        );
-        let candles = aggregate_all_trades(&trades_s, &mut aggregator);
-        assert_eq!(candles.len(), 395);
-
+        // And make sure they produce the same number of candles given the differing timestamp resolutions.
         let mut aggregator = GenericAggregator::<OhlcCandle, TimeRule, Trade>::new(
             TimeRule::new(M15, TimestampResolution::Millisecond),
             false,
         );
         let candles = aggregate_all_trades(&trades_ms, &mut aggregator);
-        assert_eq!(candles.len(), 395);
+        assert_eq!(candles.len(), 396);
 
         let mut aggregator = GenericAggregator::<OhlcCandle, TimeRule, Trade>::new(
             TimeRule::new(M15, TimestampResolution::Microsecond),
             false,
         );
         let candles = aggregate_all_trades(&trades_micros, &mut aggregator);
-        assert_eq!(candles.len(), 395);
+        assert_eq!(candles.len(), 396);
 
         let mut aggregator = GenericAggregator::<OhlcCandle, TimeRule, Trade>::new(
             TimeRule::new(M15, TimestampResolution::Nanosecond),
             false,
         );
         let candles = aggregate_all_trades(&trades_ns, &mut aggregator);
-        assert_eq!(candles.len(), 395);
+        assert_eq!(candles.len(), 396);
     }
 }

src/aggregation_rules/time_rule.rs

@@ -133,7 +133,7 @@ mod tests {
                 candle_counter += 1;
             }
         }
-        assert_eq!(candle_counter, 5704);
+        assert_eq!(candle_counter, 5953);
     }
 
     #[test]

src/aggregator.rs

@@ -43,7 +43,6 @@ impl<T: TakerTrade> CandleComponentUpdate<T> for TimeVelocity {
     #[inline(always)]
     fn update(&mut self, trade: &T) {
         let div = match trade.timestamp_resolution() {
-            TimestampResolution::Second => 1,
             TimestampResolution::Millisecond => 1_000,
             TimestampResolution::Microsecond => 1_000_000,
             TimestampResolution::Nanosecond => 1_000_000_000,

src/candle_components/time_velocity.rs

@@ -1,29 +1,31 @@
+use crate::MillisecondPeriod;
+
 /// 1 Minute candle period
-pub const M1: i64 = 60;
+pub const M1: MillisecondPeriod = MillisecondPeriod::from_non_zero_secs(60);
 
 /// 5 Minute candle period
-pub const M5: i64 = 300;
+pub const M5: MillisecondPeriod = MillisecondPeriod::from_non_zero_secs(300);
 
 /// 15 Minute candle period
-pub const M15: i64 = 900;
+pub const M15: MillisecondPeriod = MillisecondPeriod::from_non_zero_secs(900);
 
 /// 30 Minute candle period
-pub const M30: i64 = 1800;
+pub const M30: MillisecondPeriod = MillisecondPeriod::from_non_zero_secs(1800);
 
 /// 1 Hour candle period
-pub const H1: i64 = 3600;
+pub const H1: MillisecondPeriod = MillisecondPeriod::from_non_zero_secs(3600);
 
 /// 2 Hour candle period
-pub const H2: i64 = 7200;
+pub const H2: MillisecondPeriod = MillisecondPeriod::from_non_zero_secs(7200);
 
 /// 4 Hour candle period
-pub const H4: i64 = 14400;
+pub const H4: MillisecondPeriod = MillisecondPeriod::from_non_zero_secs(14400);
 
 /// 8 Hour candle period
-pub const H8: i64 = 28800;
+pub const H8: MillisecondPeriod = MillisecondPeriod::from_non_zero_secs(28800);
 
 /// 12 Hour candle period
-pub const H12: i64 = 43200;
+pub const H12: MillisecondPeriod = MillisecondPeriod::from_non_zero_secs(43200);
 
 /// 1 Day candle period
-pub const D1: i64 = 86400;
+pub const D1: MillisecondPeriod = MillisecondPeriod::from_non_zero_secs(86400);

src/constants.rs

@@ -1,5 +1,3 @@
-use crate::TimestampResolution;
-
 #[derive(Default, Debug, Clone, Copy, PartialEq)]
 #[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
 /// Defines a taker trade
@@ -65,3 +63,68 @@ pub trait TakerTrade {
     /// that the trade was a sell order, taking liquidity from the bid.
     fn size(&self) -> f64;
 }
+
+/// The resolution of the "TakerTrade" timestamps
+#[derive(Debug, Clone, Copy)]
+pub enum TimestampResolution {
+    /// The timestamp of the TakerTrade is measured in milliseconds
+    Millisecond,
+
+    /// The timestamp of the TakerTrade is measured in microseconds
+    Microsecond,
+
+    /// The timestamp of the TakerTrade is measured in nanoseconds
+    Nanosecond,
+}
+
+/// A period measured in milliseconds which must be non-zero.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
+pub struct MillisecondPeriod(u64);
+
+impl MillisecondPeriod {
+    /// Try to create the `MillisecondPeriod` from millisecond units.
+    /// # Panics:
+    /// If `millis` is zero, the contract was violated.
+    pub fn from_non_zero(millis: u64) -> Self {
+        assert!(millis > 0, "`millis` must be non-zero that was the deal");
+        Self(millis)
+    }
+
+    /// Try to create the `MillisecondPeriod` from seconds.
+    /// # Panics:
+    /// Because this is used in a const context, it is not yet possible to do `Option::unwrap` and thus
+    /// it is asserted that `seconds` is non-zero
+    pub const fn from_non_zero_secs(seconds: u64) -> Self {
+        assert!(seconds > 0, "`seconds` must be non-zero that was the deal");
+        Self(seconds * 1_000)
+    }
+
+    /// Get the inner value
+    pub fn get(self) -> u64 {
+        self.0
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+
+    #[test]
+    #[should_panic]
+    fn millisecond_period_from_non_zero_secs_panic() {
+        // panics as it cannot be zero
+        MillisecondPeriod::from_non_zero_secs(0);
+    }
+
+    #[test]
+    fn millisecond_period_from_non_zero_secs() {
+        assert_eq!(
+            MillisecondPeriod::from_non_zero_secs(1),
+            MillisecondPeriod(1_000)
+        );
+        assert_eq!(
+            MillisecondPeriod::from_non_zero_secs(60),
+            MillisecondPeriod(60_000)
+        );
+    }
+}