dekun/gold_monitor
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix stats dirty data: today filter, exclude synthetic closes, scrub frozen float PnL.

Browse Source 
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dekun
2026-10-21 19:28:01 +08:00
parent 391851a2b0
commit 1d2d4bbaff
3 changed files with 153 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/db.py
+118 -23
View File
@@ -6,10 +6,15 @@ import secrets
import sqlite3
import time
from contextlib import contextmanager
from datetime import datetime, timedelta, timezone
from typing import Any, Iterator, Optional
from app.config import Config
BJT = timezone(timedelta(hours=8))
# 对账/去重产生的假平仓，不计入盈亏统计
SYNTHETIC_CLOSE_EVENTS = frozenset({"reconcile_closed", "dedupe_closed"})
def _connect() -> sqlite3.Connection:
Config.DATABASE_PATH.parent.mkdir(parents=True, exist_ok=True)
@@ -125,6 +130,50 @@ def now() -> float:
return time.time()
def day_bounds_bjt(day: Optional[str] = None) -> tuple[float, float, str]:
"""北京时间某日 [start, end)；day=YYYY-MM-DD，默认今天。"""
if day:
d0 = datetime.strptime(day.strip(), "%Y-%m-%d").replace(
hour=0, minute=0, second=0, microsecond=0, tzinfo=BJT
)
else:
d0 = datetime.now(BJT).replace(hour=0, minute=0, second=0, microsecond=0)
d1 = d0 + timedelta(days=1)
return d0.timestamp(), d1.timestamp(), d0.strftime("%Y-%m-%d")
def _snapshot_event(snapshot_json: Any) -> str:
try:
snap = json.loads(snapshot_json or "{}")
except (TypeError, json.JSONDecodeError):
return ""
if not isinstance(snap, dict):
return ""
return str(snap.get("event") or "")
def is_synthetic_close(snapshot_json: Any) -> bool:
return _snapshot_event(snapshot_json) in SYNTHETIC_CLOSE_EVENTS
def scrub_synthetic_trade_profits(conn: sqlite3.Connection) -> int:
"""把历史假平仓上的「冻结浮盈」清掉，避免脏数据继续进统计。"""
rows = conn.execute(
"""
SELECT id, snapshot_json, profit FROM trades
WHERE close_time IS NOT NULL AND close_time != ''
AND profit IS NOT NULL
"""
).fetchall()
n = 0
for r in rows:
if not is_synthetic_close(r["snapshot_json"]):
continue
conn.execute("UPDATE trades SET profit=NULL WHERE id=?", (r["id"],))
n += 1
return n
def row_to_dict(row: Optional[sqlite3.Row]) -> Optional[dict]:
if row is None:
return None
@@ -449,19 +498,27 @@ def trade_stats(
conn: sqlite3.Connection,
account_id: Optional[int] = None,
machine_id: Optional[int] = None,
day: Optional[str] = None,
period: str = "today",
) -> 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
t.volume, t.machine_id, t.entry_style, t.snapshot_json,
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()]
raw_rows = [row_to_dict(r) for r in conn.execute(sql, tuple(params)).fetchall()]
now_ts = now()
period = (period or "today").strip().lower()
day_label = ""
start_ts = end_ts = None
if period != "all":
start_ts, end_ts, day_label = day_bounds_bjt(day)
bucket_labels = {
"lt_5m": "< 5 分钟",
@@ -478,6 +535,33 @@ def trade_stats(
}
by_style: dict[str, dict[str, Any]] = {s: _empty_group(s) for s in ENTRY_STYLES}
rows: list[dict] = []
skipped_synthetic = 0
for t in raw_rows:
closed_flag = t.get("close_time") not in (None, "")
if closed_flag and is_synthetic_close(t.get("snapshot_json")):
skipped_synthetic += 1
continue
if period != "all" and start_ts is not None and end_ts is not None:
# 今日：按平仓时间；仍持仓的按开仓时间计入「持仓中」但不进合计盈亏
if closed_flag:
ct = t.get("close_time")
try:
cts = float(ct)
except (TypeError, ValueError):
continue
if cts < start_ts or cts >= end_ts:
continue
else:
ot = t.get("open_time") or t.get("created_at")
try:
ots = float(ot)
except (TypeError, ValueError):
continue
if ots < start_ts or ots >= end_ts:
continue
rows.append(t)
total = len(rows)
closed = 0
open_pos = 0
@@ -488,19 +572,23 @@ def trade_stats(
hold_sum = 0.0
hold_n = 0
annotated = 0
scored = 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, "")
hold = _hold_seconds(t.get("open_time"), t.get("close_time"), now_ts)
# 合计盈亏/胜率只统计真实已平仓；持仓浮盈不进合计
profit = t.get("profit") if closed_flag else None
if closed_flag:
closed += 1
else:
open_pos += 1
if profit is not None:
if closed_flag and profit is not None:
p = float(profit)
profit_sum += p
scored += 1
if p > 1e-9:
wins += 1
gross_win += p
@@ -510,7 +598,7 @@ def trade_stats(
else:
breakeven += 1
if hold is not None:
if closed_flag and hold is not None:
hold_sum += hold
hold_n += 1
@@ -518,30 +606,34 @@ def trade_stats(
if style:
annotated += 1
# 分组也只累加已平仓盈亏，避免浮盈干扰分类
cat = entry_category(style)
_accumulate(by_category[cat], profit, hold)
_accumulate(by_category[cat], profit, hold if closed_flag else None)
if style in by_style:
_accumulate(by_style[style], profit, hold)
_accumulate(by_style[style], profit, hold if closed_flag else None)
elif style:
key = f"旧 · {style}"
if key not in by_style:
by_style[key] = _empty_group(key)
_accumulate(by_style[key], profit, hold)
_accumulate(by_style[key], profit, hold if closed_flag else None)
bk = _bucket_key(hold)
_accumulate(by_hold[bk], profit, hold)
if closed_flag:
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)
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)
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
@@ -555,7 +647,7 @@ def trade_stats(
"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,
"avg_profit": (profit_sum / scored) if scored 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),
@@ -563,6 +655,9 @@ def trade_stats(
"avg_hold": _fmt_hold_secs(avg_hold),
"annotated": annotated,
"unannotated": total - annotated,
"skipped_synthetic": skipped_synthetic,
"period": period,
"day": day_label,
"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()],