crypto_okx/scripts/repair_options_dual_pnl.py

#!/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 波动式;开/卖回价优先桥多笔 VWAP
        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,
            conn=conn,
            bridge_id=r["bridge_id"],
        )

        action = "ok"
        new_settled_u = settled_u
        new_pnl_settled = 1 if settled else 0

        status_label = "已平"
        if src == "pending_sell":
            action = "mark_pending_sell"
            new_pnl_settled = 0
            new_settled_u = None
            status_label = "待卖回"
        elif 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 in ("mark_uncomputable_A", "mark_pending_sell"):
            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())