fix: stabilize private execution streams

Author: Henry

backend_api_python/app/services/execution_streams/adapters.py

@@ -87,6 +87,7 @@ def __init__(
         self._connected = False
         self._reconnects = 0
         self._last_error = ""
+        self._last_message_at = 0.0
 
     @property
     def stream_key(self) -> str:
@@ -150,6 +151,34 @@ def handle_control(self, ws: Any, payload: Dict[str, Any]) -> bool:
             return True
         return False
 
+    def application_heartbeat_interval(self) -> float:
+        """Return the venue-level heartbeat interval, or 0 when not required."""
+        return 0.0
+
+    def application_heartbeat_payload(self) -> Any:
+        """Return the venue-level heartbeat payload sent after an idle interval."""
+        return None
+
+    def _application_heartbeat_loop(self, ws: Any, connection_stop: threading.Event) -> None:
+        interval = max(0.0, float(self.application_heartbeat_interval() or 0.0))
+        if interval <= 0:
+            return
+        while not self._stop.is_set() and not connection_stop.wait(interval):
+            if time.monotonic() - self._last_message_at < interval:
+                continue
+            payload = self.application_heartbeat_payload()
+            if payload is None:
+                return
+            try:
+                if isinstance(payload, str):
+                    ws.send(payload)
+                else:
+                    ws.send(json.dumps(payload, separators=(",", ":")))
+            except Exception:
+                # run_forever owns reconnect/error reporting; a failed heartbeat
+                # only means this connection is already on its way down.
+                return
+
     def prepare(self) -> None:
         return None
 
@@ -161,20 +190,33 @@ def _run(self) -> None:
             return
         backoff = 1.0
         while not self._stop.is_set():
+            heartbeat_stop = threading.Event()
+            heartbeat_thread: Optional[threading.Thread] = None
             try:
                 self.prepare()
                 url = self.url()
 
                 def _open(ws):
+                    nonlocal heartbeat_thread
+                    self._last_message_at = time.monotonic()
                     if self.ready_on_open():
                         self.mark_ready()
                     else:
                         self._connected = False
                         self.on_state("authenticating", "", self._reconnects > 0)
                     for message in self.on_open_messages():
                         ws.send(json.dumps(message, separators=(",", ":")))
+                    if self.application_heartbeat_interval() > 0:
+                        heartbeat_thread = threading.Thread(
+                            target=self._application_heartbeat_loop,
+                            args=(ws, heartbeat_stop),
+                            name=f"ExecHeartbeat-{self.stream_key}",
+                            daemon=True,
+                        )
+                        heartbeat_thread.start()
 
                 def _message(ws, raw):
+                    self._last_message_at = time.monotonic()
                     payload = _json_message(raw)
                     if not payload or self.handle_control(ws, payload):
                         return
@@ -210,12 +252,17 @@ def _close(_ws, _code, reason):
                     sslopt = {"cert_reqs": ssl.CERT_NONE, "check_hostname": False}
                 elif isinstance(verify, str):
                     sslopt = {"ca_certs": verify}
-                self._ws.run_forever(
-                    ping_interval=20,
-                    ping_timeout=10,
-                    sslopt=sslopt,
-                    skip_utf8_validation=True,
-                )
+                try:
+                    self._ws.run_forever(
+                        ping_interval=20,
+                        ping_timeout=10,
+                        sslopt=sslopt,
+                        skip_utf8_validation=True,
+                    )
+                finally:
+                    heartbeat_stop.set()
+                    if heartbeat_thread and heartbeat_thread.is_alive():
+                        heartbeat_thread.join(timeout=1.0)
             except Exception as exc:
                 self._connected = False
                 self.on_state("error", str(exc), False)
@@ -233,6 +280,36 @@ def __init__(self, **kwargs: Any) -> None:
         self._listen_key_base = ""
         self._listen_key_path = ""
         self._keepalive_thread: Optional[threading.Thread] = None
+        self._subscription_request_id = ""
+
+    def _uses_spot_ws_api(self) -> bool:
+        return self.market_type == "spot" and exchange_trading_environment(
+            self.config, "binance"
+        ) != "live"
+
+    def ready_on_open(self) -> bool:
+        return not self._uses_spot_ws_api()
+
+    def on_open_messages(self) -> List[Dict[str, Any]]:
+        if not self._uses_spot_ws_api():
+            return []
+        timestamp = int(time.time() * 1000)
+        api_key = str(self.config.get("api_key") or self.config.get("apiKey") or "")
+        secret = str(self.config.get("secret_key") or self.config.get("secret") or "")
+        params: Dict[str, Any] = {
+            "apiKey": api_key,
+            "timestamp": timestamp,
+        }
+        signature_payload = urlencode(sorted(params.items()))
+        params["signature"] = hmac.new(
+            secret.encode(), signature_payload.encode(), hashlib.sha256
+        ).hexdigest()
+        self._subscription_request_id = f"qd-user-stream-{self.credential_id}-{timestamp}"
+        return [{
+            "id": self._subscription_request_id,
+            "method": "userDataStream.subscribe.signature",
+            "params": params,
+        }]
 
     def stop(self, timeout: float = 5.0) -> bool:
         main_stopped = super().stop(timeout=timeout)
@@ -242,6 +319,9 @@ def stop(self, timeout: float = 5.0) -> bool:
         return main_stopped and not bool(keepalive and keepalive.is_alive())
 
     def prepare(self) -> None:
+        if self._uses_spot_ws_api():
+            self._listen_key = ""
+            return
         api_key = str(self.config.get("api_key") or self.config.get("apiKey") or "")
         environment = exchange_trading_environment(self.config, "binance")
         if self.market_type == "spot":
@@ -291,20 +371,59 @@ def _keepalive_loop(self) -> None:
 
     def url(self) -> str:
         environment = exchange_trading_environment(self.config, "binance")
+        if self._uses_spot_ws_api():
+            if environment == "testnet":
+                return "wss://ws-api.testnet.binance.vision/ws-api/v3"
+            return "wss://demo-ws-api.binance.com/ws-api/v3"
         if self.market_type == "spot":
             host = "wss://demo-stream.binance.com/ws" if environment != "live" else "wss://stream.binance.com:9443/ws"
         else:
             host = "wss://fstream.binancefuture.com/ws" if environment != "live" else "wss://fstream.binance.com/ws"
         return f"{host}/{self._listen_key}"
 
+    def handle_control(self, ws: Any, payload: Dict[str, Any]) -> bool:
+        if self._uses_spot_ws_api() and payload.get("id") == self._subscription_request_id:
+            try:
+                status = int(payload.get("status") or 0)
+            except (TypeError, ValueError):
+                status = 0
+            result = payload.get("result") if isinstance(payload.get("result"), dict) else {}
+            if status == 200 and result.get("subscriptionId") is not None:
+                self.mark_ready()
+            else:
+                error = payload.get("error")
+                message = error.get("msg") if isinstance(error, dict) else error
+                self.on_state(
+                    "error",
+                    f"Binance user stream subscription failed: {message or payload}",
+                    False,
+                )
+                try:
+                    ws.close()
+                except Exception:
+                    pass
+            return True
+        return super().handle_control(ws, payload)
+
     def parse(self, payload: Dict[str, Any]) -> List[ExecutionEvent]:
-        return parse_binance(payload, market_type=self.market_type)
+        event = payload.get("event")
+        return parse_binance(
+            event if isinstance(event, dict) else payload,
+            market_type=self.market_type,
+        )
 
 
 class OkxExecutionAdapter(PrivateWebSocketAdapter):
     def ready_on_open(self) -> bool:
         return False
 
+    def application_heartbeat_interval(self) -> float:
+        # OKX requires the literal text frame "ping" before 30 seconds idle.
+        return 20.0
+
+    def application_heartbeat_payload(self) -> str:
+        return "ping"
+
     def url(self) -> str:
         if exchange_trading_environment(self.config, "okx") != "live":
             return "wss://wspap.okx.com:8443/ws/v5/private"
@@ -344,6 +463,12 @@ class BybitExecutionAdapter(PrivateWebSocketAdapter):
     def ready_on_open(self) -> bool:
         return False
 
+    def application_heartbeat_interval(self) -> float:
+        return 20.0
+
+    def application_heartbeat_payload(self) -> Dict[str, str]:
+        return {"op": "ping"}
+
     def url(self) -> str:
         environment = exchange_trading_environment(self.config, "bybit")
         if environment == "demo":
@@ -366,6 +491,8 @@ def on_open_messages(self) -> List[Dict[str, Any]]:
         }]
 
     def handle_control(self, ws: Any, payload: Dict[str, Any]) -> bool:
