diff --git a/crypto_monitor_binance/app.py b/crypto_monitor_binance/app.py index 1091e45..3fef9fb 100644 --- a/crypto_monitor_binance/app.py +++ b/crypto_monitor_binance/app.py @@ -9541,12 +9541,19 @@ register_trade_records_api( app_tz=APP_TZ, ) +def _dashboard_enrich_orders(items): + from lib.instance.instance_dashboard_lib import enrich_order_items_with_marks + + return enrich_order_items_with_marks(items, get_price=get_price) + + from lib.instance.instance_dashboard_register import register_instance_dashboard_routes register_instance_dashboard_routes( app, login_required=login_required, get_db=get_db, + enrich_orders=_dashboard_enrich_orders, hedge_enabled=False, ) diff --git a/crypto_monitor_gate/app.py b/crypto_monitor_gate/app.py index c392a8e..a601dc0 100644 --- a/crypto_monitor_gate/app.py +++ b/crypto_monitor_gate/app.py @@ -9375,12 +9375,19 @@ register_trade_records_api( app_tz=APP_TZ, ) +def _dashboard_enrich_orders(items): + from lib.instance.instance_dashboard_lib import enrich_order_items_with_marks + + return enrich_order_items_with_marks(items, get_price=get_price) + + from lib.instance.instance_dashboard_register import register_instance_dashboard_routes register_instance_dashboard_routes( app, login_required=login_required, get_db=get_db, + enrich_orders=_dashboard_enrich_orders, hedge_enabled=False, ) diff --git a/crypto_monitor_okx/app.py b/crypto_monitor_okx/app.py index 775ef4a..5aea19a 100644 --- a/crypto_monitor_okx/app.py +++ b/crypto_monitor_okx/app.py @@ -9050,14 +9050,17 @@ def _dashboard_fetch_options_positions(): if not isinstance(cfg, dict) or not cfg.get("enabled"): return [] try: - from lib.options.options_hub_lib import build_options_hub_snapshot + from lib.options.options_dashboard_lib import fetch_light_option_positions_for_dashboard - snap = build_options_hub_snapshot(cfg) + return fetch_light_option_positions_for_dashboard(cfg) except Exception: return [] - if not snap.get("ok"): - return [] - return list(snap.get("positions") or []) + + +def _dashboard_enrich_orders(items): + from lib.instance.instance_dashboard_lib import enrich_order_items_with_marks + + return enrich_order_items_with_marks(items, get_price=get_price) from lib.instance.instance_dashboard_register import register_instance_dashboard_routes @@ -9067,6 +9070,7 @@ register_instance_dashboard_routes( login_required=login_required, get_db=get_db, fetch_options_positions=_dashboard_fetch_options_positions, + enrich_orders=_dashboard_enrich_orders, hedge_enabled=os.getenv("HEDGE_PLAN_ENABLED", "false").lower() in ("1", "true", "yes", "on"), ) diff --git a/lib/common/static/instance_dashboard.js b/lib/common/static/instance_dashboard.js index 5fce5c2..b3e7e69 100644 --- a/lib/common/static/instance_dashboard.js +++ b/lib/common/static/instance_dashboard.js @@ -5,7 +5,10 @@ (function (global) { const SECTION_ORDER = ["orders", "keys", "strategy", "options", "hedge_plan"]; let loading = false; - let timer = null; + let localDashVersion = 0; + let dashEventSource = null; + let dashReconnectTimer = null; + let booted = false; function root() { const active = document.querySelector('.embed-tab-pane.is-active-pane [data-inst-dashboard="1"]'); @@ -121,40 +124,6 @@ return ' class="inst-dash-row" data-dash-tab="' + escapeHtml(tab || "") + '" role="link" tabindex="0"'; } - function mergeOrderLive(items, orderPrices) { - const map = {}; - (orderPrices || []).forEach(function (p) { - if (p && p.id != null) map[String(p.id)] = p; - }); - return (items || []).map(function (it) { - const live = map[String(it.id)] || {}; - const mark = - live.exchange_mark_price != null - ? live.exchange_mark_price - : live.price != null - ? live.price - : it.mark_price; - const contracts = - live.contracts != null - ? live.contracts - : live.order_amount != null - ? live.order_amount - : it.contracts; - const entry = - live.avg_entry_price != null - ? live.avg_entry_price - : it.entry; - return Object.assign({}, it, { - entry: entry, - mark_price: mark, - mark_display: live.price_display || null, - contracts: contracts, - tp_profit: live.reward_at_tp_usdt != null ? live.reward_at_tp_usdt : it.tp_profit, - float_pnl: live.float_pnl != null ? live.float_pnl : it.float_pnl, - }); - }); - } - function renderOrdersTable(items) { const rows = items .map(function (it) { @@ -294,7 +263,7 @@ ); }) .join(""); - return tableWrap(["合约", "来源", "类型", "张数", "到期时间", "目标监控", "盈亏"], rows); + return tableWrap(["合约", "来源", "类型", "张数", "到期时间", "目标监控", "净盈亏"], rows); } function renderHedgeTable(items) { @@ -391,31 +360,32 @@ const status = el.querySelector("#inst-dash-status") || document.getElementById("inst-dash-status"); const sections = el.querySelector("#inst-dash-sections") || document.getElementById("inst-dash-sections"); const updated = el.querySelector("#inst-dash-updated") || document.getElementById("inst-dash-updated"); - if (loading) return; + const options = opts || {}; + if (loading && !options.force) return; loading = true; - if (status && !(opts && opts.silent)) status.textContent = "加载中…"; + if (status && !options.silent) status.textContent = "同步中…"; try { - const [dashRes, priceRes] = await Promise.all([ - fetch("/api/instance/dashboard", { credentials: "same-origin" }), - fetch("/api/price_snapshot", { credentials: "same-origin" }).catch(function () { - return null; - }), - ]); + const dashRes = await fetch("/api/instance/dashboard", { credentials: "same-origin" }); + if (dashRes.status === 401) { + location.href = "/login?next=" + encodeURIComponent(location.pathname); + return; + } const data = await dashRes.json().catch(function () { return {}; }); if (!dashRes.ok || !data.ok) { - throw new Error(data.msg || dashRes.statusText || "加载失败"); - } - let orderPrices = []; - if (priceRes && priceRes.ok) { - try { - const snap = await priceRes.json(); - orderPrices = snap.order_prices || []; - } catch (_) {} + if ( + data.aggregating || + (data.msg && String(data.msg).indexOf("尚未就绪") >= 0) + ) { + if (status) status.textContent = "后台聚合中…"; + return; + } + throw new Error(data.msg || data.error || dashRes.statusText || "加载失败"); } + const ver = Number(data.dashboard_version) || 0; + if (ver) localDashVersion = ver; if (data.orders && Array.isArray(data.orders.items)) { - data.orders.items = mergeOrderLive(data.orders.items, orderPrices); data.orders.count = data.orders.items.length; } if (updated) updated.textContent = "更新 " + (data.updated_at || "—"); @@ -435,7 +405,12 @@ } } } - if (status) status.textContent = ""; + const sec = Number(data.poll_interval_sec) || 5; + if (status) { + status.textContent = options.silent + ? "SSE 已连接 · 后台每 " + sec + "s 聚合" + : "已更新 · 后台每 " + sec + "s 聚合"; + } } catch (e) { if (status) status.textContent = e.message || "加载失败"; } finally { @@ -443,42 +418,81 @@ } } - function stopAuto() { - if (timer) { - clearInterval(timer); - timer = null; + function closeDashboardStream() { + if (dashEventSource) { + dashEventSource.close(); + dashEventSource = null; + } + if (dashReconnectTimer) { + clearTimeout(dashReconnectTimer); + dashReconnectTimer = null; } } - function startAuto() { - stopAuto(); - timer = setInterval(function () { - const el = root(); - if (!el) return; - const pane = el.closest(".embed-tab-pane"); - if (pane && !pane.classList.contains("is-active-pane")) return; - load({ silent: true }); - }, 15000); + function connectDashboardStream() { + const el = root(); + if (!el) return; + closeDashboardStream(); + dashEventSource = new EventSource("/api/instance/dashboard/stream"); + dashEventSource.addEventListener("dashboard", function (ev) { + try { + const st = JSON.parse(ev.data || "{}"); + const ver = Number(st.dashboard_version) || 0; + if (ver && ver !== localDashVersion) { + load({ silent: true }); + } else if (st.aggregating) { + const status = el.querySelector("#inst-dash-status") || document.getElementById("inst-dash-status"); + if (status) status.textContent = "后台聚合中…"; + } + } catch (_) {} + }); + dashEventSource.onerror = function () { + closeDashboardStream(); + const status = el.querySelector("#inst-dash-status") || document.getElementById("inst-dash-status"); + if (status) status.textContent = "SSE 断开,8s 后重连…"; + dashReconnectTimer = setTimeout(function () { + if (booted) { + connectDashboardStream(); + load({ silent: true }); + } + }, 8000); + }; + } + + async function requestDashboardRefresh() { + try { + await fetch("/api/instance/dashboard/refresh", { + method: "POST", + credentials: "same-origin", + }); + } catch (_) {} + load({ force: true }); + } + + function stopAuto() { + closeDashboardStream(); } function init(force) { const el = root(); if (!el) return; if (!force && el.getAttribute("data-dash-booted") === "1") { + booted = true; load({ silent: true }); - startAuto(); + connectDashboardStream(); return; } el.setAttribute("data-dash-booted", "1"); + booted = true; const btn = el.querySelector("#inst-dash-refresh") || document.getElementById("inst-dash-refresh"); if (btn && !btn.getAttribute("data-bound")) { btn.setAttribute("data-bound", "1"); btn.addEventListener("click", function () { - load({}); + requestDashboardRefresh(); }); } load({}); - startAuto(); + connectDashboardStream(); } function refreshSoft(opts) { diff --git a/lib/common/static/instance_live.js b/lib/common/static/instance_live.js index f942c0d..f51b3be 100644 --- a/lib/common/static/instance_live.js +++ b/lib/common/static/instance_live.js @@ -30,9 +30,7 @@ if (tab === "options" && global.OptionsPanelLive && typeof global.OptionsPanelLive.refreshSoft === "function") { global.OptionsPanelLive.refreshSoft(options); } - if (tab === "dashboard" && global.InstanceDashboard && typeof global.InstanceDashboard.refreshSoft === "function") { - global.InstanceDashboard.refreshSoft(options); - } + // 数据看板自有 SSE + 快照,不跟 embed live tick 重拉. } function scheduleRefresh(opts) { diff --git a/lib/common/static/instance_page.css b/lib/common/static/instance_page.css index 0b56f37..d1b95a2 100644 --- a/lib/common/static/instance_page.css +++ b/lib/common/static/instance_page.css @@ -34,7 +34,7 @@ .order-preview-rr.order-preview-rr-ok strong{color:#8fc8ff} .form-row > button,.form-row > label{flex:0 0 auto} .form-grid{display:grid;grid-template-columns:repeat(4,1fr);gap:8px} - /* 复盘表单:长下拉文案需可收缩,否则会撑破四列网格 */ + /* 复盘表单:长下拉文案需可收?否则会撑破四列网?*/ .journal-card .form-grid{gap:10px} .journal-card .form-grid > input, .journal-card .form-grid > select{ @@ -270,10 +270,12 @@ #embed-page-root{min-height:120px;position:relative} .embed-tab-pane[hidden]{display:none!important} .inst-dash-card{grid-column:1/-1} -.inst-dash-head{display:flex;align-items:flex-start;justify-content:space-between;gap:12px;flex-wrap:wrap;margin-bottom:8px} -.inst-dash-desc{margin:0;font-size:.82rem} -.inst-dash-head-actions{display:flex;align-items:center;gap:10px;flex-wrap:wrap} -.inst-dash-status{min-height:1.2em;margin:0 0 10px} +.inst-dash-head{display:flex;align-items:flex-start;justify-content:space-between;gap:12px;flex-wrap:wrap;margin-bottom:6px} +.inst-dash-card > .inst-dash-head h2{font-size:.88rem;margin:0 0 2px;font-weight:600} +.inst-dash-desc{margin:0;font-size:.72rem;line-height:1.35} +.inst-dash-head-actions{display:flex;align-items:center;gap:8px;flex-wrap:wrap} +.inst-dash-updated{font-size:.72rem} +.inst-dash-status{min-height:1.1em;margin:0 0 8px;font-size:.72rem} .inst-dash-sections{display:flex;flex-direction:column;gap:14px} .inst-dash-section{padding:12px;background:#141923;border:1px solid #2a3150;border-radius:10px} .inst-dash-section-head{display:flex;align-items:center;justify-content:space-between;gap:10px;margin-bottom:10px} diff --git a/lib/instance/instance_dashboard_cache.py b/lib/instance/instance_dashboard_cache.py new file mode 100644 index 0000000..bcb9faf --- /dev/null +++ b/lib/instance/instance_dashboard_cache.py @@ -0,0 +1,175 @@ +"""实例数据看板:后台定时聚合,内存快照,SSE 版本通知(对齐中控 dashboard_store).""" +from __future__ import annotations + +import json +import os +import queue +import threading +from collections.abc import Callable, Iterator +from typing import Any + +INSTANCE_DASHBOARD_POLL_SEC = float(os.getenv("INSTANCE_DASHBOARD_POLL_SEC", "5")) +INSTANCE_DASHBOARD_SSE_HEARTBEAT_SEC = float(os.getenv("INSTANCE_DASHBOARD_SSE_HEARTBEAT_SEC", "25")) + +BuildFn = Callable[[], dict[str, Any]] + + +class InstanceDashboardStore: + def __init__(self) -> None: + self._lock = threading.RLock() + self.version = 0 + self.payload: dict[str, Any] | None = None + self.aggregating = False + self.last_error: str | None = None + self._subscribers: list[queue.Queue[str | None]] = [] + self._stop = threading.Event() + self._refresh = threading.Event() + self._thread: threading.Thread | None = None + self._build_fn: BuildFn | None = None + + def start(self, build_fn: BuildFn) -> None: + self._build_fn = build_fn + if self._thread and self._thread.is_alive(): + return + self._stop.clear() + self._thread = threading.Thread( + target=self._loop, + daemon=True, + name="instance-dashboard-poll", + ) + 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: + p = dict(self.payload or {}) + ver = self.version + aggregating = self.aggregating + err = self.last_error + if not p: + return { + "ok": False, + "dashboard_version": ver, + "aggregating": aggregating, + "error": err, + "msg": err or "看板快照尚未就绪", + "poll_interval_sec": INSTANCE_DASHBOARD_POLL_SEC, + } + return { + **p, + "dashboard_version": ver, + "aggregating": aggregating, + "error": err or p.get("error"), + "poll_interval_sec": INSTANCE_DASHBOARD_POLL_SEC, + } + + def event_dict(self) -> dict[str, Any]: + with self._lock: + p = self.payload or {} + return { + "dashboard_version": self.version, + "updated_at": p.get("updated_at"), + "aggregating": self.aggregating, + "ok": p.get("ok", True) if self.payload else False, + "error": self.last_error or p.get("error"), + } + + def _loop(self) -> None: + assert self._build_fn is not None + while not self._stop.is_set(): + self._aggregate_once(self._build_fn) + if self._stop.is_set(): + break + self._refresh.clear() + # 周期等待,可被 request_refresh 提前唤醒 + self._refresh.wait(timeout=INSTANCE_DASHBOARD_POLL_SEC) + + def _aggregate_once(self, build_fn: BuildFn) -> None: + with self._lock: + self.aggregating = True + self._broadcast() + try: + result = build_fn() + if not isinstance(result, dict): + result = {"ok": False, "msg": "聚合返回无效"} + except Exception as e: + result = {"ok": False, "msg": str(e), "error": "aggregate_failed"} + with self._lock: + self.version += 1 + prev = self.payload if isinstance(self.payload, dict) else None + if result.get("ok") is False and prev and prev.get("ok"): + self.payload = prev + self.last_error = str(result.get("msg") or result.get("error") or "aggregate_failed") + else: + self.payload = result + self.last_error = ( + None + if result.get("ok") is not False + else str(result.get("msg") or result.get("error") or "aggregate_failed") + ) + self.aggregating = False + 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=INSTANCE_DASHBOARD_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: dashboard\ndata: {body}\n\n" + + +instance_dashboard_store = InstanceDashboardStore() diff --git a/lib/instance/instance_dashboard_lib.py b/lib/instance/instance_dashboard_lib.py index 8f5b8a6..58bece7 100644 --- a/lib/instance/instance_dashboard_lib.py +++ b/lib/instance/instance_dashboard_lib.py @@ -64,6 +64,7 @@ def _format_order_item(od: dict[str, Any]) -> dict[str, Any]: "title": title, "subtitle": subtitle, "symbol": sym, + "price_symbol": od.get("symbol") or sym, "direction": direction, "direction_label": _dir_label(direction), "entry": entry, @@ -135,15 +136,14 @@ def _format_options_item(p: dict[str, Any], *, conn=None) -> dict[str, Any]: inst = str(p.get("inst_id") or p.get("instId") or "-").strip() or "-" opt_type = str(p.get("opt_type") or p.get("optType") or "").upper() label = "Call" if opt_type == "C" else "Put" if opt_type == "P" else (opt_type or "OPT") - upl = _safe_float(p.get("upl")) - net = None + # 看板期权列固定用净盈亏(买一回收−权利金);残档买一则空. + pnl = None try: from lib.options.options_positions_lib import net_pnl_from_display_row - net = net_pnl_from_display_row(p) + pnl = net_pnl_from_display_row(p) except Exception: - net = None - pnl = net if net is not None else upl + pnl = None pos = _safe_float(p.get("pos")) exp_ms = p.get("exp_time_ms") if exp_ms is None: @@ -350,6 +350,38 @@ def collect_options_items( return out +def enrich_order_items_with_marks( + items: list[dict[str, Any]], + *, + get_price: Optional[Callable[[str], Any]] = None, +) -> list[dict[str, Any]]: + """后台聚合时补标记价(不打全量 fetch_positions;浮盈仍由实盘页口径负责).""" + if not items or not callable(get_price): + return items + out: list[dict[str, Any]] = [] + for it in items: + row = dict(it) + sym = str(row.get("price_symbol") or row.get("symbol") or "").strip() + if not sym: + out.append(row) + continue + try: + px = get_price(sym) + except Exception: + px = None + mark = _safe_float(px) + if mark is None and ":" in sym: + try: + px = get_price(sym.split(":", 1)[0]) + except Exception: + px = None + mark = _safe_float(px) + if mark is not None: + row["mark_price"] = mark + out.append(row) + return out + + def build_instance_dashboard_payload( conn, *, diff --git a/lib/instance/instance_dashboard_register.py b/lib/instance/instance_dashboard_register.py index d38142a..cbfe20a 100644 --- a/lib/instance/instance_dashboard_register.py +++ b/lib/instance/instance_dashboard_register.py @@ -1,9 +1,9 @@ -"""注册 GET /api/instance/dashboard(三所共用).""" +"""注册实例数据看板:快照 GET + SSE + 手动刷新(对齐中控).""" from __future__ import annotations from typing import Any, Callable, Optional -from flask import Flask, jsonify +from flask import Flask, Response, jsonify, stream_with_context def register_instance_dashboard_routes( @@ -13,12 +13,12 @@ def register_instance_dashboard_routes( get_db: Callable, fetch_options_positions: Optional[Callable[[], list[dict[str, Any]]]] = None, hedge_enabled: bool = False, + enrich_orders: Optional[Callable[[list[dict[str, Any]]], list[dict[str, Any]]]] = None, ) -> None: + from lib.instance.instance_dashboard_cache import instance_dashboard_store from lib.instance.instance_dashboard_lib import build_instance_dashboard_payload - @app.route("/api/instance/dashboard") - @login_required - def api_instance_dashboard(): + def _build() -> dict[str, Any]: conn = get_db() try: payload = build_instance_dashboard_payload( @@ -26,6 +26,45 @@ def register_instance_dashboard_routes( fetch_options_positions=fetch_options_positions, hedge_enabled=bool(hedge_enabled), ) - return jsonify(payload) + if callable(enrich_orders) and payload.get("ok") and isinstance(payload.get("orders"), dict): + items = list(payload["orders"].get("items") or []) + try: + enriched = enrich_orders(items) or items + except Exception: + enriched = items + payload["orders"]["items"] = enriched + payload["orders"]["count"] = len(enriched) + return payload finally: conn.close() + + instance_dashboard_store.start(_build) + + @app.route("/api/instance/dashboard") + @login_required + def api_instance_dashboard(): + return jsonify(instance_dashboard_store.snapshot_dict()) + + @app.route("/api/instance/dashboard/stream") + @login_required + def api_instance_dashboard_stream(): + return Response( + stream_with_context(instance_dashboard_store.iter_sse()), + mimetype="text/event-stream", + headers={ + "Cache-Control": "no-cache", + "Connection": "keep-alive", + "X-Accel-Buffering": "no", + }, + ) + + @app.route("/api/instance/dashboard/refresh", methods=["POST"]) + @login_required + def api_instance_dashboard_refresh(): + instance_dashboard_store.request_refresh() + return jsonify( + { + "ok": True, + "dashboard_version": instance_dashboard_store.version, + } + ) diff --git a/lib/instance/templates/dashboard_panel.html b/lib/instance/templates/dashboard_panel.html index a47bafe..6b7342b 100644 --- a/lib/instance/templates/dashboard_panel.html +++ b/lib/instance/templates/dashboard_panel.html @@ -3,7 +3,7 @@

数据看板

-

本户活跃监控总览 · 只读 · 无数据的区块不显示 · 有数据按表格展示

+

本户活跃监控总览 · 只读 · 后台快照 + SSE · 无数据的区块不显示

— diff --git a/lib/instance/templates/embed_shell.html b/lib/instance/templates/embed_shell.html index 8b454b2..6e69f0a 100644 --- a/lib/instance/templates/embed_shell.html +++ b/lib/instance/templates/embed_shell.html @@ -6,7 +6,7 @@ - + @@ -111,7 +111,7 @@ const ORDER_ENTRY_MODEL_CODE_TO_CATEGORY = {{ entry_model_code_to_category | toj {% include 'embed_boot_scripts.html' %} - + diff --git a/lib/instance/templates/index.html b/lib/instance/templates/index.html index ae5cc44..b81ebdc 100644 --- a/lib/instance/templates/index.html +++ b/lib/instance/templates/index.html @@ -16,7 +16,7 @@ {{ pwa_app_name }} - + @@ -2011,7 +2011,7 @@ setInterval(refreshPriceSnapshotConditional, {{ price_refresh_seconds * 1000 }}) - +