From 5ffcee3f5695b5a195bf6df4055d2a38682e3e22 Mon Sep 17 00:00:00 2001 From: dekun Date: Wed, 7 Oct 2026 14:50:52 +0800 Subject: [PATCH] =?UTF-8?q?=E6=9D=A1=E4=BB=B6=E5=8D=95=E5=88=97=E8=A1=A8?= =?UTF-8?q?=E6=94=B9=E4=B8=BA=E4=B8=AD=E6=8E=A7=E5=90=8E=E5=8F=B0=E5=B9=B6?= =?UTF-8?q?=E8=A1=8C=E7=BC=93=E5=AD=98=E5=B9=B6=E4=BB=A5=20SSE=20=E6=8E=A8?= =?UTF-8?q?=E6=B5=8F=E8=A7=88=E5=99=A8=EF=BC=8C=E4=B8=8B=E5=8F=91=E4=BB=8D?= =?UTF-8?q?=E8=B5=B0=E4=B8=AD=E6=8E=A7=E5=88=B0=E5=AE=9E=E4=BE=8B=E3=80=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Cursor --- .env.example | 2 + app.py | 120 +++++++++++------- lib/bridge_client_lib.py | 34 +++++ lib/cond_orders_cache_lib.py | 232 +++++++++++++++++++++++++++++++++++ static/cond_orders.js | 106 ++++++++++++++-- templates/cond_orders.html | 5 +- 6 files changed, 444 insertions(+), 55 deletions(-) create mode 100644 lib/cond_orders_cache_lib.py diff --git a/.env.example b/.env.example index fe14951..e9c8a3b 100644 --- a/.env.example +++ b/.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 diff --git a/app.py b/app.py index 7d46f1d..98c043b 100644 --- a/app.py +++ b/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( - { - "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 jsonify({"ok": True, "instances": out}) + """可选客户端:读热缓存(总览快照推导),不现场拉实例.""" + 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( + { + "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, + } + ) @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//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") diff --git a/lib/bridge_client_lib.py b/lib/bridge_client_lib.py index 6bdb645..d4f70a5 100644 --- a/lib/bridge_client_lib.py +++ b/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: diff --git a/lib/cond_orders_cache_lib.py b/lib/cond_orders_cache_lib.py new file mode 100644 index 0000000..8b52a7e --- /dev/null +++ b/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() diff --git a/static/cond_orders.js b/static/cond_orders.js index 183bedf..a5ec452 100644 --- a/static/cond_orders.js +++ b/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); diff --git a/templates/cond_orders.html b/templates/cond_orders.html index 9398a30..02eb052 100644 --- a/templates/cond_orders.html +++ b/templates/cond_orders.html @@ -4,7 +4,8 @@

条件单

-

中控只下发；盯盘、买币、开平仓都在客户端。有仓或已占用的机器不可选。

+

中控下发到实例；列表由中控后台缓存并推送到浏览器。盯盘与开平仓在客户端。有仓或已占用的机器不可选。

+

读取缓存…

@@ -93,6 +94,6 @@
{% endblock %} {% block scripts %} - + {% endblock %}