crypto_okx/scripts/repair_options_spot_vwap_from_sim_ledger.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""从模拟盘流水重建买/卖回 VWAP,并按真盈亏公式重算 realized_pnl_usdt.

用于 close_sell_* 漏记、开仓价被 budget/premium 抬飞的历史单.
默认 dry-run;加 --apply 才写库.

  python scripts/repair_options_spot_vwap_from_sim_ledger.py --db crypto.db
  python scripts/repair_options_spot_vwap_from_sim_ledger.py --db crypto.db --apply
"""
from __future__ import annotations

import argparse
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 _ledger_buy_sell(
    conn: sqlite3.Connection,
    *,
    underly: str,
    inst_id: str,
    open_ts: str,
    close_ts: str | None,
) -> tuple[float | None, float | None, float | None, float | None]:
    """在开仓前后窗口内汇总 USDT→币 与 平仓后币→USDT."""
    u = (underly or "").upper()
    inst = str(inst_id or "").strip()
    buy_usdt = 0.0
    buy_coin = 0.0
    opens = conn.execute(
        """
        SELECT id, ts FROM sim_ledger_entries
        WHERE kind='option_open' AND note LIKE ?
        ORDER BY id
        """,
        (f"%{inst}%",),
    ).fetchall() if inst else []
    if opens:
        open_id = int(opens[0]["id"])
        # 开仓前最近一笔 USDT→币(含缓冲买币)
        buy_row = conn.execute(
            """
            SELECT id, amount, ts FROM sim_ledger_entries
            WHERE ccy='USDT' AND account='trading' AND kind='convert'
              AND amount < 0 AND note LIKE ? AND id < ?
            ORDER BY id DESC LIMIT 1
            """,
            (f"{u}/USDT%", open_id),
        ).fetchone()
        if buy_row:
            coin_row = conn.execute(
                """
                SELECT amount FROM sim_ledger_entries
                WHERE ccy=? AND account='trading' AND kind='convert' AND amount > 0
                  AND ts=? AND note LIKE 'from usdt%' AND id < ?
                ORDER BY id DESC LIMIT 1
                """,
                (u, buy_row["ts"], open_id),
            ).fetchone()
            if coin_row:
                buy_usdt = abs(float(buy_row["amount"]))
                buy_coin = float(coin_row["amount"])
    sell_usdt = 0.0
    sell_coin = 0.0
    # 该合约 option_close 的 id 区间:本笔最后一笔 close → 下一笔任意 option_open/close 之前
    closes = conn.execute(
        """
        SELECT id, ts FROM sim_ledger_entries
        WHERE kind='option_close' AND note LIKE ?
        ORDER BY id
        """,
        (f"%{inst}%",),
    ).fetchall() if inst else []
    if closes:
        start_id = int(closes[-1]["id"])
        nxt = conn.execute(
            """
            SELECT id FROM sim_ledger_entries
            WHERE id > ? AND kind IN ('option_open','option_close')
            ORDER BY id LIMIT 1
            """,
            (start_id,),
        ).fetchone()
        end_id = int(nxt["id"]) if nxt else 10**12
        sells = conn.execute(
            """
            SELECT id, amount, ts FROM sim_ledger_entries
            WHERE ccy=? AND account='trading' AND kind='convert'
              AND amount < 0 AND note LIKE ?
              AND id > ? AND id < ?
            ORDER BY id
            """,
            (u, f"{u}/USDT%", start_id, end_id),
        ).fetchall()
        # 同秒多笔卖回时,按「币卖出 id → 下一笔 USDT 入账」配对,避免重复吃到同一笔 USDT
        used_usdt_ids: set[int] = set()
        for row in sells:
            sid = int(row["id"])
            usdt_row = conn.execute(
                """
                SELECT id, amount FROM sim_ledger_entries
                WHERE id = ? AND ccy='USDT' AND account='trading' AND kind='convert'
                  AND amount > 0 AND note LIKE 'from %'
                """,
                (sid + 1,),
            ).fetchone()
            if usdt_row is None or int(usdt_row["id"]) in used_usdt_ids:
                cands = conn.execute(
                    """
                    SELECT id, amount FROM sim_ledger_entries
                    WHERE ccy='USDT' AND account='trading' AND kind='convert'
                      AND amount > 0 AND note LIKE 'from %'
                      AND id > ? AND id < ?
                    ORDER BY id
                    """,
                    (sid, end_id),
                ).fetchall()
                usdt_row = next(
                    (r for r in cands if int(r["id"]) not in used_usdt_ids),
                    None,
                )
            if not usdt_row:
                continue
            uid = int(usdt_row["id"])
            used_usdt_ids.add(uid)
            sell_coin += abs(float(row["amount"]))
            sell_usdt += float(usdt_row["amount"])
    return (
        buy_usdt if buy_usdt > 0 else None,
        buy_coin if buy_coin > 0 else None,
        sell_usdt if sell_usdt > 0 else None,
        sell_coin if sell_coin > 0 else None,
    )


def repair(conn: sqlite3.Connection, *, apply: bool) -> list[dict[str, Any]]:
    sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
    from lib.options.options_pnl_settle_lib import true_pnl_usdt_premium_move
    from lib.options.options_spot_bridge_lib import ensure_bridge_table

    ensure_bridge_table(conn)
    rows = conn.execute(
        """
        SELECT id, inst_id, underlying, premium_paid, premium_received, realized_pnl,
               open_index_px, close_index_px, realized_pnl_usdt, bridge_id,
               created_at, closed_at, budget_usdt
        FROM options_trades
        WHERE status='closed'
        ORDER BY id
        """
    ).fetchall()
    report: list[dict[str, Any]] = []
    for r in rows:
        tid = int(r["id"])
        underly = str(r["underlying"] or (str(r["inst_id"] or "").split("-")[0]) or "").upper()
        open_ts = str(r["created_at"] or "")
        close_ts = str(r["closed_at"] or "") or None
        buy_u, buy_c, sell_u, sell_c = _ledger_buy_sell(
            conn,
            underly=underly,
            inst_id=str(r["inst_id"] or ""),
            open_ts=open_ts,
            close_ts=close_ts,
        )
        open_px = (buy_u / buy_c) if buy_u and buy_c and buy_c > 0 else None
        close_px = (sell_u / sell_c) if sell_u and sell_c and sell_c > 0 else None
        paid = _safe_float(r["premium_paid"])
        coin = _safe_float(r["realized_pnl"])
        old_open = _safe_float(r["open_index_px"])
        old_close = _safe_float(r["close_index_px"])
        old_u = _safe_float(r["realized_pnl_usdt"])
        new_u = true_pnl_usdt_premium_move(
            coin_pnl=coin,
            premium_paid=paid,
            open_index_px=open_px or old_open,
            sell_back_px=close_px or old_close,
        )
        item = {
            "id": tid,
            "inst_id": r["inst_id"],
            "buy_vwap": round(open_px, 4) if open_px else None,
            "sell_vwap": round(close_px, 4) if close_px else None,
            "old_open": old_open,
            "old_close": old_close,
            "old_u": old_u,
            "new_u": new_u,
            "bridge_id": r["bridge_id"],
            "sell_usdt": sell_u,
            "sell_coin": sell_c,
            "buy_usdt": buy_u,
            "buy_coin": buy_c,
        }
        report.append(item)
        if not apply:
            continue
        bid = r["bridge_id"]
        if bid is not None and buy_u and buy_c:
            conn.execute(
                """
                UPDATE options_spot_bridge
                SET buy_usdt_spent = ?, buy_coin_accum = ?
                WHERE id = ?
                """,
                (float(buy_u), float(buy_c), int(bid)),
            )
        if bid is not None and sell_u and sell_c:
            conn.execute(
                """
                UPDATE options_spot_bridge
                SET close_sell_usdt = ?, close_sell_coin = ?,
                    usdt_recovered = ?, coin_sold = ?
                WHERE id = ?
                """,
                (float(sell_u), float(sell_c), float(sell_u), float(sell_c), int(bid)),
            )
        conn.execute(
            """
            UPDATE options_trades
            SET open_index_px = COALESCE(?, open_index_px),
                close_index_px = COALESCE(?, close_index_px),
                realized_pnl_usdt = COALESCE(?, realized_pnl_usdt),
                initial_premium_usdt = CASE
                    WHEN ? IS NOT NULL AND premium_paid IS NOT NULL
                    THEN round(premium_paid * ?, 4)
                    ELSE initial_premium_usdt
                END,
                pnl_settled = 1
            WHERE id = ?
            """,
            (
                open_px,
                close_px,
                new_u,
                open_px,
                open_px,
                tid,
            ),
        )
    if apply:
        conn.commit()
        # 复盘行跟着刷
        try:
            from lib.options.options_review_lib import sync_options_from_local_trades

            sync_options_from_local_trades(conn, None)
            conn.commit()
        except Exception as e:
            print("review_sync_warn", e)
    return report


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--db", required=True)
    ap.add_argument("--apply", action="store_true")
    args = ap.parse_args()
    path = Path(args.db)
    if not path.is_file():
        print("db missing", path)
        return 1
    conn = sqlite3.connect(str(path))
    conn.row_factory = sqlite3.Row
    report = repair(conn, apply=bool(args.apply))
    conn.close()
    import json

    print(json.dumps(report, ensure_ascii=False, indent=2))
    print("apply" if args.apply else "dry-run", "rows", len(report))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())