nautilus_databento/decode/

market_data.rs

// -------------------------------------------------------------------------------------------------
//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
//  https://nautechsystems.io
//
//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
//  You may not use this file except in compliance with the License.
//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
//
//  Unless required by applicable law or agreed to in writing, software
//  distributed under the License is distributed on an "AS IS" BASIS,
//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//  See the License for the specific language governing permissions and
//  limitations under the License.
// -------------------------------------------------------------------------------------------------

use core::fmt::NumBuffer;
use std::{collections::VecDeque, ffi::c_char, num::NonZeroUsize};

use ahash::AHashMap;
use databento::dbn;
use nautilus_core::{UnixNanos, datetime::NANOSECONDS_IN_SECOND};
use nautilus_model::{
    data::{
        Bar, BarSpecification, BarType, BookOrder, DEPTH10_LEN, Data, InstrumentStatus,
        OrderBookDelta, OrderBookDepth10, QuoteTick, TradeTick,
    },
    enums::{AggregationSource, BarAggregation, FromU16, MarketStatusAction, OrderSide, PriceType},
    identifiers::{InstrumentId, TradeId},
};

use super::primitives::{
    decode_price_or_undef, decode_quantity, parse_aggressor_side, parse_book_action,
    parse_optional_bool, parse_order_side, parse_status_reason, parse_status_trading_event,
};

const STEP_ONE: NonZeroUsize = NonZeroUsize::new(1).unwrap();

const BAR_SPEC_1S: BarSpecification = BarSpecification {
    step: STEP_ONE,
    aggregation: BarAggregation::Second,
    price_type: PriceType::Last,
};
const BAR_SPEC_1M: BarSpecification = BarSpecification {
    step: STEP_ONE,
    aggregation: BarAggregation::Minute,
    price_type: PriceType::Last,
};
const BAR_SPEC_1H: BarSpecification = BarSpecification {
    step: STEP_ONE,
    aggregation: BarAggregation::Hour,
    price_type: PriceType::Last,
};
const BAR_SPEC_1D: BarSpecification = BarSpecification {
    step: STEP_ONE,
    aggregation: BarAggregation::Day,
    price_type: PriceType::Last,
};

pub(super) const BAR_CLOSE_ADJUSTMENT_1S: u64 = NANOSECONDS_IN_SECOND;
pub(super) const BAR_CLOSE_ADJUSTMENT_1M: u64 = NANOSECONDS_IN_SECOND * 60;
pub(super) const BAR_CLOSE_ADJUSTMENT_1H: u64 = NANOSECONDS_IN_SECOND * 60 * 60;
pub(super) const BAR_CLOSE_ADJUSTMENT_1D: u64 = NANOSECONDS_IN_SECOND * 60 * 60 * 24;

// FNV-1a 64-bit constants (see http://www.isthe.com/chongo/tech/comp/fnv/).
const FNV_OFFSET_BASIS: u64 = 0xcbf2_9ce4_8422_2325;
const FNV_PRIME: u64 = 0x0100_0000_01b3;

fn fnv1a_mix_bytes(hash: &mut u64, bytes: &[u8]) {
    for &byte in bytes {
        *hash ^= u64::from(byte);
        *hash = hash.wrapping_mul(FNV_PRIME);
    }
}

fn fnv1a_finish_field(hash: &mut u64) {
    *hash ^= 0xff;
    *hash = hash.wrapping_mul(FNV_PRIME);
}

fn fnv1a_mix(hash: &mut u64, bytes: &[u8]) {
    fnv1a_mix_bytes(hash, bytes);
    fnv1a_finish_field(hash);
}

/// Derives a deterministic [`TradeId`] for Databento schemas that do not
/// publish a native trade identifier (e.g. CMBP1, TCBBO).
///
/// The hash combines the instrument, timestamps, price, size, and aggressor
/// side so that replayed data yields the same identifier across runs.
pub(super) fn derive_cmbp_trade_id(
    instrument_id: InstrumentId,
    ts_event: u64,
    ts_recv: u64,
    price: i64,
    size: u32,
    side: c_char,
) -> TradeId {
    let mut hash: u64 = FNV_OFFSET_BASIS;
    fnv1a_mix_bytes(&mut hash, instrument_id.symbol.as_str().as_bytes());
    fnv1a_mix_bytes(&mut hash, b".");
    fnv1a_mix_bytes(&mut hash, instrument_id.venue.as_str().as_bytes());
    fnv1a_finish_field(&mut hash);
    fnv1a_mix(&mut hash, &ts_event.to_le_bytes());
    fnv1a_mix(&mut hash, &ts_recv.to_le_bytes());
    fnv1a_mix(&mut hash, &price.to_le_bytes());
    fnv1a_mix(&mut hash, &size.to_le_bytes());
    fnv1a_mix(&mut hash, &[side as u8]);
    trade_id_from_hash(hash)
}

fn trade_id_from_hash(hash: u64) -> TradeId {
    const HEX: &[u8; 16] = b"0123456789abcdef";

    let mut bytes = [0u8; 16];
    let mut value = hash;
    for byte in bytes.iter_mut().rev() {
        *byte = HEX[(value & 0x0f) as usize];
        value >>= 4;
    }

    TradeId::from_bytes(&bytes).expect("16 lowercase hex bytes are valid TradeId")
}

#[inline(always)]
#[must_use]
pub(super) fn is_trade_msg(action: c_char) -> bool {
    action as u8 as char == 'T'
}

/// Returns `true` if both bid and ask prices are defined (not `i64::MAX`).
///
/// Databento uses `i64::MAX` as a sentinel value for undefined/null prices.
/// A valid quote requires both sides to be defined.
#[inline(always)]
#[must_use]
fn has_valid_bid_ask(bid_px: i64, ask_px: i64) -> bool {
    bid_px != i64::MAX && ask_px != i64::MAX
}

/// Decodes a Databento MBO (Market by Order) message into an order book delta or trade.
///
/// Returns a tuple containing either an `OrderBookDelta` or a `TradeTick`, depending on
/// whether the message represents an order book update or a trade execution.
///
/// # Errors
///
/// Returns an error if decoding the MBO message fails.
pub fn decode_mbo_msg(
    msg: &dbn::MboMsg,
    instrument_id: InstrumentId,
    price_precision: u8,
    ts_init: Option<UnixNanos>,
    include_trades: bool,
) -> anyhow::Result<(Option<OrderBookDelta>, Option<TradeTick>)> {
    let side = parse_order_side(msg.side);
    if is_trade_msg(msg.action) {
        if include_trades && msg.size > 0 {
            let price = decode_price_or_undef(msg.price, price_precision);
            let size = decode_quantity(msg.size as u64);
            let aggressor_side = parse_aggressor_side(msg.side);
            let trade_id = TradeId::new(msg.sequence.format_into(&mut NumBuffer::new()));
            let ts_event = msg.ts_recv.into();
            let ts_init = ts_init.unwrap_or(ts_event);

            let trade = TradeTick::new(
                instrument_id,
                price,
                size,
                aggressor_side,
                trade_id,
                ts_event,
                ts_init,
            );
            return Ok((None, Some(trade)));
        }

        return Ok((None, None));
    }

    // `Action::Fill` and `Action::None` never affect the book: a fill is
    // attribution for a resting order, and its book impact arrives as the
    // explicit Cancel/Modify records of the same match event (CME MDP3
    // semantics as normalized by Databento). Decoding fills as deltas
    // corrupts the book - an iceberg hidden-part fill carries an order ID
    // that was never Added, so `BookAction::Update` materializes a phantom
    // order which nothing ever deletes (observed as a crossed book on GLBX).
    if matches!(msg.action(), Ok(dbn::Action::Fill | dbn::Action::None)) {
        return Ok((None, None));
    }

    let action = parse_book_action(msg.action)?;
    let price = decode_price_or_undef(msg.price, price_precision);
    let size = decode_quantity(msg.size as u64);
    let order = BookOrder::new(side, price, size, msg.order_id);

    let ts_event = msg.ts_recv.into();
    let ts_init = ts_init.unwrap_or(ts_event);

    // A replayed snapshot can carry source packet sequences in non-monotonic order,
    // so it must not advance the book's incremental high-water mark.
    let sequence = if msg.flags.is_snapshot() {
        0
    } else {
        msg.sequence
    };

    let delta = OrderBookDelta::new(
        instrument_id,
        action,
        order,
        msg.flags.raw(),
        sequence.into(),
        ts_event,
        ts_init,
    );

    Ok((Some(delta), None))
}

