#!/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]]: """按最新口径 A 重算历史:币盈亏×卖回价 − 权利金×(开仓价−卖回价).""" from lib.options.options_pnl_settle_lib import resolve_true_pnl_usdt_from_trade _ensure_columns(conn) rows = conn.execute( """ SELECT id, inst_id, premium_paid, premium_received, realized_pnl, open_index_px, close_index_px, budget_usdt, open_fee, close_fee, premium_ccy, 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"]) settled_u = _safe_float(r["realized_pnl_usdt"]) settled = int(r["pnl_settled"] or 0) == 1 initial = _safe_float(r["initial_premium_usdt"]) ccy = str(r["premium_ccy"] or "").strip().upper() or None 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) # 口径 B:历史币×卖回价(数据统计) 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) # 口径 A:权利金 USDT 波动式(忽略现货多买/账户差分) # 真盈亏必须同时有开仓价与卖回价;缺一则标记缺价未核(不用开仓价冒充卖回价) true_a, src = resolve_true_pnl_usdt_from_trade( premium_paid=paid, premium_received=recv, realized_pnl=coin_pnl, open_index_px=open_idx if open_idx and open_idx > 0 else None, close_index_px=close_idx if close_idx and close_idx > 0 else None, open_fee=r["open_fee"], close_fee=r["close_fee"], premium_ccy=ccy, ) action = "ok" new_settled_u = settled_u new_pnl_settled = 1 if settled else 0 status_label = "已平" if true_a is not None and src != "none": if settled_u is None or abs(float(settled_u) - float(true_a)) > 0.05: action = "rewrite_A_premium_move" new_settled_u = round(float(true_a), 4) new_pnl_settled = 1 else: action = "ok_premium_move" new_settled_u = round(float(true_a), 4) new_pnl_settled = 1 else: # 缺开仓价/卖回价/币盈亏等:无法按新公式更新 → 标记「缺价未核」 action = "mark_uncomputable_A" new_pnl_settled = 0 new_settled_u = None status_label = "缺价未核" 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, "pnl_source": src, "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, "premium_paid": paid, "premium_received": recv, "open_index_px": open_idx, "close_index_px": close_idx, } 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 == "mark_uncomputable_A": conn.execute( """ UPDATE options_trades SET pnl_settled=0, realized_pnl_usdt=NULL WHERE id=? """, (tid,), ) elif action in ("rewrite_A_premium_move", "ok_premium_move") and new_settled_u is not None: conn.execute( """ UPDATE options_trades SET realized_pnl_usdt=?, pnl_settled=1 WHERE id=? """, (new_settled_u, tid), ) # 复盘表同步最新 A;无法更新的打「缺价未核」 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, status_label, 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("action") not in ("ok", "ok_premium_move")] 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']} src={x.get('pnl_source')}" ) 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())