dekun
d1ab2d5172
Show average hold duration for wins and losses, open positions, and CSS ring/bar visualizations in the stats tab. Co-authored-by: Cursor <cursoragent@cursor.com>
104 lines
3.2 KiB
Python
104 lines
3.2 KiB
Python
"""期权本地交易统计(胜率 / 盈亏 / 持仓时长)."""
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
from typing import Any
|
|
|
|
from lib.instance.instance_embed_context_lib import profit_loss_ratio_from_averages
|
|
from lib.options.options_db import init_options_tables
|
|
|
|
|
|
def _parse_ts(raw: Any) -> datetime | None:
|
|
if raw is None or raw == "":
|
|
return None
|
|
s = str(raw).strip().replace(" ", "T", 1)
|
|
try:
|
|
return datetime.fromisoformat(s)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def _hold_seconds(created_at: Any, closed_at: Any) -> float | None:
|
|
start = _parse_ts(created_at)
|
|
end = _parse_ts(closed_at)
|
|
if start is None or end is None:
|
|
return None
|
|
sec = (end - start).total_seconds()
|
|
return sec if sec >= 0 else None
|
|
|
|
|
|
def _avg_seconds(values: list[float]) -> float | None:
|
|
if not values:
|
|
return None
|
|
return round(sum(values) / len(values), 1)
|
|
|
|
|
|
def compute_options_stats(get_db) -> dict[str, Any]:
|
|
conn = get_db()
|
|
try:
|
|
init_options_tables(conn)
|
|
closed_rows = conn.execute(
|
|
"""
|
|
SELECT realized_pnl, created_at, closed_at
|
|
FROM options_trades
|
|
WHERE status = 'closed' AND realized_pnl IS NOT NULL
|
|
"""
|
|
).fetchall()
|
|
open_rows = conn.execute(
|
|
"""
|
|
SELECT created_at FROM options_trades WHERE status = 'open'
|
|
"""
|
|
).fetchall()
|
|
finally:
|
|
conn.close()
|
|
|
|
wins: list[float] = []
|
|
losses: list[float] = []
|
|
win_holds: list[float] = []
|
|
loss_holds: list[float] = []
|
|
all_holds: list[float] = []
|
|
now = datetime.now()
|
|
|
|
for row in closed_rows:
|
|
pnl = float(row["realized_pnl"])
|
|
hold = _hold_seconds(row["created_at"], row["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)
|
|
|
|
open_holds: list[float] = []
|
|
for row in open_rows:
|
|
start = _parse_ts(row["created_at"])
|
|
if start is None:
|
|
continue
|
|
sec = (now - start).total_seconds()
|
|
if sec >= 0:
|
|
open_holds.append(sec)
|
|
|
|
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),
|
|
"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),
|
|
}
|