#[derive(Debug)]
struct QueuedMboDelta {
    delta: OrderBookDelta,
    ready: bool,
}

// TODO: Consider consolidating this boundary framing with the live client while keeping
// source-order queuing specific to historical data.
/// Preserves source order while retaining each instrument's unresolved tail until its raw
/// `F_LAST` boundary can be attached.
#[derive(Debug, Default)]
pub(crate) struct MboDeltaBuffer {
    queue: VecDeque<QueuedMboDelta>,
    tail: Option<(InstrumentId, u64)>,
    tails: AHashMap<InstrumentId, u64>,
    head: u64,
    next: u64,
}

impl MboDeltaBuffer {
    pub(crate) fn push(
        &mut self,
        msg: &dbn::MboMsg,
        instrument_id: InstrumentId,
        delta: Option<OrderBookDelta>,
    ) {
        if let Some(delta) = delta {
            self.release(instrument_id, 0);

            let index = self.next;
            self.next = self.next.checked_add(1).expect("MBO delta index overflow");
            let ready = msg.flags.is_last();
            self.queue.push_back(QueuedMboDelta { delta, ready });

            if !ready
                && let Some((tail_instrument_id, tail)) = self.tail.replace((instrument_id, index))
            {
                self.tails.insert(tail_instrument_id, tail);
            }
        } else if msg.flags.is_last() {
            self.release(instrument_id, dbn::flags::LAST);
        }
    }

    pub(crate) fn pop_ready(&mut self) -> Option<OrderBookDelta> {
        if !self.queue.front().is_some_and(|queued| queued.ready) {
            return None;
        }

        self.head = self.head.checked_add(1).expect("MBO delta index overflow");
        self.queue.pop_front().map(|queued| queued.delta)
    }

    pub(crate) fn finish(&mut self) {
        self.tail = None;
        self.tails.clear();

        for queued in &mut self.queue {
            queued.ready = true;
        }
    }

    fn release(&mut self, instrument_id: InstrumentId, flags: u8) {
        let tail = match self.tail {
            Some((tail_instrument_id, tail)) if tail_instrument_id == instrument_id => {
                self.tail = None;
                tail
            }
            _ => {
                let Some(tail) = self.tails.remove(&instrument_id) else {
                    return;
                };
                tail
            }
        };
        let offset = tail
            .checked_sub(self.head)
            .and_then(|offset| usize::try_from(offset).ok())
            .expect("MBO delta tail must be queued");
        let queued = self
            .queue
            .get_mut(offset)
            .expect("MBO delta tail must be queued");
        queued.delta.flags |= flags;
        queued.ready = true;
    }
}

/// Decodes a Databento Trade message into a `TradeTick`.
///
/// # Errors
///
/// Returns an error if decoding the Trade message fails.
pub fn decode_trade_msg(
    msg: &dbn::TradeMsg,
    instrument_id: InstrumentId,
    price_precision: u8,
    ts_init: Option<UnixNanos>,
) -> anyhow::Result<TradeTick> {
    let ts_event = msg.ts_recv.into();
    let ts_init = ts_init.unwrap_or(ts_event);

    let trade = TradeTick::new(
        instrument_id,
        decode_price_or_undef(msg.price, price_precision),
        decode_quantity(msg.size as u64),
        parse_aggressor_side(msg.side),
        TradeId::new(msg.sequence.format_into(&mut NumBuffer::new())),
        ts_event,
        ts_init,
    );

    Ok(trade)
}

