"""对冲计划结算辅助:到期内在价值与期权腿收口.""" from __future__ import annotations import os import time from datetime import datetime from typing import Any, Callable, Optional from zoneinfo import ZoneInfo 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 _APP_TZ = ZoneInfo((os.getenv("APP_TIMEZONE") or os.getenv("TZ") or "Asia/Shanghai").strip() or "Asia/Shanghai") def _sf(v: Any) -> Optional[float]: try: if v is None or v == "": return None return float(v) except (TypeError, ValueError): return None def leg_exp_ms(leg: dict[str, Any]) -> Optional[int]: return normalize_option_exp_ms(leg.get("exp_time"), str(leg.get("inst_id") or "")) def leg_is_expired(leg: dict[str, Any], *, now_ms: Optional[int] = None) -> bool: exp = leg_exp_ms(leg) if exp is None: return False now = int(now_ms if now_ms is not None else time.time() * 1000) return now >= int(exp) def settle_option_leg_at_spot(leg: dict[str, Any], spot: float) -> float: """按到期结算口径估算腿盈亏(USDC).""" premium = float(leg.get("premium") or 0) strike = _sf(leg.get("strike")) if strike is None: return -premium sheets = float(leg.get("size") or 1) # ct_mult 未入库时默认 0.01 ct = float(leg.get("ct_mult") or 0.01) return float( option_expiry_pnl( opt_type=str(leg.get("opt_type") or "P"), strike=float(strike), spot=float(spot), sheets=sheets, ct_mult=ct, premium_paid=premium, ) ) def all_option_legs_expired(legs: list[dict[str, Any]], *, now_ms: Optional[int] = None) -> bool: opts = [ x for x in legs if str(x.get("leg_role") or "").startswith("option") and str(x.get("status") or "") in ("open", "hold_to_expiry") ] 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]: """墙钟开仓时间 → UTC ms.库内时间为业务时区(默认 Asia/Shanghai),不可当 UTC.""" if raw is None or raw == "": return None s = str(raw).strip() if not s: return None for fmt, ln in (("%Y-%m-%d %H:%M:%S", 19), ("%Y-%m-%d %H:%M:%f", 26), ("%Y-%m-%d %H:%M", 16)): try: dt = datetime.strptime(s[:ln], fmt).replace(tzinfo=_APP_TZ) 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: close_ms = _parse_opened_ms(leg.get("closed_at")) sheets = _sf(leg.get("size")) or _sf(leg.get("sheets")) info = resolve_option_close_from_history( rows, open_ms=open_ms, close_ms=close_ms, sheets=sheets ) 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}