nautilus_data/engine/
book.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
16use std::{cell::RefCell, num::NonZeroUsize, rc::Rc};
17
18use indexmap::IndexMap;
19use nautilus_common::{
20    cache::Cache,
21    msgbus::{self, Handler, MStr, Topic, switchboard},
22    timer::TimeEvent,
23};
24use nautilus_model::{
25    data::{OrderBookDeltas, OrderBookDepth10, QuoteTick},
26    enums::InstrumentClass,
27    identifiers::{InstrumentId, Venue},
28    instruments::Instrument,
29    orderbook::OrderBook,
30};
31use ustr::Ustr;
32
33/// Contains information for creating snapshots of specific order books.
34#[derive(Clone, Debug)]
35pub struct BookSnapshotInfo {
36    pub instrument_id: InstrumentId,
37    pub venue: Venue,
38    /// Parent expansion components `(root, class)` when this snapshot subscription
39    /// targets a parent symbol. `None` for concrete (exact-instrument) subscriptions.
40    pub parent: Option<(Ustr, InstrumentClass)>,
41    pub topic: MStr<Topic>,
42    pub interval_ms: NonZeroUsize,
43}
44
45/// Reference-counted map of per-instrument book snapshot descriptors.
46///
47/// Shared between the engine (which populates it on subscribe) and the
48/// [`BookSnapshotter`] timer callback (which iterates it on each tick).
49pub(crate) type BookSnapshotInfos = Rc<RefCell<IndexMap<InstrumentId, BookSnapshotInfo>>>;
50
51/// Reference count key for a book snapshot subscription.
52pub(crate) type BookSnapshotKey = (InstrumentId, NonZeroUsize);
53
54/// Outcome of decrementing a book snapshot subscription.
55pub(crate) enum BookSnapshotUnsubscribeResult {
56    /// No matching subscription was found.
57    NotSubscribed,
58    /// The reference count was decremented but other consumers remain.
59    Decremented,
60    /// The last consumer was removed; tear down associated state.
61    Removed,
62}
63
64/// Handles order book updates and delta processing for a specific instrument.
65///
66/// The `BookUpdater` processes incoming order book deltas and maintains
67/// the current state of an order book. It can handle both incremental
68/// updates and full snapshots for the instrument it's assigned to.
69#[derive(Debug)]
70pub struct BookUpdater {
71    pub id: Ustr,
72    pub instrument_id: InstrumentId,
73    pub cache: Rc<RefCell<Cache>>,
74    pub emit_quotes_from_book: bool,
75}
76
77impl BookUpdater {
78    /// Creates a new [`BookUpdater`] instance.
79    pub fn new(
80        instrument_id: &InstrumentId,
81        cache: Rc<RefCell<Cache>>,
82        emit_quotes_from_book: bool,
83    ) -> Self {
84        Self {
85            id: Ustr::from(&format!("{}-{}", stringify!(BookUpdater), instrument_id)),
86            instrument_id: *instrument_id,
87            cache,
88            emit_quotes_from_book,
89        }
90    }
91}
92
93impl Handler<OrderBookDeltas> for BookUpdater {
94    fn id(&self) -> Ustr {
95        self.id
96    }
97
98    fn handle(&self, deltas: &OrderBookDeltas) {
99        let mut emit: Option<QuoteTick> = None;
100        {
101            let mut cache = self.cache.borrow_mut();
102            if let Some(book) = cache.order_book_mut(&deltas.instrument_id) {
103                if let Err(e) = book.apply_deltas(deltas) {
104                    log::error!("Failed to apply deltas: {e}");
105                    return;
106                }
107
108                if self.emit_quotes_from_book {
109                    emit = derive_quote_from_book(book);
110                }
111            }
112        }
113
114        if let Some(quote) = emit {
115            publish_quote_if_changed(&self.cache, quote);
116        }
117    }
118}
119
120impl Handler<OrderBookDepth10> for BookUpdater {
121    fn id(&self) -> Ustr {
122        self.id
123    }
124
125    fn handle(&self, depth: &OrderBookDepth10) {
126        let mut emit: Option<QuoteTick> = None;
127        {
128            let mut cache = self.cache.borrow_mut();
129            if let Some(book) = cache.order_book_mut(&depth.instrument_id) {
130                if let Err(e) = book.apply_depth(depth) {
131                    log::error!("Failed to apply depth: {e}");
132                    return;
133                }
134
135                if self.emit_quotes_from_book {
136                    emit = derive_quote_from_book(book);
137                }
138            }
139        }
140
141        if let Some(quote) = emit {
142            publish_quote_if_changed(&self.cache, quote);
143        }
144    }
145}
146
147fn derive_quote_from_book(book: &OrderBook) -> Option<QuoteTick> {
148    let bid_price = book.best_bid_price()?;
149    let ask_price = book.best_ask_price()?;
150    let bid_size = book.best_bid_size()?;
151    let ask_size = book.best_ask_size()?;
152
153    if bid_size.raw == 0 || ask_size.raw == 0 {
154        return None;
155    }
156
157    Some(QuoteTick::new(
158        book.instrument_id,
159        bid_price,
160        ask_price,
161        bid_size,
162        ask_size,
163        book.ts_last,
164        book.ts_last,
165    ))
166}
167
168/// Publishes the derived `QuoteTick` if top-of-book changed.
169///
170/// Writes to cache and republishes only when bid/ask price or size differs
171/// from the cached quote.
172pub(crate) fn publish_quote_if_changed(cache: &Rc<RefCell<Cache>>, quote: QuoteTick) {
173    let publish = {
174        let cache_ref = cache.borrow();
175        match cache_ref.quote(&quote.instrument_id) {
176            None => true,
177            Some(last) => {
178                last.bid_price != quote.bid_price
179                    || last.ask_price != quote.ask_price
180                    || last.bid_size != quote.bid_size
181                    || last.ask_size != quote.ask_size
182            }
183        }
184    };
185
186    if !publish {
187        return;
188    }
189
190    if let Err(e) = cache.borrow_mut().add_quote(quote) {
191        log::error!("Error on cache insert: {e}");
192    }
193
194    let topic = switchboard::get_quotes_topic(quote.instrument_id);
195    msgbus::publish_quote(topic, &quote);
196}
197
198/// Creates periodic snapshots of order books at configured intervals.
199///
200/// The `BookSnapshotter` generates order book snapshots on timer events,
201/// publishing them as market data. This is useful for providing periodic
202/// full order book state updates in addition to incremental delta updates.
203#[derive(Debug)]
204pub struct BookSnapshotter {
205    pub timer_name: Ustr,
206    pub interval_ms: NonZeroUsize,
207    pub snapshot_infos: Rc<RefCell<IndexMap<InstrumentId, BookSnapshotInfo>>>,
208    pub cache: Rc<RefCell<Cache>>,
209}
210
211impl BookSnapshotter {
212    /// Creates a new [`BookSnapshotter`] instance.
213    pub fn new(
214        interval_ms: NonZeroUsize,
215        snapshot_infos: Rc<RefCell<IndexMap<InstrumentId, BookSnapshotInfo>>>,
216        cache: Rc<RefCell<Cache>>,
217    ) -> Self {
218        let timer_name = format!("OrderBookSnapshots|{interval_ms}");
219
220        Self {
221            timer_name: Ustr::from(&timer_name),
222            interval_ms,
223            snapshot_infos,
224            cache,
225        }
226    }
227
228    /// Publishes a snapshot for each subscribed book.
229    ///
230    /// Books are cloned out of the cache inside a scoped borrow before publishing,
231    /// so subscribers can mutably borrow the cache (e.g. a strategy submitting an
232    /// order from `on_book`).
233    pub fn snapshot(&self, _event: TimeEvent) {
234        let snapshot_infos: Vec<BookSnapshotInfo> =
235            self.snapshot_infos.borrow().values().cloned().collect();
236
237        log::debug!(
238            "BookSnapshotter.snapshot called for {} subscriptions at {}ms",
239            snapshot_infos.len(),
240            self.interval_ms,
241        );
242
243        let books: Vec<(MStr<Topic>, OrderBook)> = {
244            let cache = self.cache.borrow();
245            let mut books = Vec::new();
246
247            for snap_info in &snapshot_infos {
248                self.collect_snapshot(snap_info, &cache, &mut books);
249            }
250
251            books
252        };
253
254        for (topic, book) in books {
255            msgbus::publish_book(topic, &book);
256        }
257    }
258
259    fn collect_snapshot(
260        &self,
261        snap_info: &BookSnapshotInfo,
262        cache: &Cache,
263        books: &mut Vec<(MStr<Topic>, OrderBook)>,
264    ) {
265        if let Some((root, class)) = snap_info.parent {
266            let topic = snap_info.topic;
267            for instrument in cache.instruments_by_parent(&snap_info.venue, &root, class) {
268                self.collect_order_book(&instrument.id(), topic, cache, books);
269            }
270        } else {
271            self.collect_order_book(&snap_info.instrument_id, snap_info.topic, cache, books);
272        }
273    }
274
275    fn collect_order_book(
276        &self,
277        instrument_id: &InstrumentId,
278        topic: MStr<Topic>,
279        cache: &Cache,
280        books: &mut Vec<(MStr<Topic>, OrderBook)>,
281    ) {
282        let book = match cache.try_order_book(instrument_id) {
283            Ok(book) => book,
284            Err(e) => {
285                log::error!("Cannot publish OrderBook snapshot: {e}");
286                return;
287            }
288        };
289
290        if book.update_count == 0 {
291            log::debug!("OrderBook not yet updated for snapshot: {instrument_id}");
292            return;
293        }
294        log::debug!(
295            "Publishing OrderBook snapshot for {instrument_id} (update_count={})",
296            book.update_count
297        );
298
299        books.push((topic, book.clone()));
300    }
301}
302
303#[cfg(test)]
304mod tests {
305    use nautilus_common::msgbus::TypedHandler;
306    use nautilus_core::{UUID4, UnixNanos};
307    use nautilus_model::{
308        data::BookOrder,
309        enums::{BookType, OrderSide},
310        types::{Price, Quantity},
311    };
312    use rstest::rstest;
313
314    use super::*;
315
316    #[rstest]
317    fn snapshot_skips_missing_order_book() {
318        let instrument_id = InstrumentId::from("AUD/USD.SIM");
319        let interval_ms = NonZeroUsize::new(100).unwrap();
320        let topic = switchboard::get_book_snapshots_topic(instrument_id, interval_ms);
321        let snapshot_infos = Rc::new(RefCell::new(IndexMap::new()));
322
323        snapshot_infos.borrow_mut().insert(
324            instrument_id,
325            BookSnapshotInfo {
326                instrument_id,
327                venue: Venue::new("SIM"),
328                parent: None,
329                topic,
330                interval_ms,
331            },
332        );
333
334        let snapshotter = BookSnapshotter::new(
335            interval_ms,
336            snapshot_infos,
337            Rc::new(RefCell::new(Cache::default())),
338        );
339        let event = TimeEvent::new(
340            Ustr::from("TEST"),
341            UUID4::new(),
342            UnixNanos::default(),
343            UnixNanos::default(),
344        );
345
346        snapshotter.snapshot(event);
347    }
348
349    #[rstest]
350    fn snapshot_allows_subscriber_to_mutably_borrow_cache() {
351        let instrument_id = InstrumentId::from("AUD/USD.SIM");
352        let interval_ms = NonZeroUsize::new(100).unwrap();
353        let topic = switchboard::get_book_snapshots_topic(instrument_id, interval_ms);
354        let snapshot_infos = Rc::new(RefCell::new(IndexMap::new()));
355
356        snapshot_infos.borrow_mut().insert(
357            instrument_id,
358            BookSnapshotInfo {
359                instrument_id,
360                venue: Venue::new("SIM"),
361                parent: None,
362                topic,
363                interval_ms,
364            },
365        );
366
367        let cache = Rc::new(RefCell::new(Cache::default()));
368        let mut book = OrderBook::new(instrument_id, BookType::L2_MBP);
369        book.add(
370            BookOrder::new(OrderSide::Buy, Price::from("100.00"), Quantity::from(10), 0),
371            0,
372            1,
373            UnixNanos::default(),
374        );
375        cache.borrow_mut().add_order_book(book).unwrap();
376
377        let received = Rc::new(RefCell::new(Vec::new()));
378        let handler = CacheWritingBookHandler {
379            id: Ustr::from("CacheWritingBookHandler"),
380            cache: cache.clone(),
381            received: received.clone(),
382        };
383        msgbus::subscribe_book_snapshots(topic.into(), TypedHandler::new(handler), None);
384
385        let snapshotter = BookSnapshotter::new(interval_ms, snapshot_infos, cache);
386        let event = TimeEvent::new(
387            Ustr::from("TEST"),
388            UUID4::new(),
389            UnixNanos::default(),
390            UnixNanos::default(),
391        );
392
393        snapshotter.snapshot(event);
394
395        let received = received.borrow();
396        assert_eq!(received.len(), 1);
397        assert_eq!(received[0].instrument_id, instrument_id);
398        assert_eq!(received[0].best_bid_price(), Some(Price::from("100.00")));
399    }
400
401    #[rstest]
402    fn snapshot_skips_book_with_no_updates() {
403        let instrument_id = InstrumentId::from("AUD/USD.SIM");
404        let interval_ms = NonZeroUsize::new(100).unwrap();
405        let topic = switchboard::get_book_snapshots_topic(instrument_id, interval_ms);
406        let snapshot_infos = Rc::new(RefCell::new(IndexMap::new()));
407
408        snapshot_infos.borrow_mut().insert(
409            instrument_id,
410            BookSnapshotInfo {
411                instrument_id,
412                venue: Venue::new("SIM"),
413                parent: None,
414                topic,
415                interval_ms,
416            },
417        );
418
419        let cache = Rc::new(RefCell::new(Cache::default()));
420        cache
421            .borrow_mut()
422            .add_order_book(OrderBook::new(instrument_id, BookType::L2_MBP))
423            .unwrap();
424
425        let received = Rc::new(RefCell::new(Vec::new()));
426        let handler = CacheWritingBookHandler {
427            id: Ustr::from("CacheWritingBookHandler-NoUpdates"),
428            cache: cache.clone(),
429            received: received.clone(),
430        };
431        msgbus::subscribe_book_snapshots(topic.into(), TypedHandler::new(handler), None);
432
433        let snapshotter = BookSnapshotter::new(interval_ms, snapshot_infos, cache);
434        let event = TimeEvent::new(
435            Ustr::from("TEST"),
436            UUID4::new(),
437            UnixNanos::default(),
438            UnixNanos::default(),
439        );
440
441        snapshotter.snapshot(event);
442
443        assert!(received.borrow().is_empty());
444    }
445
446    struct CacheWritingBookHandler {
447        id: Ustr,
448        cache: Rc<RefCell<Cache>>,
449        received: Rc<RefCell<Vec<OrderBook>>>,
450    }
451
452    impl Handler<OrderBook> for CacheWritingBookHandler {
453        fn id(&self) -> Ustr {
454            self.id
455        }
456
457        fn handle(&self, book: &OrderBook) {
458            // Mirrors a strategy writing to the cache from `on_book`
459            let mut cache = self.cache.borrow_mut();
460            let _ = cache.order_book_mut(&book.instrument_id);
461            self.received.borrow_mut().push(book.clone());
462        }
463    }
464}
nautilus_data/engine/book.rs

nautilus_data/engine/
book.rs
