Failure loading live MBO/L3 futures data from Databento with snapshot

[JackWCollins]

Bug Report

Expected Behavior

I am trying to get the full MBO/L3 data working with Databento for live data. I have successfully subscribed to subscribe_order_book_at_interval and logged the result via on_order_book. I have done the same for subscribe_trade_ticks and on_trade_tick, so the integration is generally working and streaming the expected data.

Actual Behavior

However, when I try to call subscribe_order_book_deltas I receive this error:

thread 'tokio-runtime-worker' panicked at [crates/model/src/data/trade.rs:119:10](https://github.com/nautechsystems/nautilus_trader/blob/develop/crates/model/src/data/trade.rs#L115):
Condition failed: invalid u128 for 'size.raw' not positive, was 0

I stepped through the code as best I could and I know we're requesting a snapshot with the MBO subscription (which makes sense). The first part of these messages will be a snapshot message with flag: SNAPSHOT and size: 0 (docs here).

I also found this commit which added the trade.size validation fairly recently.

Is it possible this validation is failing on the first snapshot message we receive from Databento that has a size: 0?

Steps to Reproduce the Problem

1. Create a file databento_mbo_test.py with the contents below
2. Add your DATABENTO_API_KEY as an ENV variable with dotenv
3. Run the file via python databent_mbo_test.py

Specifications

• OS platform: macOS 15.1.1
• Python version: 3.12.3
• nautilus_trader version: 1.212.0

Problematic file

from typing import Any

from nautilus_trader.adapters.databento import DATABENTO
from nautilus_trader.adapters.databento import DATABENTO_CLIENT_ID
from nautilus_trader.adapters.databento import DatabentoDataClientConfig
from nautilus_trader.adapters.databento import DatabentoLiveDataClientFactory
from nautilus_trader.cache.config import CacheConfig
from nautilus_trader.common.config import DatabaseConfig
from nautilus_trader.common.enums import LogColor
from nautilus_trader.config import InstrumentProviderConfig
from nautilus_trader.config import LiveExecEngineConfig
from nautilus_trader.config import LoggingConfig
from nautilus_trader.config import StrategyConfig
from nautilus_trader.config import TradingNodeConfig
from nautilus_trader.live.node import TradingNode
from nautilus_trader.model.enums import BookType
from nautilus_trader.model.book import OrderBook
from nautilus_trader.model.data import Bar
from nautilus_trader.model.data import OrderBookDeltas
from nautilus_trader.model.data import QuoteTick
from nautilus_trader.model.data import TradeTick
from nautilus_trader.model.identifiers import InstrumentId
from nautilus_trader.model.identifiers import TraderId
from nautilus_trader.trading.strategy import Strategy
from dotenv import load_dotenv

import os

load_dotenv()

instrument_ids = [
    InstrumentId.from_str("ESM5.GLBX"),
]

print(f"instrument_ids: {instrument_ids}")

# Configure the trading node
config_node = TradingNodeConfig(
    trader_id=TraderId("TRADER-001"),
    data_clients={
        DATABENTO: DatabentoDataClientConfig(
          api_key=os.getenv("DATABENTO_API_KEY"),
          http_gateway=None,
          mbo_subscriptions_delay=10.0,
          instrument_ids=instrument_ids,
          parent_symbols={"GLBX.MDP3": {"ES.FUT"}},
        ),
    },
)

# Instantiate the node with a configuration
node = TradingNode(config=config_node)


class TestStrategyConfig(StrategyConfig, frozen=True):
    """
    Configuration for ``TestStrategy`` instances.

    Parameters
    ----------
    instrument_ids : list[InstrumentId]
        The instrument IDs to subscribe to.

    """

    instrument_ids: list[InstrumentId]


class TestStrategy(Strategy):
    """
    An example strategy which subscribes to live data.

    Parameters
    ----------
    config : TestStrategyConfig
        The configuration for the instance.

    """

    def __init__(self, config: TestStrategyConfig) -> None:
        super().__init__(config)

    def on_start(self) -> None:
        """
        Actions to be performed when the strategy is started.

        Here we specify the 'DATABENTO' client_id for subscriptions.

        """
        for instrument_id in self.config.instrument_ids:
            # This does not work 
            self.subscribe_order_book_deltas(
                instrument_id=instrument_id,
                book_type=BookType.L3_MBO,
                client_id=DATABENTO_CLIENT_ID,
                managed=False
            )

            # This works
            # self.subscribe_order_book_at_interval(
            #     instrument_id=instrument_id,
            #     book_type=BookType.L2_MBP,
            #     depth=10,
            #     client_id=DATABENTO_CLIENT_ID,
            #     interval_ms=1000,
            # )

            # This also works
            # self.subscribe_trade_ticks(instrument_id, client_id=DATABENTO_CLIENT_ID)

    def on_stop(self) -> None:
        """
        Actions to be performed when the strategy is stopped.
        """
        # Databento does not support live data unsubscribing

    def on_historical_data(self, data: Any) -> None:
        self.log.info(repr(data), LogColor.CYAN)

    def on_order_book_deltas(self, deltas: OrderBookDeltas) -> None:
        """
        Actions to be performed when the strategy is running and receives order book
        deltas.

        Parameters
        ----------
        deltas : OrderBookDeltas
            The order book deltas received.

        """
        self.log.info(repr(deltas), LogColor.CYAN)

    def on_order_book(self, order_book: OrderBook) -> None:
        """
        Actions to be performed when an order book update is received.
        """
        self.log.info(f"\n{order_book.instrument_id}\n{order_book.pprint(10)}", LogColor.CYAN)

    def on_quote_tick(self, tick: QuoteTick) -> None:
        """
        Actions to be performed when the strategy is running and receives a quote tick.

        Parameters
        ----------
        tick : QuoteTick
            The tick received.

        """
        self.log.info(repr(tick), LogColor.CYAN)

    def on_trade_tick(self, tick: TradeTick) -> None:
        """
        Actions to be performed when the strategy is running and receives a trade tick.

        Parameters
        ----------
        tick : TradeTick
            The tick received.

        """
        self.log.info(repr(tick), LogColor.CYAN)

    def on_bar(self, bar: Bar) -> None:
        """
        Actions to be performed when the strategy is running and receives a bar.

        Parameters
        ----------
        bar : Bar
            The bar received.

        """
        self.log.info(repr(bar), LogColor.CYAN)


# Configure and initialize your strategy
strat_config = TestStrategyConfig(instrument_ids=instrument_ids)
strategy = TestStrategy(config=strat_config)

# Add your strategies and modules
node.trader.add_strategy(strategy)

# Register your client factories with the node (can take user-defined factories)
node.add_data_client_factory(DATABENTO, DatabentoLiveDataClientFactory)
print("Building node")
node.build()

# Stop and dispose of the node with SIGINT/CTRL+C
if __name__ == "__main__":
    try:
        print("Running node")
        node.run()
    finally:
        node.dispose()

[cjdsellers]

Hi @JackWCollins

Thanks for reaching out with the report and MRE.

I think your theory is probably correct. I should have a chance to look at this in the next day or so.
I'd anticipate this should be a fairly easy fix.

[cjdsellers]

I can reproduce using the current live subscriber example, just uncommenting the book subscription:
https://github.com/nautechsystems/nautilus_trader/blob/develop/examples/live/databento/databento_subscriber.py#L146

2025-03-31T23:00:55.946634Z  INFO nautilus_databento::live: Connected
2025-03-31T23:00:55.946645Z DEBUG nautilus_databento::live: Received command: Subscribe(Subscription { symbols: Symbols(["ES.c.0"]), schema: Mbo, stype_in: Continuous, start: None, use_snapshot: true })
2025-03-31T23:00:55.946711Z DEBUG LiveClient{dataset=GLBX.MDP3 session_id="1743523115"}:subscribe: databento::live::protocol: Sending subscription request sub_req=schema=mbo|stype_in=continuous|symbols=ES.c.0|snapshot=1
2025-03-31T23:00:55.946777Z DEBUG nautilus_databento::live: Received command: Start
2025-03-31T23:00:55.946787Z  INFO LiveClient{dataset=GLBX.MDP3 session_id="1743523115"}:start: databento::live::client: Starting session
2025-03-31T23:00:56.962580Z DEBUG nautilus_databento::live: Started

thread 'tokio-runtime-worker' panicked at crates/model/src/data/trade.rs:115:10:
Condition failed: Condition failed: invalid `Quantity` for 'size' not positive, was 0
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace

[cjdsellers]

Hi @JackWCollins

This is now fixed on develop branch from commit 3405de5, and will be included in the next release.

The solution is rather simplistic and probably not as efficient as it could be. We just maintain a set of "initialized books" where we have seen the initial snapshot already, and avoid attempting to decode the snapshot with a trade which results in failing that zero size condition check.