/// Decodes a Databento TBBO (Top of Book with Trade) message into quote and trade ticks.
///
/// Returns `None` for the quote if either bid or ask price is undefined (`i64::MAX`).
/// The trade is always returned.
///
/// # Errors
///
/// Returns an error if decoding the TBBO message fails.
pub fn decode_tbbo_msg(
    msg: &dbn::TbboMsg,
    instrument_id: InstrumentId,
    price_precision: u8,
    ts_init: Option<UnixNanos>,
) -> anyhow::Result<(Option<QuoteTick>, TradeTick)> {
    let top_level = &msg.levels[0];
    let ts_event = msg.ts_recv.into();
    let ts_init = ts_init.unwrap_or(ts_event);

    let maybe_quote = if has_valid_bid_ask(top_level.bid_px, top_level.ask_px) {
        Some(QuoteTick::new(
            instrument_id,
            decode_price_or_undef(top_level.bid_px, price_precision),
            decode_price_or_undef(top_level.ask_px, price_precision),
            decode_quantity(top_level.bid_sz as u64),
            decode_quantity(top_level.ask_sz as u64),
            ts_event,
            ts_init,
        ))
    } else {
        None
    };

    let trade = TradeTick::new(
        instrument_id,
        decode_price_or_undef(msg.price, price_precision),
        decode_quantity(msg.size as u64),
        parse_aggressor_side(msg.side),
        TradeId::new(msg.sequence.format_into(&mut NumBuffer::new())),
        ts_event,
        ts_init,
    );

    Ok((maybe_quote, trade))
}

/// Decodes a Databento MBP1 (Market by Price Level 1) message into quote and optional trade ticks.
///
/// Returns `None` for the quote if either bid or ask price is undefined (`i64::MAX`).
///
/// # Errors
///
/// Returns an error if decoding the MBP1 message fails.
pub fn decode_mbp1_msg(
    msg: &dbn::Mbp1Msg,
    instrument_id: InstrumentId,
    price_precision: u8,
    ts_init: Option<UnixNanos>,
    include_trades: bool,
) -> anyhow::Result<(Option<QuoteTick>, Option<TradeTick>)> {
    let top_level = &msg.levels[0];
    let ts_event = msg.ts_recv.into();
    let ts_init = ts_init.unwrap_or(ts_event);

    let maybe_quote = if has_valid_bid_ask(top_level.bid_px, top_level.ask_px) {
        Some(QuoteTick::new(
            instrument_id,
            decode_price_or_undef(top_level.bid_px, price_precision),
            decode_price_or_undef(top_level.ask_px, price_precision),
            decode_quantity(top_level.bid_sz as u64),
            decode_quantity(top_level.ask_sz as u64),
            ts_event,
            ts_init,
        ))
    } else {
        None
    };

    let maybe_trade = if include_trades && is_trade_msg(msg.action) {
        Some(TradeTick::new(
            instrument_id,
            decode_price_or_undef(msg.price, price_precision),
            decode_quantity(msg.size as u64),
            parse_aggressor_side(msg.side),
            TradeId::new(msg.sequence.format_into(&mut NumBuffer::new())),
            ts_event,
            ts_init,
        ))
    } else {
        None
    };

    Ok((maybe_quote, maybe_trade))
}

