Fix databento venue conversion for GLBX (#2121)

* Fix databento venue conversion for GLBX
* Add start as params for databento market data subscriptions

Author: faysou

examples/live/databento/databento_historical_data.py

@@ -97,7 +97,7 @@ def on_start(self) -> None:
         start_time = time_object_to_dt("2024-05-09T10:00")
         end_time = time_object_to_dt("2024-05-09T10:05")
         self.request_quote_ticks(
-            InstrumentId.from_str("ESM4.GLBX"),
+            InstrumentId.from_str("ESM4.XCME"),  # or "ESM4.GLBX"
             start_time,
             end_time,
             params={"schema": "bbo-1m"},

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

@@ -119,8 +119,9 @@ impl DatabentoHistoricalClient {
         })
     }
 
+    #[allow(clippy::too_many_arguments)]
     #[pyo3(name = "get_range_instruments")]
-    #[pyo3(signature = (dataset, instrument_ids, start, end=None, limit=None))]
+    #[pyo3(signature = (dataset, instrument_ids, start, end=None, limit=None, use_exchange_as_venue=false))]
     fn py_get_range_instruments<'py>(
         &self,
         py: Python<'py>,
@@ -129,6 +130,7 @@ impl DatabentoHistoricalClient {
         start: u64,
         end: Option<u64>,
         limit: Option<u64>,
+        use_exchange_as_venue: bool,
     ) -> PyResult<Bound<'py, PyAny>> {
         let client = self.inner.clone();
         let mut symbol_venue_map = self.symbol_venue_map.write().unwrap();
