fix: prevent trading lease starvation and unblock order dispatch

Author: tianhe

backend_api_python/app/routes/strategy.py

@@ -228,10 +228,15 @@ def stop_strategy(strategy_id: int):
         strategy_id,
         close_positions=close_positions,
     )
-    get_strategy_service().update_strategy_status(strategy_id, "stopped", user_id=int(g.user_id))
+    status = str(result.get("status") or "")
+    if status in {"stopping", "stopped"}:
+        get_strategy_service().update_strategy_status(strategy_id, "stopped", user_id=int(g.user_id))
     data = {"id": strategy_id, **result}
     if not result.get("success"):
-        return _error("strategyV2.stopClosePartialFailure", 409, data=data)
+        message = "strategyV2.stopClosePartialFailure" if close_positions and status == "stopped" else "strategyV2.stopFailed"
+        return _error(message, 409, data=data)
+    if status == "stopping":
+        return _ok(data, "strategyV2.stopQueued"), 202
     message = "strategyV2.stoppedAndCloseQueued" if close_positions else "strategyV2.paused"
     return _ok(data, message)
 

backend_api_python/app/services/execution_streams/supervisor.py

@@ -378,22 +378,26 @@ def _symbols_by_credential() -> Dict[int, Set[str]]:
             cur = db.cursor()
             cur.execute(
                 """
-                SELECT COALESCE(NULLIF(exchange_config->>'credential_id', ''),
-                                NULLIF(exchange_config->>'credentials_id', ''), '0')::INTEGER AS credential_id,
-                       symbol
+                SELECT exchange_config::text AS exchange_config, 0 AS credential_id, symbol
                 FROM qd_strategies_trading
                 WHERE COALESCE(symbol, '') <> ''
                   AND LOWER(COALESCE(status, '')) = 'running'
                   AND LOWER(COALESCE(execution_mode, 'signal')) = 'live'
                 UNION
-                SELECT credential_id, symbol
+                SELECT NULL AS exchange_config, credential_id, symbol
                 FROM pending_orders
                 WHERE credential_id > 0 AND COALESCE(symbol, '') <> ''
                   AND status IN ('pending','processing','sent','syncing')
                 """
             )
             for row in cur.fetchall() or []:
-                credential_id = int(row.get("credential_id") or 0)
+                try:
+                    config = json.loads(row.get("exchange_config") or "{}")
+                    if not isinstance(config, dict):
+                        config = {}
+                    credential_id = int(row.get("credential_id") or config.get("credential_id") or config.get("credentials_id") or 0)
+                except (TypeError, ValueError):
+                    continue
                 if credential_id > 0:
                     out.setdefault(credential_id, set()).add(str(row.get("symbol") or ""))
             cur.close()

backend_api_python/app/services/grid/engine.py

