Continue support for GLBX exchange venue mappings

Not yet operational for continuous contracts and parent symbols.

Author: cjdsellers

examples/live/databento/databento_subscriber.py

@@ -45,11 +45,11 @@
 # For correct subscription operation, you must specify all instruments to be immediately
 # subscribed for as part of the data client configuration
 instrument_ids = [
-    InstrumentId.from_str("ES.c.0.GLBX"),
-    # InstrumentId.from_str("ESZ4.GLBX"),
-    # InstrumentId.from_str("ES.FUT.GLBX"),
-    # InstrumentId.from_str("CL.FUT.GLBX"),
-    # InstrumentId.from_str("LO.OPT.GLBX"),
+    InstrumentId.from_str("ES.c.0.GLBX"),  # TODO: Continuous contracts only work with GLBX for now
+    # InstrumentId.from_str("ESZ4.XCME"),
+    # InstrumentId.from_str("ES.FUT.XCME"),
+    # InstrumentId.from_str("CL.FUT.NYMEX"),
+    # InstrumentId.from_str("LO.OPT.NYMEX"),
     # InstrumentId.from_str("AAPL.XNAS"),
     # InstrumentId.from_str("AAPL.IEXG"),
 ]
@@ -87,9 +87,10 @@
             api_key=None,  # 'DATABENTO_API_KEY' env var
             http_gateway=None,
             instrument_provider=InstrumentProviderConfig(load_all=True),
+            use_exchange_as_venue=True,
+            mbo_subscriptions_delay=10.0,
             instrument_ids=instrument_ids,
             parent_symbols={"GLBX.MDP3": {"ES.FUT"}},
-            mbo_subscriptions_delay=10.0,
         ),
     },
     timeout_connection=30.0,
@@ -154,6 +155,7 @@ def on_start(self) -> None:
             #     interval_ms=1000,
             # )
 
+            # self.subscribe_instrument(parent_symbol, client_id=DATABENTO_CLIENT_ID)
             self.subscribe_quote_ticks(instrument_id, client_id=DATABENTO_CLIENT_ID)
             self.subscribe_trade_ticks(instrument_id, client_id=DATABENTO_CLIENT_ID)
             # self.subscribe_bars(BarType.from_str(f"{instrument_id}-1-SECOND-LAST-EXTERNAL"))

nautilus_core/adapters/databento/src/live.rs

@@ -19,15 +19,11 @@ use std::{
 };
 
 use databento::{
-    dbn,
-    dbn::{PitSymbolMap, Record, SymbolIndex, VersionUpgradePolicy},
+    dbn::{self, PitSymbolMap, Publisher, Record, SymbolIndex, VersionUpgradePolicy},
     live::Subscription,
 };
 use indexmap::IndexMap;
