Refine usage of databento use_exchange_as_venue (#2487)

Author: faysou

crates/adapters/databento/src/live.rs

@@ -20,7 +20,7 @@ use std::{
 
 use ahash::{HashSet, HashSetExt};
 use databento::{
-    dbn::{self, PitSymbolMap, Publisher, Record, SymbolIndex, VersionUpgradePolicy},
+    dbn::{self, PitSymbolMap, Record, SymbolIndex, VersionUpgradePolicy},
     live::Subscription,
 };
 use indexmap::IndexMap;
@@ -75,6 +75,7 @@ pub struct DatabentoFeedHandler {
     publisher_venue_map: IndexMap<PublisherId, Venue>,
     symbol_venue_map: Arc<RwLock<HashMap<Symbol, Venue>>>,
     replay: bool,
+    use_exchange_as_venue: bool,
 }
 
 impl DatabentoFeedHandler {
@@ -87,6 +88,7 @@ impl DatabentoFeedHandler {
         tx: tokio::sync::mpsc::Sender<LiveMessage>,
         publisher_venue_map: IndexMap<PublisherId, Venue>,
         symbol_venue_map: Arc<RwLock<HashMap<Symbol, Venue>>>,
+        use_exchange_as_venue: bool,
     ) -> Self {
         Self {
             key,
@@ -96,6 +98,7 @@ impl DatabentoFeedHandler {
             publisher_venue_map,
             symbol_venue_map,
             replay: false,
+            use_exchange_as_venue,
         }
     }
 
@@ -214,9 +217,7 @@ 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 {
+                if self.use_exchange_as_venue {
                     update_instrument_id_map_with_exchange(
                         &symbol_map,
                         &self.symbol_venue_map,
@@ -390,7 +391,11 @@ fn update_instrument_id_map_with_exchange(
     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);
+    symbol_venue_map
+        .write()
+        .unwrap()
+        .entry(symbol)
+        .or_insert(venue);
     instrument_id_map.insert(raw_instrument_id, instrument_id);
     instrument_id
 }

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

@@ -71,12 +71,17 @@ pub struct DatabentoHistoricalClient {
     inner: Arc<Mutex<databento::HistoricalClient>>,
     publisher_venue_map: Arc<IndexMap<PublisherId, Venue>>,
     symbol_venue_map: Arc<RwLock<HashMap<Symbol, Venue>>>,
+    use_exchange_as_venue: bool,
 }
 
 #[pymethods]
 impl DatabentoHistoricalClient {
     #[new]
-    fn py_new(key: String, publishers_filepath: PathBuf) -> PyResult<Self> {
+    fn py_new(
+        key: String,
+        publishers_filepath: PathBuf,
+        use_exchange_as_venue: bool,
+    ) -> PyResult<Self> {
         let client = databento::HistoricalClient::builder()
             .key(key.clone())
             .map_err(to_pyvalue_err)?
@@ -98,6 +103,7 @@ impl DatabentoHistoricalClient {
             publisher_venue_map: Arc::new(publisher_venue_map),
             symbol_venue_map: Arc::new(RwLock::new(HashMap::new())),
             key,
+            use_exchange_as_venue,
         })
     }
 
@@ -127,7 +133,7 @@ impl DatabentoHistoricalClient {
     }
 
     #[pyo3(name = "get_range_instruments")]
-    #[pyo3(signature = (dataset, instrument_ids, start, end=None, limit=None, use_exchange_as_venue=false))]
+    #[pyo3(signature = (dataset, instrument_ids, start, end=None, limit=None))]
     #[allow(clippy::too_many_arguments)]
     fn py_get_range_instruments<'py>(
         &self,
@@ -137,7 +143,6 @@ 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();
@@ -165,6 +170,7 @@ impl DatabentoHistoricalClient {
         let publisher_venue_map = self.publisher_venue_map.clone();
         let symbol_venue_map = self.symbol_venue_map.clone();
         let ts_init = self.clock.get_time_ns();
+        let use_exchange_as_venue = self.use_exchange_as_venue;
 
         pyo3_async_runtimes::tokio::future_into_py(py, async move {
             let mut client = client.lock().await; // TODO: Use a client pool

crates/adapters/databento/src/python/live.rs

@@ -54,6 +54,7 @@ pub struct DatabentoLiveClient {
     buffer_size: usize,
     publisher_venue_map: IndexMap<u16, Venue>,
     symbol_venue_map: Arc<RwLock<HashMap<Symbol, Venue>>>,
+    use_exchange_as_venue: bool,
 }
 
 impl DatabentoLiveClient {
@@ -127,7 +128,12 @@ fn call_python(py: Python, callback: &PyObject, py_obj: PyObject) {
 #[pymethods]
 impl DatabentoLiveClient {
     #[new]
-    pub fn py_new(key: String, dataset: String, publishers_filepath: PathBuf) -> PyResult<Self> {
+    pub fn py_new(
+        key: String,
+        dataset: String,
+        publishers_filepath: PathBuf,
+        use_exchange_as_venue: bool,
+    ) -> PyResult<Self> {
         let publishers_json = fs::read_to_string(publishers_filepath)?;
         let publishers_vec: Vec<DatabentoPublisher> =
             serde_json::from_str(&publishers_json).map_err(to_pyvalue_err)?;
@@ -151,6 +157,7 @@ impl DatabentoLiveClient {
             is_closed: false,
             publisher_venue_map,
             symbol_venue_map: Arc::new(RwLock::new(HashMap::new())),
+            use_exchange_as_venue,
         })
     }
 
@@ -231,6 +238,7 @@ impl DatabentoLiveClient {
             msg_tx,
             self.publisher_venue_map.clone(),
             self.symbol_venue_map.clone(),
+            self.use_exchange_as_venue,
         );
 
         self.send_command(LiveCommand::Start)?;

nautilus_trader/adapters/databento/data.py

@@ -284,6 +284,7 @@ def _get_live_client(self, dataset: Dataset) -> nautilus_pyo3.DatabentoLiveClien
                 key=self._live_api_key,
                 dataset=dataset,
                 publishers_filepath=str(PUBLISHERS_FILEPATH),
+                use_exchange_as_venue=self._use_exchange_as_venue,
             )
             self._live_clients[dataset] = live_client
 
@@ -298,6 +299,7 @@ def _get_live_client_mbo(self, dataset: Dataset) -> nautilus_pyo3.DatabentoLiveC
                 key=self._live_api_key,
                 dataset=dataset,
                 publishers_filepath=str(PUBLISHERS_FILEPATH),
+                use_exchange_as_venue=self._use_exchange_as_venue,
             )
             self._live_clients_mbo[dataset] = live_client
 
@@ -874,7 +876,6 @@ async def _request_instrument(self, request: RequestInstrument) -> None:
             instrument_ids=[instrument_id_to_pyo3(request.instrument_id)],
             start=start.value,
             end=end.value,
-            use_exchange_as_venue=self._use_exchange_as_venue,
         )
 
         instruments = instruments_from_pyo3(pyo3_instruments)
@@ -899,8 +900,6 @@ async def _request_instruments(self, request: RequestInstruments) -> None:
             LogColor.BLUE,
         )
 
-        use_exchange_as_venue = request.params.get("use_exchange_as_venue", False)
-
         # parent_symbols can be equal to ["ES.OPT", "ES.FUT"] for example in order to not query all instruments of an exchange
         parent_symbols = request.params.get("parent_symbols") or [ALL_SYMBOLS]
         pyo3_instrument_ids = [
@@ -913,7 +912,6 @@ async def _request_instruments(self, request: RequestInstruments) -> None:
             instrument_ids=pyo3_instrument_ids,
             start=start.value,
             end=end.value,
-            use_exchange_as_venue=use_exchange_as_venue,
         )
         instruments = instruments_from_pyo3(pyo3_instruments)
 

nautilus_trader/adapters/databento/factories.py

@@ -34,6 +34,7 @@
 def get_cached_databento_http_client(
     key: str | None = None,
     gateway: str | None = None,
+    use_exchange_as_venue: bool = True,
 ) -> nautilus_pyo3.DatabentoHistoricalClient:
     """
     Cache and return a Databento historical HTTP client with the given key and gateway.
@@ -46,6 +47,8 @@ def get_cached_databento_http_client(
         The Databento API secret key for the client.
     gateway : str, optional
         The Databento historical HTTP client gateway override.
+    use_exchange_as_venue : bool, default True
+        If the `exchange` field will be used as the venue for instrument IDs.
 
     Returns
     -------
@@ -55,6 +58,7 @@ def get_cached_databento_http_client(
     return nautilus_pyo3.DatabentoHistoricalClient(
         key=key or get_env_key("DATABENTO_API_KEY"),
         publishers_filepath=str(PUBLISHERS_FILEPATH),
+        use_exchange_as_venue=use_exchange_as_venue,
     )
 
 
@@ -66,6 +70,7 @@ def get_cached_databento_instrument_provider(
     live_gateway: str | None = None,
     loader: DatabentoDataLoader | None = None,
     config: InstrumentProviderConfig | None = None,
+    use_exchange_as_venue=True,
 ) -> DatabentoInstrumentProvider:
     """
     Cache and return a Databento instrument provider.
@@ -87,6 +92,8 @@ def get_cached_databento_instrument_provider(
         The loader for the provider.
     config : InstrumentProviderConfig
         The configuration for the instrument provider.
+    use_exchange_as_venue : bool, default True
+        If the `exchange` field will be used as the venue for instrument IDs.
 
     Returns
     -------
@@ -100,6 +107,7 @@ def get_cached_databento_instrument_provider(
         live_gateway=live_gateway,
         loader=loader,
         config=config,
+        use_exchange_as_venue=use_exchange_as_venue,
     )
 
 
@@ -144,6 +152,7 @@ def create(  # type: ignore
         http_client = get_cached_databento_http_client(
             key=config.api_key,
             gateway=config.http_gateway,
+            use_exchange_as_venue=config.use_exchange_as_venue,
         )
 
         loader = DatabentoDataLoader(config.venue_dataset_map)
@@ -154,6 +163,7 @@ def create(  # type: ignore
             live_gateway=config.live_gateway,
             loader=loader,
             config=config.instrument_provider,
+            use_exchange_as_venue=config.use_exchange_as_venue,
         )
 
         return DatabentoDataClient(

nautilus_trader/adapters/databento/providers.py

@@ -66,6 +66,7 @@ def __init__(
         live_gateway: str | None = None,
         loader: DatabentoDataLoader | None = None,
         config: InstrumentProviderConfig | None = None,
+        use_exchange_as_venue: bool = True,
     ) -> None:
         super().__init__(config=config)
 
@@ -76,6 +77,7 @@ def __init__(
 
         self._http_client = http_client
         self._loader = loader or DatabentoDataLoader()
+        self._use_exchange_as_venue = use_exchange_as_venue
 
     async def load_all_async(self, filters: dict | None = None) -> None:
         raise RuntimeError(
@@ -122,11 +124,10 @@ async def load_ids_async(  # noqa: C901 (too complex)
             key=self._live_api_key,
             dataset=dataset,
             publishers_filepath=str(PUBLISHERS_FILEPATH),
+            use_exchange_as_venue=self._use_exchange_as_venue,
         )
 
         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 = []
 
         success_msg = "All instruments received and decoded"
@@ -157,7 +158,6 @@ 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:
@@ -265,7 +265,6 @@ 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.
@@ -274,7 +273,6 @@ async def get_range(
             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)
         instruments = sorted(instruments, key=lambda x: x.ts_init)

nautilus_trader/core/nautilus_pyo3.pyi

@@ -4911,6 +4911,7 @@ class DatabentoHistoricalClient:
         self,
         key: str,
         publishers_filepath: str,
+        use_exchange_as_venue: bool,
     ) -> None: ...
     @property
     def key(self) -> str: ...
@@ -4922,7 +4923,6 @@ 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,
@@ -4982,6 +4982,7 @@ class DatabentoLiveClient:
         key: str,
         dataset: str,
         publishers_filepath: str,
+        use_exchange_as_venue: bool,
     ) -> None: ...
     @property
     def key(self) -> str: ...