nautilus_databento/decode/
market_data.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 core::fmt::NumBuffer;
17use std::{collections::VecDeque, ffi::c_char, num::NonZeroUsize};
18
19use ahash::AHashMap;
20use databento::dbn;
21use nautilus_core::{DurationNanos, UnixNanos};
22use nautilus_model::{
23    data::{
24        Bar, BarSpecification, BarType, BookOrder, DEPTH10_LEN, Data, InstrumentStatus,
25        OrderBookDelta, OrderBookDepth, QuoteTick, TradeTick,
26    },
27    enums::{AggregationSource, BarAggregation, FromU16, MarketStatusAction, OrderSide, PriceType},
28    identifiers::{InstrumentId, TradeId},
29};
30
31use super::primitives::{
32    decode_price_or_undef, decode_quantity, parse_aggressor_side, parse_book_action,
33    parse_optional_bool, parse_order_side, parse_status_reason, parse_status_trading_event,
34};
35
36const STEP_ONE: NonZeroUsize = NonZeroUsize::new(1).unwrap();
37
38const BAR_SPEC_1S: BarSpecification = BarSpecification {
39    step: STEP_ONE,
40    aggregation: BarAggregation::Second,
41    price_type: PriceType::Last,
42};
43const BAR_SPEC_1M: BarSpecification = BarSpecification {
44    step: STEP_ONE,
45    aggregation: BarAggregation::Minute,
46    price_type: PriceType::Last,
47};
48const BAR_SPEC_1H: BarSpecification = BarSpecification {
49    step: STEP_ONE,
50    aggregation: BarAggregation::Hour,
51    price_type: PriceType::Last,
52};
53const BAR_SPEC_1D: BarSpecification = BarSpecification {
54    step: STEP_ONE,
55    aggregation: BarAggregation::Day,
56    price_type: PriceType::Last,
57};
58
59pub(super) const BAR_CLOSE_ADJUSTMENT_1S: DurationNanos = DurationNanos::from_secs(1);
60pub(super) const BAR_CLOSE_ADJUSTMENT_1M: DurationNanos = DurationNanos::from_mins(1);
61pub(super) const BAR_CLOSE_ADJUSTMENT_1H: DurationNanos = DurationNanos::from_hours(1);
62pub(super) const BAR_CLOSE_ADJUSTMENT_1D: DurationNanos = DurationNanos::from_days(1);
63
64// FNV-1a 64-bit constants (see http://www.isthe.com/chongo/tech/comp/fnv/).
65const FNV_OFFSET_BASIS: u64 = 0xcbf2_9ce4_8422_2325;
66const FNV_PRIME: u64 = 0x0100_0000_01b3;
67
68fn fnv1a_mix_bytes(hash: &mut u64, bytes: &[u8]) {
69    for &byte in bytes {
70        *hash ^= u64::from(byte);
71        *hash = hash.wrapping_mul(FNV_PRIME);
72    }
73}
74
75fn fnv1a_finish_field(hash: &mut u64) {
76    *hash ^= 0xff;
77    *hash = hash.wrapping_mul(FNV_PRIME);
78}
79
80fn fnv1a_mix(hash: &mut u64, bytes: &[u8]) {
81    fnv1a_mix_bytes(hash, bytes);
82    fnv1a_finish_field(hash);
83}
84
85/// Derives a deterministic [`TradeId`] for Databento schemas that do not
86/// publish a native trade identifier (e.g. CMBP1, TCBBO).
87///
88/// The hash combines the instrument, timestamps, price, size, and aggressor
89/// side so that replayed data yields the same identifier across runs.
90pub(super) fn derive_cmbp_trade_id(
91    instrument_id: InstrumentId,
92    ts_event: u64,
93    ts_recv: u64,
94    price: i64,
95    size: u32,
96    side: c_char,
97) -> TradeId {
98    let mut hash: u64 = FNV_OFFSET_BASIS;
99    fnv1a_mix_bytes(&mut hash, instrument_id.symbol.as_str().as_bytes());
100    fnv1a_mix_bytes(&mut hash, b".");
101    fnv1a_mix_bytes(&mut hash, instrument_id.venue.as_str().as_bytes());
102    fnv1a_finish_field(&mut hash);
103    fnv1a_mix(&mut hash, &ts_event.to_le_bytes());
104    fnv1a_mix(&mut hash, &ts_recv.to_le_bytes());
105    fnv1a_mix(&mut hash, &price.to_le_bytes());
106    fnv1a_mix(&mut hash, &size.to_le_bytes());
107    fnv1a_mix(&mut hash, &[side as u8]);
108    trade_id_from_hash(hash)
109}
110
111fn trade_id_from_hash(hash: u64) -> TradeId {
112    const HEX: &[u8; 16] = b"0123456789abcdef";
113
114    let mut bytes = [0u8; 16];
115    let mut value = hash;
116    for byte in bytes.iter_mut().rev() {
117        *byte = HEX[(value & 0x0f) as usize];
118        value >>= 4;
119    }
120
121    TradeId::from_bytes(&bytes).expect("16 lowercase hex bytes are valid TradeId")
122}
123
124#[inline(always)]
125#[must_use]
126pub(super) fn is_trade_msg(action: c_char) -> bool {
127    action as u8 as char == 'T'
128}
129
130/// Returns `true` if both bid and ask prices are defined (not `i64::MAX`).
131///
132/// Databento uses `i64::MAX` as a sentinel value for undefined/null prices.
133/// A valid quote requires both sides to be defined.
134#[inline(always)]
135#[must_use]
136fn has_valid_bid_ask(bid_px: i64, ask_px: i64) -> bool {
137    bid_px != i64::MAX && ask_px != i64::MAX
138}
139
140/// Decodes a Databento MBO (Market by Order) message into an order book delta or trade.
141///
142/// Returns a tuple containing either an `OrderBookDelta` or a `TradeTick`, depending on
143/// whether the message represents an order book update or a trade execution.
144///
145/// # Errors
146///
147/// Returns an error if decoding the MBO message fails.
148pub fn decode_mbo_msg(
149    msg: &dbn::MboMsg,
150    instrument_id: InstrumentId,
151    price_precision: u8,
152    ts_init: Option<UnixNanos>,
153    include_trades: bool,
154) -> anyhow::Result<(Option<OrderBookDelta>, Option<TradeTick>)> {
155    let side = parse_order_side(msg.side);
156    if is_trade_msg(msg.action) {
157        if include_trades && msg.size > 0 {
158            let price = decode_price_or_undef(msg.price, price_precision);
159            let size = decode_quantity(msg.size as u64);
160            let aggressor_side = parse_aggressor_side(msg.side);
161            let trade_id = TradeId::new(msg.sequence.format_into(&mut NumBuffer::new()));
162            let ts_event = msg.ts_recv.into();
163            let ts_init = ts_init.unwrap_or(ts_event);
164
165            let trade = TradeTick::new(
166                instrument_id,
167                price,
168                size,
169                aggressor_side,
170                trade_id,
171                ts_event,
172                ts_init,
173            );
174            return Ok((None, Some(trade)));
175        }
176
177        return Ok((None, None));
178    }
179
180    // `Action::Fill` and `Action::None` never affect the book: a fill is
181    // attribution for a resting order, and its book impact arrives as the
182    // explicit Cancel/Modify records of the same match event (CME MDP3
183    // semantics as normalized by Databento). Decoding fills as deltas
184    // corrupts the book - an iceberg hidden-part fill carries an order ID
185    // that was never Added, so `BookAction::Update` materializes a phantom
186    // order which nothing ever deletes (observed as a crossed book on GLBX).
187    if matches!(msg.action(), Ok(dbn::Action::Fill | dbn::Action::None)) {
188        return Ok((None, None));
189    }
190
191    let action = parse_book_action(msg.action)?;
192    let price = decode_price_or_undef(msg.price, price_precision);
193    let size = decode_quantity(msg.size as u64);
194    let order = BookOrder::new(side, price, size, msg.order_id);
195
196    let ts_event = msg.ts_recv.into();
197    let ts_init = ts_init.unwrap_or(ts_event);
198
199    // A replayed snapshot can carry source packet sequences in non-monotonic order,
200    // so it must not advance the book's incremental high-water mark.
201    let sequence = if msg.flags.is_snapshot() {
202        0
203    } else {
204        msg.sequence
205    };
206
207    let delta = OrderBookDelta::new(
208        instrument_id,
209        action,
210        order,
211        msg.flags.raw(),
212        sequence.into(),
213        ts_event,
214        ts_init,
215    );
216
217    Ok((Some(delta), None))
218}
219
220#[derive(Debug)]
221struct QueuedMboDelta {
222    delta: OrderBookDelta,
223    ready: bool,
224}
225
226// TODO: Consider consolidating this boundary framing with the live client while keeping
227// source-order queuing specific to historical data.
228/// Preserves source order while retaining each instrument's unresolved tail until its raw
229/// `F_LAST` boundary can be attached.
230#[derive(Debug, Default)]
231pub(crate) struct MboDeltaBuffer {
232    queue: VecDeque<QueuedMboDelta>,
233    tail: Option<(InstrumentId, u64)>,
234    tails: AHashMap<InstrumentId, u64>,
235    head: u64,
236    next: u64,
237}
238
239impl MboDeltaBuffer {
240    pub(crate) fn push(
241        &mut self,
242        msg: &dbn::MboMsg,
243        instrument_id: InstrumentId,
244        delta: Option<OrderBookDelta>,
245    ) {
246        if let Some(delta) = delta {
247            self.release(instrument_id, 0);
248
249            let index = self.next;
250            self.next = self.next.checked_add(1).expect("MBO delta index overflow");
251            let ready = msg.flags.is_last();
252            self.queue.push_back(QueuedMboDelta { delta, ready });
253
254            if !ready
255                && let Some((tail_instrument_id, tail)) = self.tail.replace((instrument_id, index))
256            {
257                self.tails.insert(tail_instrument_id, tail);
258            }
259        } else if msg.flags.is_last() {
260            self.release(instrument_id, dbn::flags::LAST);
261        }
262    }
263
264    pub(crate) fn pop_ready(&mut self) -> Option<OrderBookDelta> {
265        if !self.queue.front().is_some_and(|queued| queued.ready) {
266            return None;
267        }
268
269        self.head = self.head.checked_add(1).expect("MBO delta index overflow");
270        self.queue.pop_front().map(|queued| queued.delta)
271    }
272
273    pub(crate) fn finish(&mut self) {
274        self.tail = None;
275        self.tails.clear();
276
277        for queued in &mut self.queue {
278            queued.ready = true;
279        }
280    }
281
282    fn release(&mut self, instrument_id: InstrumentId, flags: u8) {
283        let tail = match self.tail {
284            Some((tail_instrument_id, tail)) if tail_instrument_id == instrument_id => {
285                self.tail = None;
286                tail
287            }
288            _ => {
289                let Some(tail) = self.tails.remove(&instrument_id) else {
290                    return;
291                };
292                tail
293            }
294        };
295        let offset = tail
296            .checked_sub(self.head)
297            .and_then(|offset| usize::try_from(offset).ok())
298            .expect("MBO delta tail must be queued");
299        let queued = self
300            .queue
301            .get_mut(offset)
302            .expect("MBO delta tail must be queued");
303        queued.delta.flags |= flags;
304        queued.ready = true;
305    }
306}
307
308/// Decodes a Databento Trade message into a `TradeTick`.
309///
310/// # Errors
311///
312/// Returns an error if decoding the Trade message fails.
313pub fn decode_trade_msg(
314    msg: &dbn::TradeMsg,
315    instrument_id: InstrumentId,
316    price_precision: u8,
317    ts_init: Option<UnixNanos>,
318) -> anyhow::Result<TradeTick> {
319    let ts_event = msg.ts_recv.into();
320    let ts_init = ts_init.unwrap_or(ts_event);
321
322    let trade = TradeTick::new(
323        instrument_id,
324        decode_price_or_undef(msg.price, price_precision),
325        decode_quantity(msg.size as u64),
326        parse_aggressor_side(msg.side),
327        TradeId::new(msg.sequence.format_into(&mut NumBuffer::new())),
328        ts_event,
329        ts_init,
330    );
331
332    Ok(trade)
333}
334
335/// Decodes a Databento TBBO (Top of Book with Trade) message into quote and trade ticks.
336///
337/// Returns `None` for the quote if either bid or ask price is undefined (`i64::MAX`).
338/// The trade is always returned.
339///
340/// # Errors
341///
342/// Returns an error if decoding the TBBO message fails.
343pub fn decode_tbbo_msg(
344    msg: &dbn::TbboMsg,
345    instrument_id: InstrumentId,
346    price_precision: u8,
347    ts_init: Option<UnixNanos>,
348) -> anyhow::Result<(Option<QuoteTick>, TradeTick)> {
349    let top_level = &msg.levels[0];
350    let ts_event = msg.ts_recv.into();
351    let ts_init = ts_init.unwrap_or(ts_event);
352
353    let maybe_quote = if has_valid_bid_ask(top_level.bid_px, top_level.ask_px) {
354        Some(QuoteTick::new(
355            instrument_id,
356            decode_price_or_undef(top_level.bid_px, price_precision),
357            decode_price_or_undef(top_level.ask_px, price_precision),
358            decode_quantity(top_level.bid_sz as u64),
359            decode_quantity(top_level.ask_sz as u64),
360            ts_event,
361            ts_init,
362        ))
363    } else {
364        None
365    };
366
367    let trade = TradeTick::new(
368        instrument_id,
369        decode_price_or_undef(msg.price, price_precision),
370        decode_quantity(msg.size as u64),
371        parse_aggressor_side(msg.side),
372        TradeId::new(msg.sequence.format_into(&mut NumBuffer::new())),
373        ts_event,
374        ts_init,
375    );
376
377    Ok((maybe_quote, trade))
378}
379
380/// Decodes a Databento MBP1 (Market by Price Level 1) message into quote and optional trade ticks.
381///
382/// Returns `None` for the quote if either bid or ask price is undefined (`i64::MAX`).
383///
384/// # Errors
385///
386/// Returns an error if decoding the MBP1 message fails.
387pub fn decode_mbp1_msg(
388    msg: &dbn::Mbp1Msg,
389    instrument_id: InstrumentId,
390    price_precision: u8,
391    ts_init: Option<UnixNanos>,
392    include_trades: bool,
393) -> anyhow::Result<(Option<QuoteTick>, Option<TradeTick>)> {
394    let top_level = &msg.levels[0];
395    let ts_event = msg.ts_recv.into();
396    let ts_init = ts_init.unwrap_or(ts_event);
397
398    let maybe_quote = if has_valid_bid_ask(top_level.bid_px, top_level.ask_px) {
399        Some(QuoteTick::new(
400            instrument_id,
401            decode_price_or_undef(top_level.bid_px, price_precision),
402            decode_price_or_undef(top_level.ask_px, price_precision),
403            decode_quantity(top_level.bid_sz as u64),
404            decode_quantity(top_level.ask_sz as u64),
405            ts_event,
406            ts_init,
407        ))
408    } else {
409        None
410    };
411
412    let maybe_trade = if include_trades && is_trade_msg(msg.action) {
413        Some(TradeTick::new(
414            instrument_id,
415            decode_price_or_undef(msg.price, price_precision),
416            decode_quantity(msg.size as u64),
417            parse_aggressor_side(msg.side),
418            TradeId::new(msg.sequence.format_into(&mut NumBuffer::new())),
419            ts_event,
420            ts_init,
421        ))
422    } else {
423        None
424    };
425
426    Ok((maybe_quote, maybe_trade))
427}
428
429/// Decodes a Databento BBO (Best Bid and Offer) message into a `QuoteTick`.
430///
431/// Returns `None` if either bid or ask price is undefined (`i64::MAX`).
432///
433/// # Errors
434///
435/// Returns an error if decoding the BBO message fails.
436pub fn decode_bbo_msg(
437    msg: &dbn::BboMsg,
438    instrument_id: InstrumentId,
439    price_precision: u8,
440    ts_init: Option<UnixNanos>,
441) -> anyhow::Result<Option<QuoteTick>> {
442    let top_level = &msg.levels[0];
443    if !has_valid_bid_ask(top_level.bid_px, top_level.ask_px) {
444        return Ok(None);
445    }
446
447    let ts_event = msg.ts_recv.into();
448    let ts_init = ts_init.unwrap_or(ts_event);
449
450    let quote = QuoteTick::new(
451        instrument_id,
452        decode_price_or_undef(top_level.bid_px, price_precision),
453        decode_price_or_undef(top_level.ask_px, price_precision),
454        decode_quantity(top_level.bid_sz as u64),
455        decode_quantity(top_level.ask_sz as u64),
456        ts_event,
457        ts_init,
458    );
459
460    Ok(Some(quote))
461}
462
463/// Decodes a Databento MBP10 (Market by Price 10 levels) message into an `OrderBookDepth`.
464///
465/// This function does not return an error; the result type matches the common decode API.
466#[allow(
467    clippy::missing_errors_doc,
468    reason = "this decoder constructs the depth without fallible operations"
469)]
470pub fn decode_mbp10_msg(
471    msg: &dbn::Mbp10Msg,
472    instrument_id: InstrumentId,
473    price_precision: u8,
474    ts_init: Option<UnixNanos>,
475) -> anyhow::Result<OrderBookDepth> {
476    let mut bids = [BookOrder::default(); DEPTH10_LEN];
477    let mut asks = [BookOrder::default(); DEPTH10_LEN];
478    let mut bid_counts = [0u32; DEPTH10_LEN];
479    let mut ask_counts = [0u32; DEPTH10_LEN];
480
481    for (index, level) in msg.levels.iter().enumerate() {
482        // Empty Databento levels carry the i64::MAX sentinel; leave NULL_ORDER
483        // so the matching engine's precision check skips them.
484        if level.bid_px != i64::MAX {
485            bids[index] = BookOrder::new(
486                OrderSide::Buy,
487                decode_price_or_undef(level.bid_px, price_precision),
488                decode_quantity(level.bid_sz as u64),
489                0,
490            );
491        }
492
493        if level.ask_px != i64::MAX {
494            asks[index] = BookOrder::new(
495                OrderSide::Sell,
496                decode_price_or_undef(level.ask_px, price_precision),
497                decode_quantity(level.ask_sz as u64),
498                0,
499            );
500        }
501
502        bid_counts[index] = level.bid_ct;
503        ask_counts[index] = level.ask_ct;
504    }
505
506    let ts_event = msg.ts_recv.into();
507    let ts_init = ts_init.unwrap_or(ts_event);
508
509    let depth = OrderBookDepth::new(
510        instrument_id,
511        bids,
512        asks,
513        bid_counts,
514        ask_counts,
515        msg.flags.raw(),
516        msg.sequence.into(),
517        ts_event,
518        ts_init,
519    );
520
521    Ok(depth)
522}
523
524/// Decodes a Databento CMBP1 (Consolidated Market by Price Level 1) message.
525///
526/// Returns a tuple containing an optional `QuoteTick` and an optional `TradeTick`.
527/// Returns `None` for the quote if either bid or ask price is undefined (`i64::MAX`).
528///
529/// # Errors
530///
531/// Returns an error if decoding the CMBP1 message fails.
532pub fn decode_cmbp1_msg(
533    msg: &dbn::Cmbp1Msg,
534    instrument_id: InstrumentId,
535    price_precision: u8,
536    ts_init: Option<UnixNanos>,
537    include_trades: bool,
538) -> anyhow::Result<(Option<QuoteTick>, Option<TradeTick>)> {
539    let top_level = &msg.levels[0];
540    let ts_event = msg.ts_recv.into();
541    let ts_init = ts_init.unwrap_or(ts_event);
542
543    let maybe_quote = if has_valid_bid_ask(top_level.bid_px, top_level.ask_px) {
544        Some(QuoteTick::new(
545            instrument_id,
546            decode_price_or_undef(top_level.bid_px, price_precision),
547            decode_price_or_undef(top_level.ask_px, price_precision),
548            decode_quantity(top_level.bid_sz as u64),
549            decode_quantity(top_level.ask_sz as u64),
550            ts_event,
551            ts_init,
552        ))
553    } else {
554        None
555    };
556
557    let maybe_trade = if include_trades && is_trade_msg(msg.action) {
558        // CMBP1 does not publish a native trade ID; derive a deterministic one
559        let trade_id = derive_cmbp_trade_id(
560            instrument_id,
561            msg.hd.ts_event,
562            msg.ts_recv,
563            msg.price,
564            msg.size,
565            msg.side,
566        );
567        Some(TradeTick::new(
568            instrument_id,
569            decode_price_or_undef(msg.price, price_precision),
570            decode_quantity(msg.size as u64),
571            parse_aggressor_side(msg.side),
572            trade_id,
573            ts_event,
574            ts_init,
575        ))
576    } else {
577        None
578    };
579
580    Ok((maybe_quote, maybe_trade))
581}
582
583/// Decodes a Databento CBBO (Consolidated Best Bid and Offer) message.
584///
585/// Returns `None` if either bid or ask price is undefined (`i64::MAX`).
586///
587/// # Errors
588///
589/// Returns an error if decoding the CBBO message fails.
590pub fn decode_cbbo_msg(
591    msg: &dbn::CbboMsg,
592    instrument_id: InstrumentId,
593    price_precision: u8,
594    ts_init: Option<UnixNanos>,
595) -> anyhow::Result<Option<QuoteTick>> {
596    let top_level = &msg.levels[0];
597    if !has_valid_bid_ask(top_level.bid_px, top_level.ask_px) {
598        return Ok(None);
599    }
600
601    let ts_event = msg.ts_recv.into();
602    let ts_init = ts_init.unwrap_or(ts_event);
603
604    let quote = QuoteTick::new(
605        instrument_id,
606        decode_price_or_undef(top_level.bid_px, price_precision),
607        decode_price_or_undef(top_level.ask_px, price_precision),
608        decode_quantity(top_level.bid_sz as u64),
609        decode_quantity(top_level.ask_sz as u64),
610        ts_event,
611        ts_init,
612    );
613
614    Ok(Some(quote))
615}
616
617/// Decodes a Databento TCBBO (Consolidated Top of Book with Trade) message.
618///
619/// Returns `None` for the quote if either bid or ask price is undefined (`i64::MAX`).
620/// The trade is always returned.
621///
622/// # Errors
623///
624/// Returns an error if decoding the TCBBO message fails.
625pub fn decode_tcbbo_msg(
626    msg: &dbn::TcbboMsg,
627    instrument_id: InstrumentId,
628    price_precision: u8,
629    ts_init: Option<UnixNanos>,
630) -> anyhow::Result<(Option<QuoteTick>, TradeTick)> {
631    let (maybe_quote, maybe_trade) =
632        decode_cmbp1_msg(msg, instrument_id, price_precision, ts_init, true)?;
633    let trade = maybe_trade.ok_or_else(|| {
634        anyhow::anyhow!(
635            "Invalid `TcbboMsg`: expected trade action, was {}",
636            msg.action as u8 as char
637        )
638    })?;
639
640    Ok((maybe_quote, trade))
641}
642
643/// # Errors
644///
645/// Returns an error if `rtype` is not a supported bar aggregation.
646pub fn decode_bar_type(
647    msg: &dbn::OhlcvMsg,
648    instrument_id: InstrumentId,
649) -> anyhow::Result<BarType> {
650    let bar_type = match msg.hd.rtype {
651        32 => {
652            // ohlcv-1s
653            BarType::new(instrument_id, BAR_SPEC_1S, AggregationSource::External)
654        }
655        33 => {
656            // ohlcv-1m
657            BarType::new(instrument_id, BAR_SPEC_1M, AggregationSource::External)
658        }
659        34 => {
660            // ohlcv-1h
661            BarType::new(instrument_id, BAR_SPEC_1H, AggregationSource::External)
662        }
663        35 => {
664            // ohlcv-1d
665            BarType::new(instrument_id, BAR_SPEC_1D, AggregationSource::External)
666        }
667        36 => {
668            // ohlcv-eod
669            BarType::new(instrument_id, BAR_SPEC_1D, AggregationSource::External)
670        }
671        _ => anyhow::bail!(
672            "`rtype` is not a supported bar aggregation, was {}",
673            msg.hd.rtype
674        ),
675    };
676
677    Ok(bar_type)
678}
679
680/// # Errors
681///
682/// Returns an error if `rtype` is not a supported bar aggregation.
683pub fn decode_ts_event_adjustment(msg: &dbn::OhlcvMsg) -> anyhow::Result<DurationNanos> {
684    let adjustment = match msg.hd.rtype {
685        32 => {
686            // ohlcv-1s
687            BAR_CLOSE_ADJUSTMENT_1S
688        }
689        33 => {
690            // ohlcv-1m
691            BAR_CLOSE_ADJUSTMENT_1M
692        }
693        34 => {
694            // ohlcv-1h
695            BAR_CLOSE_ADJUSTMENT_1H
696        }
697        35 | 36 => {
698            // ohlcv-1d and ohlcv-eod
699            BAR_CLOSE_ADJUSTMENT_1D
700        }
701        _ => anyhow::bail!(
702            "`rtype` is not a supported bar aggregation, was {}",
703            msg.hd.rtype
704        ),
705    };
706
707    Ok(adjustment)
708}
709
710/// # Errors
711///
712/// Returns an error if decoding the OHLCV message fails.
713pub fn decode_ohlcv_msg(
714    msg: &dbn::OhlcvMsg,
715    instrument_id: InstrumentId,
716    price_precision: u8,
717    ts_init: Option<UnixNanos>,
718    timestamp_on_close: bool,
719) -> anyhow::Result<Bar> {
720    let bar_type = decode_bar_type(msg, instrument_id)?;
721    let ts_event_adjustment = decode_ts_event_adjustment(msg)?;
722
723    let ts_event_raw = msg.hd.ts_event.into();
724    let ts_close = ts_event_raw + ts_event_adjustment;
725    let ts_init = ts_init.unwrap_or(ts_close); // received time or close time
726
727    let ts_event = if timestamp_on_close {
728        ts_close
729    } else {
730        ts_event_raw
731    };
732
733    let bar = Bar::new(
734        bar_type,
735        decode_price_or_undef(msg.open, price_precision),
736        decode_price_or_undef(msg.high, price_precision),
737        decode_price_or_undef(msg.low, price_precision),
738        decode_price_or_undef(msg.close, price_precision),
739        decode_quantity(msg.volume),
740        ts_event,
741        ts_init,
742    );
743
744    Ok(bar)
745}
746
747/// Decodes a Databento status message into an `InstrumentStatus` event.
748///
749/// # Errors
750///
751/// Returns an error if decoding the status message fails or if `msg.action` is not a valid `MarketStatusAction`.
752pub fn decode_status_msg(
753    msg: &dbn::StatusMsg,
754    instrument_id: InstrumentId,
755    ts_init: Option<UnixNanos>,
756) -> anyhow::Result<InstrumentStatus> {
757    let ts_event = msg.hd.ts_event.into();
758    let ts_init = ts_init.unwrap_or(ts_event);
759
760    let action = MarketStatusAction::from_u16(msg.action)
761        .ok_or_else(|| anyhow::anyhow!("Invalid `MarketStatusAction` value: {}", msg.action))?;
762
763    let status = InstrumentStatus::new(
764        instrument_id,
765        action,
766        ts_event,
767        ts_init,
768        parse_status_reason(msg.reason)?,
769        parse_status_trading_event(msg.trading_event)?,
770        parse_optional_bool(msg.is_trading),
771        parse_optional_bool(msg.is_quoting),
772        parse_optional_bool(msg.is_short_sell_restricted),
773    );
774
775    Ok(status)
776}
777
778/// # Errors
779///
780/// Returns an error if decoding the record type fails or encounters unsupported message.
781pub fn decode_record(
782    record: &dbn::RecordRef,
783    instrument_id: InstrumentId,
784    price_precision: u8,
785    ts_init: Option<UnixNanos>,
786    include_trades: bool,
787    bars_timestamp_on_close: bool,
788) -> anyhow::Result<(Option<Data>, Option<Data>)> {
789    let result = if let Some(msg) = record.get::<dbn::MboMsg>() {
790        let ts_init = determine_timestamp(ts_init, msg.ts_recv.into());
791        let result = decode_mbo_msg(
792            msg,
793            instrument_id,
794            price_precision,
795            Some(ts_init),
796            include_trades,
797        )?;
798
799        match result {
800            (Some(delta), None) => (Some(Data::BookDelta(delta)), None),
801            (None, Some(trade)) => (Some(Data::Trade(trade)), None),
802            (None, None) => (None, None),
803            _ => anyhow::bail!("Invalid `MboMsg` parsing combination"),
804        }
805    } else if let Some(msg) = record.get::<dbn::TradeMsg>() {
806        let ts_init = determine_timestamp(ts_init, msg.ts_recv.into());
807        let trade = decode_trade_msg(msg, instrument_id, price_precision, Some(ts_init))?;
808        (Some(Data::Trade(trade)), None)
809    } else if let Some(msg) = record.get::<dbn::Mbp1Msg>() {
810        let ts_init = determine_timestamp(ts_init, msg.ts_recv.into());
811        let (maybe_quote, maybe_trade) = decode_mbp1_msg(
812            msg,
813            instrument_id,
814            price_precision,
815            Some(ts_init),
816            include_trades,
817        )?;
818        (maybe_quote.map(Data::Quote), maybe_trade.map(Data::Trade))
819    } else if let Some(msg) = record.get::<dbn::Bbo1SMsg>() {
820        let ts_init = determine_timestamp(ts_init, msg.ts_recv.into());
821        let maybe_quote = decode_bbo_msg(msg, instrument_id, price_precision, Some(ts_init))?;
822        (maybe_quote.map(Data::Quote), None)
823    } else if let Some(msg) = record.get::<dbn::Bbo1MMsg>() {
824        let ts_init = determine_timestamp(ts_init, msg.ts_recv.into());
825        let maybe_quote = decode_bbo_msg(msg, instrument_id, price_precision, Some(ts_init))?;
826        (maybe_quote.map(Data::Quote), None)
827    } else if let Some(msg) = record.get::<dbn::Mbp10Msg>() {
828        let ts_init = determine_timestamp(ts_init, msg.ts_recv.into());
829        let depth = decode_mbp10_msg(msg, instrument_id, price_precision, Some(ts_init))?;
830        (Some(Data::from(depth)), None)
831    } else if let Some(msg) = record.get::<dbn::OhlcvMsg>() {
832        // if ts_init is None (like with historical data) we don't want it to be equal to ts_event
833        // it will be set correctly in decode_ohlcv_msg instead
834        let bar = decode_ohlcv_msg(
835            msg,
836            instrument_id,
837            price_precision,
838            ts_init,
839            bars_timestamp_on_close,
840        )?;
841        (Some(Data::Bar(bar)), None)
842    } else if let Some(msg) = record.get::<dbn::Cmbp1Msg>() {
843        let ts_init = determine_timestamp(ts_init, msg.ts_recv.into());
844        if msg.hd.rtype == dbn::enums::rtype::TCBBO {
845            let (maybe_quote, trade) =
846                decode_tcbbo_msg(msg, instrument_id, price_precision, Some(ts_init))?;
847            (maybe_quote.map(Data::Quote), Some(Data::Trade(trade)))
848        } else {
849            let (maybe_quote, maybe_trade) = decode_cmbp1_msg(
850                msg,
851                instrument_id,
852                price_precision,
853                Some(ts_init),
854                include_trades,
855            )?;
856            (maybe_quote.map(Data::Quote), maybe_trade.map(Data::Trade))
857        }
858    } else if let Some(msg) = record.get::<dbn::TbboMsg>() {
859        // TBBO always has a trade, quote may be skipped if prices undefined
860        let ts_init = determine_timestamp(ts_init, msg.ts_recv.into());
861        let (maybe_quote, trade) =
862            decode_tbbo_msg(msg, instrument_id, price_precision, Some(ts_init))?;
863        (maybe_quote.map(Data::Quote), Some(Data::Trade(trade)))
864    } else if let Some(msg) = record.get::<dbn::CbboMsg>() {
865        let ts_init = determine_timestamp(ts_init, msg.ts_recv.into());
866        let maybe_quote = decode_cbbo_msg(msg, instrument_id, price_precision, Some(ts_init))?;
867        (maybe_quote.map(Data::Quote), None)
868    } else {
869        anyhow::bail!("DBN message type is not currently supported")
870    };
871
872    Ok(result)
873}
874
875const fn determine_timestamp(ts_init: Option<UnixNanos>, msg_timestamp: UnixNanos) -> UnixNanos {
876    match ts_init {
877        Some(ts_init) => ts_init,
878        None => msg_timestamp,
879    }
880}
nautilus_databento/decode/market_data.rs

nautilus_databento/decode/
market_data.rs
