Add OKX WebSocket reconnect on stale order book

  okx_order_book: track last_update_time on snapshot and update messages.
  liquidation_bid_bot: check staleness at top of each loop iteration and
  reconnect if no update received in 60s.

Author: nimcon

keeper_bots/liquidation_bid_bot.py

@@ -599,6 +599,27 @@ async def run_liquidation_bid_bot():
 
     log.info("Running liquidation bid bot in %s environment", trade_env.name)
 
+    async def reconnect_order_book() -> bool:
+        """Reconnect the OKX order book WebSocket. Returns True on success."""
+        log.warning("OKX order book stale — reconnecting WebSocket")
+        okx_order_book.initialized = False
+        okx_order_book.book = {}
+        try:
+            await okx_order_book.connect()
+            await okx_order_book.subscribe()
+        except Exception as err:
+            log.error("OKX WebSocket reconnect failed: %s", err)
+            return False
+        waited = 0.0
+        while not okx_order_book.initialized and waited < 30:
+            await asyncio.sleep(0.5)
+            waited += 0.5
+        if okx_order_book.initialized:
+            log.info("OKX order book reconnected  mid=%.4f", okx_order_book.mid_price())
+            return True
+        log.error("OKX order book did not receive snapshot after reconnect")
+        return False
+
     stablecoin_symbol = "BYC"  # Circuit stablecoin (BYC)
 
     # BYC proxy asset to be used for hedging
@@ -653,27 +674,38 @@ async def run_liquidation_bid_bot():
 
     try:
         while True:
+            if time.monotonic() - okx_order_book.last_update_time > 60:
+                if not await reconnect_order_book():
+                    await asyncio.sleep(CONTINUE_DELAY)
+                    continue
+
             await rpc_client.set_fee_per_cost()
 
             if time.monotonic() - last_okx_heartbeat >= HEARTBEAT_INTERVAL:
                 try:
                     await accountAPI.get_account_balance()
                     log.info("OKX API heartbeat: balance query successful")
                 except Exception as err:
-                    log.warning("OKX API heartbeat failed. %s: %s", type(err).__name__, err)
+                    log.warning(
+                        "OKX API heartbeat failed. %s: %s", type(err).__name__, err
+                    )
                 last_okx_heartbeat = time.monotonic()
 
             try:
                 state = await rpc_client.upkeep_state(vaults=True)
             except Exception as err:
-                log.error("Failed to get state of vaults. %s: %s", type(err).__name__, err)
+                log.error(
+                    "Failed to get state of vaults. %s: %s", type(err).__name__, err
+                )
                 await asyncio.sleep(CONTINUE_DELAY)
                 continue
 
             vaults_in_liquidation = state.get("vaults_in_liquidation", [])
 
             if not vaults_in_liquidation:
-                log.info("No vaults in liquidation. Sleeping for %s seconds", RUN_INTERVAL)
+                log.info(
+                    "No vaults in liquidation. Sleeping for %s seconds", RUN_INTERVAL
+                )
                 await asyncio.sleep(RUN_INTERVAL)
                 continue
 
@@ -683,7 +715,9 @@ async def run_liquidation_bid_bot():
             try:
                 balances = await rpc_client.wallet_balances()
             except Exception as err:
-                log.error("Failed to get wallet balances. %s: %s", type(err).__name__, err)
+                log.error(
+                    "Failed to get wallet balances. %s: %s", type(err).__name__, err
+                )
 
             available_xch_amount = balances.get("xch", None)
             available_byc_amount = balances.get("byc", None)
@@ -758,7 +792,9 @@ async def run_liquidation_bid_bot():
                             statutes["implemented_statutes"]["VAULT_MINIMUM_DEBT"]
                         )
                         liquidation_ratio_pct = int(
-                            statutes["implemented_statutes"]["VAULT_LIQUIDATION_RATIO_PCT"]
+                            statutes["implemented_statutes"][
+                                "VAULT_LIQUIDATION_RATIO_PCT"
+                            ]
                         )
                     except Exception:
                         log.error(
@@ -808,7 +844,9 @@ async def run_liquidation_bid_bot():
                         response = await rpc_client.vault_deposit(deposit_amount)
                     except Exception as err:
                         log.error(
-                            "Failed to deposit to vault. %s: %s", type(err).__name__, err
+                            "Failed to deposit to vault. %s: %s",
+                            type(err).__name__,
+                            err,
                         )
                     else:
                         if response.get("status") != "success":
@@ -818,12 +856,16 @@ async def run_liquidation_bid_bot():
                                 response,
                             )
                         else:
-                            log.info("Deposited %.12f XCH", deposit_amount / MOJOS_PER_XCH)
+                            log.info(
+                                "Deposited %.12f XCH", deposit_amount / MOJOS_PER_XCH
+                            )
                             try:
                                 response = await rpc_client.vault_borrow(borrow_amount)
                             except Exception as err:
                                 log.error(
-                                    "Failed to borrow BYC. %s: %s", type(err).__name__, err
+                                    "Failed to borrow BYC. %s: %s",
+                                    type(err).__name__,
+                                    err,
                                 )
                             if response.get("status") != "success":
                                 log.error(

keeper_bots/okx_order_book.py

@@ -1,5 +1,6 @@
 import json
 import logging
+import time
 from datetime import datetime
 from pprint import pprint
 import asyncio
@@ -62,6 +63,7 @@ def __init__(self, sym, uquote, url, verbose=False, logger=None):
         self.book = {}
         self.url = url
         self.initialized = False
+        self.last_update_time: float = 0.0
         self._lock = asyncio.Lock()
         self.logger = logger or logging.getLogger(__name__)
 
@@ -162,6 +164,7 @@ def __call__(self, message):
                         self.book["bids"][price_float] = depth[1]
 
                 self.initialized = True
+                self.last_update_time = time.monotonic()
 
             elif message["action"] == "update":
                 if self.verbose:
@@ -185,7 +188,6 @@ def __call__(self, message):
                             else:
                                 # Remove price level if volume is 0
                                 self.book[side].pop(price_float, None)
-
                     ## Validate order book depth
                     # asks_depth = len(self.book.get("asks", {}))
                     # bids_depth = len(self.book.get("bids", {}))
@@ -198,7 +200,7 @@ def __call__(self, message):
                     #     self.logger.warning(
                     #         f"OKX order book bid depth is {bids_depth}, expected 400"
                     #     )
-
+                self.last_update_time = time.monotonic()
             else:
                 raise Exception(
                     f"Unknown action {message['action']} returned in callback"