新增实例数据统计页:复盘 KPI 上、资金曲线下,系统设置可关。

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dekun
2026-10-07 15:28:25 +08:00
parent 48b919bb2b
commit 23bd59c87f
12 changed files with 666 additions and 50 deletions
+158
View File
@@ -0,0 +1,158 @@
"""实例资金曲线采样:现金余额(不含权利金/浮盈),供数据统计页使用."""
from __future__ import annotations
import datetime as dt
import os
import threading
import time
from typing import Any, Callable, Optional
_lock = threading.Lock()
_last_sample_at = 0.0
_thread: threading.Thread | None = None
_stop = threading.Event()
def cash_sample_interval_sec() -> float:
try:
return max(30.0, float(os.getenv("INSTANCE_CASH_SAMPLE_SEC") or 60))
except (TypeError, ValueError):
return 60.0
def cash_sample_keep_days() -> int:
try:
return max(7, int(float(os.getenv("INSTANCE_CASH_SAMPLE_KEEP_DAYS") or 90)))
except (TypeError, ValueError):
return 90
def init_cash_sample_table(conn) -> None:
conn.execute(
"""
CREATE TABLE IF NOT EXISTS instance_cash_samples (
ts_ms INTEGER PRIMARY KEY,
cash_usdt REAL NOT NULL
)
"""
)
def _safe_float(v: Any) -> Optional[float]:
if v is None or v == "":
return None
try:
return float(v)
except (TypeError, ValueError):
return None
def cash_usdt_from_balances(bal: dict[str, Any] | None) -> Optional[float]:
"""现金 = 交易账户 + 资金账户(优先 USDT,否则 USDC);不含权利金/浮盈."""
if not isinstance(bal, dict):
return None
parts = [_safe_float(bal.get("trading_usdt")), _safe_float(bal.get("funding_usdt"))]
if all(x is None for x in parts):
parts = [_safe_float(bal.get("trading_usdc")), _safe_float(bal.get("funding_usdc"))]
known = [x for x in parts if x is not None]
if not known:
return None
return round(sum(known), 4)
def record_cash_sample(conn, cash_usdt: float, *, ts_ms: int | None = None) -> None:
init_cash_sample_table(conn)
t = int(ts_ms if ts_ms is not None else time.time() * 1000)
conn.execute(
"""
INSERT INTO instance_cash_samples(ts_ms, cash_usdt)
VALUES (?, ?)
ON CONFLICT(ts_ms) DO UPDATE SET cash_usdt=excluded.cash_usdt
""",
(t, float(cash_usdt)),
)
keep_ms = cash_sample_keep_days() * 86400 * 1000
cutoff = t - keep_ms
conn.execute("DELETE FROM instance_cash_samples WHERE ts_ms < ?", (cutoff,))
def _range_start_ms(range_key: str) -> tuple[str, int]:
rk = (range_key or "7d").strip().lower()
now_ms = int(time.time() * 1000)
if rk in ("today", "1d"):
start = dt.datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)
return "today", int(start.timestamp() * 1000)
if rk in ("30d", "30"):
return "30d", now_ms - 30 * 86400 * 1000
if rk in ("all", "0"):
return "all", 0
return "7d", now_ms - 7 * 86400 * 1000
def query_cash_series_with_db(conn, *, range_key: str = "7d") -> dict[str, Any]:
rk, start_ms = _range_start_ms(range_key)
now_ms = int(time.time() * 1000)
init_cash_sample_table(conn)
rows = conn.execute(
"""
SELECT ts_ms, cash_usdt
FROM instance_cash_samples
WHERE ts_ms >= ?
ORDER BY ts_ms ASC
""",
(start_ms,),
).fetchall()
points = [{"t": int(r["ts_ms"]), "v": float(r["cash_usdt"])} for r in rows]
return {
"ok": True,
"range": rk,
"points": points,
"start_ms": start_ms,
"end_ms": now_ms,
"latest": points[-1]["v"] if points else None,
}
def sample_once(get_db: Callable, fetch_balances: Callable) -> bool:
"""拉取余额并写入采样;成功返回 True."""
global _last_sample_at
try:
bal = fetch_balances()
except Exception:
return False
cash = cash_usdt_from_balances(bal if isinstance(bal, dict) else None)
if cash is None:
return False
with _lock:
conn = get_db()
try:
record_cash_sample(conn, cash)
conn.commit()
finally:
conn.close()
_last_sample_at = time.time()
return True
def start_cash_sample_loop(get_db: Callable, fetch_balances: Callable) -> None:
global _thread
if _thread and _thread.is_alive():
return
_stop.clear()
def _loop() -> None:
# 启动后立刻采一笔,避免曲线长期空白
try:
sample_once(get_db, fetch_balances)
except Exception:
pass
while not _stop.is_set():
if _stop.wait(cash_sample_interval_sec()):
break
try:
sample_once(get_db, fetch_balances)
except Exception:
pass
_thread = threading.Thread(target=_loop, daemon=True, name="instance-cash-sample")
_thread.start()