Compute option stats from exchange history instead of local DB.

Share history loading between history and stats APIs so average profit/loss matches the option history tab.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dekun
2026-07-11 09:34:49 +08:00
parent 68733eb4f9
commit baf94b4feb
5 changed files with 198 additions and 87 deletions
+60
View File
@@ -33,6 +33,66 @@ def _avg_seconds(values: list[float]) -> float | None:
return round(sum(values) / len(values), 1)
def compute_options_stats_from_history(history: list[dict[str, Any]]) -> dict[str, Any]:
"""基于期权历史列表(交易所)计算统计."""
wins: list[float] = []
losses: list[float] = []
win_holds: list[float] = []
loss_holds: list[float] = []
all_holds: list[float] = []
open_holds: list[float] = []
now = datetime.now()
for row in history:
if row.get("status") == "open":
start = _parse_ts(row.get("created_at"))
if start is not None:
sec = (now - start).total_seconds()
if sec >= 0:
open_holds.append(sec)
continue
pnl_raw = row.get("realized_pnl")
if pnl_raw is None:
continue
try:
pnl = float(pnl_raw)
except (TypeError, ValueError):
continue
hold = _hold_seconds(row.get("created_at"), row.get("closed_at"))
if hold is not None:
all_holds.append(hold)
if pnl > 0:
wins.append(pnl)
if hold is not None:
win_holds.append(hold)
elif pnl < 0:
losses.append(pnl)
if hold is not None:
loss_holds.append(hold)
total_closed = len(wins) + len(losses)
win_rate = round(len(wins) / total_closed * 100, 2) if total_closed else 0
avg_win = sum(wins) / len(wins) if wins else None
avg_loss = sum(losses) / len(losses) if losses else None
return {
"total_closed": total_closed,
"win_count": len(wins),
"loss_count": len(losses),
"win_rate": win_rate,
"profit_loss_ratio": profit_loss_ratio_from_averages(avg_win, avg_loss),
"avg_win": round(avg_win, 4) if avg_win is not None else None,
"avg_loss": round(abs(avg_loss), 4) if avg_loss is not None else None,
"total_profit": round(sum(wins), 4) if wins else 0.0,
"total_loss": round(abs(sum(losses)), 4) if losses else 0.0,
"avg_hold_sec": _avg_seconds(all_holds),
"avg_win_hold_sec": _avg_seconds(win_holds),
"avg_loss_hold_sec": _avg_seconds(loss_holds),
"open_count": len(open_holds),
"avg_open_hold_sec": _avg_seconds(open_holds),
}
def compute_options_stats(get_db) -> dict[str, Any]:
conn = get_db()
try: