diff --git a/lib/bridge_client_lib.py b/lib/bridge_client_lib.py index d4f70a5..eff0be0 100644 --- a/lib/bridge_client_lib.py +++ b/lib/bridge_client_lib.py @@ -325,6 +325,77 @@ def pull_all_stats(window: str = "7d") -> list[dict[str, Any]]: return results +def fetch_one_pnl_curve(inst: dict[str, Any], *, range_key: str = "7d") -> dict[str, Any]: + """拉取单机累计已实现盈亏曲线.""" + base = (inst.get("base_url") or "").rstrip("/") + token = (inst.get("bridge_token") or "").strip() + out: dict[str, Any] = { + "id": inst.get("id"), + "name": inst.get("name"), + "online": False, + "error": "", + "points": [], + "trade_count": 0, + "latest": None, + "metric": "realized_pnl", + } + if not base or not token: + out["error"] = "缺少 URL 或 Token" + return out + url = f"{base}/api/hub/options/pnl_curve" + try: + r = requests.get( + url, + headers={"X-Hub-Token": token, "Accept": "application/json"}, + params={"range": range_key}, + timeout=max(_timeout(), 15.0), + ) + data = {} + try: + data = r.json() + except Exception: + data = {} + if r.status_code != 200 or not isinstance(data, dict) or not data.get("ok"): + out["error"] = (data.get("msg") if isinstance(data, dict) else None) or f"HTTP {r.status_code}" + return out + out["online"] = True + pts = data.get("points") if isinstance(data.get("points"), list) else [] + out["points"] = pts + out["trade_count"] = int(data.get("trade_count") or 0) + out["latest"] = data.get("latest") + out["metric"] = data.get("metric") or "realized_pnl" + out["start_ms"] = data.get("start_ms") + out["end_ms"] = data.get("end_ms") + return out + except requests.Timeout: + out["error"] = "超时" + return out + except Exception as e: + out["error"] = str(e) or "拉取失败" + return out + + +def pull_all_pnl_curves(range_key: str = "7d") -> list[dict[str, Any]]: + instances = list_instances(enabled_only=True) + if not instances: + return [] + results: list[dict[str, Any]] = [] + with ThreadPoolExecutor(max_workers=min(8, len(instances))) as pool: + futs = { + pool.submit(fetch_one_pnl_curve, inst, range_key=range_key): inst["id"] + for inst in instances + } + by_id: dict[Any, dict[str, Any]] = {} + for fut in as_completed(futs): + row = fut.result() + by_id[row.get("id")] = row + for inst in instances: + results.append( + by_id.get(inst["id"]) or fetch_one_pnl_curve(inst, range_key=range_key) + ) + return results + + def fetch_client_version(inst: dict[str, Any]) -> dict[str, Any]: """探测客户端 git 版本 / 在线.""" base = (inst.get("base_url") or "").rstrip("/") diff --git a/lib/stats_agg_lib.py b/lib/stats_agg_lib.py index d66432b..e8109c5 100644 --- a/lib/stats_agg_lib.py +++ b/lib/stats_agg_lib.py @@ -1,12 +1,12 @@ -"""中控统计汇总:KPI 对比 + 余额曲线.""" +"""中控统计汇总:KPI 对比 + 已实现盈亏曲线.""" from __future__ import annotations +import time from typing import Any -from lib.bridge_client_lib import pull_all_enabled, pull_all_stats +from lib.bridge_client_lib import pull_all_enabled, pull_all_pnl_curves, pull_all_stats from lib.cash_sample_lib import ( first_cash_by_instance, - query_cash_series, record_cash_samples_from_overview, ) @@ -185,7 +185,84 @@ def build_stats_summary( def build_stats_curve(range_key: str = "7d") -> dict[str, Any]: - # 曲线只读本地采样库;余额采样由总览常驻 pull 负责 - data = query_cash_series(range_key=range_key) - data["ok"] = True - return data + """多机累计已实现盈亏曲线(各机台阶叠加;划转/提现不影响).""" + rk = normalize_window(range_key) + now_ms = int(time.time() * 1000) + rows = pull_all_pnl_curves(rk) + series: list[dict[str, Any]] = [] + by_id: dict[Any, list[dict[str, float]]] = {} + start_ms = None + end_ms = now_ms + for row in rows: + iid = row.get("id") + pts_raw = row.get("points") if isinstance(row.get("points"), list) else [] + pts: list[dict[str, float]] = [] + for p in pts_raw: + if not isinstance(p, dict): + continue + try: + t = int(p.get("t")) + v = float(p.get("v")) + except (TypeError, ValueError): + continue + pts.append({"t": t, "v": v}) + pts.sort(key=lambda x: x["t"]) + if row.get("start_ms") is not None: + try: + sm = int(row.get("start_ms")) + start_ms = sm if start_ms is None else min(start_ms, sm) + except (TypeError, ValueError): + pass + if row.get("end_ms") is not None: + try: + end_ms = max(end_ms, int(row.get("end_ms"))) + except (TypeError, ValueError): + pass + by_id[iid] = pts + series.append( + { + "id": iid, + "name": row.get("name") or str(iid), + "online": bool(row.get("online")), + "error": row.get("error") or "", + "points": pts, + "trade_count": int(row.get("trade_count") or 0), + "latest": row.get("latest"), + } + ) + + # 合计:按 1 分钟桶,各机该桶最后值求和并前向填充 + last_in_bucket: dict[int, dict[Any, float]] = {} + for iid, pts in by_id.items(): + for p in pts: + bt = (int(p["t"]) // 60000) * 60000 + last_in_bucket.setdefault(bt, {})[iid] = float(p["v"]) + running: dict[Any, float] = {} + total_pts: list[dict[str, float]] = [] + for bt in sorted(last_in_bucket.keys()): + running.update(last_in_bucket[bt]) + if running: + total_pts.append({"t": bt, "v": round(sum(running.values()), 4)}) + + if start_ms is None: + if rk in ("today", "1d"): + import datetime as dt + + start = dt.datetime.now().replace(hour=0, minute=0, second=0, microsecond=0) + start_ms = int(start.timestamp() * 1000) + elif rk in ("30d", "30"): + start_ms = now_ms - 30 * 86400 * 1000 + elif rk in ("all", "0"): + start_ms = 0 + else: + start_ms = now_ms - 7 * 86400 * 1000 + + return { + "ok": True, + "metric": "realized_pnl", + "series": series, + "total": total_pts, + "range": rk, + "start_ms": start_ms, + "end_ms": end_ms, + } diff --git a/templates/stats.html b/templates/stats.html index 491a3c8..33c0f35 100644 --- a/templates/stats.html +++ b/templates/stats.html @@ -1,77 +1,77 @@ -{% extends "base.html" %} -{% block title %}数据统计 · 期权中控{% endblock %} -{% block body %} -
-
-

Analytics

-

数据统计

-

多机 KPI 对比与余额曲线(不含权利金/浮盈)。后台常驻刷新,打开即读缓存。

-
-
- - - - -
-
- -

读取缓存…

- -
-
-
- -
-
-
-

机器对比

-

实时资金状态 + 当前窗口交易 KPI。排序：模拟在上、实盘在下。收益率 = 窗口已结算盈亏 ÷ 初始资金。

-
-
-
- - - - - - - - - - - - - - - - - - - - -
机器状态模式余额权利金浮盈笔数胜率已实现收益率均持仓
加载中…
-
-
- -
-
-
-

余额曲线

-

主指标为账户现金余额,不含已用权利金与浮盈。可勾选机器叠加对比。

-
- -
-
-
- - -
-
-
-{% endblock %} -{% block scripts %} - -{% endblock %} +{% extends "base.html" %} +{% block title %}数据统计 · 期权中控{% endblock %} +{% block body %} +
+
+

Analytics

+

数据统计

+

多机 KPI 对比与累计已实现盈亏曲线。划转/提现不影响曲线。后台常驻刷新,打开即读缓存。

+
+
+ + + + +
+
+ +

读取缓存…

+ +
+
+
+ +
+
+
+

机器对比

+

实时资金状态 + 当前窗口交易 KPI。排序：模拟在上、实盘在下。收益率 = 窗口已结算盈亏 ÷ 初始资金。

+
+
+
+ + + + + + + + + + + + + + + + + + + + +
机器状态模式余额权利金浮盈笔数胜率已实现收益率均持仓
加载中…
+
+
+ +
+
+
+

已实现盈亏曲线

+

主指标为各机累计已实现盈亏(期权平仓真盈亏U累加)。可勾选机器叠加对比。

+
+ +
+
+
+ + +
+
+
+{% endblock %} +{% block scripts %} + +{% endblock %} diff --git a/tests/test_stats_realized_pnl_curve.py b/tests/test_stats_realized_pnl_curve.py new file mode 100644 index 0000000..e513588 --- /dev/null +++ b/tests/test_stats_realized_pnl_curve.py @@ -0,0 +1,52 @@ +"""中控统计曲线:各机累计已实现盈亏叠加合计.""" +from __future__ import annotations + +import unittest +from unittest.mock import patch + + +class TestStatsRealizedPnlCurve(unittest.TestCase): + def test_build_stats_curve_sums_machines(self): + from lib.stats_agg_lib import build_stats_curve + + t0 = 1_700_000_040_000 # 已对齐到分钟桶 + t1 = t0 + 60_000 + t2 = t0 + 120_000 + fake_rows = [ + { + "id": "a", + "name": "A", + "online": True, + "error": "", + "points": [{"t": t0, "v": 10.0}, {"t": t2, "v": 15.0}], + "trade_count": 2, + "latest": 15.0, + "start_ms": t0, + "end_ms": t2, + }, + { + "id": "b", + "name": "B", + "online": True, + "error": "", + "points": [{"t": t1, "v": 5.0}], + "trade_count": 1, + "latest": 5.0, + "start_ms": t1, + "end_ms": t2, + }, + ] + with patch("lib.stats_agg_lib.pull_all_pnl_curves", return_value=fake_rows): + out = build_stats_curve("all") + self.assertTrue(out["ok"]) + self.assertEqual(out["metric"], "realized_pnl") + self.assertEqual(len(out["series"]), 2) + totals = {p["t"]: p["v"] for p in out["total"]} + # t0: only A=10; t1: A=10 + B=5; t2: A=15 + B=5 + self.assertAlmostEqual(totals[t0], 10.0) + self.assertAlmostEqual(totals[t1], 15.0) + self.assertAlmostEqual(totals[t2], 20.0) + + +if __name__ == "__main__": + unittest.main()