session logon

.github/workflows/python-app.yml

@@ -34,7 +34,7 @@ jobs:
   build:
     needs: lint
     runs-on: ubuntu-22.04
-    timeout-minutes: 60
+    timeout-minutes: 20
     env:
       PROXY: "http://51.83.140.52:16301"
       TEST_TESTNET: "true"
@@ -67,19 +67,18 @@ jobs:
       run: pyright
     - name: Test with tox
       run: tox -e py
-  # comment due to coveralls maintenance of April 6 2025: https://status.coveralls.io/
-  #   - name: Coveralls Parallel
-  #     uses: coverallsapp/github-action@v2
-  #     with:
-  #       flag-name: run-${{ join(matrix.*, '-') }}
-  #       parallel: true
-  # finish:
-  #   needs: build
-  #   if: ${{ always() }}
-  #   runs-on: ubuntu-latest
-  #   timeout-minutes: 5
-  #   steps:
-  #   - name: Coveralls Finished
-  #     uses: coverallsapp/github-action@v2
-  #     with:
-  #       parallel-finished: true
+    - name: Coveralls Parallel
+      uses: coverallsapp/github-action@v2
+      with:
+        flag-name: run-${{ join(matrix.*, '-') }}
+        parallel: true
+  finish:
+    needs: build
+    if: ${{ always() }}
+    runs-on: ubuntu-latest
+    timeout-minutes: 5
+    steps:
+    - name: Coveralls Finished
+      uses: coverallsapp/github-action@v2
+      with:
+        parallel-finished: true

binance/async_client.py

@@ -3396,6 +3396,57 @@ async def create_oco_order(self, **params):
     # WebSocket API methods
     ############################################################
 
