条件单列表改为中控后台并行缓存并以 SSE 推浏览器，下发仍走中控到实例。

Co-authored-by: Cursor <cursoragent@cursor.com>

.env.example

@@ -11,6 +11,8 @@ HUB_PULL_TIMEOUT_SEC=5
 HUB_SSE_HEARTBEAT_SEC=25
 # 数据统计后台刷新间隔(秒);不设则回落 HUB_PULL_INTERVAL_SEC
 HUB_STATS_POLL_INTERVAL_SEC=15
+# 条件单列表后台并行拉取间隔(秒)
+HUB_COND_POLL_INTERVAL_SEC=10
 
 # 中控企业微信(离线/故障告警,与客户端通知分开)
 HUB_WECHAT_ENABLED=false

app.py

@@ -43,13 +43,13 @@ from lib.bridge_client_lib import (
     fetch_backup_file,
     fetch_backup_list,
     fetch_client_version,
-    fetch_cond_orders,
     post_client_update,
     post_close_full,
     post_cond_cancel,
     post_cond_exit,
     pull_all_enabled,
 )
+from lib.cond_orders_cache_lib import cond_orders_store
 from lib.cash_sample_lib import record_cash_samples_from_overview
 from lib.db_lib import db_path as hub_db_path, init_db
 from lib.env_lib import apply_env_updates, env_get
@@ -560,8 +560,13 @@ def start_stats_cache() -> None:
     stats_store.start()
 
 
+def start_cond_orders_cache() -> None:
+    cond_orders_store.start()
+
+
 start_overview_cache()
 start_stats_cache()
