Add support for bbo-1s and bbo-1m quotes in databento adapter

examples/backtest/databento_option_greeks.py

@@ -1,13 +1,25 @@
+# ---
+# jupyter:
+#   jupytext:
+#     formats: py:percent
+#     text_representation:
+#       extension: .py
+#       format_name: percent
+#       format_version: '1.3'
+#       jupytext_version: 1.16.4
+#   kernelspec:
+#     display_name: Python 3 (ipykernel)
+#     language: python
+#     name: python3
+# ---
+
 # %% [markdown]
 # ## imports
 
 # %%
 # Note: Use the python extension jupytext to be able to open this python file in jupyter as a notebook
 
 # %%
-# from nautilus_trader.adapters.databento.data_utils import init_databento_client
-import nautilus_trader.adapters.databento.data_utils as db_data_utils
-from nautilus_trader import PACKAGE_ROOT
 from nautilus_trader.adapters.databento.data_utils import data_path
 from nautilus_trader.adapters.databento.data_utils import databento_data
 from nautilus_trader.adapters.databento.data_utils import load_catalog
@@ -43,8 +55,6 @@
 # ## parameters
 
 # %%
-db_data_utils.DATA_PATH = PACKAGE_ROOT / "tests" / "test_data" / "databento"
-
 catalog_folder = "option_catalog_example"
 catalog = load_catalog(catalog_folder)
 

nautilus_core/adapters/src/databento/decode.rs

@@ -615,6 +615,26 @@ pub fn decode_mbp1_msg(
     Ok((quote, maybe_trade))
 }
 
