Add support for Binance's Mark Price Stream across all markets (#2670)

Author: sunlei

nautilus_trader/adapters/binance/data.py

@@ -170,6 +170,10 @@ def __init__(
         self._connect_websockets_delay: float = 0.0  # Delay for bulk subscriptions to come in
         self._connect_websockets_task: asyncio.Task | None = None
 
+        self._subscribe_allow_no_instrument_id = [
+            BinanceFuturesMarkPriceUpdate,
+        ]
+
         # HTTP API
         self._http_client = client
         self._http_market = market
@@ -296,7 +300,10 @@ def _should_retry(self, error_code: BinanceErrorCode, retries: int) -> bool:
 
     async def _subscribe(self, command: SubscribeData) -> None:
         instrument_id: InstrumentId | None = command.data_type.metadata.get("instrument_id")
-        if instrument_id is None:
+        if (
+            instrument_id is None
+            and command.data_type.type not in self._subscribe_allow_no_instrument_id
+        ):
             self._log.error(
                 f"Cannot subscribe to `{command.data_type.type}` no instrument ID in `data_type` metadata",
             )
@@ -307,21 +314,25 @@ async def _subscribe(self, command: SubscribeData) -> None:
         elif command.data_type.type == BinanceFuturesMarkPriceUpdate:
             if not self._binance_account_type.is_futures:
                 self._log.error(
-                    f"Cannot subscribe to `BinanceFuturesMarkPriceUpdate` "
+                    "Cannot subscribe to `BinanceFuturesMarkPriceUpdate` "
                     f"for {self._binance_account_type.value} account types",
                 )
                 return
-            await self._ws_client.subscribe_mark_price(instrument_id.symbol.value, speed=1000)
+            mark_price_symbol = instrument_id.symbol.value if instrument_id else None
+            await self._ws_client.subscribe_mark_price(mark_price_symbol, speed=1000)
         else:
             self._log.error(
                 f"Cannot subscribe to {command.data_type.type} (not implemented)",
             )
 
     async def _unsubscribe(self, command: UnsubscribeData) -> None:
         instrument_id: InstrumentId | None = command.data_type.metadata.get("instrument_id")
-        if instrument_id is None:
+        if (
+            instrument_id is None
+            and command.data_type.type not in self._subscribe_allow_no_instrument_id
+        ):
             self._log.error(
-                "Cannot subscribe to `BinanceFuturesMarkPriceUpdate` no instrument ID in `data_type` metadata",
+                "Cannot unsubscribe to `BinanceFuturesMarkPriceUpdate` no instrument ID in `data_type` metadata",
             )
             return
 

nautilus_trader/adapters/binance/futures/data.py

@@ -22,6 +22,8 @@
 from nautilus_trader.adapters.binance.data import BinanceCommonDataClient
 from nautilus_trader.adapters.binance.futures.enums import BinanceFuturesEnumParser
 from nautilus_trader.adapters.binance.futures.http.market import BinanceFuturesMarketHttpAPI
+from nautilus_trader.adapters.binance.futures.schemas.market import BinanceFuturesMarkPriceAllMsg
+from nautilus_trader.adapters.binance.futures.schemas.market import BinanceFuturesMarkPriceData
 from nautilus_trader.adapters.binance.futures.schemas.market import BinanceFuturesMarkPriceMsg
 from nautilus_trader.adapters.binance.futures.schemas.market import BinanceFuturesTradeMsg
 from nautilus_trader.adapters.binance.futures.types import BinanceFuturesMarkPriceUpdate
@@ -111,10 +113,14 @@ def __init__(
 
         # Register additional futures websocket handlers
         self._ws_handlers["@markPrice"] = self._handle_mark_price
+        self._ws_handlers["!markPrice@arr"] = self._handle_mark_price_all
 
         # Websocket msgspec decoders
         self._decoder_futures_trade_msg = msgspec.json.Decoder(BinanceFuturesTradeMsg)
         self._decoder_futures_mark_price_msg = msgspec.json.Decoder(BinanceFuturesMarkPriceMsg)
+        self._decoder_futures_mark_price_all_msg = msgspec.json.Decoder(
+            BinanceFuturesMarkPriceAllMsg,
+        )
 
     # -- WEBSOCKET HANDLERS ---------------------------------------------------------------------------------
 
@@ -149,10 +155,9 @@ def _handle_trade(self, raw: bytes) -> None:
         else:
             self._handle_data(trade_tick)
 
-    def _handle_mark_price(self, raw: bytes) -> None:
-        msg = self._decoder_futures_mark_price_msg.decode(raw)
-        instrument_id: InstrumentId = self._get_cached_instrument_id(msg.data.s)
-        data = msg.data.parse_to_binance_futures_mark_price_update(
+    def _handle_mark_price_data(self, data: BinanceFuturesMarkPriceData) -> None:
+        instrument_id: InstrumentId = self._get_cached_instrument_id(data.s)
+        data = data.parse_to_binance_futures_mark_price_update(
             instrument_id=instrument_id,
             ts_init=self._clock.timestamp_ns(),
         )
@@ -161,6 +166,7 @@ def _handle_mark_price(self, raw: bytes) -> None:
             metadata={"instrument_id": instrument_id},
         )
         generic = CustomData(data_type=data_type, data=data)
+
         self._handle_data(generic)
         self._handle_data(
             MarkPriceUpdate(
@@ -170,3 +176,12 @@ def _handle_mark_price(self, raw: bytes) -> None:
                 data.ts_init,
             ),
         )
+
+    def _handle_mark_price(self, raw: bytes) -> None:
+        msg = self._decoder_futures_mark_price_msg.decode(raw)
+        self._handle_mark_price_data(msg.data)
+
+    def _handle_mark_price_all(self, raw: bytes) -> None:
+        msg = self._decoder_futures_mark_price_all_msg.decode(raw)
+        for data in msg.data:
+            self._handle_mark_price_data(data)

nautilus_trader/adapters/binance/futures/schemas/market.py

@@ -248,3 +248,12 @@ class BinanceFuturesMarkPriceMsg(msgspec.Struct, frozen=True):
 
     stream: str
     data: BinanceFuturesMarkPriceData
+
+
+class BinanceFuturesMarkPriceAllMsg(msgspec.Struct, frozen=True):
+    """
+    WebSocket message from Binance Futures All Mark Price Update events.
+    """
+
+    stream: str
+    data: list[BinanceFuturesMarkPriceData]

nautilus_trader/adapters/binance/futures/types.py

@@ -13,6 +13,8 @@
 #  limitations under the License.
 # -------------------------------------------------------------------------------------------------
 
+from __future__ import annotations
+
 from decimal import Decimal
 from typing import Any
 
@@ -47,7 +49,7 @@ class BinanceFuturesMarkPriceUpdate(Data):
 
     References
     ----------
-    https://binance-docs.github.io/apidocs/futures/en/#mark-price-stream
+    https://developers.binance.com/docs/derivatives/usds-margined-futures/websocket-market-streams/Mark-Price-Stream
 
     """
 
@@ -109,7 +111,7 @@ def ts_init(self) -> int:
         return self._ts_init
 
     @staticmethod
-    def from_dict(values: dict[str, Any]) -> "BinanceFuturesMarkPriceUpdate":
+    def from_dict(values: dict[str, Any]) -> BinanceFuturesMarkPriceUpdate:
         """
         Return a Binance Futures mark price update parsed from the given values.
 
@@ -135,7 +137,7 @@ def from_dict(values: dict[str, Any]) -> "BinanceFuturesMarkPriceUpdate":
         )
 
     @staticmethod
-    def to_dict(obj: "BinanceFuturesMarkPriceUpdate") -> dict[str, Any]:
+    def to_dict(obj: BinanceFuturesMarkPriceUpdate) -> dict[str, Any]:
         """
         Return a dictionary representation of this object.
 

nautilus_trader/adapters/binance/websocket/client.py

@@ -225,7 +225,7 @@ async def _connect_client(self, client_id: int, streams: list[str]) -> None:
             msg = self._create_subscribe_msg(streams=streams[1:])
             await self._send(client_id, msg)
             self._log.debug(
-                f"Client {client_id}: Subscribed to additional {len(streams)-1} streams",
+                f"Client {client_id}: Subscribed to additional {len(streams) - 1} streams",
             )
 
     def _handle_ping(self, client_id: int, raw: bytes) -> None:
@@ -557,12 +557,17 @@ async def subscribe_mark_price(
         """
         Subscribe to aggregate mark price stream.
         """
-        if speed not in (1000, 3000):
+        if speed and speed not in (1000, 3000):
             raise ValueError(f"`speed` options are 1000ms or 3000ms only, was {speed}")
+
         if symbol is None:
             stream = "!markPrice@arr"
         else:
-            stream = f"{BinanceSymbol(symbol).lower()}@markPrice@{int(speed / 1000)}s"
+            stream = f"{BinanceSymbol(symbol).lower()}@markPrice"
+
+        if speed:
+            stream += f"@{int(speed / 1000)}s"
+
         await self._subscribe(stream)
 
     async def unsubscribe_mark_price(