Refine LiveTimer accuracy

RELEASES.md

@@ -6,6 +6,7 @@ Released on TBD (UTC).
 - Implemented `FeeModel` including `FixedFeeModel` and `MakerTakerFeeModel` (#1584), thanks @rsmb7z
 - Implemented `TradeTickDataWrangler.process_bar_data` (#1585), thanks @rsmb7z
 - Implemented multiple timeframe bar execution (will use lowest timeframe per instrument)
+- Optimized `LiveTimer` efficiency and accuracy with `tokio` timer under the hood
 - Optimized `QuoteTickDataWrangler` and `TradeTickDataWrangler` (#1590), thanks @rsmb7z
 - Standardized adapter client logging (handle more logging from client base classes)
 - Simplified and consolidated Rust `OrderBook` design

examples/live/binance/binance_spot_market_maker.py

@@ -44,7 +44,7 @@
         log_level="INFO",
         # log_level_file="DEBUG",
         # log_file_format="json",
-        use_pyo3=False,
+        use_pyo3=True,
     ),
     exec_engine=LiveExecEngineConfig(
         reconciliation=True,

nautilus_core/common/src/timer.rs

@@ -25,6 +25,7 @@ use std::{
 
 use nautilus_core::{
     correctness::{check_positive_u64, check_valid_string},
+    datetime::floor_to_nearest_microsecond,
     nanos::{TimedeltaNanos, UnixNanos},
     time::get_atomic_clock_realtime,
     uuid::UUID4,
@@ -35,7 +36,7 @@ use tokio::{
     sync::oneshot,
     time::{Duration, Instant},
 };
-use tracing::error;
+use tracing::{debug, error, trace};
 use ustr::Ustr;
 
 use crate::{handlers::EventHandler, runtime::get_runtime};
@@ -241,6 +242,7 @@ impl LiveTimer {
         check_valid_string(name, stringify!(name))?;
         check_positive_u64(interval_ns, stringify!(interval_ns))?;
 
+        debug!("Creating timer '{}'", name);
         Ok(Self {
             name: Ustr::from(name),
             interval_ns,
@@ -259,12 +261,16 @@ impl LiveTimer {
 
     pub fn start(&mut self) {
         let event_name = self.name;
-        let start_time_ns = self.start_time_ns;
         let stop_time_ns = self.stop_time_ns;
+        let mut start_time_ns = self.start_time_ns;
+        let next_time_ns = self.next_time_ns;
         let interval_ns = self.interval_ns;
         let is_expired = self.is_expired.clone();
         let callback = self.callback.clone();
 
+        // Floor the next time to the nearest microsecond which is within the timers accuracy
+        let mut next_time_ns = UnixNanos::from(floor_to_nearest_microsecond(next_time_ns.into()));
+
         // Setup oneshot channel for cancelling timer task
         let (cancel_tx, mut cancel_rx) = oneshot::channel();
         self.canceler = Some(cancel_tx);
@@ -273,22 +279,38 @@ impl LiveTimer {
         rt.spawn(async move {
             let clock = get_atomic_clock_realtime();
             let now_ns = clock.get_time_ns();
-            let start_instant = if start_time_ns <= now_ns {
+
+            if start_time_ns == 0 {
+                // No start was specified so start immediately
+                start_time_ns = now_ns;
+            }
+
+            let start = if next_time_ns <= now_ns {
                 Instant::now()
             } else {
-                let delay_duration = Duration::from_nanos((start_time_ns - now_ns).into());
-                Instant::now() + delay_duration
+                // Timer initialization delay
+                let delay = Duration::from_millis(1);
+                let diff: u64 = (next_time_ns - now_ns).into();
+                Instant::now() + Duration::from_nanos(diff) - delay
+            };
+
+            if let Some(stop_time_ns) = stop_time_ns {
+                assert!(stop_time_ns > now_ns, "stop_time was < now_ns");
+                assert!(
+                    start_time_ns + interval_ns <= stop_time_ns,
+                    "start_time + interval was > stop_time"
+                )
             };
 
-            let mut timer = tokio::time::interval_at(start_instant, Duration::from_millis(interval_ns));
-            let mut next_time_ns = start_time_ns + interval_ns;
+            let mut timer = tokio::time::interval_at(start, Duration::from_nanos(interval_ns));
 
             loop {
                 // SAFETY: `timer.tick` is cancellation safe, if the cancel branch completes
                 // first then no tick has been consumed (no event was ready).
                 tokio::select! {
                     _ = timer.tick() => {
-                        call_python_with_time_event(event_name, next_time_ns, clock.get_time_ns(), &callback);
+                        let now_ns = clock.get_time_ns();
+                        call_python_with_time_event(event_name, next_time_ns, now_ns, &callback);
 
                         // Prepare next time interval
                         next_time_ns += interval_ns;
@@ -301,6 +323,7 @@ impl LiveTimer {
                         }
                     },
                     _ = (&mut cancel_rx) => {
+                        trace!("Received timer cancel");
                         break; // Timer canceled
                     },
                 }
@@ -314,6 +337,7 @@ impl LiveTimer {
 
     /// Cancels the timer (the timer will not generate an event).
     pub fn cancel(&mut self) -> anyhow::Result<()> {
+        debug!("Cancel timer '{}'", self.name);
         if let Some(sender) = self.canceler.take() {
             // Send cancellation signal
             sender.send(()).map_err(|e| anyhow::anyhow!("{:?}", e))?;

nautilus_core/core/src/datetime.rs

@@ -92,6 +92,11 @@ pub fn unix_nanos_to_iso8601(unix_nanos: UnixNanos) -> String {
     dt.to_rfc3339_opts(SecondsFormat::Nanos, true)
 }
 
+/// Floor the given UNIX nanoseconds to the nearest microsecond.
+pub fn floor_to_nearest_microsecond(unix_nanos: u64) -> u64 {
+    (unix_nanos / NANOSECONDS_IN_MICROSECOND) * NANOSECONDS_IN_MICROSECOND
+}
+
 pub fn last_weekday_nanos(year: i32, month: u32, day: u32) -> anyhow::Result<UnixNanos> {
     let date =
         NaiveDate::from_ymd_opt(year, month, day).ok_or_else(|| anyhow::anyhow!("Invalid date"))?;

nautilus_trader/common/component.pyx

@@ -771,14 +771,6 @@ cdef class LiveClock(Clock):
         if callback is not None:
             callback = create_pyo3_conversion_wrapper(callback)
 
-        cdef uint64_t ts_now = self.timestamp_ns()  # Call here for greater accuracy
-
-        if start_time_ns == 0:
-            start_time_ns = ts_now
-        if stop_time_ns:
-            Condition.true(stop_time_ns > ts_now, "stop_time was < ts_now")
-            Condition.true(start_time_ns + interval_ns <= stop_time_ns, "start_time + interval was > stop_time")
-
         live_clock_set_timer(
             &self._mem,
             pystr_to_cstr(name),