"""中控统计汇总:KPI 对比 + 已实现盈亏曲线.""" from __future__ import annotations import time from typing import Any 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, record_cash_samples_from_overview, ) def _safe_float(v: Any) -> float | None: if v is None or v == "": return None try: return float(v) except (TypeError, ValueError): return None def _safe_int(v: Any) -> int: try: return int(float(v)) except (TypeError, ValueError): return 0 def normalize_window(raw: Any) -> str: w = str(raw or "7d").strip().lower() if w in ("1d", "day", "今日"): return "today" if w in ("7", "week"): return "7d" if w in ("30", "month"): return "30d" if w in ("0", "全部", "total"): return "all" return w if w in ("today", "7d", "30d", "all") else "7d" def build_stats_summary( window: str = "7d", *, snaps: list[dict[str, Any]] | None = None, ) -> dict[str, Any]: w = normalize_window(window) # 实时状态(余额/权利金/浮盈)来自快照;窗口 KPI 来自各机 stats # snaps 可由热缓存传入,避免统计页再现场重复拉全机 if snaps is None: snaps = pull_all_enabled() try: record_cash_samples_from_overview(snaps) except Exception: pass stats_rows = pull_all_stats(w) stats_by_id = {r.get("id"): r for r in stats_rows} try: initial_fallback = first_cash_by_instance([s.get("id") for s in snaps]) except Exception: initial_fallback = {} machines: list[dict[str, Any]] = [] sum_cash = 0.0 sum_premium = 0.0 sum_unreal = 0.0 sum_net = 0.0 sum_profit = 0.0 sum_loss = 0.0 sum_closed = 0 sum_wins = 0 sum_losses = 0 online_n = 0 cash_known = False premium_known = False unreal_known = False for snap in snaps: iid = snap.get("id") st = stats_by_id.get(iid) or {} stats = st.get("stats") if isinstance(st.get("stats"), dict) else {} body = snap.get("snapshot") if isinstance(snap.get("snapshot"), dict) else {} premium = body.get("premium") if isinstance(body.get("premium"), dict) else {} pnl = body.get("pnl") if isinstance(body.get("pnl"), dict) else {} brief = body.get("positions_brief") if isinstance(body.get("positions_brief"), list) else [] cash = _safe_float(premium.get("cash_usdt")) paid = _safe_float(premium.get("paid_usdt")) ratio = _safe_float(premium.get("ratio_pct")) unreal = _safe_float(pnl.get("unrealized_usdt")) net = _safe_float(stats.get("net_realized_pnl")) profit = _safe_float(stats.get("total_profit")) loss = _safe_float(stats.get("total_loss")) closed = _safe_int(stats.get("total_closed")) wins = _safe_int(stats.get("win_count")) losses = _safe_int(stats.get("loss_count")) win_rate = _safe_float(stats.get("win_rate")) avg_hold = _safe_float(stats.get("avg_hold_sec")) online = bool(snap.get("online")) mode = ((body.get("instance") or {}) if isinstance(body.get("instance"), dict) else {}).get( "mode" ) # 收益率 = 窗口已结算盈亏 / 初始资金(非当前余额) initial = _safe_float(premium.get("initial_cash_usdt")) if initial is None or initial <= 0: initial = _safe_float(initial_fallback.get(iid)) return_pct = None if online and stats.get("ok") and net is not None and initial is not None and initial > 0: return_pct = round(float(net) / float(initial) * 100.0, 2) if online: online_n += 1 if cash is not None: sum_cash += cash cash_known = True if paid is not None: sum_premium += paid premium_known = True if unreal is not None: sum_unreal += unreal unreal_known = True if online and stats.get("ok"): if net is not None: sum_net += net if profit is not None: sum_profit += profit if loss is not None: sum_loss += loss sum_closed += closed sum_wins += wins sum_losses += losses machines.append( { "id": iid, "name": snap.get("name"), "online": online, "error": snap.get("error") or st.get("error") or "", "mode": mode, "cash_usdt": cash, "premium_usdt": paid, "ratio_pct": ratio, "unrealized_usdt": unreal, "initial_cash_usdt": initial, "pos_count": len(brief), "total_closed": closed if online and stats.get("ok") else None, "win_count": wins if online and stats.get("ok") else None, "loss_count": losses if online and stats.get("ok") else None, "win_rate": win_rate if online and stats.get("ok") else None, "net_realized_pnl": net if online and stats.get("ok") else None, "return_rate_pct": return_pct, "total_profit": profit if online and stats.get("ok") else None, "total_loss": loss if online and stats.get("ok") else None, "avg_hold_sec": avg_hold if online and stats.get("ok") else None, "stats_ok": bool(online and stats.get("ok")), } ) def _mode_rank(m: dict[str, Any]) -> tuple[int, str]: mode_v = str(m.get("mode") or "").strip().lower() # 上方 sim,下方 live rank = 0 if mode_v == "sim" else 1 if mode_v == "live" else 2 return (rank, str(m.get("name") or m.get("id") or "")) machines.sort(key=_mode_rank) agg_win_rate = round(sum_wins / sum_closed * 100, 2) if sum_closed else 0.0 return { "ok": True, "window": w, "totals": { "online": online_n, "machines": len(machines), "cash_usdt": round(sum_cash, 4) if cash_known else None, "premium_usdt": round(sum_premium, 4) if premium_known else None, "unrealized_usdt": round(sum_unreal, 4) if unreal_known else None, "net_realized_pnl": round(sum_net, 4), "total_profit": round(sum_profit, 4), "total_loss": round(sum_loss, 4), "total_closed": sum_closed, "win_count": sum_wins, "loss_count": sum_losses, "win_rate": agg_win_rate, }, "machines": machines, } def build_stats_curve(range_key: str = "7d") -> dict[str, Any]: """多机累计已实现盈亏曲线(各机台阶叠加;划转/提现不影响).""" 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, }