/// Decodes a Databento BBO (Best Bid and Offer) message into a `QuoteTick`.
///
/// Returns `None` if either bid or ask price is undefined (`i64::MAX`).
///
/// # Errors
///
/// Returns an error if decoding the BBO message fails.
pub fn decode_bbo_msg(
    msg: &dbn::BboMsg,
    instrument_id: InstrumentId,
    price_precision: u8,
    ts_init: Option<UnixNanos>,
) -> anyhow::Result<Option<QuoteTick>> {
    let top_level = &msg.levels[0];
    if !has_valid_bid_ask(top_level.bid_px, top_level.ask_px) {
        return Ok(None);
    }

    let ts_event = msg.ts_recv.into();
    let ts_init = ts_init.unwrap_or(ts_event);

    let quote = QuoteTick::new(
        instrument_id,
        decode_price_or_undef(top_level.bid_px, price_precision),
        decode_price_or_undef(top_level.ask_px, price_precision),
        decode_quantity(top_level.bid_sz as u64),
        decode_quantity(top_level.ask_sz as u64),
        ts_event,
        ts_init,
    );

    Ok(Some(quote))
}

/// Decodes a Databento MBP10 (Market by Price 10 levels) message into an `OrderBookDepth10`.
///
/// # Errors
///
/// This function does not currently return an error; the result type matches the common decode API.
pub fn decode_mbp10_msg(
    msg: &dbn::Mbp10Msg,
    instrument_id: InstrumentId,
    price_precision: u8,
    ts_init: Option<UnixNanos>,
) -> anyhow::Result<OrderBookDepth10> {
    let mut bids = [BookOrder::default(); DEPTH10_LEN];
    let mut asks = [BookOrder::default(); DEPTH10_LEN];
    let mut bid_counts = [0u32; DEPTH10_LEN];
    let mut ask_counts = [0u32; DEPTH10_LEN];

    for (index, level) in msg.levels.iter().enumerate() {
        // Empty Databento levels carry the i64::MAX sentinel; leave NULL_ORDER
        // so the matching engine's precision check skips them.
        if level.bid_px != i64::MAX {
            bids[index] = BookOrder::new(
                OrderSide::Buy,
                decode_price_or_undef(level.bid_px, price_precision),
                decode_quantity(level.bid_sz as u64),
                0,
            );
        }

        if level.ask_px != i64::MAX {
            asks[index] = BookOrder::new(
                OrderSide::Sell,
                decode_price_or_undef(level.ask_px, price_precision),
                decode_quantity(level.ask_sz as u64),
                0,
            );
        }

        bid_counts[index] = level.bid_ct;
        ask_counts[index] = level.ask_ct;
    }

    let ts_event = msg.ts_recv.into();
    let ts_init = ts_init.unwrap_or(ts_event);

    let depth = OrderBookDepth10::new(
        instrument_id,
        bids,
        asks,
        bid_counts,
        ask_counts,
        msg.flags.raw(),
        msg.sequence.into(),
        ts_event,
        ts_init,
    );

    Ok(depth)
}

/// Decodes a Databento CMBP1 (Consolidated Market by Price Level 1) message.
///
/// Returns a tuple containing an optional `QuoteTick` and an optional `TradeTick`.
/// Returns `None` for the quote if either bid or ask price is undefined (`i64::MAX`).
///
/// # Errors
///
/// Returns an error if decoding the CMBP1 message fails.
pub fn decode_cmbp1_msg(
    msg: &dbn::Cmbp1Msg,
    instrument_id: InstrumentId,
    price_precision: u8,
    ts_init: Option<UnixNanos>,
    include_trades: bool,
) -> anyhow::Result<(Option<QuoteTick>, Option<TradeTick>)> {
    let top_level = &msg.levels[0];
    let ts_event = msg.ts_recv.into();
    let ts_init = ts_init.unwrap_or(ts_event);

    let maybe_quote = if has_valid_bid_ask(top_level.bid_px, top_level.ask_px) {
        Some(QuoteTick::new(
            instrument_id,
            decode_price_or_undef(top_level.bid_px, price_precision),
            decode_price_or_undef(top_level.ask_px, price_precision),
            decode_quantity(top_level.bid_sz as u64),
            decode_quantity(top_level.ask_sz as u64),
            ts_event,
            ts_init,
        ))
    } else {
        None
    };

    let maybe_trade = if include_trades && is_trade_msg(msg.action) {
        // CMBP1 does not publish a native trade ID; derive a deterministic one
        let trade_id = derive_cmbp_trade_id(
            instrument_id,
            msg.hd.ts_event,
            msg.ts_recv,
            msg.price,
            msg.size,
            msg.side,
        );
        Some(TradeTick::new(
            instrument_id,
            decode_price_or_undef(msg.price, price_precision),
            decode_quantity(msg.size as u64),
            parse_aggressor_side(msg.side),
            trade_id,
            ts_event,
            ts_init,
        ))
    } else {
        None
    };

    Ok((maybe_quote, maybe_trade))
}

