dekun
104 lines
3.3 KiB
Python
104 lines
3.3 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Annotated, Any
|
|
|
|
from fastapi import APIRouter, Depends
|
|
|
|
from ..config import get_settings
|
|
from ..models.db import get_db
|
|
from ..sim.pnl import summarize_fills_pnl
|
|
from .auth import require_user
|
|
|
|
router = APIRouter(prefix="/api/stats", tags=["stats"])
|
|
|
|
|
|
def _current_loss_streak(curve: list[dict[str, Any]]) -> int:
|
|
"""从最近已平仓组往前数连续亏损次数。"""
|
|
streak = 0
|
|
for item in reversed(curve):
|
|
if float(item.get("realized_pnl") or 0) < 0:
|
|
streak += 1
|
|
else:
|
|
break
|
|
return streak
|
|
|
|
|
|
def build_stats_summary(db: Any | None = None) -> dict[str, Any]:
|
|
"""已平仓组汇总（策略页统计 / 中控 Fleet 共用）。"""
|
|
database = db or get_db()
|
|
s = get_settings()
|
|
mode = "LIVE" if not s.is_sim else "SIM"
|
|
rows = database.fetchall("SELECT * FROM groups WHERE status='closed'")
|
|
n = len(rows)
|
|
wins = sum(1 for r in rows if float(r["realized_pnl"] or 0) > 0)
|
|
total_pnl = sum(float(r["realized_pnl"] or 0) for r in rows)
|
|
pnls = [float(r["realized_pnl"] or 0) for r in rows]
|
|
max_single_loss = min(pnls) if pnls else 0.0
|
|
if max_single_loss > 0:
|
|
max_single_loss = 0.0
|
|
|
|
fees_perp = 0.0
|
|
fees_option = 0.0
|
|
total_slip = 0.0
|
|
for r in rows:
|
|
fills = database.fetchall(
|
|
"SELECT * FROM fills WHERE group_id=? ORDER BY id ASC",
|
|
(r["group_id"],),
|
|
)
|
|
summary = summarize_fills_pnl(list(fills))
|
|
fees_perp += float(summary.get("fees_perp") or 0)
|
|
fees_option += float(summary.get("fees_option") or 0)
|
|
# LIVE 不展示、不计入滑点；按组成交模式判断（可混有历史 SIM 组）
|
|
exec_mode = str(r["exec_mode"] or mode).upper()
|
|
if exec_mode != "LIVE":
|
|
total_slip += float(summary.get("slip_total") or 0)
|
|
|
|
total_fees = fees_perp + fees_option
|
|
reasons: dict[str, int] = {}
|
|
for r in rows:
|
|
k = str(r["close_reason"] or "unknown")
|
|
reasons[k] = reasons.get(k, 0) + 1
|
|
curve = [
|
|
{
|
|
"group_id": r["group_id"],
|
|
"realized_pnl": float(r["realized_pnl"] or 0),
|
|
"close_at_ms": r["close_at_ms"],
|
|
}
|
|
for r in sorted(rows, key=lambda x: int(x["close_at_ms"] or 0))
|
|
]
|
|
latest_funds = 0.0
|
|
try:
|
|
from ..sim.funds_wallets import FundsWallets
|
|
|
|
latest_funds = float(FundsWallets(database).total_usdt_equiv())
|
|
except Exception:
|
|
try:
|
|
from ..sim.ledger import Ledger
|
|
|
|
latest_funds = float(Ledger(database).snapshot().get("equity") or 0)
|
|
except Exception:
|
|
latest_funds = 0.0
|
|
|
|
return {
|
|
"mode": mode,
|
|
"show_slip": mode == "SIM",
|
|
"groups": n,
|
|
"wins": wins,
|
|
"win_rate": (wins / n) if n else 0.0,
|
|
"total_pnl": total_pnl,
|
|
"fees_perp": fees_perp,
|
|
"fees_option": fees_option,
|
|
"total_fees": total_fees,
|
|
"total_slip": total_slip if mode == "SIM" else 0.0,
|
|
"close_reasons": reasons,
|
|
"equity_curve": curve,
|
|
"latest_funds": latest_funds,
|
|
"max_single_loss": max_single_loss,
|
|
"loss_streak": _current_loss_streak(curve),
|
|
}
|
|
|
|
|
|
@router.get("/summary")
|
|
async def stats_summary(_user: Annotated[str, Depends(require_user)]) -> dict:
|
|
return build_stats_summary()
|