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

nautilus_data/engine/
book.rs