/// Decodes a Databento CBBO (Consolidated Best Bid and Offer) message.
///
/// Returns `None` if either bid or ask price is undefined (`i64::MAX`).
///
/// # Errors
///
/// Returns an error if decoding the CBBO message fails.
pub fn decode_cbbo_msg(
    msg: &dbn::CbboMsg,
    instrument_id: InstrumentId,
    price_precision: u8,
    ts_init: Option<UnixNanos>,
) -> anyhow::Result<Option<QuoteTick>> {
    let top_level = &msg.levels[0];
    if !has_valid_bid_ask(top_level.bid_px, top_level.ask_px) {
        return Ok(None);
    }

    let ts_event = msg.ts_recv.into();
    let ts_init = ts_init.unwrap_or(ts_event);

    let quote = QuoteTick::new(
        instrument_id,
        decode_price_or_undef(top_level.bid_px, price_precision),
        decode_price_or_undef(top_level.ask_px, price_precision),
        decode_quantity(top_level.bid_sz as u64),
        decode_quantity(top_level.ask_sz as u64),
        ts_event,
        ts_init,
    );

    Ok(Some(quote))
}

/// Decodes a Databento TCBBO (Consolidated Top of Book with Trade) message.
///
/// Returns `None` for the quote if either bid or ask price is undefined (`i64::MAX`).
/// The trade is always returned.
///
/// # Errors
///
/// Returns an error if decoding the TCBBO message fails.
pub fn decode_tcbbo_msg(
    msg: &dbn::TcbboMsg,
    instrument_id: InstrumentId,
    price_precision: u8,
    ts_init: Option<UnixNanos>,
) -> anyhow::Result<(Option<QuoteTick>, TradeTick)> {
    let (maybe_quote, maybe_trade) =
        decode_cmbp1_msg(msg, instrument_id, price_precision, ts_init, true)?;
    let trade = maybe_trade.ok_or_else(|| {
        anyhow::anyhow!(
            "Invalid `TcbboMsg`: expected trade action, was {}",
            msg.action as u8 as char
        )
    })?;

    Ok((maybe_quote, trade))
}

/// # Errors
///
/// Returns an error if `rtype` is not a supported bar aggregation.
pub fn decode_bar_type(
    msg: &dbn::OhlcvMsg,
    instrument_id: InstrumentId,
) -> anyhow::Result<BarType> {
    let bar_type = match msg.hd.rtype {
        32 => {
            // ohlcv-1s
            BarType::new(instrument_id, BAR_SPEC_1S, AggregationSource::External)
        }
        33 => {
            // ohlcv-1m
            BarType::new(instrument_id, BAR_SPEC_1M, AggregationSource::External)
        }
        34 => {
            // ohlcv-1h
            BarType::new(instrument_id, BAR_SPEC_1H, AggregationSource::External)
        }
        35 => {
            // ohlcv-1d
            BarType::new(instrument_id, BAR_SPEC_1D, AggregationSource::External)
        }
        36 => {
            // ohlcv-eod
            BarType::new(instrument_id, BAR_SPEC_1D, AggregationSource::External)
        }
        _ => anyhow::bail!(
            "`rtype` is not a supported bar aggregation, was {}",
            msg.hd.rtype
        ),
    };

    Ok(bar_type)
}