+    async def ws_logon(self, **params):
+        """Authenticate WebSocket connection using the provided API key.
+        https://developers.binance.com/docs/binance-spot-api-docs/websocket-api/authentication-requests
+
+        :returns: WS response
+        .. code-block:: python
+            {
+                "apiKey": "vmPUZE6mv9SD5VNHk4HlWFsOr6aKE2zvsw0MuIgwCIPy6utIco14y7Ju91duEh8A",
+                "authorizedSince": 1649729878532,
+                "connectedSince": 1649729873021,
+                "returnRateLimits": false,
+                "serverTime": 1649729878630,
+                "userDataStream": false
+            }
+        """
+        return await self._ws_api_request("session.logon", True, params)
+
+    async def ws_user_data_stream_subscribe(self, **params):
+        """Subscribe to the User Data Stream in the current WebSocket connection.
+        https://developers.binance.com/docs/binance-spot-api-docs/websocket-api/user-data-stream-requests#subscribe-to-user-data-stream-user_stream
+
+        Note:
+        - This method requires an authenticated WebSocket connection using Ed25519 keys. Please use ws_logon first.
+        - To check the subscription status, use ws_session_status, see the userDataStream flag.
+        - User Data Stream events are available in both JSON and SBE sessions.
+
+        :returns: WS response
+        .. code-block:: python
+            {
+                "id": "d3df8a21-98ea-4fe0-8f4e-0fcea5d418b7",
+                "status": 200,
+                "result": {}
+            }
+        """
+        return await self._ws_api_request("userDataStream.subscribe", True, params)
+
+    async def ws_user_data_stream_unsubscribe(self, **params):
+        """Stop listening to the User Data Stream in the current WebSocket connection.
+        https://developers.binance.com/docs/binance-spot-api-docs/websocket-api/user-data-stream-requests#unsubscribe-from-user-data-stream-user_stream
+
+        :returns: WS response
+        .. code-block:: python
+            {
+                "id": "d3df8a21-98ea-4fe0-8f4e-0fcea5d418b7",
+                "status": 200,
+                "result": {}
+            }
+        """
+        return await self._ws_api_request("userDataStream.unsubscribe", False, params)
+
+
     async def ws_create_test_order(self, **params):
         """Test new order creation and signature/recvWindow long. Creates and validates a new order but does not send it into the matching engine.
         https://binance-docs.github.io/apidocs/websocket_api/en/#test-new-order-trade

binance/base_client.py

@@ -405,6 +405,8 @@ async def _ws_api_request(self, method: str, signed: bool, params: dict):
             payload["params"] = self._sign_ws_params(
                 params, self._generate_ws_api_signature
             )
+        if method == "userDataStream.subscribe":
+            del payload["params"]
         return await self.ws_api.request(id, payload)
 
     def _ws_api_request_sync(self, method: str, signed: bool, params: dict):

binance/client.py

@@ -10898,6 +10898,55 @@ def __del__(self):
     ############################################################
     # WebSocket API methods
     ############################################################
+    def ws_logon(self, **params):
+        """Authenticate WebSocket connection using the provided API key.
+        https://developers.binance.com/docs/binance-spot-api-docs/websocket-api/authentication-requests
+
+        :returns: WS response
+        .. code-block:: python
+            {
+                "apiKey": "vmPUZE6mv9SD5VNHk4HlWFsOr6aKE2zvsw0MuIgwCIPy6utIco14y7Ju91duEh8A",
+                "authorizedSince": 1649729878532,
+                "connectedSince": 1649729873021,
+                "returnRateLimits": false,
+                "serverTime": 1649729878630,
+                "userDataStream": false
+            }
+        """
+        return self._ws_api_request_sync("session.logon", True, params)
+
+    def ws_user_data_stream_subscribe(self, **params):
+        """Subscribe to the User Data Stream in the current WebSocket connection.
+        https://developers.binance.com/docs/binance-spot-api-docs/websocket-api/user-data-stream-requests#subscribe-to-user-data-stream-user_stream
+
+        Note:
+        - This method requires an authenticated WebSocket connection using Ed25519 keys. Please use ws_logon first.
+        - To check the subscription status, use ws_session_status, see the userDataStream flag.
+        - User Data Stream events are available in both JSON and SBE sessions.
+
+        :returns: WS response
+        .. code-block:: python
+            {
+                "id": "d3df8a21-98ea-4fe0-8f4e-0fcea5d418b7",
+                "status": 200,
+                "result": {}
+            }
+        """
+        return self._ws_api_request_sync("userDataStream.subscribe", True, params)
+
+    def ws_user_data_stream_unsubscribe(self, **params):
+        """Stop listening to the User Data Stream in the current WebSocket connection.
+        https://developers.binance.com/docs/binance-spot-api-docs/websocket-api/user-data-stream-requests#unsubscribe-from-user-data-stream-user_stream
+
+        :returns: WS response
+        .. code-block:: python
+            {
+                "id": "d3df8a21-98ea-4fe0-8f4e-0fcea5d418b7",
+                "status": 200,
+                "result": {}
+            }
+        """
+        return self._ws_api_request_sync("userDataStream.unsubscribe", False, params)
 
     def ws_create_test_order(self, **params):
         """Test new order creation and signature/recvWindow long. Creates and validates a new order but does not send it into the matching engine.

binance/ws/streams.py

@@ -1,4 +1,5 @@
 import asyncio
+import threading
 import time
 from enum import Enum
 from typing import Optional, List, Dict, Callable, Any
@@ -13,7 +14,9 @@
 from binance.enums import FuturesType
 from binance.enums import ContractType
 from binance.helpers import get_loop
+import logging
 
+logger = logging.getLogger(__name__)
 
 class BinanceSocketType(str, Enum):
     SPOT = "Spot"
@@ -907,6 +910,21 @@
         stream_url = self.STREAM_URL
         if self.testnet:
             stream_url = self.STREAM_TESTNET_URL
+        client = self._client
+        if client.PRIVATE_KEY and not client._is_rsa:
+            async def setup_ws():
+                try:
+                    res = await client.ws_logon()
+                    logger.debug("Logged on: %s", res)
+                    res = await client.ws_user_data_stream_subscribe()
+                    logger.debug("Subscribed to user data stream: %s", res)
+                except Exception as e:
+                    logger.error("Error setting up user data stream: %s", e)
+            self._client.loop.call_soon_threadsafe(
+                asyncio.create_task, setup_ws()
+            )
+            # TODO: Fix to wait for setup_ws to complete
+            return client.ws_api
         return self._get_account_socket("user", stream_url=stream_url)
 
     def futures_user_socket(self):
