#!/usr/bin/env python3 # -*- coding: utf-8 -*- """修复期权双口径历史数据(规格 docs/规格-期权交易账户盈亏结算.md §9). 默认 dry-run,打印将要修改的行;加 --apply 才写库. 用法: python scripts/repair_options_dual_pnl.py --db path/to/instance.db python scripts/repair_options_dual_pnl.py --db path/to/instance.db --apply python scripts/repair_options_dual_pnl.py --db path/to/instance.db --csv out.csv --apply """ from __future__ import annotations import argparse import csv import sqlite3 import sys from pathlib import Path from typing import Any def _safe_float(v: Any) -> float | None: if v is None or v == "": return None try: return float(v) except (TypeError, ValueError): return None def _connect(db_path: Path) -> sqlite3.Connection: conn = sqlite3.connect(str(db_path)) conn.row_factory = sqlite3.Row return conn def _ensure_columns(conn: sqlite3.Connection) -> None: # 与业务库迁移一致,可重复执行 for ddl in ( "ALTER TABLE options_trades ADD COLUMN initial_premium_usdt REAL", "ALTER TABLE options_trades ADD COLUMN trading_usdt_before REAL", "ALTER TABLE options_trades ADD COLUMN trading_usdt_after REAL", "ALTER TABLE options_trades ADD COLUMN pnl_settled INTEGER DEFAULT 0", "ALTER TABLE options_review_trades ADD COLUMN initial_premium_usdt REAL", "ALTER TABLE options_review_trades ADD COLUMN profit_rate REAL", ): try: conn.execute(ddl) except sqlite3.OperationalError: pass def repair(conn: sqlite3.Connection, *, apply: bool) -> list[dict[str, Any]]: _ensure_columns(conn) rows = conn.execute( """ SELECT id, inst_id, premium_paid, premium_received, realized_pnl, open_index_px, close_index_px, budget_usdt, trading_usdt_before, trading_usdt_after, realized_pnl_usdt, pnl_settled, bridge_id, initial_premium_usdt FROM options_trades WHERE status='closed' ORDER BY id """ ).fetchall() report: list[dict[str, Any]] = [] for r in rows: tid = int(r["id"]) paid = _safe_float(r["premium_paid"]) recv = _safe_float(r["premium_received"]) coin_pnl = _safe_float(r["realized_pnl"]) open_idx = _safe_float(r["open_index_px"]) close_idx = _safe_float(r["close_index_px"]) before = _safe_float(r["trading_usdt_before"]) after = _safe_float(r["trading_usdt_after"]) settled_u = _safe_float(r["realized_pnl_usdt"]) settled = int(r["pnl_settled"] or 0) == 1 initial = _safe_float(r["initial_premium_usdt"]) new_initial = initial if new_initial is None and open_idx and paid and open_idx > 0 and paid > 0: new_initial = round(open_idx * paid, 4) stats_u = None if coin_pnl is not None and close_idx is not None and close_idx > 0: stats_u = round(coin_pnl * close_idx, 4) elif coin_pnl is not None and open_idx is not None and open_idx > 0: stats_u = round(coin_pnl * open_idx, 4) # 错账:币盈亏为正、账户冻结为负(典型卖回封顶) suspicious = bool( settled and coin_pnl is not None and coin_pnl > 0 and settled_u is not None and settled_u < 0 ) # 桥侧:coin_sold ≈ premium 且远小于 recv → 未卖盈利币 bridge_hint = "" bid = r["bridge_id"] coin_bought = coin_sold = None if bid is not None: try: b = conn.execute( "SELECT coin_bought, coin_sold FROM options_spot_bridge WHERE id=?", (int(bid),), ).fetchone() if b: coin_bought = _safe_float(b["coin_bought"]) coin_sold = _safe_float(b["coin_sold"]) if ( paid and coin_sold is not None and recv and coin_sold <= paid * 1.05 and recv > paid * 1.2 ): bridge_hint = "sell_capped_at_premium" suspicious = True except Exception: pass action = "ok" new_settled_u = settled_u new_pnl_settled = 1 if settled else 0 if suspicious: # 无法可靠重建 after 时解除错误冻结,待卖回后重结算 action = "unsettle_bad_account_freeze" new_pnl_settled = 0 new_settled_u = None elif before is not None and after is not None and not settled: action = "settle_from_snapshots" new_settled_u = round(after - before, 4) new_pnl_settled = 1 profit_rate = None a_for_rate = new_settled_u if new_pnl_settled == 1 else None if a_for_rate is not None and new_initial and abs(new_initial) > 1e-12: profit_rate = round(a_for_rate / new_initial, 6) rec = { "trade_id": tid, "inst_id": r["inst_id"], "action": action, "suspicious": suspicious, "bridge_hint": bridge_hint, "coin_pnl": coin_pnl, "stats_pnl_usdt_B": stats_u, "old_realized_pnl_usdt_A": settled_u, "new_realized_pnl_usdt_A": new_settled_u, "old_pnl_settled": int(r["pnl_settled"] or 0), "new_pnl_settled": new_pnl_settled, "initial_premium_usdt": new_initial, "profit_rate": profit_rate, "coin_bought": coin_bought, "coin_sold": coin_sold, "premium_paid": paid, "premium_received": recv, } report.append(rec) if not apply: continue if new_initial is not None and initial != new_initial: conn.execute( "UPDATE options_trades SET initial_premium_usdt=? WHERE id=?", (new_initial, tid), ) if action == "unsettle_bad_account_freeze": conn.execute( """ UPDATE options_trades SET pnl_settled=0, realized_pnl_usdt=NULL, trading_usdt_after=NULL WHERE id=? AND COALESCE(pnl_settled,0)=1 """, (tid,), ) elif action == "settle_from_snapshots" and new_settled_u is not None: conn.execute( """ UPDATE options_trades SET realized_pnl_usdt=?, pnl_settled=1 WHERE id=? AND COALESCE(pnl_settled,0)=0 """, (new_settled_u, tid), ) # 复盘表 hk = f"local_opt:{tid}" rev = conn.execute( "SELECT id FROM options_review_trades WHERE history_key=?", (hk,) ).fetchone() if rev: pnl_total = new_settled_u if new_pnl_settled == 1 else None conn.execute( """ UPDATE options_review_trades SET initial_premium_usdt=COALESCE(?, initial_premium_usdt), realized_pnl_total=?, realized_pnl=?, profit_rate=?, status_raw=? WHERE history_key=? """, ( new_initial, pnl_total, pnl_total, profit_rate, "待结算" if new_pnl_settled == 0 and before is not None else "已平", hk, ), ) # 实盘/统计冻结表:口径 B if stats_u is not None: try: conn.execute( """ INSERT INTO options_exchange_pnl_freeze( history_key, realized_pnl, realized_pnl_usdt, close_index_px, premium_ccy, updated_at ) VALUES (?, ?, ?, ?, ?, datetime('now')) ON CONFLICT(history_key) DO UPDATE SET realized_pnl=excluded.realized_pnl, realized_pnl_usdt=excluded.realized_pnl_usdt, close_index_px=excluded.close_index_px, updated_at=excluded.updated_at """, ( hk, coin_pnl, stats_u, close_idx or open_idx, "BTC" if "BTC" in str(r["inst_id"] or "") else "ETH", ), ) except Exception: pass if apply: conn.commit() return report def main() -> int: ap = argparse.ArgumentParser(description="修复期权双口径历史盈亏") ap.add_argument("--db", required=True, help="实例 SQLite 路径") ap.add_argument("--apply", action="store_true", help="实际写库(默认 dry-run)") ap.add_argument("--csv", default="", help="导出报告 CSV 路径") args = ap.parse_args() db_path = Path(args.db) if not db_path.is_file(): print(f"db not found: {db_path}", file=sys.stderr) return 2 conn = _connect(db_path) try: report = repair(conn, apply=bool(args.apply)) finally: conn.close() bad = [x for x in report if x.get("suspicious") or x.get("action") != "ok"] print(f"scanned={len(report)} need_attention={len(bad)} apply={bool(args.apply)}") for x in bad[:50]: print( f" #{x['trade_id']} {x['inst_id']} action={x['action']} " f"A:{x['old_realized_pnl_usdt_A']}->{x['new_realized_pnl_usdt_A']} " f"B={x['stats_pnl_usdt_B']} hint={x['bridge_hint']}" ) if args.csv: out = Path(args.csv) if report: with out.open("w", newline="", encoding="utf-8-sig") as f: w = csv.DictWriter(f, fieldnames=list(report[0].keys())) w.writeheader() w.writerows(report) print(f"wrote {out}") return 0 if __name__ == "__main__": raise SystemExit(main())