+start_cond_orders_cache()
 
 try:
     start_hub_backup_scheduler(
@@ -690,49 +695,66 @@ def cond_orders_page():
 @app.route("/api/cond_orders/candidates")
 @login_required
 def api_cond_order_candidates():
-    """可选客户端:在线且无持仓、未被条件单锁定."""
-    rows = pull_all_enabled()
-    out = []
-    for row in rows:
-        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 = ""
-        out.append(
+    """可选客户端:读热缓存(总览快照推导),不现场拉实例."""
+    force = (request.args.get("force") or "").strip().lower() in ("1", "true", "yes")
+    if force:
+        cond_orders_store.request_refresh()
+        overview_store.request_refresh()
+    snap = cond_orders_store.snapshot_dict()
+    return jsonify(
         {
-                "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 "",
+            "ok": True,
+            "instances": snap.get("candidates") or [],
+            "cond_version": snap.get("cond_version"),
+            "updated_at": snap.get("updated_at"),
+            "refreshing": snap.get("refreshing"),
+            "from_cache": True,
         }
     )
-    return jsonify({"ok": True, "instances": out})
 
 
 @app.route("/api/cond_orders", methods=["GET"])
 @login_required
 def api_cond_orders_list():
-    instances = list_instances(enabled_only=True)
-    all_orders = []
-    for inst in instances:
-        data = fetch_cond_orders(inst)
-        for o in data.get("orders") or []:
-            if not isinstance(o, dict):
-                continue
-            item = dict(o)
-            item["instance_id"] = inst.get("id")
-            item["instance_name"] = inst.get("name")
-            all_orders.append(item)
-    return jsonify({"ok": True, "orders": all_orders})
+    """条件单列表只读热缓存;后台常驻并行从实例拉取."""
+    force = (request.args.get("force") or "").strip().lower() in ("1", "true", "yes")
+    if force:
+        cond_orders_store.request_refresh()
+    snap = cond_orders_store.snapshot_dict()
+    return jsonify(
+        {
+            "ok": True,
+            "orders": snap.get("orders") or [],
+            "cond_version": snap.get("cond_version"),
+            "updated_at": snap.get("updated_at"),
+            "refreshing": snap.get("refreshing"),
+            "from_cache": True,
+            "poll_interval_sec": snap.get("poll_interval_sec"),
+            "error": snap.get("error"),
+        }
+    )
+
+
+@app.route("/api/cond_orders/refresh", methods=["POST"])
+@login_required
+def api_cond_orders_refresh():
+    cond_orders_store.request_refresh()
+    overview_store.request_refresh()
+    return jsonify({"ok": True, "msg": "已触发条件单后台刷新"})
+
+
+@app.route("/api/cond_orders/stream")
+@login_required
+def api_cond_orders_stream():
+    return Response(
+        stream_with_context(cond_orders_store.iter_sse()),
+        mimetype="text/event-stream",
+        headers={
+            "Cache-Control": "no-cache",
+            "Connection": "keep-alive",
+            "X-Accel-Buffering": "no",
+        },
+    )
 
 
 @app.route("/api/cond_orders", methods=["POST"])
@@ -749,8 +771,9 @@ def api_cond_orders_create():
     if not instance_ids:
         return jsonify({"ok": False, "msg": "请选择客户端"})
 
-    # 再次按快照过滤有仓/锁定
-    rows = {r.get("id"): r for r in pull_all_enabled()}
+    # 再次按热缓存快照过滤有仓/锁定(下发本身仍 POST 到实例)
+    ov = overview_store.snapshot_dict()
+    rows = {r.get("id"): r for r in (ov.get("instances") or [])}
     allowed = []
     rejected = []
     for iid in instance_ids:
@@ -792,6 +815,9 @@ def api_cond_orders_create():
     }
     results = dispatch_cond_order_to_ids(allowed, payload)
     ok_n = sum(1 for r in results if r.get("ok"))
+    if ok_n > 0:
+        cond_orders_store.request_refresh()
+        overview_store.request_refresh()
     return jsonify(
         {
             "ok": ok_n > 0,
@@ -813,7 +839,11 @@ def api_cond_orders_cancel(hub_order_id: str):
     inst = get_instance(int(iid))
     if not inst:
         return jsonify({"ok": False, "msg": "实例不存在"}), 404
-    return jsonify(post_cond_cancel(inst, hub_order_id))
+    result = post_cond_cancel(inst, hub_order_id)
+    if isinstance(result, dict) and result.get("ok"):
+        cond_orders_store.request_refresh()
+        overview_store.request_refresh()
+    return jsonify(result)
 
 
 @app.route("/api/cond_orders/<hub_order_id>/exit", methods=["POST"])
@@ -831,7 +861,11 @@ def api_cond_orders_exit(hub_order_id: str):
         "profit_exit_enabled": data.get("profit_exit_enabled"),
         "profit_exit_mult": data.get("profit_exit_mult"),
     }
-    return jsonify(post_cond_exit(inst, hub_order_id, payload))
+    result = post_cond_exit(inst, hub_order_id, payload)
+    if isinstance(result, dict) and result.get("ok"):
+        cond_orders_store.request_refresh()
+        overview_store.request_refresh()
+    return jsonify(result)
 
 
 @app.route("/manifest.webmanifest")

lib/bridge_client_lib.py

@@ -215,6 +215,40 @@ def fetch_cond_orders(inst: dict[str, Any]) -> dict[str, Any]:
         return {"ok": False, "msg": str(e) or "拉取失败", "orders": [], "instance_id": inst.get("id")}
 
 
+def pull_all_cond_orders() -> list[dict[str, Any]]:
+    """并行拉取各机条件单并展平为带 instance 字段的订单列表."""
+    instances = list_instances(enabled_only=True)
+    if not instances:
+        return []
+    by_id: dict[Any, dict[str, Any]] = {}
+    with ThreadPoolExecutor(max_workers=min(8, len(instances))) as pool:
+        futs = {pool.submit(fetch_cond_orders, inst): inst for inst in instances}
+        for fut in as_completed(futs):
+            inst = futs[fut]
+            try:
+                data = fut.result()
+            except Exception as e:
+                data = {
+                    "ok": False,
+                    "msg": str(e) or "拉取失败",
+                    "orders": [],
+                    "instance_id": inst.get("id"),
+                    "instance_name": inst.get("name"),
+                }
+            by_id[inst.get("id")] = data if isinstance(data, dict) else {}
+    all_orders: list[dict[str, Any]] = []
+    for inst in instances:
+        data = by_id.get(inst["id"]) or {}
+        for o in data.get("orders") or []:
+            if not isinstance(o, dict):
+                continue
+            item = dict(o)
+            item["instance_id"] = inst.get("id")
+            item["instance_name"] = inst.get("name")
+            all_orders.append(item)
+    return all_orders
+
+
 def pull_all_enabled() -> list[dict[str, Any]]:
     instances = list_instances(enabled_only=True)
     if not instances:

lib/cond_orders_cache_lib.py

@@ -0,0 +1,232 @@
+"""条件单热缓存:后台常驻并行从实例拉取;浏览器只读缓存 + 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 _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 = ""
+        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 "",
+            }
+        )
+    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()

static/cond_orders.js

@@ -117,7 +117,7 @@
         });
         const dd = await rr.json();
         alert(dd.msg || (dd.ok ? "已取消" : "失败"));
-        refresh();
+        requestCondRefresh();
       });
     });
     box.querySelectorAll("[data-exit]").forEach(function (btn) {
@@ -148,7 +148,7 @@
         });
         const dd = await rr.json();
         alert(dd.msg || (dd.ok ? "已更新" : "失败"));
-        refresh();
+        requestCondRefresh();
       });
     });
   }
@@ -414,13 +414,25 @@
     fillOrderBox(document.getElementById("cond-list-done"), lastDone, "暂无已结束的条件单", "done");
   }
 
-  async function loadOrders() {
+  let condVersion = 0;
+  let condLoading = false;
+  let condEventSource = null;
+  let condReconnectTimer = null;
+
+  function setCondStatus(msg) {
+    const el = document.getElementById("cond-status");
+    if (el) el.textContent = msg || "";
+  }
+
+  async function loadOrders(opts) {
+    opts = opts || {};
     const activeBox = document.getElementById("cond-list-active");
     const doneBox = document.getElementById("cond-list-done");
     if (!activeBox && !doneBox) return;
-    const r = await fetch("/api/cond_orders");
+    const r = await fetch("/api/cond_orders", { credentials: "same-origin" });
     const d = await r.json();
     const list = d.orders || [];
+    condVersion = Number(d.cond_version) || condVersion;
     lastActive = [];
     lastDone = [];
     list.forEach(function (o) {
@@ -428,6 +440,19 @@
       else lastDone.push(o);
     });
     paintCachedOrders();
+    if (!opts.silent) {
+      const live = condEventSource && condEventSource.readyState === 1;
+      const sec = Number(d.poll_interval_sec) || 10;
+      setCondStatus(
+        (live ? "SSE 已连接 · " : "") +
+          "进行中 " +
+          lastActive.length +
+          " · 已结束 " +
+          lastDone.length +
+          (d.updated_at ? " · " + d.updated_at : "") +
+          (d.refreshing ? " · 后台拉取中" : " · 后台每 " + sec + "s 并行拉")
+      );
+    }
   }
 
   async function createOrder() {
@@ -467,16 +492,72 @@
       const tog = document.getElementById("cond-add-toggle");
       if (tog) tog.checked = false;
     }
-    refresh();
+    requestCondRefresh();
   }
 
-  async function refresh() {
-    await Promise.all([loadCandidates(), loadOrders()]);
+  async function refresh(opts) {
+    opts = opts || {};
+    if (condLoading) return;
+    condLoading = true;
+    if (!opts.silent) setCondStatus("读取缓存…");
+    try {
+      await Promise.all([loadCandidates(), loadOrders({ silent: !!opts.silent })]);
+    } finally {
+      condLoading = false;
+    }
+  }
+
+  async function requestCondRefresh() {
+    try {
+      await fetch("/api/cond_orders/refresh", {
+        method: "POST",
+        credentials: "same-origin",
+      });
+    } catch (_) {}
+    refresh({ silent: true });
+  }
+
+  function closeCondStream() {
+    if (condEventSource) {
+      condEventSource.close();
+      condEventSource = null;
+    }
+    if (condReconnectTimer) {
+      clearTimeout(condReconnectTimer);
+      condReconnectTimer = null;
+    }
+  }
+
+  function connectCondStream() {
+    if (typeof EventSource === "undefined") return;
+    closeCondStream();
+    condEventSource = new EventSource("/api/cond_orders/stream");
+    condEventSource.addEventListener("cond", function (ev) {
+      try {
+        const st = JSON.parse(ev.data || "{}");
+        const ver = Number(st.cond_version) || 0;
+        if (ver && ver !== condVersion) {
+          refresh({ silent: true });
+        } else if (st.refreshing) {
+          setCondStatus("后台并行拉取条件单…");
+        }
+      } catch (_) {}
+    });
+    condEventSource.onerror = function () {
+      closeCondStream();
+      setCondStatus("SSE 断开,8s 后重连…");
+      condReconnectTimer = setTimeout(function () {
+        connectCondStream();
+        refresh({ silent: true });
+      }, 8000);
+    };
   }
 
   function boot() {
     const btn = document.getElementById("btn-cond-refresh");
-    if (btn) btn.addEventListener("click", refresh);
+    if (btn) btn.addEventListener("click", function () {
+      requestCondRefresh();
+    });
     const create = document.getElementById("btn-cond-create");
     if (create) create.addEventListener("click", createOrder);
     const typeEl = document.getElementById("cf-opt-type");
@@ -484,9 +565,14 @@
     if (typeEl) typeEl.addEventListener("change", syncTriggerHint);
     if (pxEl) pxEl.addEventListener("input", syncTriggerHint);
     syncTriggerHint();
+    // 读缓存秒开 + SSE;不再前端定时现场拉实例
     refresh();
-    setInterval(refresh, 10000);
+    connectCondStream();
   }
 
-  global.HubCondOrders = { boot: boot, refresh: refresh };
+  global.HubCondOrders = {
+    boot: boot,
+    refresh: refresh,
+    requestRefresh: requestCondRefresh,
+  };
 })(window);

templates/cond_orders.html

@@ -4,7 +4,8 @@
 <section class="page-head cond-head">
   <div>
     <h1>条件单</h1>
-    <p class="muted cond-lead">中控只下发；盯盘、买币、开平仓都在客户端。有仓或已占用的机器不可选。</p>
+    <p class="muted cond-lead">中控下发到实例；列表由中控后台缓存并推送到浏览器。盯盘与开平仓在客户端。有仓或已占用的机器不可选。</p>
+    <p class="muted" id="cond-status">读取缓存…</p>
   </div>
   <div class="cond-head-actions">
     <button type="button" class="btn" id="btn-cond-refresh">刷新</button>
@@ -93,6 +94,6 @@
 </div>
 {% endblock %}
 {% block scripts %}
-<script src="{{ url_for('static', filename='cond_orders.js') }}?v=6"></script>
+<script src="{{ url_for('static', filename='cond_orders.js') }}?v=7"></script>
 <script>HubCondOrders.boot();</script>
 {% endblock %}