/// # Errors
///
/// Returns an error if `rtype` is not a supported bar aggregation.
pub fn decode_ts_event_adjustment(msg: &dbn::OhlcvMsg) -> anyhow::Result<UnixNanos> {
    let adjustment = match msg.hd.rtype {
        32 => {
            // ohlcv-1s
            BAR_CLOSE_ADJUSTMENT_1S
        }
        33 => {
            // ohlcv-1m
            BAR_CLOSE_ADJUSTMENT_1M
        }
        34 => {
            // ohlcv-1h
            BAR_CLOSE_ADJUSTMENT_1H
        }
        35 | 36 => {
            // ohlcv-1d and ohlcv-eod
            BAR_CLOSE_ADJUSTMENT_1D
        }
        _ => anyhow::bail!(
            "`rtype` is not a supported bar aggregation, was {}",
            msg.hd.rtype
        ),
    };

    Ok(adjustment.into())
}

/// # Errors
///
/// Returns an error if decoding the OHLCV message fails.
pub fn decode_ohlcv_msg(
    msg: &dbn::OhlcvMsg,
    instrument_id: InstrumentId,
    price_precision: u8,
    ts_init: Option<UnixNanos>,
    timestamp_on_close: bool,
) -> anyhow::Result<Bar> {
    let bar_type = decode_bar_type(msg, instrument_id)?;
    let ts_event_adjustment = decode_ts_event_adjustment(msg)?;

    let ts_event_raw = msg.hd.ts_event.into();
    let ts_close = ts_event_raw + ts_event_adjustment;
    let ts_init = ts_init.unwrap_or(ts_close); // received time or close time

    let ts_event = if timestamp_on_close {
        ts_close
    } else {
        ts_event_raw
    };

    let bar = Bar::new(
        bar_type,
        decode_price_or_undef(msg.open, price_precision),
        decode_price_or_undef(msg.high, price_precision),
        decode_price_or_undef(msg.low, price_precision),
        decode_price_or_undef(msg.close, price_precision),
        decode_quantity(msg.volume),
        ts_event,
        ts_init,
    );

    Ok(bar)
}

/// Decodes a Databento status message into an `InstrumentStatus` event.
///
/// # Errors
///
/// Returns an error if decoding the status message fails or if `msg.action` is not a valid `MarketStatusAction`.
pub fn decode_status_msg(
    msg: &dbn::StatusMsg,
    instrument_id: InstrumentId,
    ts_init: Option<UnixNanos>,
) -> anyhow::Result<InstrumentStatus> {
    let ts_event = msg.hd.ts_event.into();
    let ts_init = ts_init.unwrap_or(ts_event);

    let action = MarketStatusAction::from_u16(msg.action)
        .ok_or_else(|| anyhow::anyhow!("Invalid `MarketStatusAction` value: {}", msg.action))?;

    let status = InstrumentStatus::new(
        instrument_id,
        action,
        ts_event,
        ts_init,
        parse_status_reason(msg.reason)?,
        parse_status_trading_event(msg.trading_event)?,
        parse_optional_bool(msg.is_trading),
        parse_optional_bool(msg.is_quoting),
        parse_optional_bool(msg.is_short_sell_restricted),
    );

    Ok(status)
}

