Add hedge history delete/detail modal and typed stats cards.

History shows contract names with clickable fill details; stats split perp vs options plans for win rate, profit factor, max win/loss, and drawdown.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dekun
2026-07-14 14:04:48 +08:00
parent e6907dfbbe
commit f6ef06f659
8 changed files with 700 additions and 51 deletions
+152 -7
View File
@@ -151,8 +151,124 @@ def get_plan_legs(conn: sqlite3.Connection, plan_id: int) -> list[dict[str, Any]
return [dict(r) for r in rows]
def delete_plan(conn: sqlite3.Connection, plan_id: int) -> dict[str, Any]:
"""删除已结束/失败/取消的计划及其腿;活跃计划拒绝删除."""
plan = get_plan(conn, int(plan_id))
if not plan:
return {"ok": False, "msg": "计划不存在"}
st = str(plan.get("status") or "")
if st in ("opening", "active", "partial"):
return {"ok": False, "msg": "进行中的计划不可删除,请先结束"}
conn.execute("DELETE FROM hedge_plan_legs WHERE plan_id=?", (int(plan_id),))
conn.execute("DELETE FROM hedge_plans WHERE id=?", (int(plan_id),))
return {"ok": True, "deleted_id": int(plan_id)}
def legs_contract_summary(legs: list[dict[str, Any]]) -> str:
parts: list[str] = []
for leg in legs:
role = str(leg.get("leg_role") or "")
if role == "perp":
name = str(leg.get("symbol") or "永续")
parts.append(f"永续 {name}")
else:
inst = str(leg.get("inst_id") or "")
ot = str(leg.get("opt_type") or "").upper()
strike = leg.get("strike")
label = inst or (f"{ot}{strike}" if ot or strike is not None else role)
parts.append(label)
return " · ".join(parts) if parts else ""
def attach_legs_to_plans(conn: sqlite3.Connection, plans: list[dict[str, Any]]) -> list[dict[str, Any]]:
out: list[dict[str, Any]] = []
for p in plans:
legs = get_plan_legs(conn, int(p["id"]))
row = dict(p)
row["legs"] = legs
row["contracts_summary"] = legs_contract_summary(legs)
out.append(row)
return out
def _sf(v: Any) -> Optional[float]:
try:
if v is None or v == "":
return None
return float(v)
except (TypeError, ValueError):
return None
def _metrics_from_pnls(rows: list[dict[str, Any]]) -> dict[str, Any]:
"""对一组已结束计划计算胜率/盈亏比/最大盈亏/最大回撤."""
pnls: list[float] = []
timed: list[tuple[str, float]] = []
for r in rows:
pnl = _sf(r.get("realized_pnl_total"))
if pnl is None:
continue
pnls.append(pnl)
t = str(r.get("closed_at") or r.get("opened_at") or r.get("created_at") or "")
timed.append((t, pnl))
n = len(pnls)
if n == 0:
return {
"count": 0,
"wins": 0,
"losses": 0,
"win_rate": None,
"net_pnl": 0.0,
"avg_pnl": None,
"avg_premium": None,
"profit_factor": None,
"max_profit": None,
"max_loss": None,
"max_drawdown": None,
}
wins = [x for x in pnls if x > 0]
losses = [x for x in pnls if x < 0]
gross_win = sum(wins)
gross_loss = abs(sum(losses))
if gross_loss > 0:
profit_factor = round(gross_win / gross_loss, 4)
elif gross_win > 0:
profit_factor = None # 全胜,标无限
else:
profit_factor = 0.0
timed.sort(key=lambda x: x[0] or "")
cum = 0.0
peak = 0.0
mdd = 0.0
for _, p in timed:
cum += p
if cum > peak:
peak = cum
dd = peak - cum
if dd > mdd:
mdd = dd
premiums = [_sf(r.get("premium_total")) for r in rows]
premiums_f = [x for x in premiums if x is not None]
return {
"count": n,
"wins": len(wins),
"losses": len(losses),
"win_rate": round(len(wins) / n, 4),
"net_pnl": round(sum(pnls), 4),
"avg_pnl": round(sum(pnls) / n, 4),
"avg_premium": round(sum(premiums_f) / len(premiums_f), 4) if premiums_f else None,
"profit_factor": profit_factor,
"profit_factor_infinite": bool(gross_loss <= 0 and gross_win > 0),
"max_profit": round(max(pnls), 4),
"max_loss": round(min(pnls), 4),
"max_drawdown": round(mdd, 4),
}
def stats_summary(conn: sqlite3.Connection) -> dict[str, Any]:
rows = conn.execute(
reason_rows = conn.execute(
"""
SELECT plan_type, close_reason, COUNT(1) AS n,
COALESCE(SUM(realized_pnl_total), 0) AS pnl
@@ -161,13 +277,42 @@ def stats_summary(conn: sqlite3.Connection) -> dict[str, Any]:
GROUP BY plan_type, close_reason
"""
).fetchall()
closed = conn.execute(
"SELECT COUNT(1) AS c, COALESCE(SUM(realized_pnl_total),0) AS pnl FROM hedge_plans WHERE status='closed'"
).fetchone()
closed_rows = [
dict(r)
for r in conn.execute(
"SELECT * FROM hedge_plans WHERE status='closed' ORDER BY COALESCE(closed_at, opened_at, created_at), id"
).fetchall()
]
active = count_active_plans(conn)
overall = _metrics_from_pnls(closed_rows)
by_type = {
"perp_options": _metrics_from_pnls(
[r for r in closed_rows if r.get("plan_type") == "perp_options"]
),
"options_options": _metrics_from_pnls(
[r for r in closed_rows if r.get("plan_type") == "options_options"]
),
}
# 永期止盈/止损分桶
po = [r for r in closed_rows if r.get("plan_type") == "perp_options"]
by_type["perp_options"]["buckets"] = {
"tp": _metrics_from_pnls([r for r in po if r.get("close_reason") == "perp_tp"]),
"sl": _metrics_from_pnls([r for r in po if r.get("close_reason") == "perp_sl"]),
}
oo = [r for r in closed_rows if r.get("plan_type") == "options_options"]
by_type["options_options"]["buckets"] = {
"expiry_loss": _metrics_from_pnls(
[r for r in oo if r.get("close_reason") == "oo_expiry_loss"]
),
"expiry_win": _metrics_from_pnls(
[r for r in oo if r.get("close_reason") == "oo_expiry_win"]
),
}
return {
"active": active,
"closed_count": int((closed["c"] if closed else 0) or 0),
"closed_pnl_total": float((closed["pnl"] if closed else 0) or 0),
"by_reason": [dict(r) for r in rows],
"closed_count": overall["count"],
"closed_pnl_total": overall["net_pnl"],
"overall": overall,
"by_type": by_type,
"by_reason": [dict(r) for r in reason_rows],
}