Add fleet stats table with funds, fees, loss streak, and totals.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dekun
2026-08-02 12:25:30 +08:00
parent 5982cb7414
commit 8745d8e6d1
5 changed files with 298 additions and 2 deletions
+31
View File
@@ -12,6 +12,17 @@ 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()
@@ -21,6 +32,10 @@ def build_stats_summary(db: Any | None = None) -> dict[str, Any]:
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
@@ -51,6 +66,19 @@ def build_stats_summary(db: Any | None = None) -> dict[str, Any]:
}
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",
@@ -64,6 +92,9 @@ def build_stats_summary(db: Any | None = None) -> dict[str, Any]:
"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),
}
+28
View File
@@ -0,0 +1,28 @@
"""统计汇总:最大单笔亏损、连亏次数。"""
from __future__ import annotations
from app.api.stats import _current_loss_streak
def test_current_loss_streak() -> None:
assert _current_loss_streak([]) == 0
assert (
_current_loss_streak(
[
{"realized_pnl": 10},
{"realized_pnl": -1},
{"realized_pnl": -2},
]
)
== 2
)
assert (
_current_loss_streak(
[
{"realized_pnl": -5},
{"realized_pnl": 1},
]
)
== 0
)