Merge branch 'master' of github.com:sammchardy/python-binance

Author: carlosmiei

.github/workflows/python-app.yml

@@ -34,7 +34,7 @@ jobs:
   build:
     needs: lint
     runs-on: ubuntu-22.04
-    timeout-minutes: 60
+    timeout-minutes: 40
     env:
       PROXY: "http://51.83.140.52:16301"
       TEST_TESTNET: "true"

binance/async_client.py

@@ -64,7 +64,10 @@ async def create(
         testnet: bool = False,
         loop=None,
         session_params: Optional[Dict[str, Any]] = None,
+        private_key: Optional[Union[str, Path]] = None,
+        private_key_pass: Optional[str] = None,
         https_proxy: Optional[str] = None,
+        time_unit: Optional[str] = None,
     ):
         self = cls(
             api_key,
@@ -75,6 +78,10 @@ async def create(
             testnet,
             loop,
             session_params,
+            private_key,
+            private_key_pass,
+            https_proxy,
+            time_unit
         )
         self.https_proxy = https_proxy  # move this to the constructor
 
@@ -162,7 +169,8 @@ async def _handle_response(self, response: aiohttp.ClientResponse):
         if not str(response.status).startswith("2"):
             raise BinanceAPIException(response, response.status, await response.text())
 
-        if response.text == "":
+        text = await response.text()
+        if text == "":
             return {}
 
         try:

binance/base_client.py

@@ -36,7 +36,7 @@ class BaseClient:
     OPTIONS_TESTNET_URL = "https://testnet.binanceops.{}/eapi"
     PAPI_URL = "https://papi.binance.{}/papi"
     WS_API_URL = "wss://ws-api.binance.{}/ws-api/v3"
-    WS_API_TESTNET_URL = "wss://testnet.binance.vision/ws-api/v3"
+    WS_API_TESTNET_URL = "wss://ws-api.testnet.binance.vision/ws-api/v3"
     WS_FUTURES_URL = "wss://ws-fapi.binance.{}/ws-fapi/v1"
     WS_FUTURES_TESTNET_URL = "wss://testnet.binancefuture.com/ws-fapi/v1"
     PUBLIC_API_VERSION = "v1"

binance/helpers.py

@@ -6,7 +6,7 @@
 import dateparser
 import pytz
 
-from datetime import datetime
+from datetime import datetime, timezone
 
 from binance.exceptions import UnknownDateFormat
 
@@ -21,7 +21,7 @@ def date_to_milliseconds(date_str: str) -> int:
     :param date_str: date in readable format, i.e. "January 01, 2018", "11 hours ago UTC", "now UTC"
     """
     # get epoch value in UTC
-    epoch: datetime = datetime.utcfromtimestamp(0).replace(tzinfo=pytz.utc)
+    epoch: datetime = datetime.fromtimestamp(0,timezone.utc)
     # parse our date string
     d: Optional[datetime] = dateparser.parse(date_str, settings={"TIMEZONE": "UTC"})
     if not d:

binance/ws/depthcache.py

@@ -185,15 +185,17 @@ async def __aenter__(self):
         return self
 
     async def __aexit__(self, *args, **kwargs):
+        self._log.debug(f"Exiting depth cache manager for {self._symbol}")
         await self._socket.__aexit__(*args, **kwargs)
 
     async def recv(self):
         dc = None
         while not dc:
             try:
                 res = await asyncio.wait_for(self._socket.recv(), timeout=self.TIMEOUT)
+                self._log.debug(f"Received message: {res}")
             except Exception as e:
-                self._log.warning(e)
+                self._log.warning(f"Exception recieving message: {e.__class__.__name__} (e) ")
             else:
                 dc = await self._depth_event(res)
         return dc
@@ -203,7 +205,7 @@ async def _init_cache(self):
 
         :return:
         """
-
+        self._log.debug(f"Initialising depth cache for {self._symbol}")
         # initialise or clear depth cache
         self._depth_cache = DepthCache(self._symbol, conv_type=self._conv_type)
 
@@ -228,16 +230,15 @@ async def _depth_event(self, msg):
         :return:
 
         """
+        self._log.debug(f"Received depth event: {msg}")
 
         if not msg:
             return None
 
         if "e" in msg and msg["e"] == "error":
-            # close the socket
-            await self.close()
-
-            # notify the user by returning a None value
-            return None
+            # notify user by return msg with error
+            self._log.error(f"Error in depth event restarting cache: {msg}")
+            return msg
 
         return await self._process_depth_message(msg)
 

binance/ws/reconnecting_websocket.py

@@ -47,7 +47,6 @@ class ReconnectingWebsocket:
     MIN_RECONNECT_WAIT = 0.1
     TIMEOUT = 10
     NO_MESSAGE_RECONNECT_TIMEOUT = 60
-    MAX_QUEUE_SIZE = 100
 
     def __init__(
         self,
@@ -57,6 +56,7 @@ def __init__(
         is_binary: bool = False,
         exit_coro=None,
         https_proxy: Optional[str] = None,
+        max_queue_size: int = 100,
         **kwargs,
     ):
         self._loop = get_loop()
@@ -75,6 +75,7 @@ def __init__(
         self._handle_read_loop = None
         self._https_proxy = https_proxy
         self._ws_kwargs = kwargs
+        self.max_queue_size = max_queue_size
 
     def json_dumps(self, msg) -> str:
         if orjson:
@@ -203,17 +204,22 @@ async def _read_loop(self):
                         res = self._handle_message(res)
                         self._log.debug(f"Received message: {res}")
                         if res:
-                            if self._queue.qsize() < self.MAX_QUEUE_SIZE:
+                            if self._queue.qsize() < self.max_queue_size:
                                 await self._queue.put(res)
                             else:
                                 raise BinanceWebsocketQueueOverflow(
-                                    f"Message queue size {self._queue.qsize()} exceeded maximum {self.MAX_QUEUE_SIZE}"
+                                    f"Message queue size {self._queue.qsize()} exceeded maximum {self.max_queue_size}"
                                 )
                 except asyncio.TimeoutError:
                     self._log.debug(f"no message in {self.TIMEOUT} seconds")
                     # _no_message_received_reconnect
                 except asyncio.CancelledError as e:
                     self._log.debug(f"_read_loop cancelled error {e}")
+                    await self._queue.put({
+                        "e": "error",
+                        "type": f"{e.__class__.__name__}",
+                        "m": f"{e}",
+                    })
                     break
                 except (
                     asyncio.IncompleteReadError,
@@ -234,7 +240,7 @@ async def _read_loop(self):
                     Exception,
                 ) as e:
                     # reports errors and break the loop
-                    self._log.error(f"Unknown exception ({e})")
+                    self._log.error(f"Unknown exception: {e.__class__.__name__} ({e})")
                     await self._queue.put({
                         "e": "error",
                         "type": e.__class__.__name__,

binance/ws/streams.py

@@ -25,7 +25,7 @@ class BinanceSocketType(str, Enum):
 
 class BinanceSocketManager:
     STREAM_URL = "wss://stream.binance.{}:9443/"
-    STREAM_TESTNET_URL = "wss://testnet.binance.vision/"
+    STREAM_TESTNET_URL = "wss://stream.testnet.binance.vision/"
     FSTREAM_URL = "wss://fstream.binance.{}/"
     FSTREAM_TESTNET_URL = "wss://stream.binancefuture.com/"
     DSTREAM_URL = "wss://dstream.binance.{}/"
@@ -36,12 +36,19 @@ class BinanceSocketManager:
     WEBSOCKET_DEPTH_10 = "10"
     WEBSOCKET_DEPTH_20 = "20"
 
-    def __init__(self, client: AsyncClient, user_timeout=KEEPALIVE_TIMEOUT):
+    def __init__(
+        self, 
+        client: AsyncClient, 
+        user_timeout=KEEPALIVE_TIMEOUT,
+        max_queue_size: int = 100,
+    ):
         """Initialise the BinanceSocketManager
 
         :param client: Binance API client
         :type client: binance.AsyncClient
-
+        :param user_timeout: Timeout for user socket in seconds
+        :param max_queue_size: Max size of the websocket queue, defaults to 100
+        :type max_queue_size: int
         """
         self.STREAM_URL = self.STREAM_URL.format(client.tld)
         self.FSTREAM_URL = self.FSTREAM_URL.format(client.tld)
@@ -52,8 +59,8 @@ def __init__(self, client: AsyncClient, user_timeout=KEEPALIVE_TIMEOUT):
         self._loop = get_loop()
         self._client = client
         self._user_timeout = user_timeout
-
         self.testnet = self._client.testnet
+        self._max_queue_size = max_queue_size
         self.ws_kwargs = {}
 
     def _get_stream_url(self, stream_url: Optional[str] = None):
@@ -84,6 +91,7 @@ def _get_socket(
                 exit_coro=lambda p: self._exit_socket(f"{socket_type}_{p}"),
                 is_binary=is_binary,
                 https_proxy=self._client.https_proxy,
+                max_queue_size=self._max_queue_size,
                 **self.ws_kwargs,
             )
 
@@ -1202,6 +1210,7 @@ def __init__(
         session_params: Optional[Dict[str, Any]] = None,
         https_proxy: Optional[str] = None,
         loop: Optional[asyncio.AbstractEventLoop] = None,
+        max_queue_size: int = 100,
     ):
         super().__init__(
             api_key,
@@ -1214,10 +1223,14 @@ def __init__(
             loop,
         )
         self._bsm: Optional[BinanceSocketManager] = None
+        self._max_queue_size = max_queue_size
 
     async def _before_socket_listener_start(self):
         assert self._client
-        self._bsm = BinanceSocketManager(client=self._client)
+        self._bsm = BinanceSocketManager(
+            client=self._client,
+            max_queue_size=self._max_queue_size
+        )
 
     def _start_async_socket(
         self,
@@ -1226,7 +1239,10 @@ def _start_async_socket(
         params: Dict[str, Any],
         path: Optional[str] = None,
     ) -> str:
+        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")
             time.sleep(0.1)
         socket = getattr(self._bsm, socket_name)(**params)
         socket_path: str = path or socket._path  # noqa

binance/ws/threaded_stream.py

@@ -1,4 +1,5 @@
 import asyncio
+import logging
 import threading
 from typing import Optional, Dict, Any
 
@@ -24,6 +25,7 @@ def __init__(
         self._client: Optional[AsyncClient] = None
         self._running: bool = True
         self._socket_running: Dict[str, bool] = {}
+        self._log = logging.getLogger(__name__)
         self._client_params = {
             "api_key": api_key,
             "api_secret": api_secret,
@@ -37,12 +39,17 @@ def __init__(
     async def _before_socket_listener_start(self): ...
 
     async def socket_listener(self):
-        self._client = await AsyncClient.create(loop=self._loop, **self._client_params)
-        await self._before_socket_listener_start()
+        try:
+            self._client = await AsyncClient.create(loop=self._loop, **self._client_params)
+            await self._before_socket_listener_start()
+        except Exception as e:
+            self._log.error(f"Failed to create client: {e}")
+            self.stop()
         while self._running:
             await asyncio.sleep(0.2)
         while self._socket_running:
             await asyncio.sleep(0.2)
+        self._log.info("Socket listener stopped")
 
     async def start_listener(self, socket, path: str, callback):
         async with socket as s:
@@ -52,13 +59,19 @@ async def start_listener(self, socket, path: str, callback):
                 except asyncio.TimeoutError:
                     ...
                     continue
+                except Exception as e:
+                    self._log.error(f"Error receiving message: {e}")
+                    msg = {
+                        "e": "error",
+                        "type": e.__class__.__name__,
+                        "m": f"{e}",
+                    }
+                if not msg:
+                    continue  # Handle both async and sync callbacks
+                if asyncio.iscoroutinefunction(callback):
+                    asyncio.create_task(callback(msg))
                 else:
-                    if not msg:
-                        continue  # Handle both async and sync callbacks
-                    if asyncio.iscoroutinefunction(callback):
-                        await callback(msg)
-                    else:
-                        callback(msg)
+                    callback(msg)
         del self._socket_running[path]
 
     def run(self):
@@ -74,6 +87,7 @@ async def stop_client(self):
         await self._client.close_connection()
 
     def stop(self):
+        self._log.debug("Stopping ThreadedApiManager")
         if not self._running:
             return
         self._running = False
@@ -85,6 +99,6 @@ def stop(self):
                 future.result(timeout=5)  # Add timeout to prevent hanging
             except Exception as e:
                 # Log the error but don't raise it
-                print(f"Error stopping client: {e}")
+                self._log.error(f"Error stopping client: {e}")
         for socket_name in self._socket_running.keys():
             self._socket_running[socket_name] = False

docs/depth_cache.rst

@@ -153,6 +153,31 @@ Websocket Errors
 ----------------
 
 If the underlying websocket is disconnected and is unable to reconnect None is returned for the depth_cache parameter.
+If the underlying websocket is disconnected an error msg is passed to the callback and to recv() containing the error message.
+In the case the BinanceWebsocketClosed is returned, the websocket will attempt to reconnect 5 times before returning a BinanceUnableToConnect error.
+Example:
+
+.. code:: python
+
+            depth_cache = await dcm.recv()
+            if isinstance(depth_cache, dict) and depth_cache.get('e') == 'error':
+                logger.error(f"Received depth cache error in callback: {depth_cache}")
+                if type == 'BinanceWebsocketClosed':
+                    # ignore as attempts to reconnect
+                    continue
+                break
+
+.. code:: python
+            def handle_depth_cache(depth_cache):
+                if isinstance(depth_cache, dict) and depth_cache.get('e') == 'error':
+                    logger.error(f"Received depth cache error in callback: {depth_cache}")
+                    type = depth_cache.get('type')
+                    if type == 'BinanceWebsocketClosed':
+                        # Automatically attempts to reconnect
+                        return
+                    dcm.stop()
+                    return
+                # handle non error cases here
 
 Examples
 --------