ADD: Add Python client support for statistics

Author: nmacholl

CHANGELOG.md

@@ -1,11 +1,12 @@
 # Changelog
 
 ## 0.13.0 - TBD
+- Added support for `statistics` schema
+- Upgraded `databento-dbn` to 0.5.1
 - Renamed `booklevel` MBP field to `levels` for brevity and consistent naming
 - Changed `flags` field to an unsigned int
 - Changed default of `ts_out` to `False` for `Live` client
-- Removed `open_interest_qty` and `cleared_volume` fields that were always unset from
-  definition schema
+- Removed `open_interest_qty` and `cleared_volume` fields that were always unset from definition schema
 
 ## 0.12.0 - 2023-05-01
 - Added `Live` client for connecting to Databento's live service

databento/common/data.py

@@ -170,6 +170,20 @@ def get_deriv_ba_types(level: int) -> List[Tuple[str, Union[type, str]]]:
     ("dummy", "S1"),
 ]
 
+STATISTICS_MSG: List[Tuple[str, Union[type, str]]] = RECORD_HEADER + [
+    ("ts_recv", np.uint64),
+    ("ts_ref", np.uint64),
+    ("price", np.int64),
+    ("quantity", np.int32),
+    ("sequence", np.uint32),
+    ("ts_in_delta", np.int32),
+    ("stat_type", np.uint16),
+    ("channel_id", np.uint16),
+    ("update_action", np.uint8),
+    ("stat_flags", np.uint8),
+    ("dummy", "S6"),
+]
+
 
 STRUCT_MAP: Dict[Schema, List[Tuple[str, Union[type, str]]]] = {
     Schema.MBO: MBO_MSG,
@@ -193,6 +207,7 @@ def get_deriv_ba_types(level: int) -> List[Tuple[str, Union[type, str]]]:
     Schema.OHLCV_1D: OHLCV_MSG,
     Schema.DEFINITION: DEFINITION_MSG,
     Schema.IMBALANCE: IMBALANCE_MSG,
+    Schema.STATISTICS: STATISTICS_MSG,
 }
 
 
@@ -288,6 +303,13 @@ def get_deriv_ba_fields(level: int) -> List[str]:
     "dummy",
 ]
 
+STATISTICS_DROP_COLUMNS = [
+    "ts_recv",
+    "length",
+    "rtype",
+    "dummy",
+]
+
 DEFINITION_COLUMNS = [
     x
     for x in (np.dtype(DEFINITION_MSG).names or ())
@@ -298,6 +320,12 @@ def get_deriv_ba_fields(level: int) -> List[str]:
     x for x in (np.dtype(IMBALANCE_MSG).names or ()) if x not in IMBALANCE_DROP_COLUMNS
 ]
 
+STATISTICS_COLUMNS = [
+    x
+    for x in (np.dtype(STATISTICS_MSG).names or ())
+    if x not in STATISTICS_DROP_COLUMNS
+]
+
 COLUMNS = {
     Schema.MBO: [
         "ts_event",
@@ -333,4 +361,5 @@ def get_deriv_ba_fields(level: int) -> List[str]:
     Schema.OHLCV_1D: OHLCV_HEADER_COLUMNS,
     Schema.DEFINITION: DEFINITION_COLUMNS,
     Schema.IMBALANCE: IMBALANCE_COLUMNS,
+    Schema.STATISTICS: STATISTICS_COLUMNS,
 }

databento/common/dbnstore.py

@@ -383,11 +383,19 @@ def _apply_pretty_ts(self, df: pd.DataFrame) -> pd.DataFrame:
         df.index = pd.to_datetime(df.index, utc=True)
         for column in df.columns:
             if column.startswith("ts_") and "delta" not in column:
-                df[column] = pd.to_datetime(df[column], utc=True)
+                df[column] = pd.to_datetime(df[column], errors="coerce", utc=True)
 
         if self.schema == Schema.DEFINITION:
-            df["expiration"] = pd.to_datetime(df["expiration"], utc=True)
-            df["activation"] = pd.to_datetime(df["activation"], utc=True)
+            df["expiration"] = pd.to_datetime(
+                df["expiration"],
+                errors="coerce",
+                utc=True,
+            )
+            df["activation"] = pd.to_datetime(
+                df["activation"],
+                errors="coerce",
+                utc=True,
+            )
 
         return df
 

databento/common/enums.py

@@ -10,6 +10,7 @@
     MBP1Msg,
     MBP10Msg,
     OHLCVMsg,
+    StatMsg,
     TradeMsg,
 )
 
@@ -192,6 +193,7 @@ class Schema(StringyMixin, str, Enum):
     OHLCV_1D = "ohlcv-1d"
     DEFINITION = "definition"
     IMBALANCE = "imbalance"
+    STATISTICS = "statistics"
 
     def get_record_type(self) -> Type[DBNRecord]:
         if self == Schema.MBO:
@@ -216,6 +218,8 @@ def get_record_type(self) -> Type[DBNRecord]:
             return InstrumentDefMsg
         if self == Schema.IMBALANCE:
             return ImbalanceMsg
+        if self == Schema.STATISTICS:
+            return StatMsg
         raise NotImplementedError(f"No message type for {self}")
 
 

databento/live/dbn.py

@@ -22,6 +22,7 @@
     databento_dbn.OHLCVMsg,
     databento_dbn.ImbalanceMsg,
     databento_dbn.InstrumentDefMsg,
+    databento_dbn.StatMsg,
     databento_dbn.SymbolMappingMsg,
     databento_dbn.SystemMsg,
     databento_dbn.ErrorMsg,

requirements.txt

@@ -1,5 +1,5 @@
 aiohttp>=3.7.2,<4.0.0
-databento-dbn==0.5.0
+databento-dbn==0.5.1
 numpy>=1.17.0
 pandas>=1.1.3
 requests>=2.24.0

tests/data/test_data.statistics.dbn.zst

@@ -256,21 +256,7 @@ def test_to_df_across_schemas_returns_identical_dimension_dfs(
 
 @pytest.mark.parametrize(
     "schema",
-    [
-        pytest.param(schema, id=str(schema))
-        for schema in (
-            Schema.MBO,
-            Schema.MBP_1,
-            Schema.MBP_10,
-            Schema.TBBO,
-            Schema.TRADES,
-            Schema.OHLCV_1S,
-            Schema.OHLCV_1M,
-            Schema.OHLCV_1H,
-            Schema.OHLCV_1D,
-            Schema.DEFINITION,
-        )
-    ],
+    [pytest.param(schema, id=str(schema)) for schema in Schema],
 )
 def test_to_df_drop_columns(
     test_data: Callable[[Schema], bytes],