2.1.16

CHANGELOG.md

@@ -1,6 +1,10 @@
-## 2.1.16 2024-09-xx
+## 2.1.16 2024-09-13
 ### Fix
-* OKX: get_exchange_info: <class 'decimal.ConversionSyntax'> [#82](https://github.com/DogsTailFarmer/martin-binance/issues/82#issue-2467548368)
+* `OKX`: get_exchange_info: <class 'decimal.ConversionSyntax'> [#82](https://github.com/DogsTailFarmer/martin-binance/issues/82#issue-2467548368)
+
+### Update
+* `HTX`: monitoring PING interval from server for each WSS channel and restart if timing out
+* Dependency
 
 ## 2.1.15 2024-07-12
 ### Added for new features

exchanges_wrapper/__init__.py

@@ -12,7 +12,7 @@
 __contact__ = "https://github.com/DogsTailFarmer"
 __email__ = "[email protected]"
 __credits__ = ["https://github.com/DanyaSWorlD"]
-__version__ = "2.1.16b0"
+__version__ = "2.1.16"
 
 from pathlib import Path
 import shutil

exchanges_wrapper/web_sockets.py

@@ -17,10 +17,7 @@
 from crypto_ws_api.ws_session import generate_signature
 from exchanges_wrapper import LOG_PATH
 
-logger = logging.getLogger('exch_srv_logger')
-
-logger_ws = logging.getLogger(__name__)
-logger_ws.level = logging.INFO
+logger = logging.getLogger(__name__)
 formatter = logging.Formatter(fmt="[%(asctime)s: %(levelname)s] %(message)s")
 #
 fh = logging.handlers.RotatingFileHandler(Path(LOG_PATH, 'websockets.log'), maxBytes=1000000, backupCount=10)
@@ -31,8 +28,9 @@
 sh.setFormatter(formatter)
 sh.setLevel(logging.INFO)
 
-logger_ws.addHandler(fh)
-logger_ws.addHandler(sh)
+logger.addHandler(fh)
+logger.addHandler(sh)
+logger.propagate = False
 
 sys.tracebacklimit = 0
 
@@ -50,12 +48,13 @@ def __init__(self, client, endpoint, exchange, trade_id):
         self._price = None
         self.tasks = set()
         self.wss_started = False
+        self.ping = 0
 
     async def start(self):
         ping_interval = None if self.exchange == 'huobi' else 20
         async for self.websocket in websockets.connect(
                 self.endpoint,
-                logger=logger_ws,
+                logger=logger,
                 ping_interval=ping_interval
         ):
             try:
@@ -66,7 +65,7 @@ async def start(self):
                     logger.info(f"WSS closed for {self.exchange}:{self.trade_id}")
                     break
                 else:
-                    logger.warning(f"Restart WSS for {self.exchange}")
+                    logger.warning(f"Restart WSS for {self.exchange}: {ex}")
                     continue
             except Exception as ex:
                 self.tasks_cancel()
@@ -187,9 +186,11 @@ async def _handle_messages(self, msg, symbol=None, ch_type=str()):
                 logger.debug(f"Bitfinex undefined WSS: symbol: {symbol}, ch_type: {ch_type}, msg_data: {msg_data}")
         elif self.exchange == 'huobi':
             if ping := msg_data.get('ping'):
+                self.ping = 0
                 await self.websocket.send(json.dumps({"pong": ping}))
                 await asyncio.sleep(0)
             elif msg_data.get('action') == 'ping':
+                self.ping = 0
                 pong = {
                     "action": "pong",
                     "data": {
@@ -229,6 +230,17 @@ async def bybit_heartbeat(self, req_id, interval=20):
             await asyncio.sleep(interval)
             await self.websocket.send(json.dumps({"req_id": req_id, "op": "ping"}))
 
+    async def htx_keepalive(self, interval=60):
+        await asyncio.sleep(interval * 10)
+        while True:
+            await asyncio.sleep(interval)
+            if self.ping:
+                break
+            else:
+                self.ping = 1
+        logger.warning("From HTX server PING timeout exceeded")
+        await self.websocket.close()
+
 
 class MarketEventsDataStream(EventsDataStream):
     def __init__(self, client, endpoint, exchange, trade_id, channel=None):
@@ -300,6 +312,8 @@ async def start_wss(self):
                 elif ch_type == 'depth5':
                     request = {'sub': f"market.{symbol}.depth.step0"}
 
+                self.tasks_manage(self.htx_keepalive(interval=30))
+
         await self.ws_listener(request, symbol, ch_type)
 
     async def _handle_event(self, content, symbol=None, ch_type=str()):
@@ -404,6 +418,7 @@ async def start_wss(self):
             "action": "sub",
             "ch": f"trade.clearing#{self.symbol.lower()}#0"
         }
+        self.tasks_manage(self.htx_keepalive())
         await self.ws_listener(request, symbol=self.symbol)
 
     async def _handle_event(self, msg_data, *args):

requirements.txt

@@ -1,10 +1,10 @@
 crypto-ws-api==2.0.12
 pyotp==2.9.0
-simplejson==3.19.2
+simplejson==3.19.3
 toml~=0.10.2
-aiohttp~=3.9.5
-websockets==12.0
+aiohttp~=3.10.5
+websockets==13.0.1
 expiringdict~=1.2.2
 ujson~=5.10.0
-betterproto==2.0.0b6
+betterproto==2.0.0b7
 grpclib~=0.4.7