#!/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() buys = conn.execute( """ SELECT amount, ts FROM sim_ledger_entries WHERE ccy='USDT' AND account='trading' AND kind='convert' AND amount < 0 AND note LIKE ? AND ts >= datetime(?, '-14 hour') AND ts <= datetime(?, '+14 hour') ORDER BY id """, (f"{u}/USDT%", open_ts, open_ts), ).fetchall() buy_usdt = 0.0 buy_coin = 0.0 for row in buys: 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%' ORDER BY id LIMIT 1 """, (u, row["ts"]), ).fetchone() if not coin_row: continue buy_usdt += abs(float(row["amount"])) buy_coin += float(coin_row["amount"]) sell_usdt = 0.0 sell_coin = 0.0 last_close = None if inst: last_close = conn.execute( """ SELECT ts FROM sim_ledger_entries WHERE kind='option_close' AND note LIKE ? ORDER BY id DESC LIMIT 1 """, (f"%{inst}%",), ).fetchone() if last_close or close_ts: close_anchor = str(last_close["ts"]) if last_close else str(close_ts) sells = conn.execute( """ SELECT amount, ts FROM sim_ledger_entries WHERE ccy=? AND account='trading' AND kind='convert' AND amount < 0 AND note LIKE ? AND ts >= ? ORDER BY id """, (u, f"{u}/USDT%", close_anchor), ).fetchall() for row in sells: usdt_row = conn.execute( """ SELECT amount FROM sim_ledger_entries WHERE ccy='USDT' AND account='trading' AND kind='convert' AND amount > 0 AND ts=? AND note LIKE 'from %' ORDER BY id LIMIT 1 """, (row["ts"],), ).fetchone() if not usdt_row: continue 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())