+pub fn decode_bbo_msg(
+    msg: &dbn::BboMsg,
+    instrument_id: InstrumentId,
+    price_precision: u8,
+    ts_init: UnixNanos,
+) -> anyhow::Result<QuoteTick> {
+    let top_level = &msg.levels[0];
+    let quote = QuoteTick::new(
+        instrument_id,
+        Price::from_raw(top_level.bid_px, price_precision),
+        Price::from_raw(top_level.ask_px, price_precision),
+        Quantity::from_raw(u64::from(top_level.bid_sz) * FIXED_SCALAR as u64, 0),
+        Quantity::from_raw(u64::from(top_level.ask_sz) * FIXED_SCALAR as u64, 0),
+        msg.ts_recv.into(),
+        ts_init,
+    );
+
+    Ok(quote)
+}
+
 pub fn decode_mbp10_msg(
     msg: &dbn::Mbp10Msg,
     instrument_id: InstrumentId,
@@ -801,6 +821,14 @@ pub fn decode_record(
             (quote, None) => (Some(Data::Quote(quote)), None),
             (quote, Some(trade)) => (Some(Data::Quote(quote)), Some(Data::Trade(trade))),
         }
+    } else if let Some(msg) = record.get::<dbn::Bbo1SMsg>() {
+        let ts_init = determine_timestamp(ts_init, msg.ts_recv.into());
+        let quote = decode_bbo_msg(msg, instrument_id, price_precision, ts_init)?;
+        (Some(Data::Quote(quote)), None)
+    } else if let Some(msg) = record.get::<dbn::Bbo1MMsg>() {
+        let ts_init = determine_timestamp(ts_init, msg.ts_recv.into());
+        let quote = decode_bbo_msg(msg, instrument_id, price_precision, ts_init)?;
+        (Some(Data::Quote(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, ts_init)?;
@@ -1248,6 +1276,48 @@ mod tests {
         assert_eq!(quote.ts_init, 0);
     }
 
+    #[rstest]
+    fn test_decode_bbo_1s_msg() {
+        let path = PathBuf::from(format!("{TEST_DATA_PATH}/test_data.bbo-1s.dbn.zst"));
+        let mut dbn_stream = Decoder::from_zstd_file(path)
+            .unwrap()
+            .decode_stream::<dbn::BboMsg>();
+        let msg = dbn_stream.next().unwrap().unwrap();
+
+        let instrument_id = InstrumentId::from("ESM4.GLBX");
+        let quote = decode_bbo_msg(msg, instrument_id, 2, 0.into()).unwrap();
+
+        assert_eq!(quote.instrument_id, instrument_id);
+        assert_eq!(quote.bid_price, Price::from("5199.50"));
+        assert_eq!(quote.ask_price, Price::from("5199.75"));
+        assert_eq!(quote.bid_size, Quantity::from("26"));
+        assert_eq!(quote.ask_size, Quantity::from("23"));
+        assert_eq!(quote.ts_event, msg.ts_recv);
+        assert_eq!(quote.ts_event, 1715248801000000000);
+        assert_eq!(quote.ts_init, 0);
+    }
+
+    #[rstest]
+    fn test_decode_bbo_1m_msg() {
+        let path = PathBuf::from(format!("{TEST_DATA_PATH}/test_data.bbo-1m.dbn.zst"));
+        let mut dbn_stream = Decoder::from_zstd_file(path)
+            .unwrap()
+            .decode_stream::<dbn::BboMsg>();
+        let msg = dbn_stream.next().unwrap().unwrap();
+
+        let instrument_id = InstrumentId::from("ESM4.GLBX");
+        let quote = decode_bbo_msg(msg, instrument_id, 2, 0.into()).unwrap();
+
+        assert_eq!(quote.instrument_id, instrument_id);
+        assert_eq!(quote.bid_price, Price::from("5199.50"));
+        assert_eq!(quote.ask_price, Price::from("5199.75"));
+        assert_eq!(quote.bid_size, Quantity::from("33"));
+        assert_eq!(quote.ask_size, Quantity::from("17"));
+        assert_eq!(quote.ts_event, msg.ts_recv);
+        assert_eq!(quote.ts_event, 1715248800000000000);
+        assert_eq!(quote.ts_init, 0);
+    }
+
     #[rstest]
     fn test_decode_mbp10_msg() {
         let path = PathBuf::from(format!("{TEST_DATA_PATH}/test_data.mbp-10.dbn.zst"));

nautilus_core/adapters/src/databento/loader.rs

@@ -309,6 +309,26 @@ impl DatabentoDataLoader {
             .collect()
     }
 
+    pub fn load_bbo_quotes(
+        &self,
+        filepath: PathBuf,
+        instrument_id: Option<InstrumentId>,
+    ) -> anyhow::Result<Vec<QuoteTick>> {
+        self.read_records::<dbn::BboMsg>(filepath, instrument_id, false)?
+            .filter_map(|result| match result {
+                Ok((Some(item1), _)) => {
+                    if let Data::Quote(quote) = item1 {
+                        Some(Ok(quote))
+                    } else {
+                        None
+                    }
+                }
+                Ok((None, _)) => None,
+                Err(e) => Some(Err(e)),
+            })
+            .collect()
+    }
+
     pub fn load_tbbo_trades(
         &self,
         filepath: PathBuf,

nautilus_core/adapters/src/databento/python/loader.rs

@@ -170,6 +170,31 @@ impl DatabentoDataLoader {
         exhaust_data_iter_to_pycapsule(py, iter).map_err(to_pyvalue_err)
     }
 
+    #[pyo3(name = "load_bbo_quotes")]
+    fn py_load_bbo_quotes(
+        &self,
+        filepath: PathBuf,
+        instrument_id: Option<InstrumentId>,
+    ) -> PyResult<Vec<QuoteTick>> {
+        Ok(self
+            .load_bbo_quotes(filepath, instrument_id)
+            .map_err(to_pyvalue_err)?)
+    }
+
+    #[pyo3(name = "load_bbo_quotes_as_pycapsule")]
+    fn py_load_bbo_quotes_as_pycapsule(
+        &self,
+        py: Python,
+        filepath: PathBuf,
+        instrument_id: Option<InstrumentId>,
+    ) -> PyResult<PyObject> {
+        let iter = self
+            .read_records::<dbn::BboMsg>(filepath, instrument_id, false)
+            .map_err(to_pyvalue_err)?;
+
+        exhaust_data_iter_to_pycapsule(py, iter).map_err(to_pyvalue_err)
+    }
+
     #[pyo3(name = "load_tbbo_trades")]
     fn py_load_tbbo_trades(
         &self,

nautilus_core/adapters/src/databento/symbology.rs

@@ -47,6 +47,10 @@ pub fn get_nautilus_instrument_id_for_record(
         (msg.hd.instrument_id, msg.ts_recv)
     } else if let Some(msg) = record.get::<dbn::Mbp1Msg>() {
         (msg.hd.instrument_id, msg.ts_recv)
+    } else if let Some(msg) = record.get::<dbn::Bbo1SMsg>() {
+        (msg.hd.instrument_id, msg.ts_recv)
+    } else if let Some(msg) = record.get::<dbn::Bbo1MMsg>() {
+        (msg.hd.instrument_id, msg.ts_recv)
     } else if let Some(msg) = record.get::<dbn::Mbp10Msg>() {
         (msg.hd.instrument_id, msg.ts_recv)
     } else if let Some(msg) = record.get::<dbn::OhlcvMsg>() {

nautilus_trader/adapters/databento/data_utils.py

@@ -1,12 +1,14 @@
 from datetime import datetime
 from datetime import timedelta
-from pathlib import Path
 
+from nautilus_trader import PACKAGE_ROOT
 from nautilus_trader.adapters.databento.loaders import DatabentoDataLoader
 from nautilus_trader.persistence.catalog import ParquetDataCatalog
 
 
-DATA_PATH = Path("~/databento_data").expanduser()
+# Note: when using the functions below, change the variable below to a folder path
+# where you store all your databento data
+DATA_PATH = PACKAGE_ROOT / "tests" / "test_data" / "databento"
 
 # this variable can be modified with a valid key if downloading data is needed
 DATABENTO_API_KEY = "db-XXXXX"

nautilus_trader/adapters/databento/enums.py

@@ -24,6 +24,8 @@ class DatabentoSchema(Enum):
     MBO = "mbo"
     MBP_1 = "mbp-1"
     MBP_10 = "mbp-10"
+    BBO_1S = "bbo-1s"
+    BBO_1M = "bbo-1m"
     TBBO = "tbbo"
     TRADES = "trades"
     OHLCV_1S = "ohlcv-1s"

nautilus_trader/adapters/databento/loaders.py

@@ -210,6 +210,21 @@ def from_dbn_file(  # noqa: C901 (too complex)
                         filepath=str(path),
                         instrument_id=pyo3_instrument_id,
                     )
+            case DatabentoSchema.BBO_1S.value | DatabentoSchema.BBO_1M.value:
+                if as_legacy_cython:
+                    capsule = self._pyo3_loader.load_bbo_quotes_as_pycapsule(
+                        filepath=str(path),
+                        instrument_id=pyo3_instrument_id,
+                    )
+                    data = capsule_to_list(capsule)
+                    # Drop encapsulated `CVec` as data is now transferred
+                    drop_cvec_pycapsule(capsule)
+                    return data
+                else:
+                    return self._pyo3_loader.load_bbo_quotes(
+                        filepath=str(path),
+                        instrument_id=pyo3_instrument_id,
+                    )
             case DatabentoSchema.MBP_10.value:
                 if as_legacy_cython:
                     capsule = self._pyo3_loader.load_order_book_depth10_as_pycapsule(

nautilus_trader/core/nautilus_pyo3.pyi

@@ -3852,6 +3852,8 @@ class DatabentoDataLoader:
     def load_order_book_depth10_as_pycapsule(self, filepath: str, instrument_id: InstrumentId | None) -> object: ...
     def load_quotes(self, filepath: str, instrument_id: InstrumentId | None) -> list[QuoteTick]: ...
     def load_quotes_as_pycapsule(self, filepath: str, instrument_id: InstrumentId | None, include_trades: bool | None) -> object: ...
+    def load_bbo_quotes(self, filepath: str, instrument_id: InstrumentId | None) -> list[QuoteTick]: ...
+    def load_bbo_quotes_as_pycapsule(self, filepath: str, instrument_id: InstrumentId | None) -> object: ...
     def load_trades(self, filepath: str, instrument_id: InstrumentId | None) -> list[TradeTick]: ...
     def load_trades_as_pycapsule(self, filepath: str, instrument_id: InstrumentId | None) -> object: ...
     def load_bars(self, filepath: str, instrument_id: InstrumentId | None) -> list[Bar]: ...

tests/integration_tests/adapters/databento/test_loaders.py

@@ -274,6 +274,90 @@ def test_loader_mbp_1_pyo3() -> None:
     assert quote.ts_init == 1609160400006136329
 
 
+def test_loader_bbo_1s() -> None:
+    # Arrange
+    loader = DatabentoDataLoader()
+    path = DATABENTO_TEST_DATA_DIR / "bbo-1s.dbn.zst"
+
+    # Act
+    data = loader.from_dbn_file(path, as_legacy_cython=True)
+
+    # Assert
+    assert len(data) > 0
+    assert isinstance(data[0], QuoteTick)
+    quote = data[0]
+    assert quote.instrument_id == InstrumentId.from_str("ESM4.GLBX")
+    assert quote.bid_price == Price.from_str("5199.50")
+    assert quote.ask_price == Price.from_str("5199.75")
+    assert quote.bid_size == Quantity.from_int(26)
+    assert quote.ask_size == Quantity.from_int(23)
+    assert quote.ts_event == 1715248801000000000
+    assert quote.ts_init == 1715248801000000000
+
+
+def test_loader_bbo_1s_pyo3() -> None:
+    # Arrange
+    loader = DatabentoDataLoader()
+    path = DATABENTO_TEST_DATA_DIR / "bbo-1s.dbn.zst"
+
+    # Act
+    data = loader.from_dbn_file(path, as_legacy_cython=False)
+
+    # Assert
+    assert len(data) > 0
+    assert isinstance(data[0], nautilus_pyo3.QuoteTick)
+    quote = data[0]
+    assert quote.instrument_id == nautilus_pyo3.InstrumentId.from_str("ESM4.GLBX")
+    assert quote.bid_price == nautilus_pyo3.Price.from_str("5199.50")
+    assert quote.ask_price == nautilus_pyo3.Price.from_str("5199.75")
+    assert quote.bid_size == nautilus_pyo3.Quantity.from_int(26)
+    assert quote.ask_size == nautilus_pyo3.Quantity.from_int(23)
+    assert quote.ts_event == 1715248801000000000
+    assert quote.ts_init == 1715248801000000000
+
+
+def test_loader_bbo_1m() -> None:
+    # Arrange
+    loader = DatabentoDataLoader()
+    path = DATABENTO_TEST_DATA_DIR / "bbo-1m.dbn.zst"
+
+    # Act
+    data = loader.from_dbn_file(path, as_legacy_cython=True)
+
+    # Assert
+    assert len(data) > 0
+    assert isinstance(data[0], QuoteTick)
+    quote = data[0]
+    assert quote.instrument_id == InstrumentId.from_str("ESM4.GLBX")
+    assert quote.bid_price == Price.from_str("5199.50")
+    assert quote.ask_price == Price.from_str("5199.75")
+    assert quote.bid_size == Quantity.from_int(33)
+    assert quote.ask_size == Quantity.from_int(17)
+    assert quote.ts_event == 1715248800000000000
+    assert quote.ts_init == 1715248800000000000
+
+
+def test_loader_bbo_1m_pyo3() -> None:
+    # Arrange
+    loader = DatabentoDataLoader()
+    path = DATABENTO_TEST_DATA_DIR / "bbo-1m.dbn.zst"
+
+    # Act
+    data = loader.from_dbn_file(path, as_legacy_cython=False)
+
+    # Assert
+    assert len(data) > 0
+    assert isinstance(data[0], nautilus_pyo3.QuoteTick)
+    quote = data[0]
+    assert quote.instrument_id == nautilus_pyo3.InstrumentId.from_str("ESM4.GLBX")
+    assert quote.bid_price == nautilus_pyo3.Price.from_str("5199.50")
+    assert quote.ask_price == nautilus_pyo3.Price.from_str("5199.75")
+    assert quote.bid_size == nautilus_pyo3.Quantity.from_int(33)
+    assert quote.ask_size == nautilus_pyo3.Quantity.from_int(17)
+    assert quote.ts_event == 1715248800000000000
+    assert quote.ts_init == 1715248800000000000
+
+
 def test_loader_mbp_10() -> None:
     # Arrange
     loader = DatabentoDataLoader()