Add trade stats board: win rate, hold buckets, by machine/side.
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dekun
@@ -304,6 +304,196 @@ def list_trades(
|
||||
return rows
|
||||
|
||||
|
||||
def _hold_seconds(open_time: Any, close_time: Any, now_ts: Optional[float] = None) -> Optional[float]:
|
||||
if open_time in (None, ""):
|
||||
return None
|
||||
try:
|
||||
start = float(open_time)
|
||||
end = float(close_time) if close_time not in (None, "") else float(now_ts or now())
|
||||
except (TypeError, ValueError):
|
||||
return None
|
||||
return max(0.0, end - start)
|
||||
|
||||
|
||||
def _fmt_hold_secs(secs: Optional[float]) -> str:
|
||||
if secs is None:
|
||||
return "—"
|
||||
s = int(secs)
|
||||
days, rem = divmod(s, 86400)
|
||||
hours, rem = divmod(rem, 3600)
|
||||
mins, sec = divmod(rem, 60)
|
||||
if days > 0:
|
||||
return f"{days}天{hours}小时{mins}分"
|
||||
if hours > 0:
|
||||
return f"{hours}小时{mins}分"
|
||||
if mins > 0:
|
||||
return f"{mins}分{sec}秒"
|
||||
return f"{sec}秒"
|
||||
|
||||
|
||||
def _bucket_key(secs: Optional[float]) -> str:
|
||||
if secs is None:
|
||||
return "unknown"
|
||||
if secs < 300:
|
||||
return "lt_5m"
|
||||
if secs < 1800:
|
||||
return "m5_30"
|
||||
if secs < 7200:
|
||||
return "m30_2h"
|
||||
return "ge_2h"
|
||||
|
||||
|
||||
def _empty_group(label: str) -> dict[str, Any]:
|
||||
return {
|
||||
"label": label,
|
||||
"count": 0,
|
||||
"wins": 0,
|
||||
"losses": 0,
|
||||
"breakeven": 0,
|
||||
"profit": 0.0,
|
||||
"hold_sum": 0.0,
|
||||
"hold_n": 0,
|
||||
}
|
||||
|
||||
|
||||
def _finalize_group(g: dict[str, Any]) -> dict[str, Any]:
|
||||
n = g["count"]
|
||||
decided = g["wins"] + g["losses"]
|
||||
g["win_rate"] = (g["wins"] / decided * 100.0) if decided else 0.0
|
||||
g["avg_profit"] = (g["profit"] / n) if n else 0.0
|
||||
g["avg_hold_secs"] = (g["hold_sum"] / g["hold_n"]) if g["hold_n"] else None
|
||||
g["avg_hold"] = _fmt_hold_secs(g["avg_hold_secs"])
|
||||
return g
|
||||
|
||||
|
||||
def _accumulate(g: dict[str, Any], profit: Optional[float], hold: Optional[float]) -> None:
|
||||
g["count"] += 1
|
||||
p = float(profit) if profit is not None else 0.0
|
||||
g["profit"] += p
|
||||
if profit is not None:
|
||||
if p > 1e-9:
|
||||
g["wins"] += 1
|
||||
elif p < -1e-9:
|
||||
g["losses"] += 1
|
||||
else:
|
||||
g["breakeven"] += 1
|
||||
if hold is not None:
|
||||
g["hold_sum"] += hold
|
||||
g["hold_n"] += 1
|
||||
|
||||
|
||||
def trade_stats(
|
||||
conn: sqlite3.Connection,
|
||||
account_id: Optional[int] = None,
|
||||
machine_id: Optional[int] = None,
|
||||
) -> dict[str, Any]:
|
||||
"""程序统计看板：盈亏/胜率/持仓时长分段/按机器与方向。"""
|
||||
clause, params = _trades_where(account_id, machine_id)
|
||||
sql = f"""
|
||||
SELECT t.id, t.side, t.symbol, t.profit, t.open_time, t.close_time,
|
||||
t.volume, t.machine_id, t.entry_style, m.name AS machine_name
|
||||
FROM trades t
|
||||
LEFT JOIN machines m ON m.id = t.machine_id
|
||||
{clause}
|
||||
ORDER BY COALESCE(t.close_time, t.open_time, t.created_at) DESC
|
||||
"""
|
||||
rows = [row_to_dict(r) for r in conn.execute(sql, tuple(params)).fetchall()]
|
||||
now_ts = now()
|
||||
|
||||
bucket_labels = {
|
||||
"lt_5m": "< 5 分钟",
|
||||
"m5_30": "5–30 分钟",
|
||||
"m30_2h": "30 分钟–2 小时",
|
||||
"ge_2h": "≥ 2 小时",
|
||||
"unknown": "时长未知",
|
||||
}
|
||||
by_hold = {k: _empty_group(v) for k, v in bucket_labels.items()}
|
||||
by_machine: dict[str, dict[str, Any]] = {}
|
||||
by_side: dict[str, dict[str, Any]] = {}
|
||||
|
||||
total = len(rows)
|
||||
closed = 0
|
||||
open_pos = 0
|
||||
wins = losses = breakeven = 0
|
||||
profit_sum = 0.0
|
||||
gross_win = 0.0
|
||||
gross_loss = 0.0
|
||||
hold_sum = 0.0
|
||||
hold_n = 0
|
||||
annotated = 0
|
||||
|
||||
for t in rows:
|
||||
profit = t.get("profit")
|
||||
hold = _hold_seconds(t.get("open_time"), t.get("close_time"), now_ts)
|
||||
closed_flag = t.get("close_time") not in (None, "")
|
||||
if closed_flag:
|
||||
closed += 1
|
||||
else:
|
||||
open_pos += 1
|
||||
|
||||
if profit is not None:
|
||||
p = float(profit)
|
||||
profit_sum += p
|
||||
if p > 1e-9:
|
||||
wins += 1
|
||||
gross_win += p
|
||||
elif p < -1e-9:
|
||||
losses += 1
|
||||
gross_loss += abs(p)
|
||||
else:
|
||||
breakeven += 1
|
||||
|
||||
if hold is not None:
|
||||
hold_sum += hold
|
||||
hold_n += 1
|
||||
|
||||
if (t.get("entry_style") or "").strip():
|
||||
annotated += 1
|
||||
|
||||
bk = _bucket_key(hold)
|
||||
_accumulate(by_hold[bk], profit, hold)
|
||||
|
||||
mid = t.get("machine_id")
|
||||
mkey = str(mid) if mid is not None else "none"
|
||||
if mkey not in by_machine:
|
||||
name = t.get("machine_name") or ("未绑定" if mid is None else f"#{mid}")
|
||||
by_machine[mkey] = _empty_group(name)
|
||||
_accumulate(by_machine[mkey], profit, hold)
|
||||
|
||||
side = (t.get("side") or "—").lower()
|
||||
if side not in by_side:
|
||||
by_side[side] = _empty_group("买入" if side == "buy" else ("卖出" if side == "sell" else side))
|
||||
_accumulate(by_side[side], profit, hold)
|
||||
|
||||
decided = wins + losses
|
||||
avg_hold = (hold_sum / hold_n) if hold_n else None
|
||||
return {
|
||||
"total": total,
|
||||
"closed": closed,
|
||||
"open_pos": open_pos,
|
||||
"wins": wins,
|
||||
"losses": losses,
|
||||
"breakeven": breakeven,
|
||||
"win_rate": (wins / decided * 100.0) if decided else 0.0,
|
||||
"profit": profit_sum,
|
||||
"avg_profit": (profit_sum / total) if total else 0.0,
|
||||
"gross_win": gross_win,
|
||||
"gross_loss": gross_loss,
|
||||
"profit_factor": (gross_win / gross_loss) if gross_loss > 1e-9 else (None if gross_win <= 0 else 999.0),
|
||||
"avg_hold_secs": avg_hold,
|
||||
"avg_hold": _fmt_hold_secs(avg_hold),
|
||||
"annotated": annotated,
|
||||
"unannotated": total - annotated,
|
||||
"by_hold": [_finalize_group(by_hold[k]) for k in ("lt_5m", "m5_30", "m30_2h", "ge_2h", "unknown") if by_hold[k]["count"] or k != "unknown"],
|
||||
"by_machine": sorted(
|
||||
[_finalize_group(g) for g in by_machine.values()],
|
||||
key=lambda x: x["profit"],
|
||||
reverse=True,
|
||||
),
|
||||
"by_side": [_finalize_group(g) for g in by_side.values()],
|
||||
}
|
||||
|
||||
|
||||
def get_trade(conn: sqlite3.Connection, trade_id: int) -> Optional[dict]:
|
||||
r = conn.execute(
|
||||
"""
|
||||
|
||||
Reference in New Issue
Block a user