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::TestClock, 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(&mut self, clock: &mut TestClock, to_time_ns: UnixNanos, set_time: bool) {
97        let events = clock.advance_time(to_time_ns, set_time);
98        let handlers = clock.match_handlers(events);
99        for handler in handlers {
100            self.push(handler);
101        }
102    }
103
104    /// Peek at the next event timestamp without removing it.
105    ///
106    /// Returns `None` if the heap is empty.
107    #[must_use]
108    pub fn peek_next_time(&self) -> Option<UnixNanos> {
109        self.heap.peek().map(|h| h.0.handler.event.ts_event)
110    }
111
112    /// Pop the next event if its timestamp is at or before `ts`.
113    ///
114    /// Returns `None` if the heap is empty or the next event is after `ts`.
115    pub fn pop_next_at_or_before(&mut self, ts: UnixNanos) -> Option<TimeEventHandler> {
116        if self
117            .heap
118            .peek()
119            .is_some_and(|h| h.0.handler.event.ts_event <= ts)
120        {
121            self.heap.pop().map(|h| h.0.handler)
122        } else {
123            None
124        }
125    }
126
127    /// Check if the heap is empty.
128    #[must_use]
129    pub fn is_empty(&self) -> bool {
130        self.heap.is_empty()
131    }
132
133    /// Get the number of events in the heap.
134    #[must_use]
135    pub fn len(&self) -> usize {
136        self.heap.len()
137    }
138
139    /// Clear all events from the heap.
140    pub fn clear(&mut self) {
141        self.heap.clear();
142        self.next_sequence = 0;
143    }
144
145    /// Drain all events from the heap in timestamp order.
146    ///
147    /// This is provided for backwards compatibility with code that expects
148    /// batch processing. For iterative processing, prefer `pop_next_at_or_before`.
149    pub fn drain(&mut self) -> Vec<TimeEventHandler> {
150        let mut handlers = Vec::with_capacity(self.heap.len());
151        while let Some(scheduled) = self.heap.pop() {
152            handlers.push(scheduled.0.handler);
153        }
154        handlers
155    }
156}
157
158impl Default for TimeEventAccumulator {
159    /// Creates a new default [`TimeEventAccumulator`] instance.
160    fn default() -> Self {
161        Self::new()
162    }
163}
164
165#[cfg(test)]
166mod determinism_tests {
167    use nautilus_common::timer::{TimeEvent, TimeEventCallback};
168    use nautilus_core::UUID4;
169    use rstest::rstest;
170    use ustr::Ustr;
171
172    use super::*;
173
174    #[rstest]
175    fn test_accumulator_preserves_fifo_order_for_identical_timer_keys() {
176        let callback = TimeEventCallback::from(|_: TimeEvent| {});
177        let first_event = TimeEvent::new(
178            Ustr::from("REBALANCE"),
179            UUID4::from("ffffffff-ffff-4fff-bfff-ffffffffffff"),
180            100.into(),
181            100.into(),
182        );
183        let second_event = TimeEvent::new(
184            Ustr::from("REBALANCE"),
185            UUID4::from("00000000-0000-4000-8000-000000000000"),
186            100.into(),
187            100.into(),
188        );
189
190        let mut accumulator = TimeEventAccumulator::new();
191        accumulator.push(TimeEventHandler::new(first_event.clone(), callback.clone()));
192        accumulator.push(TimeEventHandler::new(second_event, callback));
193
194        let popped = accumulator
195            .pop_next_at_or_before(100.into())
196            .expect("Expected first accumulated timer");
197        assert_eq!(popped.event.event_id, first_event.event_id);
198    }
199}
200
201#[cfg(all(test, feature = "python"))]
202mod tests {
203    use nautilus_common::timer::{TimeEvent, TimeEventCallback};
204    use nautilus_core::UUID4;
205    use pyo3::{Py, Python, prelude::*, types::PyList};
206    use rstest::*;
207    use ustr::Ustr;
208
209    use super::*;
210
211    #[rstest]
212    fn test_accumulator_pop_in_order() {
213        Python::initialize();
214        Python::attach(|py| {
215            let py_list = PyList::empty(py);
216            let py_append = Py::from(py_list.getattr("append").unwrap());
217
218            let mut accumulator = TimeEventAccumulator::new();
219
220            let time_event1 = TimeEvent::new(
221                Ustr::from("TEST_EVENT_1"),
222                UUID4::new(),
223                100.into(),
224                100.into(),
225            );
226            let time_event2 = TimeEvent::new(
227                Ustr::from("TEST_EVENT_2"),
228                UUID4::new(),
229                300.into(),
230                300.into(),
231            );
232            let time_event3 = TimeEvent::new(
233                Ustr::from("TEST_EVENT_3"),
234                UUID4::new(),
235                200.into(),
236                200.into(),
237            );
238
239            let callback = TimeEventCallback::from(py_append.into_any());
240
241            let handler1 = TimeEventHandler::new(time_event1.clone(), callback.clone());
242            let handler2 = TimeEventHandler::new(time_event2.clone(), callback.clone());
243            let handler3 = TimeEventHandler::new(time_event3.clone(), callback);
244
245            accumulator.push(handler1);
246            accumulator.push(handler2);
247            accumulator.push(handler3);
248            assert_eq!(accumulator.len(), 3);
249
250            let popped1 = accumulator.pop_next_at_or_before(1000.into()).unwrap();
251            assert_eq!(popped1.event.ts_event, time_event1.ts_event);
252
253            let popped2 = accumulator.pop_next_at_or_before(1000.into()).unwrap();
254            assert_eq!(popped2.event.ts_event, time_event3.ts_event);
255
256            let popped3 = accumulator.pop_next_at_or_before(1000.into()).unwrap();
257            assert_eq!(popped3.event.ts_event, time_event2.ts_event);
258
259            assert!(accumulator.is_empty());
260        });
261    }
262
263    #[rstest]
264    fn test_accumulator_pop_same_timestamp_in_name_order() {
265        Python::initialize();
266        Python::attach(|py| {
267            let py_list = PyList::empty(py);
268            let py_append = Py::from(py_list.getattr("append").unwrap());
269
270            let mut accumulator = TimeEventAccumulator::new();
271            let callback = TimeEventCallback::from(py_append.into_any());
272
273            let spread_event = TimeEvent::new(
274                Ustr::from("SPREAD_QUOTE_ESM4"),
275                UUID4::new(),
276                100.into(),
277                100.into(),
278            );
279            let time_bar_event = TimeEvent::new(
280                Ustr::from("TIME_BAR_ESM4-2-MINUTE-ASK-INTERNAL"),
281                UUID4::new(),
282                100.into(),
283                100.into(),
284            );
285
286            accumulator.push(TimeEventHandler::new(
287                time_bar_event.clone(),
288                callback.clone(),
289            ));
290            accumulator.push(TimeEventHandler::new(spread_event.clone(), callback));
291
292            let popped1 = accumulator.pop_next_at_or_before(100.into()).unwrap();
293            assert_eq!(popped1.event.ts_event, spread_event.ts_event);
294            assert_eq!(popped1.event.name, spread_event.name);
295
296            let popped2 = accumulator.pop_next_at_or_before(100.into()).unwrap();
297            assert_eq!(popped2.event.ts_event, time_bar_event.ts_event);
298            assert_eq!(popped2.event.name, time_bar_event.name);
299        });
300    }
301
302    #[rstest]
303    fn test_accumulator_pop_respects_timestamp() {
304        Python::initialize();
305        Python::attach(|py| {
306            let py_list = PyList::empty(py);
307            let py_append = Py::from(py_list.getattr("append").unwrap());
308
309            let mut accumulator = TimeEventAccumulator::new();
310
311            let time_event1 = TimeEvent::new(
312                Ustr::from("TEST_EVENT_1"),
313                UUID4::new(),
314                100.into(),
315                100.into(),
316            );
317            let time_event2 = TimeEvent::new(
318                Ustr::from("TEST_EVENT_2"),
319                UUID4::new(),
320                300.into(),
321                300.into(),
322            );
323
324            let callback = TimeEventCallback::from(py_append.into_any());
325
326            accumulator.push(TimeEventHandler::new(time_event1.clone(), callback.clone()));
327            accumulator.push(TimeEventHandler::new(time_event2.clone(), callback));
328
329            let popped1 = accumulator.pop_next_at_or_before(200.into()).unwrap();
330            assert_eq!(popped1.event.ts_event, time_event1.ts_event);
331
332            // Event at 300 should not be returned with ts=200
333            assert!(accumulator.pop_next_at_or_before(200.into()).is_none());
334
335            let popped2 = accumulator.pop_next_at_or_before(300.into()).unwrap();
336            assert_eq!(popped2.event.ts_event, time_event2.ts_event);
337        });
338    }
339
340    #[rstest]
341    fn test_peek_next_time() {
342        Python::initialize();
343        Python::attach(|py| {
344            let py_list = PyList::empty(py);
345            let py_append = Py::from(py_list.getattr("append").unwrap());
346            let callback = TimeEventCallback::from(py_append.into_any());
347
348            let mut accumulator = TimeEventAccumulator::new();
349            assert!(accumulator.peek_next_time().is_none());
350
351            let time_event1 = TimeEvent::new(
352                Ustr::from("TEST_EVENT_1"),
353                UUID4::new(),
354                200.into(),
355                200.into(),
356            );
357            let time_event2 = TimeEvent::new(
358                Ustr::from("TEST_EVENT_2"),
359                UUID4::new(),
360                100.into(),
361                100.into(),
362            );
363
364            accumulator.push(TimeEventHandler::new(time_event1, callback.clone()));
365            assert_eq!(accumulator.peek_next_time(), Some(200.into()));
366
367            accumulator.push(TimeEventHandler::new(time_event2, callback));
368            assert_eq!(accumulator.peek_next_time(), Some(100.into()));
369        });
370    }
371
372    #[rstest]
373    fn test_drain_returns_in_order() {
374        Python::initialize();
375        Python::attach(|py| {
376            let py_list = PyList::empty(py);
377            let py_append = Py::from(py_list.getattr("append").unwrap());
378            let callback = TimeEventCallback::from(py_append.into_any());
379
380            let mut accumulator = TimeEventAccumulator::new();
381
382            for ts in [300u64, 100, 200] {
383                let event = TimeEvent::new(Ustr::from("TEST"), UUID4::new(), ts.into(), ts.into());
384                accumulator.push(TimeEventHandler::new(event, callback.clone()));
385            }
386
387            let handlers = accumulator.drain();
388
389            assert_eq!(handlers.len(), 3);
390            assert_eq!(handlers[0].event.ts_event.as_u64(), 100);
391            assert_eq!(handlers[1].event.ts_event.as_u64(), 200);
392            assert_eq!(handlers[2].event.ts_event.as_u64(), 300);
393            assert!(accumulator.is_empty());
394        });
395    }
396
397    #[rstest]
398    fn test_interleaved_push_pop_maintains_order() {
399        Python::initialize();
400        Python::attach(|py| {
401            let py_list = PyList::empty(py);
402            let py_append = Py::from(py_list.getattr("append").unwrap());
403            let callback = TimeEventCallback::from(py_append.into_any());
404
405            let mut accumulator = TimeEventAccumulator::new();
406            let mut popped_timestamps: Vec<u64> = Vec::new();
407
408            for ts in [100u64, 300] {
409                let event = TimeEvent::new(Ustr::from("TEST"), UUID4::new(), ts.into(), ts.into());
410                accumulator.push(TimeEventHandler::new(event, callback.clone()));
411            }
412
413            let handler = accumulator.pop_next_at_or_before(1000.into()).unwrap();
414            popped_timestamps.push(handler.event.ts_event.as_u64());
415
416            // Simulate callback scheduling new event at 150 (between popped 100 and pending 300)
417            let event = TimeEvent::new(Ustr::from("NEW"), UUID4::new(), 150.into(), 150.into());
418            accumulator.push(TimeEventHandler::new(event, callback));
419
420            while let Some(handler) = accumulator.pop_next_at_or_before(1000.into()) {
421                popped_timestamps.push(handler.event.ts_event.as_u64());
422            }
423
424            assert_eq!(popped_timestamps, vec![100, 150, 300]);
425        });
426    }
427
428    #[rstest]
429    fn test_same_timestamp_events() {
430        Python::initialize();
431        Python::attach(|py| {
432            let py_list = PyList::empty(py);
433            let py_append = Py::from(py_list.getattr("append").unwrap());
434            let callback = TimeEventCallback::from(py_append.into_any());
435
436            let mut accumulator = TimeEventAccumulator::new();
437
438            for i in 0..3 {
439                let event = TimeEvent::new(
440                    Ustr::from(&format!("EVENT_{i}")),
441                    UUID4::new(),
442                    100.into(),
443                    100.into(),
444                );
445                accumulator.push(TimeEventHandler::new(event, callback.clone()));
446            }
447
448            let mut count = 0;
449
450            while let Some(handler) = accumulator.pop_next_at_or_before(100.into()) {
451                assert_eq!(handler.event.ts_event.as_u64(), 100);
452                count += 1;
453            }
454            assert_eq!(count, 3);
455        });
456    }
457
458    #[rstest]
459    fn test_pop_at_exact_timestamp_boundary() {
460        Python::initialize();
461        Python::attach(|py| {
462            let py_list = PyList::empty(py);
463            let py_append = Py::from(py_list.getattr("append").unwrap());
464            let callback = TimeEventCallback::from(py_append.into_any());
465
466            let mut accumulator = TimeEventAccumulator::new();
467
468            let event = TimeEvent::new(Ustr::from("TEST"), UUID4::new(), 100.into(), 100.into());
469            accumulator.push(TimeEventHandler::new(event, callback));
470
471            let handler = accumulator.pop_next_at_or_before(100.into());
472            assert!(handler.is_some());
473            assert_eq!(handler.unwrap().event.ts_event.as_u64(), 100);
474
475            let event2 = TimeEvent::new(Ustr::from("TEST2"), UUID4::new(), 200.into(), 200.into());
476            accumulator.push(TimeEventHandler::new(
477                event2,
478                TimeEventCallback::from(Py::from(py_list.getattr("append").unwrap()).into_any()),
479            ));
480
481            assert!(accumulator.pop_next_at_or_before(199.into()).is_none());
482            assert!(accumulator.pop_next_at_or_before(200.into()).is_some());
483        });
484    }
485}
nautilus_backtest/accumulator.rs

nautilus_backtest/
accumulator.rs
