Fix BinanceWebSocketClient reconnects

RELEASES.md

@@ -41,6 +41,7 @@ This will be the final release with support for Python 3.9.
 - Fixed `OrderBook` pickling (did not include all attributes), thanks @limx0
 - Fixed open position snapshots race condition (added `open_only` flag)
 - Fixed `Strategy.cancel_order` for orders in `INITIALIZED` state and with an `emulation_trigger` (was not sending command to `OrderEmulator`)
+- Fixed `BinanceWebSocketClient` reconnect behavior (reconnect handler was not being called due event loop issue from Rust)
 - Fixed Binance instruments missing max notional values, thanks for reporting @AnthonyVince and thanks for fixing @filipmacek
 - Fixed Binance Futures fee rates for backtesting
 - Fixed `Timer` missing condition check for non-positive intervals

nautilus_trader/adapters/binance/common/data.py

@@ -155,6 +155,7 @@ def __init__(
             logger=logger,
             handler=self._handle_ws_message,
             base_url=base_url_ws,
+            loop=self._loop,
         )
 
         # Hot caches

nautilus_trader/adapters/binance/common/execution.py

@@ -187,6 +187,7 @@ def __init__(
             logger=logger,
             handler=self._handle_user_ws_message,
             base_url=base_url_ws,
+            loop=self._loop,
         )
 
         # Hot caches

nautilus_trader/adapters/binance/futures/execution.py

@@ -261,6 +261,9 @@ def _handle_user_ws_message(self, raw: bytes) -> None:
         # TODO(cs): Uncomment for development
         # self._log.info(str(json.dumps(msgspec.json.decode(raw), indent=4)), color=LogColor.MAGENTA)
         wrapper = self._decoder_futures_user_msg_wrapper.decode(raw)
+        if not wrapper.stream:
+            # Control message response
+            return
         try:
             self._futures_user_ws_handlers[wrapper.data.e](raw)
         except Exception as e:

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

@@ -68,8 +68,8 @@ class BinanceFuturesUserMsgWrapper(msgspec.Struct, frozen=True):
     Provides a wrapper for execution WebSocket messages from `Binance`.
     """
 
-    stream: str
-    data: BinanceFuturesUserMsgData
+    data: Optional[BinanceFuturesUserMsgData] = None
+    stream: Optional[str] = None
 
 
 class MarginCallPosition(msgspec.Struct, frozen=True):

nautilus_trader/adapters/binance/websocket/client.py

@@ -39,6 +39,8 @@ class BinanceWebSocketClient:
         The base URL for the WebSocket connection.
     handler : Callable[[bytes], None]
         The callback handler for message events.
+    loop : asyncio.AbstractEventLoop
+        The event loop for the client.
 
     References
     ----------
@@ -52,13 +54,15 @@ def __init__(
         logger: Logger,
         base_url: str,
         handler: Callable[[bytes], None],
+        loop: asyncio.AbstractEventLoop,
     ) -> None:
         self._clock = clock
         self._logger = logger
         self._log: LoggerAdapter = LoggerAdapter(type(self).__name__, logger=logger)
 
         self._base_url: str = base_url
         self._handler: Callable[[bytes], None] = handler
+        self._loop = loop
 
         self._streams: list[str] = []
         self._inner: Optional[WebSocketClient] = None
@@ -137,8 +141,7 @@ def reconnect(self) -> None:
         self._log.warning(f"Reconnected to {self._base_url}.")
 
         # Re-subscribe to all streams
-        loop = asyncio.get_event_loop()
-        loop.create_task(self._subscribe_all())
+        self._loop.create_task(self._subscribe_all())
 
     async def disconnect(self) -> None:
         """

tests/integration_tests/adapters/binance/sandbox/sandbox_ws_futures_market.py

@@ -31,6 +31,7 @@ async def test_binance_websocket_client():
         logger=Logger(clock=clock),
         handler=print,
         base_url="wss://fstream.binance.com",
+        loop=asyncio.get_event_loop(),
     )
 
     await client.connect()

tests/integration_tests/adapters/binance/sandbox/sandbox_ws_spot_user.py

@@ -46,6 +46,7 @@ async def test_binance_websocket_client():
         clock=clock,
         logger=Logger(clock=clock),
         handler=print,
+        loop=asyncio.get_event_loop(),
     )
 
     ws.subscribe(key=key)