@@ -1211,6 +1229,7 @@
         https_proxy: Optional[str] = None,
         loop: Optional[asyncio.AbstractEventLoop] = None,
         max_queue_size: int = 100,
+        private_key: Optional[str] = None,
     ):
         super().__init__(
             api_key,
@@ -1221,6 +1240,7 @@
             session_params,
             https_proxy,
             loop,
+            private_key,
         )
         self._bsm: Optional[BinanceSocketManager] = None
         self._max_queue_size = max_queue_size

binance/ws/threaded_stream.py

@@ -18,6 +18,7 @@ def __init__(
         session_params: Optional[Dict[str, Any]] = None,
         https_proxy: Optional[str] = None,
         _loop: Optional[asyncio.AbstractEventLoop] = None,
+        private_key: Optional[str] = None,
     ):
         """Initialise the BinanceSocketManager"""
         super().__init__()
@@ -34,6 +35,7 @@ def __init__(
             "testnet": testnet,
             "session_params": session_params,
             "https_proxy": https_proxy,
+            "private_key": private_key,
         }
 
     async def _before_socket_listener_start(self): ...

binance/ws/websocket_api.py

@@ -40,7 +40,7 @@ def _handle_message(self, msg):
         elif exception is not None:
             raise exception
         else:
-            self._log.warning(f"WS api receieved unknown message: {parsed_msg}")
+            return parsed_msg
 
     async def _ensure_ws_connection(self) -> None:
         """Ensure WebSocket connection is established and ready

tests/conftest.py

@@ -21,8 +21,10 @@
 futures_api_key = os.getenv("TEST_FUTURES_API_KEY")
 futures_api_secret = os.getenv("TEST_FUTURES_API_SECRET")
 testnet = os.getenv("TEST_TESTNET", "true").lower() == "true"
-api_key = "u4L8MG2DbshTfTzkx2Xm7NfsHHigvafxeC29HrExEmah1P8JhxXkoOu6KntLICUc"
-api_secret = "hBZEqhZUUS6YZkk7AIckjJ3iLjrgEFr5CRtFPp5gjzkrHKKC9DAv4OH25PlT6yq5"
+# api_key = "u4L8MG2DbshTfTzkx2Xm7NfsHHigvafxeC29HrExEmah1P8JhxXkoOu6KntLICUc"
+# api_secret = "hBZEqhZUUS6YZkk7AIckjJ3iLjrgEFr5CRtFPp5gjzkrHKKC9DAv4OH25PlT6yq5"
+api_key = "c5CTLgB3lqajkoN7OD28RG5sLOsVLeK4o0Ca9kT64mV6F8mKMwFeAB7uBJFRL8pG"
+api_secret = "-----BEGIN PRIVATE KEY-----\nMC4CAQAwBQYDK2VwBCIEIMwY1k3A4awroEHcfCdrn/Azan+ZgJqGhlNep3/Bud3e\n-----END PRIVATE KEY-----" 
 testnet = True
 futures_api_key = "227719da8d8499e8d3461587d19f259c0b39c2b462a77c9b748a6119abd74401"
 futures_api_secret = "b14b935f9cfacc5dec829008733c40da0588051f29a44625c34967b45c11d73c"
@@ -46,7 +48,7 @@ def setup_logging():
 
 @pytest.fixture(scope="function")
 def client():
-    return Client(api_key, api_secret, {"proxies": proxies}, testnet=testnet)
+    return Client(api_key, api_secret, {"proxies": proxies}, testnet=testnet, private_key=api_secret)
 
 
 @pytest.fixture(scope="function")
@@ -63,7 +65,7 @@ def futuresClient():
 
 @pytest.fixture(scope="function")
 def clientAsync():
-    return AsyncClient(api_key, api_secret, https_proxy=proxy, testnet=testnet)
+    return AsyncClient(api_key, api_secret, https_proxy=proxy, testnet=testnet, private_key=api_secret)
 
 
 @pytest.fixture(scope="function")

tests/test_async_client_ws_api.py

@@ -1,76 +1,65 @@
+import sys
 import pytest
 
+pytestmark = [pytest.mark.asyncio(), pytest.mark.skipif(sys.version_info < (3, 8), reason="websockets_proxy Python 3.8+")]
 
-@pytest.mark.asyncio()
 async def test_ws_get_order_book(clientAsync):
     await clientAsync.ws_get_order_book(symbol="BTCUSDT")
 
-
-@pytest.mark.asyncio()
 async def test_ws_get_recent_trades(clientAsync):
     await clientAsync.ws_get_recent_trades(symbol="BTCUSDT")
 
-
-@pytest.mark.asyncio()
 async def test_ws_get_historical_trades(clientAsync):
     await clientAsync.ws_get_historical_trades(symbol="BTCUSDT")
 
-
-@pytest.mark.asyncio()
 async def test_ws_get_aggregate_trades(clientAsync):
     await clientAsync.ws_get_aggregate_trades(symbol="BTCUSDT")
 
-
-@pytest.mark.asyncio()
 async def test_ws_get_klines(clientAsync):
     await clientAsync.ws_get_klines(symbol="BTCUSDT", interval="1m")
 
-
-@pytest.mark.asyncio()
 async def test_ws_get_uiKlines(clientAsync):
     await clientAsync.ws_get_uiKlines(symbol="BTCUSDT", interval="1m")
 
-
-@pytest.mark.asyncio()
 async def test_ws_get_avg_price(clientAsync):
     await clientAsync.ws_get_avg_price(symbol="BTCUSDT")
 
-
-@pytest.mark.asyncio()
 async def test_ws_get_ticker(clientAsync):
     await clientAsync.ws_get_ticker(symbol="BTCUSDT")
 
-
-@pytest.mark.asyncio()
 async def test_ws_get_trading_day_ticker(clientAsync):
     await clientAsync.ws_get_trading_day_ticker(symbol="BTCUSDT")
 
-
-@pytest.mark.asyncio()
 async def test_ws_get_symbol_ticker_window(clientAsync):
     await clientAsync.ws_get_symbol_ticker_window(symbol="BTCUSDT")
 
-
-@pytest.mark.asyncio()
 async def test_ws_get_symbol_ticker(clientAsync):
     await clientAsync.ws_get_symbol_ticker(symbol="BTCUSDT")
 
-
-@pytest.mark.asyncio()
 async def test_ws_get_orderbook_ticker(clientAsync):
     await clientAsync.ws_get_orderbook_ticker(symbol="BTCUSDT")
 
-
-@pytest.mark.asyncio()
 async def test_ws_ping(clientAsync):
     await clientAsync.ws_ping()
 
-
-@pytest.mark.asyncio()
 async def test_ws_get_time(clientAsync):
     await clientAsync.ws_get_time()
 
-
-@pytest.mark.asyncio()
 async def test_ws_get_exchange_info(clientAsync):
     await clientAsync.ws_get_exchange_info(symbol="BTCUSDT")
+
+async def test_ws_logon(clientAsync):
+    res = await clientAsync.ws_logon()
+    apiKey = res.get("apiKey")
+    assert apiKey
+
+async def test_ws_user_data_stream_subscribe(clientAsync):
+    """Test subscribing to user data stream"""
+    await clientAsync.ws_logon()
+    await clientAsync.ws_user_data_stream_subscribe()
+
+async def test_ws_user_data_stream_unsubscribe(clientAsync):
+    """Test unsubscribing from user data stream"""
+    await clientAsync.ws_logon()
+    await clientAsync.ws_user_data_stream_subscribe()
+    await clientAsync.ws_user_data_stream_unsubscribe()

tests/test_client_ws_api.py

@@ -64,6 +64,22 @@ def test_ws_get_exchange_info(client):
     client.ws_get_exchange_info(symbol="BTCUSDT")
 
 
+def test_ws_logon(client):
+    res = client.ws_logon()
+    apiKey = res.get("apiKey")
+    assert apiKey
+
+def test_ws_user_data_stream_subscribe(client):
+    """Test subscribing to user data stream"""
+    client.ws_logon()
+    client.ws_user_data_stream_subscribe()
+
+def test_ws_user_data_stream_unsubscribe(client):
+    """Test unsubscribing from user data stream"""
+    client.ws_logon()
+    client.ws_user_data_stream_subscribe()
+    client.ws_user_data_stream_unsubscribe()
+
 def test_ws_time_microseconds():
     micro_client = Client(
         api_key,