"""实例资金曲线采样:权益(现金+现货币+权利金+浮盈),供数据统计页使用.""" 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 ) """ ) try: conn.execute("ALTER TABLE instance_cash_samples ADD COLUMN equity_usdt REAL") except Exception: pass 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 spot_coin_usdt_from_balances( bal: dict[str, Any] | None, *, btc_px: float | None = None, eth_px: float | None = None, ) -> float: """交易+资金账户 BTC/ETH 现货按指数折 U(缓冲买币占用).""" if not isinstance(bal, dict): return 0.0 total = 0.0 btc = (_safe_float(bal.get("trading_btc")) or 0.0) + ( _safe_float(bal.get("funding_btc")) or 0.0 ) eth = (_safe_float(bal.get("trading_eth")) or 0.0) + ( _safe_float(bal.get("funding_eth")) or 0.0 ) if btc > 0 and btc_px is not None and btc_px > 0: total += btc * float(btc_px) if eth > 0 and eth_px is not None and eth_px > 0: total += eth * float(eth_px) return round(total, 4) def option_premium_and_upl_usdt(positions: list[dict[str, Any]] | None) -> tuple[float, float]: """持仓权利金U + 浮盈U(权利金+浮盈≈期权市值).""" prem = 0.0 upl = 0.0 if not positions: return 0.0, 0.0 from lib.options.options_positions_lib import ( display_pnl_usdt_from_option_row, premium_paid_usdt_from_option_row, ) for p in positions: if not isinstance(p, dict): continue try: pu = premium_paid_usdt_from_option_row(p) except Exception: pu = None if pu is not None: prem += float(pu) try: du = display_pnl_usdt_from_option_row(p) except Exception: du = None if du is not None: upl += float(du) return round(prem, 4), round(upl, 4) def equity_usdt_from_snapshot( bal: dict[str, Any] | None, positions: list[dict[str, Any]] | None = None, *, btc_px: float | None = None, eth_px: float | None = None, ) -> Optional[float]: """权益 = 稳定币现金 + 现货币折U + 权利金U + 浮盈U.""" cash = cash_usdt_from_balances(bal) spot = spot_coin_usdt_from_balances(bal, btc_px=btc_px, eth_px=eth_px) prem, upl = option_premium_and_upl_usdt(positions) if cash is None and spot <= 0 and prem == 0 and upl == 0: return None return round(float(cash or 0.0) + float(spot) + float(prem) + float(upl), 4) def record_cash_sample( conn, cash_usdt: float, *, equity_usdt: float | None = None, 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) eq = float(equity_usdt) if equity_usdt is not None else float(cash_usdt) conn.execute( """ INSERT INTO instance_cash_samples(ts_ms, cash_usdt, equity_usdt) VALUES (?, ?, ?) ON CONFLICT(ts_ms) DO UPDATE SET cash_usdt=excluded.cash_usdt, equity_usdt=excluded.equity_usdt """, (t, float(cash_usdt), eq), ) 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 _closed_at_to_ms(raw: Any) -> int | None: if raw is None or raw == "": return None try: from lib.options.options_stats_lib import _parse_ts d = _parse_ts(raw) if d is None: return None return int(d.timestamp() * 1000) except Exception: return None def _trade_realized_usdt(row: Any) -> float | None: u = _safe_float(row["realized_pnl_usdt"] if hasattr(row, "keys") else None) if u is not None: return round(float(u), 4) # 兼容未冻结 U:币盈亏×平仓价 try: coin = _safe_float(row["realized_pnl"]) px = _safe_float(row["close_index_px"]) except Exception: return None if coin is None or px is None or px <= 0: return None return round(float(coin) * float(px), 4) def query_realized_pnl_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) try: from lib.options.options_db import init_options_tables init_options_tables(conn) except Exception: pass rows = conn.execute( """ SELECT id, closed_at, created_at, realized_pnl_usdt, realized_pnl, close_index_px FROM options_trades WHERE status = 'closed' ORDER BY COALESCE(closed_at, created_at) ASC, id ASC """ ).fetchall() events: list[tuple[int, float]] = [] for r in rows: pnl_u = _trade_realized_usdt(r) if pnl_u is None: continue ts = _closed_at_to_ms(r["closed_at"]) or _closed_at_to_ms(r["created_at"]) if ts is None: continue events.append((ts, float(pnl_u))) cum = 0.0 all_points: list[dict[str, Any]] = [] for ts, pnl in events: cum = round(cum + pnl, 4) all_points.append({"t": int(ts), "v": cum}) baseline = 0.0 series: list[dict[str, Any]] = [] for p in all_points: if int(p["t"]) < start_ms: baseline = float(p["v"]) continue if not series: series.append({"t": int(start_ms), "v": baseline}) series.append({"t": int(p["t"]), "v": float(p["v"])}) if not series: # 窗口内无新平仓:画水平线(历史累计或 0) series = [ {"t": int(start_ms), "v": baseline}, {"t": int(now_ms), "v": baseline}, ] else: last_v = float(series[-1]["v"]) if int(series[-1]["t"]) < now_ms: series.append({"t": int(now_ms), "v": last_v}) return { "ok": True, "range": rk, "metric": "realized_pnl", "points": series, "start_ms": start_ms, "end_ms": now_ms, "latest": series[-1]["v"] if series else None, "trade_count": len(events), } def query_cash_series_with_db(conn, *, range_key: str = "7d") -> dict[str, Any]: """兼容旧名:数据统计曲线改为累计已实现盈亏.""" return query_realized_pnl_series_with_db(conn, range_key=range_key) def sample_once(get_db: Callable, fetch_snapshot: Callable) -> bool: """拉取快照并写入采样. fetch_snapshot 可返回: - float: 视为权益(兼容旧现金回调) - dict: {cash_usdt, equity_usdt} """ global _last_sample_at try: snap = fetch_snapshot() except Exception: return False cash: float | None equity: float | None if isinstance(snap, dict): cash = _safe_float(snap.get("cash_usdt")) equity = _safe_float(snap.get("equity_usdt")) else: cash = _safe_float(snap) equity = cash if equity is None and cash is None: return False if cash is None: cash = float(equity or 0.0) if equity is None: equity = float(cash) with _lock: conn = get_db() try: record_cash_sample(conn, float(cash), equity_usdt=float(equity)) conn.commit() finally: conn.close() _last_sample_at = time.time() return True def start_cash_sample_loop(get_db: Callable, fetch_snapshot: Callable) -> None: global _thread if _thread and _thread.is_alive(): return _stop.clear() def _loop() -> None: try: sample_once(get_db, fetch_snapshot) except Exception: pass while not _stop.is_set(): if _stop.wait(cash_sample_interval_sec()): break try: sample_once(get_db, fetch_snapshot) except Exception: pass _thread = threading.Thread(target=_loop, daemon=True, name="instance-cash-sample") _thread.start()