@@ -172,14 +174,21 @@ impl DatabentoHistoricalClient {
 
             while let Ok(Some(msg)) = decoder.decode_record::<dbn::InstrumentDefMsg>().await {
                 let record = dbn::RecordRef::from(msg);
-                let instrument_id = decode_nautilus_instrument_id(
+                let mut instrument_id = decode_nautilus_instrument_id(
                     &record,
                     &metadata,
                     &publisher_venue_map,
                     &symbol_venue_map.read().unwrap(),
                 )
                 .map_err(to_pyvalue_err)?;
 
+                if use_exchange_as_venue && instrument_id.venue == Venue::GLBX() {
+                    let exchange = msg.exchange().unwrap();
+                    let venue = Venue::from_code(exchange)
+                        .unwrap_or_else(|_| panic!("`Venue` not found for exchange {exchange}"));
+                    instrument_id.venue = venue;
+                }
+
                 let result = decode_instrument_def_msg(msg, instrument_id, ts_init);
                 match result {
                     Ok(instrument) => instruments.push(instrument),

nautilus_core/adapters/databento/src/symbology.rs

@@ -26,17 +26,16 @@ pub fn instrument_id_to_symbol_string(
     instrument_id: InstrumentId,
     symbol_venue_map: &mut HashMap<Symbol, Venue>,
 ) -> String {
-    if let Some(venue) = symbol_venue_map.get(&instrument_id.symbol) {
-        if venue == &Venue::CBCM()
-            || venue == &Venue::NYUM()
-            || venue == &Venue::XCBT()
-            || venue == &Venue::XCEC()
-            || venue == &Venue::XCME()
-            || venue == &Venue::XFXS()
-            || venue == &Venue::XNYM()
-        {
-            symbol_venue_map.insert(instrument_id.symbol, Venue::GLBX());
-        }
+    let venue = instrument_id.venue;
+    if venue == Venue::CBCM()
+        || venue == Venue::NYUM()
+        || venue == Venue::XCBT()
+        || venue == Venue::XCEC()
+        || venue == Venue::XCME()
+        || venue == Venue::XFXS()
+        || venue == Venue::XNYM()
+    {
+        symbol_venue_map.insert(instrument_id.symbol, venue);
     }
 
     instrument_id.symbol.to_string()
@@ -55,8 +54,6 @@ pub fn decode_nautilus_instrument_id(
         .ok_or_else(|| anyhow::anyhow!("`Venue` not found for `publisher_id` {publisher_id}"))?;
     let mut instrument_id = get_nautilus_instrument_id_for_record(record, metadata, *venue)?;
     if publisher == Publisher::GlbxMdp3Glbx {
-        // Source actual exchange from GLBX instrument
-        // definitions if they were loaded.
         if let Some(venue) = symbol_venue_map.get(&instrument_id.symbol) {
             instrument_id.venue = *venue;
         }

nautilus_trader/adapters/databento/data.py

@@ -612,11 +612,16 @@ async def _subscribe_quote_ticks(
             ]:
                 schema = DatabentoSchema.MBP_1.value
 
+            start = params.get("start") if params else None
+            start = start.value + 1 if start else None  # time in nanoseconds from pd.Timestamp
+            start = start if params and params.get("subscribe_from_start") else None
+
             dataset: Dataset = self._loader.get_dataset_for_venue(instrument_id.venue)
             live_client = self._get_live_client(dataset)
             live_client.subscribe(
                 schema=schema,
                 instrument_ids=[instrument_id_to_pyo3(instrument_id)],
+                start=start,
             )
 
             # Add trade tick subscriptions for instrument (MBP-1 data includes trades)
@@ -637,11 +642,16 @@ async def _subscribe_trade_ticks(
 
             await self._ensure_subscribed_for_instrument(instrument_id)
 
+            start = params.get("start") if params else None
+            start = start.value + 1 if start else None  # time in nanoseconds from pd.Timestamp
+            start = start if params and params.get("subscribe_from_start") else None
+
             dataset: Dataset = self._loader.get_dataset_for_venue(instrument_id.venue)
             live_client = self._get_live_client(dataset)
             live_client.subscribe(
                 schema=DatabentoSchema.TRADES.value,
                 instrument_ids=[instrument_id_to_pyo3(instrument_id)],
+                start=start,
             )
             await self._check_live_client_started(dataset, live_client)
         except asyncio.CancelledError:
@@ -661,10 +671,15 @@ async def _subscribe_bars(
                 self._log.error(f"Cannot subscribe: {e}")
                 return
 
+            start = params.get("start") if params else None
+            start = start.value + 1 if start else None  # time in nanoseconds from pd.Timestamp
+            start = start if params and params.get("subscribe_from_start") else None
+
             live_client = self._get_live_client(dataset)
             live_client.subscribe(
                 schema=schema.value,
                 instrument_ids=[instrument_id_to_pyo3(bar_type.instrument_id)],
+                start=start,
             )
             await self._check_live_client_started(dataset, live_client)
         except asyncio.CancelledError:
@@ -941,11 +956,14 @@ async def _request_instruments(
             LogColor.BLUE,
         )
 
+        use_exchange_as_venue = params is not None and params.get("use_exchange_as_venue", False)
+
         pyo3_instruments = await self._http_client.get_range_instruments(
             dataset=dataset,
             instrument_ids=[instrument_id_to_pyo3(InstrumentId.from_str(f"{ALL_SYMBOLS}.{venue}"))],
             start=start.value,
             end=end.value,
+            use_exchange_as_venue=use_exchange_as_venue,
         )
 
         instruments = instruments_from_pyo3(pyo3_instruments)

nautilus_trader/common/actor.pyx

@@ -28,7 +28,6 @@ import asyncio
 from concurrent.futures import Executor
 
 import cython
-import msgspec
 
 from nautilus_trader.common.config import ActorConfig
 from nautilus_trader.common.config import ImportableActorConfig

nautilus_trader/core/nautilus_pyo3.pyi

@@ -4235,6 +4235,7 @@ class DatabentoHistoricalClient:
         start: int,
         end: int | None = None,
         limit: int | None = None,
+        use_exchange_as_venue: bool = False
     ) -> list[Instrument]: ...
     async def get_range_quotes(
         self,

nautilus_trader/data/engine.pyx

@@ -984,6 +984,15 @@ cdef class DataEngine(Component):
             return
         Condition.not_none(client, "client")
 
+        if not params.get("start"):
+            last_timestamp: datetime | None = self._catalogs_last_timestamp(
+                QuoteTick,
+                instrument_id,
+            )[0]
+
+            params = params if params else {}
+            params["start"] = last_timestamp
+
         if instrument_id not in client.subscribed_quote_ticks():
             client.subscribe_quote_ticks(instrument_id, params)
 
@@ -1025,6 +1034,15 @@ cdef class DataEngine(Component):
             return
         Condition.not_none(client, "client")
 
+        if not params.get("start"):
+            last_timestamp: datetime | None = self._catalogs_last_timestamp(
+                TradeTick,
+                instrument_id,
+            )[0]
+
+            params = params if params else {}
+            params["start"] = last_timestamp
+
         if instrument_id not in client.subscribed_trade_ticks():
             client.subscribe_trade_ticks(instrument_id, params)
 
@@ -1076,6 +1094,15 @@ cdef class DataEngine(Component):
                 )
                 return
 
+            if not params.get("start"):
+                last_timestamp: datetime | None = self._catalogs_last_timestamp(
+                    Bar,
+                    bar_type=bar_type,
+                )[0]
+
+                params = params if params else {}
+                params["start"] = last_timestamp
+
             if bar_type not in client.subscribed_bars():
                 client.subscribe_bars(bar_type, params)
 
@@ -1090,6 +1117,12 @@ cdef class DataEngine(Component):
 
         try:
             if data_type not in client.subscribed_custom_data():
+                if not params.get("start"):
+                    last_timestamp: datetime | None = self._catalogs_last_timestamp(data_type.type)[0]
+
+                    params = params if params else {}
+                    params["start"] = last_timestamp
+
                 client.subscribe(data_type, params)
         except NotImplementedError:
             self._log.error(
@@ -1419,10 +1452,10 @@ cdef class DataEngine(Component):
         datetime end,
         dict params,
     ):
-        last_timestamp, _ = self._catalogs_last_timestamp(
+        last_timestamp = self._catalogs_last_timestamp(
             Instrument,
             instrument_id,
-        )
+        )[0]
 
         if last_timestamp:
             self._query_catalog(request)
@@ -1474,10 +1507,10 @@ cdef class DataEngine(Component):
     ):
         instrument_id = request.data_type.metadata.get("instrument_id")
 
-        last_timestamp, _ = self._catalogs_last_timestamp(
+        last_timestamp = self._catalogs_last_timestamp(
             QuoteTick,
             instrument_id,
-        )
+        )[0]
 
         if last_timestamp:
             if (now <= last_timestamp) or (end and end <= last_timestamp):
@@ -1515,10 +1548,10 @@ cdef class DataEngine(Component):
     ):
         instrument_id = request.data_type.metadata.get("instrument_id")
 
-        last_timestamp, _ = self._catalogs_last_timestamp(
+        last_timestamp = self._catalogs_last_timestamp(
             TradeTick,
             instrument_id,
-        )
+        )[0]
 
         if last_timestamp:
             if (now <= last_timestamp) or (end and end <= last_timestamp):
@@ -1556,10 +1589,10 @@ cdef class DataEngine(Component):
     ):
         bar_type = request.data_type.metadata.get("bar_type")
 
-        last_timestamp, _ = self._catalogs_last_timestamp(
+        last_timestamp = self._catalogs_last_timestamp(
             Bar,
             bar_type=bar_type,
-        )
+        )[0]
 
         if last_timestamp:
             if (now <= last_timestamp) or (end and end <= last_timestamp):
@@ -1595,9 +1628,9 @@ cdef class DataEngine(Component):
         datetime now,
         dict params,
     ):
-        last_timestamp, _ = self._catalogs_last_timestamp(
+        last_timestamp = self._catalogs_last_timestamp(
             request.data_type.type,
-        )
+        )[0]
 
         if last_timestamp:
             if (now <= last_timestamp) or (end and end <= last_timestamp):