feat: Add USD-M futures WebSocket URL category support (public/market/private)

Binance is retiring legacy USD-M futures WebSocket URLs on 2026-04-23.
This migrates all fstream methods to use the new categorized URL structure.

Author: pcriadoperez

binance/ws/keepalive_websocket.py

@@ -54,10 +54,15 @@ async def __aexit__(self, *args, **kwargs):
         await super().__aexit__(*args, **kwargs)
 
     def _build_path(self):
-        self._path = self._listen_key
+        if self._keepalive_type in ("futures", "coin_futures"):
+            self._path = f"?listenKey={self._listen_key}"
+            self._prefix = "ws"
+        else:
+            self._path = self._listen_key
         time_unit = getattr(self._client, "TIME_UNIT", None)
         if time_unit:
-            self._path = f"{self._listen_key}?timeUnit={time_unit}"
+            sep = "&" if "?" in self._path else "?"
+            self._path = f"{self._path}{sep}timeUnit={time_unit}"
 
     async def _before_connect(self):
         if self._keepalive_type == "user":

binance/ws/streams.py

@@ -73,7 +73,7 @@ def __init__(
         self.ws_kwargs = {}
 
         if verbose:
-            logging.getLogger('binance.ws').setLevel(logging.DEBUG)
+            logging.getLogger("binance.ws").setLevel(logging.DEBUG)
 
     def _get_stream_url(self, stream_url: Optional[str] = None):
         if stream_url:
@@ -136,7 +136,11 @@ def _get_account_socket(
         return self._conns[conn_id]
 
     def _get_futures_socket(
-        self, path: str, futures_type: FuturesType, prefix: str = "stream?streams="
+        self,
+        path: str,
+        futures_type: FuturesType,
+        prefix: str = "stream?streams=",
+        category: Optional[str] = None,
     ):
         socket_type: BinanceSocketType = BinanceSocketType.USD_M_FUTURES
         if futures_type == FuturesType.USD_M:
@@ -145,6 +149,8 @@ def _get_futures_socket(
                 stream_url = self.FSTREAM_TESTNET_URL
             elif self.demo:
                 stream_url = self.FSTREAM_DEMO_URL
+            if category:
+                stream_url = f"{stream_url}{category}/"
         else:
             stream_url = self.DSTREAM_URL
             if self.testnet:
@@ -340,7 +346,9 @@ def kline_futures_socket(
         """
 
         path = f"{symbol.lower()}_{contract_type.value}@continuousKline_{interval}"
-        return self._get_futures_socket(path, prefix="ws/", futures_type=futures_type)
+        return self._get_futures_socket(
+            path, prefix="ws/", futures_type=futures_type, category="market"
+        )
 
     def miniticker_socket(self, update_time: int = 1000):
         """Start a miniticker websocket for all trades
@@ -466,7 +474,7 @@ def aggtrade_futures_socket(
 
         """
         return self._get_futures_socket(
-            symbol.lower() + "@aggTrade", futures_type=futures_type
+            symbol.lower() + "@aggTrade", futures_type=futures_type, category="market"
         )
 
     def symbol_miniticker_socket(self, symbol: str):
@@ -621,7 +629,9 @@ def futures_ticker_socket(self):
                 }
             ]
         """
-        return self._get_futures_socket("!ticker@arr", FuturesType.USD_M)
+        return self._get_futures_socket(
+            "!ticker@arr", FuturesType.USD_M, category="market"
+        )
 
     def futures_coin_ticker_socket(self):
         """Start a websocket for all ticker data
@@ -707,7 +717,7 @@ def symbol_mark_price_socket(
         """
         stream_name = "@markPrice@1s" if fast else "@markPrice"
         return self._get_futures_socket(
-            symbol.lower() + stream_name, futures_type=futures_type
+            symbol.lower() + stream_name, futures_type=futures_type, category="market"
         )
 
     def all_mark_price_socket(
@@ -734,7 +744,9 @@ def all_mark_price_socket(
             ]
         """
         stream_name = "!markPrice@arr@1s" if fast else "!markPrice@arr"
-        return self._get_futures_socket(stream_name, futures_type=futures_type)
+        return self._get_futures_socket(
+            stream_name, futures_type=futures_type, category="market"
+        )
 
     def symbol_ticker_futures_socket(
         self, symbol: str, futures_type: FuturesType = FuturesType.USD_M
@@ -758,7 +770,7 @@ def symbol_ticker_futures_socket(
             ]
         """
         return self._get_futures_socket(
-            symbol.lower() + "@bookTicker", futures_type=futures_type
+            symbol.lower() + "@bookTicker", futures_type=futures_type, category="public"
         )
 
     def individual_symbol_ticker_futures_socket(
@@ -779,7 +791,7 @@ def individual_symbol_ticker_futures_socket(
             }
         """
         return self._get_futures_socket(
-            symbol.lower() + "@ticker", futures_type=futures_type
+            symbol.lower() + "@ticker", futures_type=futures_type, category="market"
         )
 
     def all_ticker_futures_socket(
@@ -811,7 +823,10 @@ def all_ticker_futures_socket(
             ]
         """
 
-        return self._get_futures_socket(channel, futures_type=futures_type)
+        category = "public" if "bookTicker" in channel else "market"
+        return self._get_futures_socket(
+            channel, futures_type=futures_type, category=category
+        )
 
     def symbol_book_ticker_socket(self, symbol: str):
         """Start a websocket for the best bid or ask's price or quantity for a specified symbol.
@@ -889,7 +904,10 @@ def options_multiplex_socket(self, streams: List[str]):
         return self._get_options_socket(stream_path, prefix="stream?")
 
     def futures_multiplex_socket(
-        self, streams: List[str], futures_type: FuturesType = FuturesType.USD_M
+        self,
+        streams: List[str],
+        futures_type: FuturesType = FuturesType.USD_M,
+        category: str = "market",
     ):
         """Start a multiplexed socket using a list of socket names.
         User stream sockets can not be included.
@@ -902,6 +920,7 @@ def futures_multiplex_socket(
 
         :param streams: list of stream names in lower case
         :param futures_type: use USD-M or COIN-M futures default USD-M
+        :param category: stream category for USD-M futures URL routing ("public", "market", or "private"), default "market"
 
         :returns: connection key string if successful, False otherwise
 
@@ -910,7 +929,7 @@ def futures_multiplex_socket(
         """
         path = f"streams={'/'.join(streams)}"
         return self._get_futures_socket(
-            path, prefix="stream?", futures_type=futures_type
+            path, prefix="stream?", futures_type=futures_type, category=category
         )
 
     def user_socket(self):
@@ -945,6 +964,7 @@ def futures_user_socket(self):
             stream_url = self.FSTREAM_TESTNET_URL
         elif self.demo:
             stream_url = self.FSTREAM_DEMO_URL
+        stream_url += "private/"
         return self._get_account_socket("futures", stream_url=stream_url)
 
     def coin_futures_user_socket(self):
@@ -989,6 +1009,7 @@ def futures_socket(self):
             stream_url = self.FSTREAM_TESTNET_URL
         elif self.demo:
             stream_url = self.FSTREAM_DEMO_URL
+        stream_url += "private/"
         return self._get_account_socket("futures", stream_url=stream_url)
 
     def coin_futures_socket(self):
@@ -1121,7 +1142,9 @@ def options_depth_socket(self, symbol: str, depth: str = "10"):
         """
         return self._get_options_socket(symbol.upper() + "@depth" + str(depth))
 
-    def futures_depth_socket(self, symbol: str, depth: str = "10", futures_type=FuturesType.USD_M):
+    def futures_depth_socket(
+        self, symbol: str, depth: str = "10", futures_type=FuturesType.USD_M
+    ):
         """Subscribe to a futures depth data stream
 
         https://binance-docs.github.io/apidocs/futures/en/#partial-book-depth-streams
@@ -1133,7 +1156,9 @@ def futures_depth_socket(self, symbol: str, depth: str = "10", futures_type=Futu
         :param futures_type: use USD-M or COIN-M futures default USD-M
         """
         return self._get_futures_socket(
-            symbol.lower() + "@depth" + str(depth), futures_type=futures_type
+            symbol.lower() + "@depth" + str(depth),
+            futures_type=futures_type,
+            category="public",
         )
 
     def futures_rpi_depth_socket(self, symbol: str, futures_type=FuturesType.USD_M):
@@ -1149,7 +1174,9 @@ def futures_rpi_depth_socket(self, symbol: str, futures_type=FuturesType.USD_M):
         :param futures_type: use USD-M or COIN-M futures default USD-M
         """
         return self._get_futures_socket(
-            symbol.lower() + "@rpiDepth@500ms", futures_type=futures_type
+            symbol.lower() + "@rpiDepth@500ms",
+            futures_type=futures_type,
+            category="public",
         )
 
     def options_new_symbol_socket(self):
@@ -1194,7 +1221,9 @@ def options_open_interest_socket(self, symbol: str, expiration_date: str):
         :param expiration_date: The expiration date (e.g., "221125" for Nov 25, 2022)
         :type expiration_date: str
         """
-        return self._get_options_socket(symbol.upper() + "@openInterest@" + expiration_date)
+        return self._get_options_socket(
+            symbol.upper() + "@openInterest@" + expiration_date
+        )
 
     def options_mark_price_socket(self, symbol: str):
         """Subscribe to an options mark price stream.
@@ -1277,8 +1306,7 @@ def __init__(
     async def _before_socket_listener_start(self):
         assert self._client
         self._bsm = BinanceSocketManager(
-            client=self._client,
-            max_queue_size=self._max_queue_size
+            client=self._client, max_queue_size=self._max_queue_size
         )
 
     def _start_async_socket(
@@ -1291,7 +1319,9 @@ def _start_async_socket(
         start_time = time.time()
         while not self._bsm:
             if time.time() - start_time > 5:
-                raise RuntimeError("Binance Socket Manager failed to initialize after 5 seconds")
+                raise RuntimeError(
+                    "Binance Socket Manager failed to initialize after 5 seconds"
+                )
             time.sleep(0.1)
         socket = getattr(self._bsm, socket_name)(**params)
         socket_path: str = path or socket._path  # noqa