diff --git a/lib/hedge_plan/hedge_plan_monitor_lib.py b/lib/hedge_plan/hedge_plan_monitor_lib.py index f6127f0..5e5bbf0 100644 --- a/lib/hedge_plan/hedge_plan_monitor_lib.py +++ b/lib/hedge_plan/hedge_plan_monitor_lib.py @@ -8,7 +8,11 @@ from typing import Any, Optional from lib.hedge_plan.hedge_plan_db import get_plan, get_plan_legs, list_plans, update_plan from lib.hedge_plan.hedge_plan_notify_lib import notify_hedge, notify_plan_end, build_hedge_alert_message from lib.hedge_plan.hedge_plan_orders_lib import _sell_option -from lib.hedge_plan.hedge_plan_settle_lib import leg_is_expired, settle_option_leg_at_spot +from lib.hedge_plan.hedge_plan_settle_lib import ( + leg_is_expired, + resolve_option_leg_realized_pnl, + settle_option_leg_at_spot, +) def _now() -> str: @@ -69,6 +73,7 @@ def tick_active_plans(cfg: dict[str, Any]) -> dict[str, Any]: return {"ok": False, "msg": "get_db missing"} conn = get_db() acted: list[dict[str, Any]] = [] + backfill_stats: dict[str, int] = {} try: from lib.hedge_plan.hedge_plan_db import init_hedge_plan_tables @@ -80,10 +85,22 @@ def tick_active_plans(cfg: dict[str, Any]) -> dict[str, Any]: acted.append(r) orphaned = _settle_orphaned_after_tp(cfg, conn) acted.extend(orphaned) + try: + ex = cfg.get("exchange_options") + if ex is not None: + from lib.exchange.okx_options_lib import fetch_all_option_positions_history + from lib.hedge_plan.hedge_plan_settle_lib import ( + backfill_hedge_option_legs_realized_pnl, + ) + + hist = fetch_all_option_positions_history(ex, limit=200) + backfill_stats = backfill_hedge_option_legs_realized_pnl(conn, hist) + except Exception: + pass conn.commit() finally: conn.close() - return {"ok": True, "acted": acted} + return {"ok": True, "acted": acted, "pnl_backfill": backfill_stats} def _notify_end_reload(cfg: dict[str, Any], conn: Any, plan_id: int) -> None: @@ -230,9 +247,10 @@ def _tick_po(cfg: dict[str, Any], conn: Any, plan: dict[str, Any], legs: list[di ask_open = _sf(opt.get("avg_open")) if bid is not None and ask_open is not None: ct = float(opt.get("ct_mult") or 0.01) - opt_pnl = (bid - ask_open) * float(opt.get("size") or 1) * ct + est = (bid - ask_open) * float(opt.get("size") or 1) * ct else: - opt_pnl = -premium + est = -premium + opt_pnl = _option_leg_pnl_after_close(cfg, opt, fallback=est) conn.execute( "UPDATE hedge_plan_legs SET status=?, close_reason=?, closed_at=?, realized_pnl=? WHERE id=?", ("closed", reason, _now(), opt_pnl, opt["id"]), @@ -284,6 +302,18 @@ def _tick_po(cfg: dict[str, Any], conn: Any, plan: dict[str, Any], legs: list[di return {"plan_id": plan["id"], "close_reason": reason, "total": total} +def _option_leg_pnl_after_close( + cfg: dict[str, Any], + leg: dict[str, Any], + *, + fallback: float, +) -> float: + """平仓后写腿盈亏:优先交易所历史,否则用估算.""" + ex = cfg.get("exchange_options") + pnl, _src = resolve_option_leg_realized_pnl(ex=ex, leg=leg, fallback=fallback) + return float(pnl if pnl is not None else fallback) + + def _estimate_leg_close_pnl(leg: dict[str, Any], idx: Optional[float], bid: Optional[float]) -> float: """残腿平仓盈亏估算:优先买一回收 − 权利金;无买一则用内在价值.""" premium = float(leg.get("premium") or 0) @@ -341,7 +371,8 @@ def _tick_oo_close_rest( update_plan(conn, int(plan["id"]), close_reason="oo_rest_closing") return {"plan_id": plan["id"], "msg": "残腿平仓失败", "close": close_r, "retry": True} bid = _sf(close_r.get("bid")) - pnl = _estimate_leg_close_pnl(leg, idx, bid) + est = _estimate_leg_close_pnl(leg, idx, bid) + pnl = _option_leg_pnl_after_close(cfg, leg, fallback=est) conn.execute( "UPDATE hedge_plan_legs SET status=?, close_reason=?, closed_at=?, realized_pnl=? WHERE id=?", ("closed", "oo_rest_close", _now(), round(pnl, 4), leg["id"]), @@ -418,9 +449,11 @@ def _tick_oo_target( ) return {"plan_id": plan["id"], "msg": "平盈利腿失败", "close": close_r} reason = "target_up_win_leg" if hit_side == "up" else "target_down_win_leg" + # 选腿用内在估算;落库优先交易所已实现盈亏 + closed_pnl = _option_leg_pnl_after_close(cfg, best, fallback=float(best_pnl)) conn.execute( "UPDATE hedge_plan_legs SET status=?, close_reason=?, closed_at=?, realized_pnl=? WHERE id=?", - ("closed", reason, _now(), best_pnl, best["id"]), + ("closed", reason, _now(), closed_pnl, best["id"]), ) rest_mode = resolve_oo_rest_close_mode(plan) update_plan(conn, int(plan["id"]), close_reason=reason) @@ -505,7 +538,8 @@ def _tick_oo_expiry( settled_sum = 0.0 for leg in pending: - pnl = settle_option_leg_at_spot(leg, float(spot)) + est = settle_option_leg_at_spot(leg, float(spot)) + pnl = _option_leg_pnl_after_close(cfg, leg, fallback=est) settled_sum += pnl conn.execute( "UPDATE hedge_plan_legs SET status=?, close_reason=?, closed_at=?, realized_pnl=? WHERE id=?", @@ -554,7 +588,11 @@ def _settle_orphaned_after_tp(cfg: dict[str, Any], conn: Any) -> list[dict[str, spot = _index_px(cfg, str(leg.get("underlying") or "ETH")) if spot is None: continue - pnl = settle_option_leg_at_spot(leg, float(spot)) + pnl_est = settle_option_leg_at_spot(leg, float(spot)) + # orphan row uses leg_id; map to id for resolver + leg_for_pnl = dict(leg) + leg_for_pnl["id"] = leg.get("leg_id") + pnl = _option_leg_pnl_after_close(cfg, leg_for_pnl, fallback=pnl_est) conn.execute( "UPDATE hedge_plan_legs SET status=?, close_reason=?, closed_at=?, realized_pnl=? WHERE id=?", ("closed", "expiry", _now(), round(pnl, 4), leg["leg_id"]), diff --git a/lib/hedge_plan/hedge_plan_settle_lib.py b/lib/hedge_plan/hedge_plan_settle_lib.py index a864e98..45af41c 100644 --- a/lib/hedge_plan/hedge_plan_settle_lib.py +++ b/lib/hedge_plan/hedge_plan_settle_lib.py @@ -2,9 +2,10 @@ from __future__ import annotations import time -from typing import Any, Optional +from datetime import datetime, timezone +from typing import Any, Callable, Optional -from lib.exchange.okx_options_lib import normalize_option_exp_ms +from lib.exchange.okx_options_lib import normalize_option_exp_ms, resolve_option_close_from_history from lib.hedge_plan.hedge_plan_calc_lib import option_expiry_pnl @@ -60,3 +61,148 @@ def all_option_legs_expired(legs: list[dict[str, Any]], *, now_ms: Optional[int] if not opts: return False return all(leg_is_expired(x, now_ms=now_ms) for x in opts) + + +def _parse_opened_ms(raw: Any) -> Optional[int]: + if raw is None or raw == "": + return None + s = str(raw).strip() + if not s: + return None + for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M:%f"): + try: + dt = datetime.strptime(s[:26], fmt).replace(tzinfo=timezone.utc) + return int(dt.timestamp() * 1000) + except ValueError: + continue + return None + + +def resolve_option_leg_realized_pnl( + *, + ex: Any = None, + leg: dict[str, Any], + fallback: Optional[float] = None, + fetch_history_fn: Optional[Callable[[str], list[dict[str, Any]]]] = None, + hist_rows: Optional[list[dict[str, Any]]] = None, +) -> tuple[Optional[float], str]: + """ + 期权腿已实现盈亏:优先 OKX positions-history realizedPnl. + 返回 (pnl, source) source=exchange|fallback|none. + """ + inst_id = str(leg.get("inst_id") or "").strip() + open_ms = _parse_opened_ms(leg.get("opened_at")) + rows = hist_rows + if rows is None and inst_id: + try: + if callable(fetch_history_fn): + rows = fetch_history_fn(inst_id) + elif ex is not None: + from lib.exchange.okx_options_lib import fetch_option_position_history + + rows = fetch_option_position_history(ex, inst_id) + except Exception: + rows = None + if rows: + info = resolve_option_close_from_history(rows, open_ms=open_ms) + pnl = _sf((info or {}).get("realized_pnl")) if info else None + if pnl is not None: + return round(float(pnl), 4), "exchange" + if fallback is not None: + return round(float(fallback), 4), "fallback" + return None, "none" + + +def backfill_hedge_option_legs_realized_pnl( + conn: Any, + hist_rows: list[dict[str, Any]], + *, + update_plan_fn: Optional[Callable[..., Any]] = None, +) -> dict[str, int]: + """用交易所历史覆盖已平期权腿盈亏,并重算已结束计划合计.""" + from lib.hedge_plan.hedge_plan_db import get_plan, get_plan_legs, update_plan + + by_inst: dict[str, list[dict[str, Any]]] = {} + for raw in hist_rows or []: + if not isinstance(raw, dict): + continue + inst = str(raw.get("instId") or "").strip() + if inst: + by_inst.setdefault(inst, []).append(raw) + + legs = conn.execute( + """ + SELECT * FROM hedge_plan_legs + WHERE status = 'closed' + AND inst_id IS NOT NULL AND TRIM(inst_id) != '' + AND (leg_role LIKE 'option%' OR opt_type IS NOT NULL) + ORDER BY id DESC + LIMIT 400 + """ + ).fetchall() + updated_legs = 0 + touched_plans: set[int] = set() + for row in legs: + leg = dict(row) + inst = str(leg.get("inst_id") or "").strip() + if not inst or inst not in by_inst: + continue + pnl, src = resolve_option_leg_realized_pnl( + leg=leg, + hist_rows=by_inst[inst], + fallback=None, + ) + if src != "exchange" or pnl is None: + continue + local = _sf(leg.get("realized_pnl")) + if local is not None and abs(local - pnl) < 1e-6: + continue + conn.execute( + "UPDATE hedge_plan_legs SET realized_pnl=? WHERE id=?", + (pnl, int(leg["id"])), + ) + updated_legs += 1 + touched_plans.add(int(leg["plan_id"])) + + updated_plans = 0 + updater = update_plan_fn or update_plan + for pid in touched_plans: + plan = get_plan(conn, pid) + if not plan or str(plan.get("status") or "") != "closed": + continue + plan_legs = get_plan_legs(conn, pid) + opt_sum = 0.0 + for lg in plan_legs: + role = str(lg.get("leg_role") or "") + if not (role.startswith("option") or lg.get("opt_type")): + continue + if str(lg.get("status") or "") != "closed": + continue + opt_sum += float(_sf(lg.get("realized_pnl")) or 0.0) + perp = float(_sf(plan.get("realized_pnl_perp")) or 0.0) + ptype = str(plan.get("plan_type") or "") + if ptype == "options_options": + total = opt_sum + kwargs: dict[str, Any] = { + "realized_pnl_options": round(opt_sum, 4), + "realized_pnl_total": round(total, 4), + } + else: + total = perp + opt_sum + kwargs = { + "realized_pnl_perp": round(perp, 4), + "realized_pnl_options": round(opt_sum, 4), + "realized_pnl_total": round(total, 4), + } + old_total = _sf(plan.get("realized_pnl_total")) + old_opts = _sf(plan.get("realized_pnl_options")) + if ( + old_total is not None + and abs(old_total - total) < 1e-6 + and old_opts is not None + and abs(old_opts - opt_sum) < 1e-6 + ): + continue + updater(conn, pid, **kwargs) + updated_plans += 1 + return {"legs": updated_legs, "plans": updated_plans} diff --git a/lib/options/options_review_lib.py b/lib/options/options_review_lib.py index 827f996..3353d6d 100644 --- a/lib/options/options_review_lib.py +++ b/lib/options/options_review_lib.py @@ -566,12 +566,16 @@ def ensure_local_review_synced( if backfill_exchange_pnl and ex is not None: try: from lib.exchange.okx_options_lib import fetch_all_option_positions_history + from lib.hedge_plan.hedge_plan_settle_lib import ( + backfill_hedge_option_legs_realized_pnl, + ) from lib.options.options_monitor_lib import ( backfill_closed_options_realized_pnl_from_history, ) hist = fetch_all_option_positions_history(ex, limit=200) backfill_closed_options_realized_pnl_from_history(conn, hist) + backfill_hedge_option_legs_realized_pnl(conn, hist) except Exception: pass return sync_all_review_sources(conn, from_exchange=False) diff --git a/tests/test_hedge_exchange_pnl.py b/tests/test_hedge_exchange_pnl.py new file mode 100644 index 0000000..aef97e3 --- /dev/null +++ b/tests/test_hedge_exchange_pnl.py @@ -0,0 +1,105 @@ +"""对冲计划期权腿盈亏与交易所对齐.""" +from __future__ import annotations + +import sqlite3 +import unittest + +from lib.hedge_plan.hedge_plan_db import get_plan, init_hedge_plan_tables, insert_leg, insert_plan +from lib.hedge_plan.hedge_plan_settle_lib import ( + backfill_hedge_option_legs_realized_pnl, + resolve_option_leg_realized_pnl, +) + + +class HedgeExchangePnlTest(unittest.TestCase): + def test_resolve_prefers_exchange(self): + leg = { + "inst_id": "ETH-USD_UM-260719-1850-P", + "opened_at": "2026-07-17 06:59:18", + "premium": 5.0, + } + hist = [ + { + "instId": "ETH-USD_UM-260719-1850-P", + "uTime": "1784400000000", + "realizedPnl": "12.00", + "closeAvgPx": "30", + } + ] + pnl, src = resolve_option_leg_realized_pnl( + leg=leg, hist_rows=hist, fallback=-5.0 + ) + self.assertEqual(src, "exchange") + self.assertEqual(pnl, 12.0) + + def test_backfill_updates_plan_total(self): + conn = sqlite3.connect(":memory:") + conn.row_factory = sqlite3.Row + init_hedge_plan_tables(conn) + pid = insert_plan( + conn, + { + "plan_type": "options_options", + "status": "closed", + "underlying": "ETH", + "premium_total": 10, + "realized_pnl_options": 6.08, + "realized_pnl_total": 6.08, + "opened_at": "2026-07-17 06:59:18", + "closed_at": "2026-07-19 16:00:13", + }, + ) + insert_leg( + conn, + { + "plan_id": pid, + "leg_role": "option_a", + "inst_id": "ETH-USD_UM-260719-1890-C", + "opt_type": "C", + "strike": 1890, + "size": 10, + "premium": 5.92, + "status": "closed", + "realized_pnl": -5.92, + "opened_at": "2026-07-17 06:59:18", + "closed_at": "2026-07-19 16:00:13", + }, + ) + insert_leg( + conn, + { + "plan_id": pid, + "leg_role": "option_b", + "inst_id": "ETH-USD_UM-260719-1850-P", + "opt_type": "P", + "strike": 1850, + "size": 10, + "premium": 4.0, + "status": "closed", + "realized_pnl": 12.0, + "opened_at": "2026-07-17 06:59:18", + "closed_at": "2026-07-19 08:00:00", + }, + ) + hist = [ + { + "instId": "ETH-USD_UM-260719-1890-C", + "uTime": "1784476813000", + "realizedPnl": "-5.50", + }, + { + "instId": "ETH-USD_UM-260719-1850-P", + "uTime": "1784448000000", + "realizedPnl": "11.80", + }, + ] + out = backfill_hedge_option_legs_realized_pnl(conn, hist) + self.assertEqual(out["legs"], 2) + self.assertEqual(out["plans"], 1) + plan = get_plan(conn, pid) + self.assertAlmostEqual(float(plan["realized_pnl_options"]), 6.3, places=4) + self.assertAlmostEqual(float(plan["realized_pnl_total"]), 6.3, places=4) + + +if __name__ == "__main__": + unittest.main()