+        if payload.get("op") == "ping" and str(payload.get("ret_msg") or "").lower() == "pong":
+            return True
         if payload.get("op") == "auth":
             if bool(payload.get("success")):
                 self.mark_ready()
@@ -383,7 +510,17 @@ class BitgetExecutionAdapter(PrivateWebSocketAdapter):
     def ready_on_open(self) -> bool:
         return False
 
+    def application_heartbeat_interval(self) -> float:
+        # Bitget recommends a literal text "ping" every 30 seconds.  Use a
+        # small safety margin so scheduling jitter cannot cross the deadline.
+        return 25.0
+
+    def application_heartbeat_payload(self) -> str:
+        return "ping"
+
     def url(self) -> str:
+        if exchange_trading_environment(self.config, "bitget") != "live":
+            return "wss://wspap.bitget.com/v2/ws/private"
         return "wss://ws.bitget.com/v2/ws/private"
 
     def on_open_messages(self) -> List[Dict[str, Any]]:

backend_api_python/app/services/execution_streams/repository.py

@@ -376,6 +376,7 @@ def update_health(
                     rest_fallback = EXCLUDED.rest_fallback,
                     last_error = EXCLUDED.last_error,
                     updated_at = NOW()
+                RETURNING stream_key
                 """,
                 (
                     str(stream_key),

backend_api_python/tests/test_db_log_noise_fixes.py

@@ -1,6 +1,7 @@
 from contextlib import contextmanager
 
 from app.services import mfa_service
+from app.services.execution_streams.repository import ExecutionEventRepository
 from app.services.strategy_command_repository import StrategyCommandRepository
 from app.utils import strategy_runtime_logs
 
@@ -104,3 +105,26 @@ def test_worker_heartbeat_returns_text_primary_key(monkeypatch):
     assert "RETURNING id" not in sql
     assert params[0:2] == ("worker-1", "trading")
     assert conn.committed
+
+
+def test_execution_stream_health_returns_stream_key(monkeypatch):
+    conn = _CaptureConn()
+    monkeypatch.setattr(
+        "app.services.execution_streams.repository.get_db_connection",
+        lambda: _capture_connection(conn),
+    )
+
+    ExecutionEventRepository().update_health(
+        stream_key="okx:9:all",
+        credential_id=9,
+        exchange_id="okx",
+        market_type="all",
+        state="connected",
+    )
+
+    sql, params = conn.cursor_obj.calls[0]
+    assert "RETURNING stream_key" in sql
+    assert "RETURNING id" not in sql
+    assert params[0] == "okx:9:all"
+    assert conn.committed
+    assert conn.cursor_obj.closed