"""实例资金曲线采样:现金余额(不含权利金/浮盈),供数据统计页使用.""" 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()