nautilus_backtest/
accumulator.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! Time event accumulation and scheduling for the backtest engine.
17
18use std::{
19    cmp::{Ordering, Reverse},
20    collections::BinaryHeap,
21};
22
23use nautilus_common::{clock::VirtualClock, timer::TimeEventHandler};
24use nautilus_core::UnixNanos;
25
26/// Provides a means of accumulating and draining time event handlers using a priority queue.
27///
28/// Events are maintained in timestamp order using a binary heap, allowing efficient
29/// retrieval of the next event to process.
30#[derive(Debug)]
31pub struct TimeEventAccumulator {
32    heap: BinaryHeap<Reverse<AccumulatedTimeEventHandler>>,
33    next_sequence: u64,
34}
35
36#[derive(Debug)]
37struct AccumulatedTimeEventHandler {
38    handler: TimeEventHandler,
39    // Replaces the handler's random event ID as the final heap tie-break.
40    sequence: u64,
41}
42
43impl AccumulatedTimeEventHandler {
44    fn cmp_key(&self, other: &Self) -> Ordering {
45        self.handler
46            .event
47            .ts_event
48            .cmp(&other.handler.event.ts_event)
49            .then_with(|| self.handler.event.name.cmp(&other.handler.event.name))
50            .then_with(|| self.handler.event.ts_init.cmp(&other.handler.event.ts_init))
51            .then_with(|| self.sequence.cmp(&other.sequence))
52    }
53}
54
55impl PartialOrd for AccumulatedTimeEventHandler {
56    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
57        Some(self.cmp(other))
58    }
59}
60
61impl PartialEq for AccumulatedTimeEventHandler {
62    fn eq(&self, other: &Self) -> bool {
63        self.cmp_key(other).is_eq()
64    }
65}
66
67impl Eq for AccumulatedTimeEventHandler {}
68
69impl Ord for AccumulatedTimeEventHandler {
70    fn cmp(&self, other: &Self) -> Ordering {
71        self.cmp_key(other)
72    }
73}
74
75impl TimeEventAccumulator {
76    /// Creates a new [`TimeEventAccumulator`] instance.
77    #[must_use]
78    pub fn new() -> Self {
79        Self {
80            heap: BinaryHeap::new(),
81            next_sequence: 0,
82        }
83    }
84
85    fn push(&mut self, handler: TimeEventHandler) {
86        let sequence = self.next_sequence;
87        self.next_sequence = self
88            .next_sequence
89            .checked_add(1)
90            .expect("Time event accumulator sequence overflow");
91        self.heap
92            .push(Reverse(AccumulatedTimeEventHandler { handler, sequence }));
93    }
94
95    /// Advance the given clock to the `to_time_ns` and push events to the heap.
96    pub fn advance_clock(
97        &mut self,
98        clock: &mut VirtualClock,
99        to_time_ns: UnixNanos,
100        set_time: bool,
101    ) {
102        let events = clock.advance_time(to_time_ns, set_time);
103        let handlers = clock.match_handlers(events);
104        for handler in handlers {
105            self.push(handler);
106        }
107    }
108
109    /// Peek at the next event timestamp without removing it.
110    ///
111    /// Returns `None` if the heap is empty.
112    #[must_use]
113    pub fn peek_next_time(&self) -> Option<UnixNanos> {
114        self.heap.peek().map(|h| h.0.handler.event.ts_event)
115    }
116
117    /// Pop the next event if its timestamp is at or before `ts_now`.
118    ///
119    /// Returns `None` if the heap is empty or the next event is after `ts_now`.
120    pub fn pop_next_at_or_before(&mut self, ts_now: UnixNanos) -> Option<TimeEventHandler> {
121        if self
122            .heap
123            .peek()
124            .is_some_and(|h| h.0.handler.event.ts_event <= ts_now)
125        {
126            self.heap.pop().map(|h| h.0.handler)
127        } else {
128            None
129        }
130    }
131
132    /// Check if the heap is empty.
133    #[must_use]
134    pub fn is_empty(&self) -> bool {
135        self.heap.is_empty()
136    }
137
138    /// Get the number of events in the heap.
139    #[must_use]
140    pub fn len(&self) -> usize {
141        self.heap.len()
142    }
143
144    /// Clear all events from the heap.
145    pub fn clear(&mut self) {
146        self.heap.clear();
147        self.next_sequence = 0;
148    }
149
150    /// Drain all events from the heap in timestamp order.
151    ///
152    /// This is provided for backwards compatibility with code that expects
153    /// batch processing. For iterative processing, prefer `pop_next_at_or_before`.
154    pub fn drain(&mut self) -> Vec<TimeEventHandler> {
155        let mut handlers = Vec::with_capacity(self.heap.len());
156        while let Some(scheduled) = self.heap.pop() {
157            handlers.push(scheduled.0.handler);
158        }
159        handlers
160    }
161}
162
163impl Default for TimeEventAccumulator {
164    /// Creates a new default [`TimeEventAccumulator`] instance.
165    fn default() -> Self {
166        Self::new()
167    }
168}
169
170#[cfg(test)]
171mod determinism_tests {
172    use nautilus_common::timer::{TimeEvent, TimeEventCallback};
173    use nautilus_core::UUID4;
174    use rstest::rstest;
175    use ustr::Ustr;
176
177    use super::*;
178
179    #[rstest]
180    fn test_accumulator_preserves_fifo_order_for_identical_timer_keys() {
181        let callback = TimeEventCallback::from(|_: TimeEvent| {});
182        let first_event = TimeEvent::new(
183            Ustr::from("REBALANCE"),
184            UUID4::from("ffffffff-ffff-4fff-bfff-ffffffffffff"),
185            100.into(),
186            100.into(),
187        );
188        let second_event = TimeEvent::new(
189            Ustr::from("REBALANCE"),
190            UUID4::from("00000000-0000-4000-8000-000000000000"),
191            100.into(),
192            100.into(),
193        );
194
195        let mut accumulator = TimeEventAccumulator::new();
196        accumulator.push(TimeEventHandler::new(first_event.clone(), callback.clone()));
197        accumulator.push(TimeEventHandler::new(second_event, callback));
198
199        let popped = accumulator
200            .pop_next_at_or_before(100.into())
201            .expect("Expected first accumulated timer");
202        assert_eq!(popped.event.event_id, first_event.event_id);
203    }
204}
205
206#[cfg(all(test, feature = "python"))]
207mod tests {
208    use nautilus_common::timer::{TimeEvent, TimeEventCallback};
209    use nautilus_core::UUID4;
210    use pyo3::{Py, Python, prelude::*, types::PyList};
211    use rstest::*;
212    use ustr::Ustr;
213
214    use super::*;
215
216    #[rstest]
217    fn test_accumulator_pop_in_order() {
218        Python::initialize();
219        Python::attach(|py| {
220            let py_list = PyList::empty(py);
221            let py_append = Py::from(py_list.getattr("append").unwrap());
222
223            let mut accumulator = TimeEventAccumulator::new();
224
225            let time_event1 = TimeEvent::new(
226                Ustr::from("TEST_EVENT_1"),
227                UUID4::new(),
228                100.into(),
229                100.into(),
230            );
231            let time_event2 = TimeEvent::new(
232                Ustr::from("TEST_EVENT_2"),
233                UUID4::new(),
234                300.into(),
235                300.into(),
236            );
237            let time_event3 = TimeEvent::new(
238                Ustr::from("TEST_EVENT_3"),
239                UUID4::new(),
240                200.into(),
241                200.into(),
242            );
243
244            let callback = TimeEventCallback::from(py_append.into_any());
245
246            let handler1 = TimeEventHandler::new(time_event1.clone(), callback.clone());
247            let handler2 = TimeEventHandler::new(time_event2.clone(), callback.clone());
248            let handler3 = TimeEventHandler::new(time_event3.clone(), callback);
249
250            accumulator.push(handler1);
251            accumulator.push(handler2);
252            accumulator.push(handler3);
253            assert_eq!(accumulator.len(), 3);
254
255            let popped1 = accumulator.pop_next_at_or_before(1000.into()).unwrap();
256            assert_eq!(popped1.event.ts_event, time_event1.ts_event);
257
258            let popped2 = accumulator.pop_next_at_or_before(1000.into()).unwrap();
259            assert_eq!(popped2.event.ts_event, time_event3.ts_event);
260
261            let popped3 = accumulator.pop_next_at_or_before(1000.into()).unwrap();
262            assert_eq!(popped3.event.ts_event, time_event2.ts_event);
263
264            assert!(accumulator.is_empty());
265        });
266    }
267
268    #[rstest]
269    fn test_accumulator_pop_same_timestamp_in_name_order() {
270        Python::initialize();
271        Python::attach(|py| {
272            let py_list = PyList::empty(py);
273            let py_append = Py::from(py_list.getattr("append").unwrap());
274
275            let mut accumulator = TimeEventAccumulator::new();
276            let callback = TimeEventCallback::from(py_append.into_any());
277
278            let spread_event = TimeEvent::new(
279                Ustr::from("SPREAD_QUOTE_ESM4"),
280                UUID4::new(),
281                100.into(),
282                100.into(),
283            );
284            let time_bar_event = TimeEvent::new(
285                Ustr::from("TIME_BAR_ESM4-2-MINUTE-ASK-INTERNAL"),
286                UUID4::new(),
287                100.into(),
288                100.into(),
289            );
290
291            accumulator.push(TimeEventHandler::new(
292                time_bar_event.clone(),
293                callback.clone(),
294            ));
295            accumulator.push(TimeEventHandler::new(spread_event.clone(), callback));
296
297            let popped1 = accumulator.pop_next_at_or_before(100.into()).unwrap();
298            assert_eq!(popped1.event.ts_event, spread_event.ts_event);
299            assert_eq!(popped1.event.name, spread_event.name);
300
301            let popped2 = accumulator.pop_next_at_or_before(100.into()).unwrap();
302            assert_eq!(popped2.event.ts_event, time_bar_event.ts_event);
303            assert_eq!(popped2.event.name, time_bar_event.name);
304        });
305    }
306
307    #[rstest]
308    fn test_accumulator_pop_respects_timestamp() {
309        Python::initialize();
310        Python::attach(|py| {
311            let py_list = PyList::empty(py);
312            let py_append = Py::from(py_list.getattr("append").unwrap());
313
314            let mut accumulator = TimeEventAccumulator::new();
315
316            let time_event1 = TimeEvent::new(
317                Ustr::from("TEST_EVENT_1"),
318                UUID4::new(),
319                100.into(),
320                100.into(),
321            );
322            let time_event2 = TimeEvent::new(
323                Ustr::from("TEST_EVENT_2"),
324                UUID4::new(),
325                300.into(),
326                300.into(),
327            );
328
329            let callback = TimeEventCallback::from(py_append.into_any());
330
331            accumulator.push(TimeEventHandler::new(time_event1.clone(), callback.clone()));
332            accumulator.push(TimeEventHandler::new(time_event2.clone(), callback));
333
334            let popped1 = accumulator.pop_next_at_or_before(200.into()).unwrap();
335            assert_eq!(popped1.event.ts_event, time_event1.ts_event);
336
337            // Event at 300 should not be returned with ts=200
338            assert!(accumulator.pop_next_at_or_before(200.into()).is_none());
339
340            let popped2 = accumulator.pop_next_at_or_before(300.into()).unwrap();
341            assert_eq!(popped2.event.ts_event, time_event2.ts_event);
342        });
343    }
344
345    #[rstest]
346    fn test_peek_next_time() {
347        Python::initialize();
348        Python::attach(|py| {
349            let py_list = PyList::empty(py);
350            let py_append = Py::from(py_list.getattr("append").unwrap());
351            let callback = TimeEventCallback::from(py_append.into_any());
352
353            let mut accumulator = TimeEventAccumulator::new();
354            assert!(accumulator.peek_next_time().is_none());
355
356            let time_event1 = TimeEvent::new(
357                Ustr::from("TEST_EVENT_1"),
358                UUID4::new(),
359                200.into(),
360                200.into(),
361            );
362            let time_event2 = TimeEvent::new(
363                Ustr::from("TEST_EVENT_2"),
364                UUID4::new(),
365                100.into(),
366                100.into(),
367            );
368
369            accumulator.push(TimeEventHandler::new(time_event1, callback.clone()));
370            assert_eq!(accumulator.peek_next_time(), Some(200.into()));
371
372            accumulator.push(TimeEventHandler::new(time_event2, callback));
373            assert_eq!(accumulator.peek_next_time(), Some(100.into()));
374        });
375    }
376
377    #[rstest]
378    fn test_drain_returns_in_order() {
379        Python::initialize();
380        Python::attach(|py| {
381            let py_list = PyList::empty(py);
382            let py_append = Py::from(py_list.getattr("append").unwrap());
383            let callback = TimeEventCallback::from(py_append.into_any());
384
385            let mut accumulator = TimeEventAccumulator::new();
386
387            for ts in [300u64, 100, 200] {
388                let event = TimeEvent::new(Ustr::from("TEST"), UUID4::new(), ts.into(), ts.into());
389                accumulator.push(TimeEventHandler::new(event, callback.clone()));
390            }
391
392            let handlers = accumulator.drain();
393
394            assert_eq!(handlers.len(), 3);
395            assert_eq!(handlers[0].event.ts_event.as_u64(), 100);
396            assert_eq!(handlers[1].event.ts_event.as_u64(), 200);
397            assert_eq!(handlers[2].event.ts_event.as_u64(), 300);
398            assert!(accumulator.is_empty());
399        });
400    }
401
402    #[rstest]
403    fn test_interleaved_push_pop_maintains_order() {
404        Python::initialize();
405        Python::attach(|py| {
406            let py_list = PyList::empty(py);
407            let py_append = Py::from(py_list.getattr("append").unwrap());
408            let callback = TimeEventCallback::from(py_append.into_any());
409
410            let mut accumulator = TimeEventAccumulator::new();
411            let mut popped_timestamps: Vec<u64> = Vec::new();
412
413            for ts in [100u64, 300] {
414                let event = TimeEvent::new(Ustr::from("TEST"), UUID4::new(), ts.into(), ts.into());
415                accumulator.push(TimeEventHandler::new(event, callback.clone()));
416            }
417
418            let handler = accumulator.pop_next_at_or_before(1000.into()).unwrap();
419            popped_timestamps.push(handler.event.ts_event.as_u64());
420
421            // Simulate callback scheduling new event at 150 (between popped 100 and pending 300)
422            let event = TimeEvent::new(Ustr::from("NEW"), UUID4::new(), 150.into(), 150.into());
423            accumulator.push(TimeEventHandler::new(event, callback));
424
425            while let Some(handler) = accumulator.pop_next_at_or_before(1000.into()) {
426                popped_timestamps.push(handler.event.ts_event.as_u64());
427            }
428
429            assert_eq!(popped_timestamps, vec![100, 150, 300]);
430        });
431    }
432
433    #[rstest]
434    fn test_same_timestamp_events() {
435        Python::initialize();
436        Python::attach(|py| {
437            let py_list = PyList::empty(py);
438            let py_append = Py::from(py_list.getattr("append").unwrap());
439            let callback = TimeEventCallback::from(py_append.into_any());
440
441            let mut accumulator = TimeEventAccumulator::new();
442
443            for i in 0..3 {
444                let event = TimeEvent::new(
445                    Ustr::from(&format!("EVENT_{i}")),
446                    UUID4::new(),
447                    100.into(),
448                    100.into(),
449                );
450                accumulator.push(TimeEventHandler::new(event, callback.clone()));
451            }
452
453            let mut count = 0;
454
455            while let Some(handler) = accumulator.pop_next_at_or_before(100.into()) {
456                assert_eq!(handler.event.ts_event.as_u64(), 100);
457                count += 1;
458            }
459            assert_eq!(count, 3);
460        });
461    }
462
463    #[rstest]
464    fn test_pop_at_exact_timestamp_boundary() {
465        Python::initialize();
466        Python::attach(|py| {
467            let py_list = PyList::empty(py);
468            let py_append = Py::from(py_list.getattr("append").unwrap());
469            let callback = TimeEventCallback::from(py_append.into_any());
470
471            let mut accumulator = TimeEventAccumulator::new();
472
473            let event = TimeEvent::new(Ustr::from("TEST"), UUID4::new(), 100.into(), 100.into());
474            accumulator.push(TimeEventHandler::new(event, callback));
475
476            let handler = accumulator.pop_next_at_or_before(100.into());
477            assert!(handler.is_some());
478            assert_eq!(handler.unwrap().event.ts_event.as_u64(), 100);
479
480            let event2 = TimeEvent::new(Ustr::from("TEST2"), UUID4::new(), 200.into(), 200.into());
481            accumulator.push(TimeEventHandler::new(
482                event2,
483                TimeEventCallback::from(Py::from(py_list.getattr("append").unwrap()).into_any()),
484            ));
485
486            assert!(accumulator.pop_next_at_or_before(199.into()).is_none());
487            assert!(accumulator.pop_next_at_or_before(200.into()).is_some());
488        });
489    }
490}
nautilus_backtest/accumulator.rs

nautilus_backtest/
accumulator.rs