@@ -50,6 +50,7 @@ def __init__(
         self.cfg = GridBotConfig.from_trading_config(self.trading_config)
         self._create_client = create_client_fn
         self._enqueue_market = enqueue_market
+        self.order_guard = None
         self._orders = GridRestingOrderRepository()
         self._cells = GridCellRepository()
         self._bootstrapped = False
@@ -636,6 +637,8 @@ def _sync_initial_market_leg(
             logger.debug("grid initial pre-place probe sid=%s: %s", self.strategy_id, e)
         try:
             client = self._create_client()
+            if self.order_guard and not self.order_guard():
+                return False
             execution = execute_grid_market_order(
                 client,
                 symbol=self.symbol,
@@ -1474,6 +1477,8 @@ def _place_limit(
                 not reduce_only
                 and self.cfg.order_mode in ("maker", "limit", "limit_first", "maker_then_market")
             )
+            if self.order_guard and not self.order_guard():
+                return False
             res = place_grid_limit_order(
                 client,
                 symbol=self.symbol,
@@ -1816,6 +1821,8 @@ def cancel_exit_orders_on_exchange(self, *, pos_side: str = "") -> None:
 
     def cancel_all_orders_on_exchange(self) -> None:
         open_orders = self._orders.list_open(self.strategy_id)
+        if not open_orders:
+            return
         try:
             client = self._create_client()
         except Exception as e:

backend_api_python/app/services/grid/exchange_orders.py

@@ -651,6 +651,11 @@ def cancel_grid_order(
 ) -> None:
     mt = str(market_type or "swap").strip().lower()
     ex_cfg = exchange_config if isinstance(exchange_config, dict) else {}
+    if isinstance(client, (GateSpotClient, GateUsdtFuturesClient)):
+        if not exchange_order_id:
+            raise LiveTradingError("Gate cancellation requires a confirmed exchange order ID")
+        client.cancel_order(order_id=str(exchange_order_id))
+        return
     if isinstance(client, OkxClient):
         client.cancel_order(
             market_type=mt,

backend_api_python/app/services/grid/runner.py

@@ -44,7 +44,7 @@ def shutdown_grid_for_strategy(strategy_id: int) -> None:
         gr.shutdown()
         return
     try:
-        from app.services.exchange_execution import load_strategy_configs, resolve_exchange_config
+        from app.services.exchange_execution import coalesce_exchange_config_from_payload, load_strategy_configs, resolve_exchange_config
         from app.services.live_trading.factory import create_client
 
         sc = load_strategy_configs(sid) or {}
@@ -58,8 +58,8 @@ def shutdown_grid_for_strategy(strategy_id: int) -> None:
         if not symbol:
             return
         user_id = int(sc.get("user_id") or 1)
-        ex_cfg = resolve_exchange_config(sc.get("exchange_config") or {}, user_id=user_id)
-        mt = str(tc.get("market_type") or "swap").strip().lower()
+        ex_cfg = resolve_exchange_config(coalesce_exchange_config_from_payload(sc), user_id=user_id)
+        mt = str(tc.get("market_type") or sc.get("market_type") or "swap").strip().lower()
 
         def _create_client():
             return create_client(ex_cfg, market_type=mt)
@@ -74,7 +74,7 @@ def _create_client():
             enqueue_market=lambda *a, **k: False,
         )
         engine.shutdown()
-        append_strategy_log(sid, "info", "Grid orders cancelled on strategy stop (no active runner)")
+        append_strategy_log(sid, "info", "strategyRuntime.gridStopCleanup")
     except Exception as e:
         logger.warning("shutdown_grid_for_strategy sid=%s: %s", sid, e)
         append_strategy_log(sid, "warning", f"Grid stop cancel failed: {e}")

backend_api_python/app/services/pending_order_loops.py

@@ -0,0 +1,49 @@
+"""Independent order dispatch and exchange reconciliation loops."""
+from app.utils.logger import get_logger
+
+logger = get_logger(__name__)
+
+
+class PendingOrderLoops:
+    def _run_loop(self) -> None:
+        while not self._stop_event.is_set():
+            try:
+                self._tick()
+            except Exception as e:
+                logger.warning(f"PendingOrderWorker tick error: {e}")
+            self._stop_event.wait(self.poll_interval_sec)
+
+    def _run_sync_loop(self) -> None:
+        while not self._stop_event.is_set():
+            for sync in (self._sync_quick_trade_orders, self._sync_alpaca_sent_orders, self._sync_live_sent_orders, self._maybe_sync_positions):
+                if self._stop_event.is_set() or (self.lease_guard and not self.lease_guard()):
+                    break
+                try:
+                    sync()
+                except Exception:
+                    logger.warning("Order reconciliation failed: %s", sync.__name__, exc_info=True)
+            self._stop_event.wait(self.poll_interval_sec)
+
+    def _tick(self) -> None:
+        if self.lease_guard and not self.lease_guard():
+            return
+        orders = self._fetch_pending_orders(limit=self.batch_size)
+        if not orders:
+            return
+
+        for o in orders:
+            if self._stop_event.is_set() or (self.lease_guard and not self.lease_guard()):
+                break
+            oid = o.get("id")
+            if not oid:
+                continue
+
+            # Mark processing (best-effort)
+            if not self._mark_processing(order_id=int(oid)):
+                continue
+
+            try:
+                self._dispatch_one(o)
+            except Exception as e:
+                self._mark_failed(order_id=int(oid), error=str(e))
+

backend_api_python/app/services/pending_order_worker.py

@@ -115,6 +115,7 @@
 from app.services.live_trading.gate import GateSpotClient, GateUsdtFuturesClient
 from app.services.live_trading.htx import HtxClient
 from app.utils.db import get_db_connection
+from app.services.pending_order_loops import PendingOrderLoops
 from app.utils.logger import get_logger
 from app.utils.strategy_runtime_logs import append_strategy_log
 from app.services.strategy_lifecycle import (
@@ -135,12 +136,14 @@
 ALPACA_FILL_DELTA_EPSILON = 1e-8
 
 
-class PendingOrderWorker(PendingOrderPositionSyncMixin):
+class PendingOrderWorker(PendingOrderLoops, PendingOrderPositionSyncMixin):
     def __init__(self, poll_interval_sec: float = 1.0, batch_size: int = 50):
         self.poll_interval_sec = float(poll_interval_sec)
         self.batch_size = int(batch_size)
         self._stop_event = threading.Event()
         self._thread: Optional[threading.Thread] = None
+        self._sync_thread: Optional[threading.Thread] = None
+        self.lease_guard = None
         self._lock = threading.Lock()
         self._notifier = SignalNotifier()
         self._instrument_rules = get_instrument_rules_provider()
@@ -177,15 +180,19 @@ def request_exchange_catchup(
 
     def start(self) -> bool:
         with self._lock:
+            if self._thread and self._thread.is_alive() and self._sync_thread and self._sync_thread.is_alive():
+                return True
             try:
                 ensure_position_ledger_schema()
             except Exception as e:
                 logger.warning("ensure_position_ledger_schema failed: %s", e)
-            if self._thread and self._thread.is_alive():
-                return True
             self._stop_event.clear()
-            self._thread = threading.Thread(target=self._run_loop, name="PendingOrderWorker", daemon=True)
-            self._thread.start()
+            if not self._thread or not self._thread.is_alive():
+                self._thread = threading.Thread(target=self._run_loop, name="PendingOrderWorker", daemon=True)
+                self._thread.start()
+            if not self._sync_thread or not self._sync_thread.is_alive():
+                self._sync_thread = threading.Thread(target=self._run_sync_loop, name="PendingOrderReconciliation", daemon=True)
+                self._sync_thread.start()
             logger.info("PendingOrderWorker started")
             return True
 
@@ -195,43 +202,10 @@ def stop(self, timeout_sec: float = 5.0) -> None:
             th = self._thread
         if th and th.is_alive():
             th.join(timeout=timeout_sec)
+        if self._sync_thread and self._sync_thread.is_alive():
+            self._sync_thread.join(timeout=timeout_sec)
         logger.info("PendingOrderWorker stopped")
 
-    def _run_loop(self) -> None:
-        while not self._stop_event.is_set():
-            try:
-                self._tick()
-            except Exception as e:
-                logger.warning(f"PendingOrderWorker tick error: {e}")
-            time.sleep(self.poll_interval_sec)
-
-    def _tick(self) -> None:
-        # logger.info(f"[PendingOrderWorker] _tick start. last_sync={self._last_position_sync_ts}")
-        self._sync_quick_trade_orders()
-        self._sync_alpaca_sent_orders()
-        self._sync_live_sent_orders()
-        orders = self._fetch_pending_orders(limit=self.batch_size)
-        # logger.info(f"[PendingOrderWorker] orders fetched: {len(orders)}")
-        if not orders:
-            self._maybe_sync_positions()
-            return
-
-        for o in orders:
-            oid = o.get("id")
-            if not oid:
-                continue
-
-            # Mark processing (best-effort)
-            if not self._mark_processing(order_id=int(oid)):
-                continue
-
-            try:
-                self._dispatch_one(o)
-            except Exception as e:
-                self._mark_failed(order_id=int(oid), error=str(e))
-
-        self._maybe_sync_positions()
-
     def _sync_quick_trade_orders(self, limit: int = 50) -> None:
         """Reconcile non-terminal Quick Trade orders and protect new fills."""
         try:

backend_api_python/app/services/strategy_command_client.py

@@ -75,32 +75,29 @@ def stop_strategy_with_policy(
                 payload={"close_positions": bool(close_positions)},
             )
             self._remember(strategy_id, "stop", command.id)
+            queued = {
+                "success": True,
+                "status": "stopping",
+                "command_id": command.id,
+                "close_requested": bool(command.payload.get("close_positions")),
+                "close_orders_queued": 0,
+                "close_errors": [],
+            }
             timeout = max(0.0, float(os.getenv("STRATEGY_COMMAND_STOP_WAIT_SEC", "5")))
             if timeout == 0:
-                return {
-                    "success": True,
-                    "status": "stopping",
-                    "close_requested": bool(close_positions),
-                    "close_orders_queued": 0,
-                    "close_errors": [],
-                }
+                return queued
             finished = self._wait(command.id, timeout)
             if finished is None or finished.status not in TERMINAL_COMMAND_STATUSES:
                 logger.warning(
                     "Strategy stop remains queued: strategy=%s command=%s",
                     strategy_id,
                     command.id,
                 )
-                return {
-                    "success": False,
-                    "status": "stopping",
-                    "close_requested": bool(close_positions),
-                    "close_orders_queued": 0,
-                    "close_errors": [],
-                }
+                return queued
             if finished.status == "succeeded":
                 result = dict(finished.result or {})
                 result.setdefault("success", True)
+                result.setdefault("command_id", command.id)
                 result.setdefault("status", "stopped")
                 result.setdefault("close_requested", bool(close_positions))
                 result.setdefault("close_orders_queued", 0)
@@ -109,6 +106,7 @@ def stop_strategy_with_policy(
             return {
                 "success": False,
                 "status": "running",
+                "command_id": command.id,
                 "close_requested": bool(close_positions),
                 "close_orders_queued": 0,
                 "close_errors": [finished.error_message] if finished.error_message else [],

backend_api_python/app/services/trading_executor.py

@@ -62,6 +62,7 @@ def __init__(self) -> None:
         self.order_gateway = StrategyV2OrderGateway()
         self._last_start_failure = ""
         self._last_exit_reason: dict[int, str] = {}
+        self.runtime_guard = None
 
     def start_strategy(self, strategy_id: int) -> bool:
         strategy_id = int(strategy_id)
@@ -80,6 +81,9 @@ def start_strategy(self, strategy_id: int) -> bool:
                 self._last_start_failure = str(exc or "strategyV2.livePreflightFailed")
                 logger.warning("Strategy %s live preflight rejected: %s", strategy_id, exc)
                 return False
+            if self.runtime_guard and not self.runtime_guard(strategy_id):
+                self._last_start_failure = "strategyRuntime.leaseLost"
+                return False
             thread = threading.Thread(
                 target=self._run_strategy_loop,
                 args=(strategy_id,),
@@ -1202,6 +1206,8 @@ def _execute_signal(self, **values: Any) -> bool:
             and inflight_check(request)
         ):
             return False
+        if self.runtime_guard and not self.runtime_guard(strategy_id):
+            return False
         pending_id = self.order_gateway.submit(request)
         if pending_id:
             order_details = [f"pending_id={pending_id}"]
@@ -1265,6 +1271,8 @@ def create_grid_client():
             return client
 
         def enqueue_market(signal_type: str, quantity: float, price: float, reason: str) -> bool:
+            if self.runtime_guard and not self.runtime_guard(strategy_id):
+                return False
             return self._execute_signal(
                 strategy_id=strategy_id,
                 strategy_name=strategy_name,
@@ -1328,6 +1336,7 @@ def evaluate_grid_risk(price: float) -> List[Dict[str, Any]]:
             create_client_fn=create_grid_client,
             risk_exit_fn=evaluate_grid_risk,
         )
+        runner.engine.order_guard = lambda: self.runtime_guard is None or self.runtime_guard(strategy_id)
         ok, message = runner.startup(initial_price, bars_df=frame)
         if not ok:
             raise RuntimeError(f"grid.startupFailed:{message}")
@@ -1984,6 +1993,9 @@ def _load_source(strategy: dict[str, Any]) -> tuple[int, str]:
         return source_id, code
 
     def _is_strategy_running(self, strategy_id: int, thread: threading.Thread) -> bool:
+        if self.runtime_guard and not self.runtime_guard(strategy_id):
+            self._last_exit_reason[strategy_id] = "strategyRuntime.leaseLost"
+            return False
         with self.lock:
             if self.running_strategies.get(strategy_id) is not thread:
                 return False