/// # Errors
///
/// Returns an error if decoding the record type fails or encounters unsupported message.
pub fn decode_record(
    record: &dbn::RecordRef,
    instrument_id: InstrumentId,
    price_precision: u8,
    ts_init: Option<UnixNanos>,
    include_trades: bool,
    bars_timestamp_on_close: bool,
) -> anyhow::Result<(Option<Data>, Option<Data>)> {
    let result = if let Some(msg) = record.get::<dbn::MboMsg>() {
        let ts_init = determine_timestamp(ts_init, msg.ts_recv.into());
        let result = decode_mbo_msg(
            msg,
            instrument_id,
            price_precision,
            Some(ts_init),
            include_trades,
        )?;

        match result {
            (Some(delta), None) => (Some(Data::Delta(delta)), None),
            (None, Some(trade)) => (Some(Data::Trade(trade)), None),
            (None, None) => (None, None),
            _ => anyhow::bail!("Invalid `MboMsg` parsing combination"),
        }
    } else if let Some(msg) = record.get::<dbn::TradeMsg>() {
        let ts_init = determine_timestamp(ts_init, msg.ts_recv.into());
        let trade = decode_trade_msg(msg, instrument_id, price_precision, Some(ts_init))?;
        (Some(Data::Trade(trade)), None)
    } else if let Some(msg) = record.get::<dbn::Mbp1Msg>() {
        let ts_init = determine_timestamp(ts_init, msg.ts_recv.into());
        let (maybe_quote, maybe_trade) = decode_mbp1_msg(
            msg,
            instrument_id,
            price_precision,
            Some(ts_init),
            include_trades,
        )?;
        (maybe_quote.map(Data::Quote), maybe_trade.map(Data::Trade))
    } else if let Some(msg) = record.get::<dbn::Bbo1SMsg>() {
        let ts_init = determine_timestamp(ts_init, msg.ts_recv.into());
        let maybe_quote = decode_bbo_msg(msg, instrument_id, price_precision, Some(ts_init))?;
        (maybe_quote.map(Data::Quote), None)
    } else if let Some(msg) = record.get::<dbn::Bbo1MMsg>() {
        let ts_init = determine_timestamp(ts_init, msg.ts_recv.into());
        let maybe_quote = decode_bbo_msg(msg, instrument_id, price_precision, Some(ts_init))?;
        (maybe_quote.map(Data::Quote), None)
    } else if let Some(msg) = record.get::<dbn::Mbp10Msg>() {
        let ts_init = determine_timestamp(ts_init, msg.ts_recv.into());
        let depth = decode_mbp10_msg(msg, instrument_id, price_precision, Some(ts_init))?;
        (Some(Data::from(depth)), None)
    } else if let Some(msg) = record.get::<dbn::OhlcvMsg>() {
        // if ts_init is None (like with historical data) we don't want it to be equal to ts_event
        // it will be set correctly in decode_ohlcv_msg instead
        let bar = decode_ohlcv_msg(
            msg,
            instrument_id,
            price_precision,
            ts_init,
            bars_timestamp_on_close,
        )?;
        (Some(Data::Bar(bar)), None)
    } else if let Some(msg) = record.get::<dbn::Cmbp1Msg>() {
        let ts_init = determine_timestamp(ts_init, msg.ts_recv.into());
        if msg.hd.rtype == dbn::enums::rtype::TCBBO {
            let (maybe_quote, trade) =
                decode_tcbbo_msg(msg, instrument_id, price_precision, Some(ts_init))?;
            (maybe_quote.map(Data::Quote), Some(Data::Trade(trade)))
        } else {
            let (maybe_quote, maybe_trade) = decode_cmbp1_msg(
                msg,
                instrument_id,
                price_precision,
                Some(ts_init),
                include_trades,
            )?;
            (maybe_quote.map(Data::Quote), maybe_trade.map(Data::Trade))
        }
    } else if let Some(msg) = record.get::<dbn::TbboMsg>() {
        // TBBO always has a trade, quote may be skipped if prices undefined
        let ts_init = determine_timestamp(ts_init, msg.ts_recv.into());
        let (maybe_quote, trade) =
            decode_tbbo_msg(msg, instrument_id, price_precision, Some(ts_init))?;
        (maybe_quote.map(Data::Quote), Some(Data::Trade(trade)))
    } else if let Some(msg) = record.get::<dbn::CbboMsg>() {
        let ts_init = determine_timestamp(ts_init, msg.ts_recv.into());
        let maybe_quote = decode_cbbo_msg(msg, instrument_id, price_precision, Some(ts_init))?;
        (maybe_quote.map(Data::Quote), None)
    } else {
        anyhow::bail!("DBN message type is not currently supported")
    };

    Ok(result)
}

const fn determine_timestamp(ts_init: Option<UnixNanos>, msg_timestamp: UnixNanos) -> UnixNanos {
    match ts_init {
        Some(ts_init) => ts_init,
        None => msg_timestamp,
    }
}