"""条件单热缓存:后台常驻并行从实例拉取;浏览器只读缓存 + SSE. 下发/撤单/改出场仍由中控 POST 到实例,成功后触发刷新. """ from __future__ import annotations import json import os import queue import threading import time from collections.abc import Iterator from typing import Any def _poll_interval_sec() -> float: try: return max( 3.0, float( os.getenv("HUB_COND_POLL_INTERVAL_SEC") or os.getenv("HUB_PULL_INTERVAL_SEC") or 10 ), ) except (TypeError, ValueError): return 10.0 def _sse_heartbeat_sec() -> float: try: return max(10.0, float(os.getenv("HUB_SSE_HEARTBEAT_SEC") or 25)) except (TypeError, ValueError): return 25.0 def _safe_float(v: Any) -> float | None: if v is None or v == "": return None try: return float(v) except (TypeError, ValueError): return None def _candidates_from_snaps(rows: list[dict[str, Any]]) -> list[dict[str, Any]]: out: list[dict[str, Any]] = [] for row in rows or []: snap = row.get("snapshot") if isinstance(row.get("snapshot"), dict) else {} brief = snap.get("positions_brief") or snap.get("positions") or [] has_pos = isinstance(brief, list) and len(brief) > 0 locked = bool(snap.get("cond_order_locked")) selectable = bool(row.get("online")) and not has_pos and not locked inst = snap.get("instance") if isinstance(snap.get("instance"), dict) else {} mode = str(inst.get("mode") or "").strip().lower() if mode not in ("sim", "live"): mode = "" funds = snap.get("funds") if isinstance(snap.get("funds"), dict) else {} # 开仓吃交易户;优先 avail,回退总额 trading_usdt = _safe_float(funds.get("trading_usdt_avail")) if trading_usdt is None: trading_usdt = _safe_float(funds.get("trading_usdt")) if trading_usdt is None: trading_usdt = _safe_float(snap.get("trading_usdt")) funding_usdt = _safe_float(funds.get("funding_usdt")) if funding_usdt is None: funding_usdt = _safe_float(snap.get("funding_usdt")) if trading_usdt is not None: trading_usdt = round(float(trading_usdt), 4) if funding_usdt is not None: funding_usdt = round(float(funding_usdt), 4) out.append( { "id": row.get("id"), "name": row.get("name"), "online": bool(row.get("online")), "has_position": has_pos, "cond_order_locked": locked, "selectable": selectable, "mode": mode, "error": row.get("error") or "", "trading_usdt": trading_usdt, "funding_usdt": funding_usdt, } ) return out class CondOrdersStore: def __init__(self) -> None: self._lock = threading.RLock() self.version = 0 self.orders: list[dict[str, Any]] = [] self.candidates: list[dict[str, Any]] = [] self.refreshing = True self.last_error: str | None = None self.updated_at: str = "" self._ready = False self._subscribers: list[queue.Queue[str | None]] = [] self._stop = threading.Event() self._refresh = threading.Event() self._thread: threading.Thread | None = None def start(self) -> None: if self._thread and self._thread.is_alive(): return self._stop.clear() with self._lock: self.refreshing = True self._ready = False self._thread = threading.Thread( target=self._loop, daemon=True, name="hub-cond-orders-pull", ) self._thread.start() def stop(self) -> None: self._stop.set() self._refresh.set() self._broadcast(close=True) def request_refresh(self) -> None: self._refresh.set() def snapshot_dict(self) -> dict[str, Any]: with self._lock: return { "ok": True, "orders": list(self.orders), "candidates": list(self.candidates), "cond_version": self.version, "refreshing": self.refreshing, "updated_at": self.updated_at, "error": self.last_error, "from_cache": True, "poll_interval_sec": _poll_interval_sec(), "ready": self._ready, } def event_dict(self) -> dict[str, Any]: with self._lock: return { "cond_version": self.version, "updated_at": self.updated_at, "refreshing": self.refreshing, "ok": True, "error": self.last_error, "poll_interval_sec": _poll_interval_sec(), } def _overview_rows(self) -> list[dict[str, Any]]: try: from lib.overview_cache_lib import overview_store snap = overview_store.snapshot_dict() rows = snap.get("instances") if isinstance(snap, dict) else None if isinstance(rows, list) and rows: return rows except Exception: pass try: from lib.bridge_client_lib import pull_all_enabled return pull_all_enabled() except Exception: return [] def _loop(self) -> None: while not self._stop.is_set(): self._refresh_once() if self._stop.is_set(): break self._refresh.clear() self._refresh.wait(timeout=_poll_interval_sec()) def _refresh_once(self) -> None: with self._lock: self.refreshing = True self._broadcast() try: from lib.bridge_client_lib import pull_all_cond_orders rows = self._overview_rows() candidates = _candidates_from_snaps(rows) orders = pull_all_cond_orders() with self._lock: self.version += 1 self.orders = orders if isinstance(orders, list) else [] self.candidates = candidates self.updated_at = time.strftime("%Y-%m-%d %H:%M:%S") self.last_error = None self.refreshing = False self._ready = True except Exception as e: with self._lock: self.version += 1 self.last_error = str(e) or "cond_refresh_failed" self.refreshing = False self._ready = True self._broadcast() def _broadcast(self, *, close: bool = False) -> None: with self._lock: subs = list(self._subscribers) event = None if close else json.dumps(self.event_dict(), ensure_ascii=False) dead: list[queue.Queue[str | None]] = [] for q in subs: try: q.put_nowait(None if close else event) except queue.Full: try: q.get_nowait() except queue.Empty: pass try: q.put_nowait(event) except queue.Full: dead.append(q) except Exception: dead.append(q) if dead: with self._lock: for q in dead: if q in self._subscribers: self._subscribers.remove(q) def iter_sse(self) -> Iterator[str]: q: queue.Queue[str | None] = queue.Queue(maxsize=32) with self._lock: self._subscribers.append(q) try: yield _sse_frame(self.event_dict()) while True: try: raw = q.get(timeout=_sse_heartbeat_sec()) except queue.Empty: yield ": heartbeat\n\n" continue if raw is None: break try: data = json.loads(raw) except Exception: data = self.event_dict() yield _sse_frame(data) finally: with self._lock: if q in self._subscribers: self._subscribers.remove(q) def _sse_frame(data: dict[str, Any]) -> str: body = json.dumps(data, ensure_ascii=False) return f"event: cond\ndata: {body}\n\n" cond_orders_store = CondOrdersStore()