-use nautilus_core::{
-    python::to_pyruntime_err,
-    time::{get_atomic_clock_realtime, AtomicTime},
-};
+use nautilus_core::{nanos::UnixNanos, python::to_pyruntime_err, time::get_atomic_clock_realtime};
 use nautilus_model::{
     data::{
         delta::OrderBookDelta,
@@ -209,6 +205,8 @@ impl DatabentoFeedHandler {
                 }
             };
 
+            let ts_init = clock.get_time_ns();
+
             // Decode record
             if let Some(msg) = record.get::<dbn::ErrorMsg>() {
                 handle_error_msg(msg);
@@ -219,14 +217,25 @@ impl DatabentoFeedHandler {
                 instrument_id_map.remove(&msg.hd.instrument_id);
                 handle_symbol_mapping_msg(msg, &mut symbol_map, &mut instrument_id_map);
             } else if let Some(msg) = record.get::<dbn::InstrumentDefMsg>() {
+                // TODO: Make this configurable so that exchange can be used as venue
+                let use_exchange_as_venue = false;
+                if use_exchange_as_venue && msg.publisher()? == Publisher::GlbxMdp3Glbx {
+                    update_instrument_id_map_with_exchange(
+                        &symbol_map,
+                        &self.symbol_venue_map,
+                        &mut instrument_id_map,
+                        msg.hd.instrument_id,
+                        msg.exchange()?,
+                    );
+                };
                 let data = handle_instrument_def_msg(
                     msg,
                     &record,
                     &symbol_map,
                     &self.publisher_venue_map,
                     &self.symbol_venue_map.read().unwrap(),
                     &mut instrument_id_map,
-                    clock,
+                    ts_init,
                 )?;
                 self.send_msg(LiveMessage::Instrument(data)).await;
             } else if let Some(msg) = record.get::<dbn::StatusMsg>() {
@@ -237,7 +246,7 @@ impl DatabentoFeedHandler {
                     &self.publisher_venue_map,
                     &self.symbol_venue_map.read().unwrap(),
                     &mut instrument_id_map,
-                    clock,
+                    ts_init,
                 )?;
                 self.send_msg(LiveMessage::Status(data)).await;
             } else if let Some(msg) = record.get::<dbn::ImbalanceMsg>() {
@@ -248,7 +257,7 @@ impl DatabentoFeedHandler {
                     &self.publisher_venue_map,
                     &self.symbol_venue_map.read().unwrap(),
                     &mut instrument_id_map,
-                    clock,
+                    ts_init,
                 )?;
                 self.send_msg(LiveMessage::Imbalance(data)).await;
             } else if let Some(msg) = record.get::<dbn::StatMsg>() {
@@ -259,17 +268,17 @@ impl DatabentoFeedHandler {
                     &self.publisher_venue_map,
                     &self.symbol_venue_map.read().unwrap(),
                     &mut instrument_id_map,
-                    clock,
+                    ts_init,
                 )?;
                 self.send_msg(LiveMessage::Statistics(data)).await;
             } else {
                 let (mut data1, data2) = match handle_record(
                     record,
                     &symbol_map,
                     &self.publisher_venue_map,
-                    &self.symbol_venue_map.read().unwrap(), // TODO: Optimize this bottleneck
+                    &self.symbol_venue_map.read().unwrap(),
                     &mut instrument_id_map,
-                    clock,
+                    ts_init,
                 ) {
                     Ok(decoded) => decoded,
                     Err(e) => {
@@ -363,6 +372,24 @@ fn handle_symbol_mapping_msg(
     instrument_id_map.remove(&msg.header().instrument_id);
 }
 
+fn update_instrument_id_map_with_exchange(
+    symbol_map: &PitSymbolMap,
+    symbol_venue_map: &RwLock<HashMap<Symbol, Venue>>,
+    instrument_id_map: &mut HashMap<u32, InstrumentId>,
+    raw_instrument_id: u32,
+    exchange: &str,
+) -> InstrumentId {
+    let raw_symbol = symbol_map
+        .get(raw_instrument_id)
+        .expect("Cannot resolve `raw_symbol` from `symbol_map`");
+    let symbol = Symbol::from(raw_symbol.as_str());
+    let venue = Venue::from(exchange);
+    let instrument_id = InstrumentId::new(symbol, venue);
+    symbol_venue_map.write().unwrap().insert(symbol, venue);
+    instrument_id_map.insert(raw_instrument_id, instrument_id);
+    instrument_id
+}
+
 fn update_instrument_id_map(
     record: &dbn::RecordRef,
     symbol_map: &PitSymbolMap,
@@ -403,7 +430,7 @@ fn handle_instrument_def_msg(
     publisher_venue_map: &IndexMap<PublisherId, Venue>,
     symbol_venue_map: &HashMap<Symbol, Venue>,
     instrument_id_map: &mut HashMap<u32, InstrumentId>,
-    clock: &AtomicTime,
+    ts_init: UnixNanos,
 ) -> anyhow::Result<InstrumentAny> {
     let instrument_id = update_instrument_id_map(
         record,
@@ -412,7 +439,6 @@ fn handle_instrument_def_msg(
         symbol_venue_map,
         instrument_id_map,
     );
-    let ts_init = clock.get_time_ns();
 
     decode_instrument_def_msg(msg, instrument_id, ts_init)
 }
@@ -424,7 +450,7 @@ fn handle_status_msg(
     publisher_venue_map: &IndexMap<PublisherId, Venue>,
     symbol_venue_map: &HashMap<Symbol, Venue>,
     instrument_id_map: &mut HashMap<u32, InstrumentId>,
-    clock: &AtomicTime,
+    ts_init: UnixNanos,
 ) -> anyhow::Result<InstrumentStatus> {
     let instrument_id = update_instrument_id_map(
         record,
@@ -433,7 +459,6 @@ fn handle_status_msg(
         symbol_venue_map,
         instrument_id_map,
     );
-    let ts_init = clock.get_time_ns();
 
     decode_status_msg(msg, instrument_id, ts_init)
 }
@@ -445,7 +470,7 @@ fn handle_imbalance_msg(
     publisher_venue_map: &IndexMap<PublisherId, Venue>,
     symbol_venue_map: &HashMap<Symbol, Venue>,
     instrument_id_map: &mut HashMap<u32, InstrumentId>,
-    clock: &AtomicTime,
+    ts_init: UnixNanos,
 ) -> anyhow::Result<DatabentoImbalance> {
     let instrument_id = update_instrument_id_map(
         record,
@@ -456,7 +481,6 @@ fn handle_imbalance_msg(
     );
 
     let price_precision = 2; // Hard-coded for now
-    let ts_init = clock.get_time_ns();
 
     decode_imbalance_msg(msg, instrument_id, price_precision, ts_init)
 }
@@ -468,7 +492,7 @@ fn handle_statistics_msg(
     publisher_venue_map: &IndexMap<PublisherId, Venue>,
     symbol_venue_map: &HashMap<Symbol, Venue>,
     instrument_id_map: &mut HashMap<u32, InstrumentId>,
-    clock: &AtomicTime,
+    ts_init: UnixNanos,
 ) -> anyhow::Result<DatabentoStatistics> {
     let instrument_id = update_instrument_id_map(
         record,
@@ -479,7 +503,6 @@ fn handle_statistics_msg(
     );
 
     let price_precision = 2; // Hard-coded for now
-    let ts_init = clock.get_time_ns();
 
     decode_statistics_msg(msg, instrument_id, price_precision, ts_init)
 }
@@ -490,7 +513,7 @@ fn handle_record(
     publisher_venue_map: &IndexMap<PublisherId, Venue>,
     symbol_venue_map: &HashMap<Symbol, Venue>,
     instrument_id_map: &mut HashMap<u32, InstrumentId>,
-    clock: &AtomicTime,
+    ts_init: UnixNanos,
 ) -> anyhow::Result<(Option<Data>, Option<Data>)> {
     let instrument_id = update_instrument_id_map(
         &record,
@@ -501,7 +524,6 @@ fn handle_record(
     );
 
     let price_precision = 2; // Hard-coded for now
-    let ts_init = clock.get_time_ns();
 
     decode_record(
         &record,

nautilus_trader/adapters/databento/config.py

@@ -30,25 +30,28 @@ class DatabentoDataClientConfig(LiveDataClientConfig, frozen=True):
         The historical HTTP client gateway override.
     live_gateway : str, optional
         The live client gateway override.
-    parent_symbols : dict[str, set[str]], optional
-        The Databento parent symbols to subscribe to instrument definitions for on start.
-        This is a map of Databento dataset keys -> to a sequence of the parent symbols,
-        e.g. {'GLBX.MDP3', ['ES.FUT', 'ES.OPT']} (for all E-mini S&P 500 futures and options products).
-    instrument_ids : list[InstrumentId], optional
-        The instrument IDs to request instrument definitions for on start.
+    use_exchange_as_venue : bool, default True
+        If the `exchange` field will be used as the venue for instrument IDs.
     timeout_initial_load : float, default 15.0
         The timeout (seconds) to wait for instruments to load (concurrently per dataset).
     mbo_subscriptions_delay : float, default 3.0
         The timeout (seconds) to wait for MBO/L3 subscriptions (concurrently per dataset).
         After the timeout the MBO order book feed will start and replay messages from the initial
         snapshot and then all deltas.
+    parent_symbols : dict[str, set[str]], optional
+        The Databento parent symbols to subscribe to instrument definitions for on start.
+        This is a map of Databento dataset keys -> to a sequence of the parent symbols,
+        e.g. {'GLBX.MDP3', ['ES.FUT', 'ES.OPT']} (for all E-mini S&P 500 futures and options products).
+    instrument_ids : list[InstrumentId], optional
+        The instrument IDs to request instrument definitions for on start.
 
     """
 
     api_key: str | None = None
     http_gateway: str | None = None
     live_gateway: str | None = None
-    instrument_ids: list[InstrumentId] | None = None
-    parent_symbols: dict[str, set[str]] | None = None
+    use_exchange_as_venue: bool = True
     timeout_initial_load: float | None = 15.0
     mbo_subscriptions_delay: float | None = 3.0  # Need to have received all definitions
+    instrument_ids: list[InstrumentId] | None = None
+    parent_symbols: dict[str, set[str]] | None = None

nautilus_trader/adapters/databento/data.py

@@ -115,10 +115,15 @@ def __init__(
         # Configuration
         self._live_api_key: str = config.api_key or http_client.key
         self._live_gateway: str | None = config.live_gateway
-        self._parent_symbols: dict[Dataset, set[str]] = defaultdict(set)
-        self._instrument_ids: dict[Dataset, set[InstrumentId]] = defaultdict(set)
+        self._use_exchange_as_venue: bool = config.use_exchange_as_venue
         self._timeout_initial_load: float | None = config.timeout_initial_load
         self._mbo_subscriptions_delay: float | None = config.mbo_subscriptions_delay
+        self._parent_symbols: dict[Dataset, set[str]] = defaultdict(set)
+        self._instrument_ids: dict[Dataset, set[InstrumentId]] = defaultdict(set)
+
+        self._log.info(f"{config.use_exchange_as_venue=}", LogColor.BLUE)
+        self._log.info(f"{config.timeout_initial_load=}", LogColor.BLUE)
+        self._log.info(f"{config.mbo_subscriptions_delay=}", LogColor.BLUE)
 
         # Clients
         self._http_client = http_client
@@ -161,7 +166,10 @@ async def _connect(self) -> None:
         coros: list[Coroutine] = []
         for dataset, instrument_ids in self._instrument_ids.items():
             loading_ids: list[InstrumentId] = sorted(instrument_ids)
-            filters = {"parent_symbols": list(self._parent_symbols.get(dataset, []))}
+            filters = {
+                "use_exchange_as_venue": self._use_exchange_as_venue,
+                "parent_symbols": list(self._parent_symbols.get(dataset, [])),
+            }
             coro = self._instrument_provider.load_ids_async(
                 instrument_ids=loading_ids,
                 filters=filters,
@@ -424,10 +432,13 @@ async def _subscribe_instrument(
     ) -> None:
         try:
             dataset: Dataset = self._loader.get_dataset_for_venue(instrument_id.venue)
+            start: int | None = params.get("start") if params else None
+
             live_client = self._get_live_client(dataset)
             live_client.subscribe(
                 schema=DatabentoSchema.DEFINITION.value,
                 instrument_ids=[instrument_id_to_pyo3(instrument_id)],
+                start=start,
             )
             await self._check_live_client_started(dataset, live_client)
         except asyncio.CancelledError:
@@ -436,17 +447,14 @@ async def _subscribe_instrument(
     async def _subscribe_parent_symbols(
         self,
         dataset: Dataset,
-        parent_instrument_ids: set[InstrumentId],
+        parent_symbols: set[InstrumentId],
     ) -> None:
         try:
             live_client = self._get_live_client(dataset)
             live_client.subscribe(
                 schema=DatabentoSchema.DEFINITION.value,
                 instrument_ids=sorted(  # type: ignore[type-var]
-                    [
-                        instrument_id_to_pyo3(instrument_id)
-                        for instrument_id in parent_instrument_ids
-                    ],
+                    [instrument_id_to_pyo3(instrument_id) for instrument_id in parent_symbols],
                 ),
                 stype_in="parent",
             )
@@ -919,6 +927,7 @@ async def _request_instrument(
             instrument_ids=[instrument_id_to_pyo3(instrument_id)],
             start=start.value,
             end=end.value,
+            use_exchange_as_venue=self._use_exchange_as_venue,
         )
 
         instruments = instruments_from_pyo3(pyo3_instruments)

nautilus_trader/adapters/databento/providers.py

@@ -126,6 +126,7 @@ async def load_ids_async(  # noqa: C901 (too complex)
         )
 
         parent_symbols = list(filters.get("parent_symbols", [])) if filters is not None else None
+        # use_exchange_as_venue = filters.get("use_exchange_as_venue", True) if filters else True
 
         pyo3_instruments = []
 
@@ -157,6 +158,7 @@ def receive_instruments(pyo3_instrument: Any) -> None:
                 [instrument_id_to_pyo3(instrument_id) for instrument_id in instrument_ids],
             ),
             start=0,  # From start of current week (latest definitions)
+            # use_exchange_as_venue=use_exchange_as_venue,  # TODO
         )
 
         if parent_symbols:
@@ -226,7 +228,7 @@ async def load_async(
         Calling this method will incur a cost to your Databento account in USD.
 
         """
-        await self.load_ids_async([instrument_id])
+        await self.load_ids_async([instrument_id], filters=filters)
 
     async def get_range(
         self,
@@ -262,12 +264,16 @@ async def get_range(
 
         """
         dataset = self._check_all_datasets_equal(instrument_ids)
+        use_exchange_as_venue = filters.get("use_exchange_as_venue", True) if filters else True
 
+        # Here the NULL venue is overridden and so is used as a
+        # placeholder to conform to instrument ID conventions.
         pyo3_instruments = await self._http_client.get_range_instruments(
             dataset=dataset,
-            instrument_ids=[instrument_id_to_pyo3(InstrumentId.from_str(f"{ALL_SYMBOLS}.GLBX"))],
+            instrument_ids=[instrument_id_to_pyo3(InstrumentId.from_str(f"{ALL_SYMBOLS}.NULL"))],
             start=pd.Timestamp(start, tz=pytz.utc).value,
             end=pd.Timestamp(end, tz=pytz.utc).value if end is not None else None,
+            use_exchange_as_venue=use_exchange_as_venue,
         )
 
         instruments = instruments_from_pyo3(pyo3_instruments)