Add start as params for databento market data subscriptions

Author: faysou

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:

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/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):