Align instance dashboard with snapshot+SSE and options net PnL.
Only dashboard path changes: background snapshot, SSE refresh, smaller header type, options net PnL. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
@@ -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()
|
||||
@@ -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,
|
||||
*,
|
||||
|
||||
@@ -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,
|
||||
}
|
||||
)
|
||||
|
||||
@@ -3,7 +3,7 @@
|
||||
<div class="inst-dash-head">
|
||||
<div>
|
||||
<h2 style="margin-bottom:4px">数据看板</h2>
|
||||
<p class="muted inst-dash-desc">本户活跃监控总览 · 只读 · 无数据的区块不显示 · 有数据按表格展示</p>
|
||||
<p class="muted inst-dash-desc">本户活跃监控总览 · 只读 · 后台快照 + SSE · 无数据的区块不显示</p>
|
||||
</div>
|
||||
<div class="inst-dash-head-actions">
|
||||
<span class="muted inst-dash-updated" id="inst-dash-updated">—</span>
|
||||
|
||||
@@ -6,7 +6,7 @@
|
||||
<script src="/static/instance_theme.js?v=50"></script>
|
||||
<link rel="stylesheet" href="/static/instance_theme_early.css?v=4">
|
||||
<link rel="stylesheet" href="/static/account_risk_badge.css?v=4">
|
||||
<link rel="stylesheet" href="/static/instance_page.css?v=9">
|
||||
<link rel="stylesheet" href="/static/instance_page.css?v=10">
|
||||
<link rel="stylesheet" href="/static/instance_theme.css?v=97">
|
||||
<script src="/static/account_risk_badge.js?v=4"></script>
|
||||
<meta name="theme-color" content="#0b0d14">
|
||||
@@ -111,7 +111,7 @@ const ORDER_ENTRY_MODEL_CODE_TO_CATEGORY = {{ entry_model_code_to_category | toj
|
||||
{% include 'embed_boot_scripts.html' %}
|
||||
<script src="/static/records_review_page.js?v=2"></script>
|
||||
<script src="/static/options_expiry_countdown.js?v=1"></script>
|
||||
<script src="/static/instance_dashboard.js?v=3"></script>
|
||||
<script src="/static/instance_dashboard.js?v=5"></script>
|
||||
<script>
|
||||
window.__INSTANCE_DISPLAY__ = {{ display | tojson }};
|
||||
</script>
|
||||
|
||||
@@ -16,7 +16,7 @@
|
||||
<link rel="apple-touch-icon" href="/static/icons/apple-touch-icon.png">
|
||||
<link rel="manifest" href="/static/icons/manifest.webmanifest">
|
||||
<title>{{ pwa_app_name }}</title>
|
||||
<link rel="stylesheet" href="/static/instance_page.css?v=9">
|
||||
<link rel="stylesheet" href="/static/instance_page.css?v=10">
|
||||
<link rel="stylesheet" href="/static/instance_theme.css?v=97">
|
||||
|
||||
</head>
|
||||
@@ -2011,7 +2011,7 @@ setInterval(refreshPriceSnapshotConditional, {{ price_refresh_seconds * 1000 }})
|
||||
</script>
|
||||
<script src="/static/records_review_page.js?v=2"></script>
|
||||
<script src="/static/options_expiry_countdown.js?v=1"></script>
|
||||
<script src="/static/instance_dashboard.js?v=3"></script>
|
||||
<script src="/static/instance_dashboard.js?v=5"></script>
|
||||
<script>
|
||||
window.__INSTANCE_DISPLAY__ = {{ display | tojson }};
|
||||
{% if page == 'dashboard' %}
|
||||
|
||||
Reference in New Issue
Block a user