"""对冲计划监控:永期 TP/SL、期期目标价、到期结算与微信收口推送.""" from __future__ import annotations import os from datetime import datetime, timezone 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, resolve_option_leg_realized_pnl, settle_option_leg_at_spot, ) def _now() -> str: return datetime.now(timezone.utc).astimezone().strftime("%Y-%m-%d %H:%M:%S") def _env_bool(key: str, default: bool = False) -> bool: raw = (os.getenv(key) or "").strip().lower() if not raw: return default return raw in ("1", "true", "yes", "on") def _sf(v: Any) -> Optional[float]: try: if v is None or v == "": return None return float(v) except (TypeError, ValueError): return None def _perp_live_contracts(cfg: dict[str, Any], symbol: str, direction: str) -> Optional[float]: fn = cfg.get("get_live_position_contracts") if not callable(fn): return None try: return fn(symbol, direction) except Exception: return None def _index_px(cfg: dict[str, Any], underlying: str) -> Optional[float]: ex = cfg.get("exchange_options") fn = cfg.get("fetch_index_price") if callable(fn) and ex is not None: try: return fn(ex, underlying) except Exception: return None # 无期权账户时回退永续 ticker ex_perp = cfg.get("exchange") if ex_perp is not None: try: base = (underlying or "ETH").upper() sym = f"{base}/USDT:USDT" t = ex_perp.fetch_ticker(sym) return _sf((t.get("info") or {}).get("markPx")) or _sf(t.get("last")) except Exception: return None return None def _plan_open_grace_sec() -> float: try: return max(0.0, float(os.getenv("HEDGE_PLAN_OPEN_GRACE_SEC") or "90")) except (TypeError, ValueError): return 90.0 def _within_open_grace(plan: dict[str, Any]) -> bool: """开仓后宽限期:仓位尚未同步到交易所时禁止按「已平」收口.""" grace = _plan_open_grace_sec() if grace <= 0: return False raw = str(plan.get("opened_at") or plan.get("created_at") or "").strip() if not raw: return True try: # "YYYY-MM-DD HH:MM:SS" 本地墙钟 opened = datetime.strptime(raw[:19], "%Y-%m-%d %H:%M:%S") age = (datetime.now() - opened).total_seconds() return age < grace except Exception: return True def _classify_po_flat_reason( *, direction: str, entry: float, mark: Optional[float], tp: Optional[float], sl: Optional[float], ) -> str: """永续已平时分类 TP/SL.歧义时偏 SL(触发强平期权),避免误判 TP 跳过强平.""" d = (direction or "long").lower() if mark is None or not entry: return "perp_flat_unknown" if sl is not None: if d == "long" and mark <= sl: return "perp_sl" if d == "short" and mark >= sl: return "perp_sl" if tp is not None: if d == "long" and mark >= tp: return "perp_tp" if d == "short" and mark <= tp: return "perp_tp" if sl is not None and tp is not None: return "perp_sl" if abs(mark - sl) <= abs(mark - tp) else "perp_tp" if sl is not None: return "perp_sl" if tp is not None: return "perp_tp" return "perp_flat_unknown" def tick_active_plans(cfg: dict[str, Any]) -> dict[str, Any]: """扫描 active/partial 计划 + 止盈后遗留期权到期收口.返回处理摘要.""" get_db = cfg.get("get_db") if not callable(get_db): 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 init_hedge_plan_tables(conn) plans = list_plans(conn, status="watching", limit=20) plans.extend(list_plans(conn, status="active", limit=40)) # partial:裸永续/半腿也需侦测永续 TP/SL plans.extend(list_plans(conn, status="partial", limit=20)) seen: set[int] = set() for plan in plans: pid = int(plan.get("id") or 0) if pid in seen: continue seen.add(pid) r = _tick_one(cfg, conn, plan) if r: 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, "pnl_backfill": backfill_stats} def _notify_end_reload(cfg: dict[str, Any], conn: Any, plan_id: int) -> None: plan = get_plan(conn, int(plan_id)) if plan: notify_plan_end(cfg, conn, plan) def resolve_oo_rest_close_mode(plan: dict[str, Any]) -> str: """盈利腿平后另一腿:close_all(残值平) / hold_expiry(到期平). - 方案C关闭 → 强制到期平 - 计划未写 oo_close_mode(旧单) → 到期平,避免误清残腿 - 新开仓默认写入 close_all(残值平:权利金≤初始20%且有买一) """ if not _env_bool("HEDGE_PLAN_OO_CLOSE_MODE_ENABLED", True): return "hold_expiry" raw = plan.get("oo_close_mode") if raw is None or str(raw).strip() == "": return "hold_expiry" v = str(raw).strip().lower() if v in ("hold_expiry", "hold_to_expiry", "expiry", "到期平"): return "hold_expiry" return "close_all" # 期期亏损腿残值平:当前买一回收 ≤ 本合约初始权利金 × 该比例 OO_LOSS_LEG_RESIDUAL_RATIO = 0.20 # 期期默认盈亏比:盈利金额 / 总权利金 OO_DEFAULT_PROFIT_RR = 2.0 def _oo_option_legs(legs: list[dict[str, Any]], *, statuses: tuple[str, ...]) -> list[dict[str, Any]]: out = [] for x in legs: if not str(x.get("leg_role") or "").startswith("option"): continue if str(x.get("status") or "") in statuses: out.append(x) return out def _oo_quote_bid(cfg: dict[str, Any], inst_id: str) -> tuple[Optional[float], Optional[float]]: quote_fn = cfg.get("quote_option_contract") ex_opt = cfg.get("exchange_options") if not callable(quote_fn) or ex_opt is None or not inst_id: return None, None try: q = quote_fn(ex_opt, inst_id) if not q.get("ok"): return None, None return _sf(q.get("bid")), _sf(q.get("bid_sz")) except Exception: return None, None def _oo_leg_mark_value(leg: dict[str, Any], bid: Optional[float]) -> Optional[float]: """买一可回收金额(USDC)= bid × 张数 × ct_mult.""" b = _sf(bid) if b is None or b < 0: return None sheets = float(leg.get("size") or 1) ct = float(leg.get("ct_mult") or 0.01) return float(b) * sheets * ct def _oo_plan_premium_total(plan: dict[str, Any], legs: list[dict[str, Any]]) -> float: """双腿总权利金:优先计划字段,否则对期权腿 premium 求和.""" total = _sf(plan.get("premium_total")) if total is not None and total > 0: return float(total) s = 0.0 for leg in legs: if not str(leg.get("leg_role") or "").startswith("option"): continue s += float(leg.get("premium") or 0) return s def _oo_leg_profit_rr( leg: dict[str, Any], bid: Optional[float], *, total_premium: float ) -> Optional[float]: """盈亏比 = 该腿盈利金额 / 总权利金;盈利金额 = 买一回收 − 本腿权利金.""" if total_premium <= 0: return None leg_prem = float(leg.get("premium") or 0) value = _oo_leg_mark_value(leg, bid) if value is None: return None return (value - leg_prem) / total_premium def _oo_resolve_profit_rr(plan: dict[str, Any]) -> Optional[float]: rr = _sf(plan.get("profit_rr")) if rr is not None and rr > 0: return rr return None def _finalize_oo_all_closed( cfg: dict[str, Any], conn: Any, plan: dict[str, Any], legs: list[dict[str, Any]], *, reason: str ) -> dict[str, Any]: closed_opts = _oo_option_legs(legs, statuses=("closed",)) total_opts = sum(float(x.get("realized_pnl") or 0) for x in closed_opts) close_reason = reason or "oo_rest_closed" bucket = "oo_target" if total_opts > 0 else "oo_expiry_loss" update_plan( conn, int(plan["id"]), status="closed", close_reason=close_reason, realized_pnl_options=round(total_opts, 4), realized_pnl_total=round(total_opts, 4), stats_bucket=bucket, closed_at=_now(), ) _notify_end_reload(cfg, conn, int(plan["id"])) return {"plan_id": plan["id"], "close_reason": close_reason, "total": total_opts} def _tick_one(cfg: dict[str, Any], conn: Any, plan: dict[str, Any]) -> Optional[dict[str, Any]]: pt = plan.get("plan_type") legs = get_plan_legs(conn, int(plan["id"])) if pt == "perp_options": from lib.hedge_plan.hedge_plan_option_primary_lib import is_option_primary if is_option_primary(plan): if str(plan.get("status") or "") == "watching": return _tick_po_option_primary_watching(cfg, conn, plan) # 期权为主:半平重试 → 到期 → 目标位分叉 r = _tick_po_option_primary_pending(cfg, conn, plan, legs) if r: return r r = _tick_po_option_primary_expiry(cfg, conn, plan, legs) if r: return r r = _tick_po_option_primary_both_expired(cfg, conn, plan, legs) if r: return r return _tick_po_option_primary(cfg, conn, plan, legs) r = _tick_po(cfg, conn, plan, legs) return r if pt == "options_options": r = _tick_oo_expiry(cfg, conn, plan, legs) if r: return r r = _tick_oo_close_rest(cfg, conn, plan, legs) if r: return r return _tick_oo_target(cfg, conn, plan, legs) return None def _tick_po_option_primary_watching( cfg: dict[str, Any], conn: Any, plan: dict[str, Any] ) -> Optional[dict[str, Any]]: """盯盘:链上出现杠杆/间隔达标合约后自动开仓.""" import json import os from lib.hedge_plan.hedge_plan_option_primary_lib import ( pick_option_primary_candidate, size_from_premium, ) from lib.hedge_plan.hedge_plan_orders_lib import execute_perp_options_start from lib.hedge_plan.hedge_plan_register import _activate_watching_po build_chain = cfg.get("build_option_chain") ex = cfg.get("exchange_options") if not callable(build_chain) or ex is None: return None body0: dict[str, Any] = {} try: raw = plan.get("preview_json") or "" blob = json.loads(raw) if raw else {} body0 = dict(blob.get("start_body") or blob or {}) except Exception: body0 = {} uly = str(plan.get("underlying") or body0.get("underlying") or "ETH").upper() direction = str(plan.get("direction") or body0.get("direction") or "long").lower() money = str(plan.get("option_moneyness") or body0.get("moneyness") or "otm").lower() interval = plan.get("strike_interval") if interval in (None, ""): interval = body0.get("strike_interval", 15) min_h = plan.get("min_option_hours") if min_h in (None, ""): min_h = body0.get("min_option_hours", 36) opt_lev = plan.get("option_leverage") if opt_lev in (None, ""): opt_lev = body0.get("option_leverage") try: chain = build_chain( ex, uly, max_dte_days=float(cfg.get("chain_max_dte") or 14), itm_only=False, itm_max_dist_usd=float(os.getenv("OKX_OPTIONS_ITM_MAX_DIST_USD") or "30"), ) except Exception as e: update_plan(conn, int(plan["id"]), note=f"盯盘拉链失败: {e}"[:500]) return None cand = pick_option_primary_candidate( chain, direction=direction, moneyness=money, strike_interval=interval, min_hours=min_h, min_opt_leverage=opt_lev, ) if not cand: return None ask = float(cand.get("ask") or 0) ct = float(cand.get("ct_mult") or body0.get("ct_mult") or 0.01) sized = size_from_premium( premium_budget=float(plan.get("premium_budget") or body0.get("premium_budget") or 0), ask=ask, ct_mult=ct, ratio=float(plan.get("option_perp_ratio") or body0.get("option_perp_ratio") or 2), contract_size=float(body0.get("contract_size") or 0.01), ) if not sized.get("ok"): update_plan(conn, int(plan["id"]), note=f"盯盘定仓失败: {sized.get('msg')}"[:500]) return None idx = float(cand.get("index_px") or chain.get("index_px") or 0) body = dict(body0) body.update( { "plan_type": "perp_options", "option_primary": True, "watch_entry": 0, "underlying": uly, "direction": direction, "moneyness": money, "opt_inst_id": cand.get("inst_id"), "opt_type": cand.get("opt_type"), "strike": cand.get("strike"), "ask": ask, "ct_mult": ct, "sheets": sized["sheets"], "contracts": sized["contracts"], "eth_qty": sized.get("eth_qty"), "index_px": idx, "entry": idx, "hours_to_expiry": cand.get("hours_to_expiry"), "strike_interval": interval, "min_option_hours": min_h, "option_leverage": opt_lev, "option_perp_ratio": plan.get("option_perp_ratio") or body0.get("option_perp_ratio"), "option_target_points": plan.get("option_target_points") or body0.get("option_target_points"), "perp_target_points": plan.get("perp_target_points") or body0.get("perp_target_points"), "premium_budget": plan.get("premium_budget") or body0.get("premium_budget"), "leverage": plan.get("leverage") or body0.get("leverage") or 100, "exchange_symbol": body0.get("exchange_symbol") or f"{uly}-USDT-SWAP", "contract_size": body0.get("contract_size") or 0.01, } ) dry = str(os.getenv("HEDGE_PLAN_DRY_RUN") or "").strip().lower() in ("1", "true", "yes", "on") out = execute_perp_options_start(cfg, body, dry_run=dry, persist=None) if not out.get("ok"): update_plan(conn, int(plan["id"]), note=f"盯盘开仓未成: {out.get('msg')}"[:500]) return {"plan_id": plan["id"], "watching_open": False, "msg": out.get("msg")} if dry: update_plan(conn, int(plan["id"]), note=f"dry_run命中 {cand.get('inst_id')}"[:500]) return {"plan_id": plan["id"], "watching_open": True, "dry_run": True, "inst_id": cand.get("inst_id")} _activate_watching_po(cfg, conn, int(plan["id"]), out, body) return { "plan_id": plan["id"], "watching_open": True, "inst_id": cand.get("inst_id"), "leverage": cand.get("leverage"), } def _tick_po_option_primary_pending( cfg: dict[str, Any], conn: Any, plan: dict[str, Any], legs: list[dict[str, Any]] ) -> Optional[dict[str, Any]]: """期权已平、永续待平(opt_target_perp_pending)时只重试平永续.""" from lib.hedge_plan.hedge_plan_option_primary_lib import perp_direction_for_view from lib.hedge_plan.hedge_plan_orders_lib import _close_perp pending = str(plan.get("close_reason") or "") if pending not in ("opt_target_perp_pending", "opt_target_pending"): return None perp = next((x for x in legs if x.get("leg_role") == "perp"), None) opt = next((x for x in legs if x.get("leg_role") == "option_hedge"), None) if not perp or str(perp.get("status") or "") != "open": return None view = str(plan.get("direction") or "long").lower() perp_dir = str(plan.get("perp_direction") or perp.get("side") or perp_direction_for_view(view)).lower() symbol = str(perp.get("symbol") or "") contracts = float(perp.get("size") or plan.get("perp_size") or 0) # 期权仍 open:继续走主路径,不在此强平 if pending == "opt_target_pending" and opt and str(opt.get("status") or "") == "open": return None # 期权已平或 already flat:只补平永续 if opt and str(opt.get("status") or "") == "open": return None perp_close = _close_perp(cfg, symbol=symbol, direction=perp_dir, contracts=contracts, dry_run=False) if not perp_close.get("ok"): notify_hedge( cfg, build_hedge_alert_message( title="期权已平·永续平仓重试失败", plan_id=plan.get("id"), detail=str(perp_close.get("msg") or perp_close), ), ) update_plan(conn, int(plan["id"]), close_reason="opt_target_perp_pending") return {"plan_id": plan["id"], "retry": True, "perp_close": perp_close} entry = _sf(plan.get("entry_mark")) or _sf(perp.get("avg_open")) or 0 mark = entry ex = cfg.get("exchange") if ex is not None and symbol: try: t = ex.fetch_ticker(symbol) mark = _sf((t.get("info") or {}).get("markPx")) or _sf(t.get("last")) or entry except Exception: pass cs = float(cfg.get("default_contract_size") or 0.01) get_cs = cfg.get("get_contract_size") if callable(get_cs) and symbol: try: cs = float(get_cs(symbol) or cs) except Exception: pass coins = contracts * cs if perp_dir == "short": perp_pnl = (float(entry or 0) - float(mark or 0)) * coins else: perp_pnl = (float(mark or 0) - float(entry or 0)) * coins opt_pnl = float(opt.get("realized_pnl") or 0) if opt else float(plan.get("realized_pnl_options") or 0) conn.execute( "UPDATE hedge_plan_legs SET status=?, close_reason=?, closed_at=?, realized_pnl=? WHERE id=?", ("closed", "opt_target_points", _now(), round(perp_pnl, 4), perp["id"]), ) total = opt_pnl + perp_pnl update_plan( conn, int(plan["id"]), status="closed", close_reason="opt_target_points", realized_pnl_perp=round(perp_pnl, 4), realized_pnl_options=round(opt_pnl, 4), realized_pnl_total=round(total, 4), stats_bucket="opt_primary", closed_at=_now(), ) _notify_end_reload(cfg, conn, int(plan["id"])) return {"plan_id": plan["id"], "close_reason": "opt_target_points", "total": total, "recovered": True} def _tick_po_option_primary_both_expired( cfg: dict[str, Any], conn: Any, plan: dict[str, Any], legs: list[dict[str, Any]] ) -> Optional[dict[str, Any]]: """两腿仍 open 但期权已到期:结算期权并市价平永续,避免裸奔.""" from lib.hedge_plan.hedge_plan_option_primary_lib import perp_direction_for_view from lib.hedge_plan.hedge_plan_orders_lib import _close_perp perp = next((x for x in legs if x.get("leg_role") == "perp"), None) opt = next((x for x in legs if x.get("leg_role") == "option_hedge"), None) if not perp or str(perp.get("status") or "") != "open": return None if not opt or str(opt.get("status") or "") != "open": return None if not leg_is_expired(opt): return None spot = _index_px(cfg, str(plan.get("underlying") or "ETH")) if spot is None: return None est = settle_option_leg_at_spot(opt, float(spot)) 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", "expiry", _now(), round(opt_pnl, 4), opt["id"]), ) view = str(plan.get("direction") or "long").lower() perp_dir = str(plan.get("perp_direction") or perp.get("side") or perp_direction_for_view(view)).lower() symbol = str(perp.get("symbol") or "") contracts = float(perp.get("size") or plan.get("perp_size") or 0) perp_close = _close_perp(cfg, symbol=symbol, direction=perp_dir, contracts=contracts, dry_run=False) entry = _sf(plan.get("entry_mark")) or _sf(perp.get("avg_open")) or float(spot) cs = float(cfg.get("default_contract_size") or 0.01) get_cs = cfg.get("get_contract_size") if callable(get_cs) and symbol: try: cs = float(get_cs(symbol) or cs) except Exception: pass coins = contracts * cs if perp_dir == "short": perp_pnl = (float(entry) - float(spot)) * coins else: perp_pnl = (float(spot) - float(entry)) * coins if not perp_close.get("ok"): notify_hedge( cfg, build_hedge_alert_message( title="期权到期后永续平仓失败(将重试)", plan_id=plan.get("id"), detail=str(perp_close.get("msg") or perp_close), ), ) update_plan( conn, int(plan["id"]), close_reason="opt_target_perp_pending", realized_pnl_options=round(opt_pnl, 4), note="期权已到期结算,永续待平", ) return {"plan_id": plan["id"], "retry": True, "perp_close": perp_close} conn.execute( "UPDATE hedge_plan_legs SET status=?, close_reason=?, closed_at=?, realized_pnl=? WHERE id=?", ("closed", "option_expired", _now(), round(perp_pnl, 4), perp["id"]), ) total = opt_pnl + perp_pnl update_plan( conn, int(plan["id"]), status="closed", close_reason="option_expired", realized_pnl_perp=round(perp_pnl, 4), realized_pnl_options=round(opt_pnl, 4), realized_pnl_total=round(total, 4), stats_bucket="opt_primary", closed_at=_now(), ) _notify_end_reload(cfg, conn, int(plan["id"])) return {"plan_id": plan["id"], "close_reason": "option_expired", "total": total} def _tick_po_option_primary_expiry( cfg: dict[str, Any], conn: Any, plan: dict[str, Any], legs: list[dict[str, Any]] ) -> Optional[dict[str, Any]]: """期权为主且永续已平、期权 hold_to_expiry → 到期结算后收口计划.""" perp = next((x for x in legs if x.get("leg_role") == "perp"), None) opt = next((x for x in legs if x.get("leg_role") == "option_hedge"), None) if not opt or str(opt.get("status") or "") != "hold_to_expiry": return None if perp and str(perp.get("status") or "") == "open": return None if not leg_is_expired(opt): return None spot = _index_px(cfg, str(plan.get("underlying") or "ETH")) if spot is None: return None est = settle_option_leg_at_spot(opt, float(spot)) 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", "expiry", _now(), round(opt_pnl, 4), opt["id"]), ) perp_pnl = float(perp.get("realized_pnl") or 0) if perp else float(plan.get("realized_pnl_perp") or 0) total = perp_pnl + opt_pnl update_plan( conn, int(plan["id"]), status="closed", close_reason="perp_target_points_expiry", realized_pnl_perp=round(perp_pnl, 4), realized_pnl_options=round(opt_pnl, 4), realized_pnl_total=round(total, 4), stats_bucket="opt_primary", closed_at=_now(), ) _notify_end_reload(cfg, conn, int(plan["id"])) return {"plan_id": plan["id"], "close_reason": "perp_target_points_expiry", "total": total} def _tick_po_option_primary( cfg: dict[str, Any], conn: Any, plan: dict[str, Any], legs: list[dict[str, Any]] ) -> Optional[dict[str, Any]]: """以期权为主:触达目标位立即执行分叉平仓规则.""" from lib.hedge_plan.hedge_plan_option_primary_lib import ( estimate_combo_net_pnl, option_bid_liquidity_ok, perp_direction_for_view, target_hit, ) from lib.hedge_plan.hedge_plan_orders_lib import _close_perp perp = next((x for x in legs if x.get("leg_role") == "perp"), None) opt = next((x for x in legs if x.get("leg_role") == "option_hedge"), None) if not perp or str(perp.get("status") or "") != "open": return None if not opt or str(opt.get("status") or "") != "open": return None if _within_open_grace(plan): return None view = str(plan.get("direction") or "long").lower() perp_dir = str(plan.get("perp_direction") or perp.get("side") or perp_direction_for_view(view)).lower() strike = _sf(opt.get("strike")) n = _sf(plan.get("option_target_points")) m = _sf(plan.get("perp_target_points")) if strike is None or strike <= 0: return None idx = _index_px(cfg, str(plan.get("underlying") or "ETH")) if idx is None: return None hit_opt = bool(n is not None and target_hit(view_side=view, index_px=idx, strike=strike, points=float(n))) hit_perp = bool(m is not None and target_hit(view_side=view, index_px=idx, strike=strike, points=float(m))) if not hit_opt and not hit_perp: return None symbol = str(perp.get("symbol") or "") mark = None ex = cfg.get("exchange") if ex is not None and symbol: try: t = ex.fetch_ticker(symbol) mark = _sf((t.get("info") or {}).get("markPx")) or _sf(t.get("last")) except Exception: mark = None mark = mark or idx entry = _sf(plan.get("entry_mark")) or _sf(perp.get("avg_open")) or mark cs = float(cfg.get("default_contract_size") or 0.01) get_cs = cfg.get("get_contract_size") if callable(get_cs) and symbol: try: cs = float(get_cs(symbol) or cs) except Exception: pass quote_fn = cfg.get("quote_option_contract") ex_opt = cfg.get("exchange_options") bid = None bid_sz = None if callable(quote_fn) and ex_opt is not None: try: q = quote_fn(ex_opt, str(opt.get("inst_id") or "")) if q.get("ok"): bid = _sf(q.get("bid")) bid_sz = _sf(q.get("bid_sz")) except Exception: bid = None ask_open = _sf(opt.get("avg_open")) or 0.0 sheets = float(opt.get("size") or 1) ct = float(opt.get("ct_mult") or 0.01) contracts = float(perp.get("size") or plan.get("perp_size") or 0) # 优先期权目标;买一不足或净利≤0 时若永续目标已触达则改走永续目标 if hit_opt: liq_ok, liq_msg = option_bid_liquidity_ok(bid, bid_sz, need_sheets=sheets) net = None if liq_ok: net = estimate_combo_net_pnl( view_side=view, strike=float(strike), index_px=float(idx), ask_open=float(ask_open), bid=float(bid or 0), sheets=sheets, ct_mult=ct, perp_direction=perp_dir, perp_entry=float(entry or 0), perp_mark=float(mark or 0), contracts=contracts, contract_size=cs, ) can_opt_exit = bool(liq_ok and net is not None and float(net.get("net") or 0) > 0) if can_opt_exit: reason = "opt_target_points" close_r = _sell_option(cfg, inst_id=str(opt.get("inst_id") or ""), sheets=sheets) if close_r.get("already_flat"): opt_pnl = _option_leg_pnl_after_close(cfg, opt, fallback=net["opt_net"]) elif not close_r.get("ok") or not close_r.get("fully_closed", True): notify_hedge( cfg, build_hedge_alert_message( title="期权目标平仓失败(将重试)", plan_id=plan.get("id"), detail=str(close_r.get("msg") or close_r), ), ) update_plan(conn, int(plan["id"]), close_reason="opt_target_pending") return {"plan_id": plan["id"], "retry": True, "close": close_r} else: opt_pnl = _option_leg_pnl_after_close(cfg, opt, fallback=net["opt_net"]) conn.execute( "UPDATE hedge_plan_legs SET status=?, close_reason=?, closed_at=?, realized_pnl=? WHERE id=?", ("closed", reason, _now(), round(opt_pnl, 4), opt["id"]), ) perp_close = _close_perp( cfg, symbol=symbol, direction=perp_dir, contracts=contracts, dry_run=False ) if not perp_close.get("ok"): notify_hedge( cfg, build_hedge_alert_message( title="期权已平但永续平仓失败(将重试)", plan_id=plan.get("id"), detail=str(perp_close.get("msg") or perp_close), ), ) update_plan(conn, int(plan["id"]), close_reason="opt_target_perp_pending") return {"plan_id": plan["id"], "retry": True, "perp_close": perp_close} perp_pnl = float(net.get("perp_net") or 0) conn.execute( "UPDATE hedge_plan_legs SET status=?, close_reason=?, closed_at=?, realized_pnl=? WHERE id=?", ("closed", reason, _now(), round(perp_pnl, 4), perp["id"]), ) total = float(opt_pnl) + float(perp_pnl) update_plan( conn, int(plan["id"]), status="closed", close_reason=reason, realized_pnl_perp=round(perp_pnl, 4), realized_pnl_options=round(opt_pnl, 4), realized_pnl_total=round(total, 4), stats_bucket="opt_primary", closed_at=_now(), ) _notify_end_reload(cfg, conn, int(plan["id"])) return {"plan_id": plan["id"], "close_reason": reason, "total": total, "net": net} if not hit_perp: return { "plan_id": plan["id"], "skip": True, "msg": (liq_msg if not liq_ok else "净利≤0,继续持有"), "net": net, } if not hit_perp: return None reason = "perp_target_points" # 永续目标:平永续,期权持有至到期 perp_close = _close_perp(cfg, symbol=symbol, direction=perp_dir, contracts=contracts, dry_run=False) if not perp_close.get("ok"): notify_hedge( cfg, build_hedge_alert_message( title="永续目标平仓失败(将重试)", plan_id=plan.get("id"), detail=str(perp_close.get("msg") or perp_close), ), ) update_plan(conn, int(plan["id"]), close_reason="perp_target_pending") return {"plan_id": plan["id"], "retry": True, "perp_close": perp_close} # 估永续已实现 coins = contracts * cs if perp_dir == "short": perp_pnl = (float(entry or 0) - float(mark or 0)) * coins else: perp_pnl = (float(mark or 0) - float(entry or 0)) * coins conn.execute( "UPDATE hedge_plan_legs SET status=?, close_reason=?, closed_at=?, realized_pnl=? WHERE id=?", ("closed", reason, _now(), round(perp_pnl, 4), perp["id"]), ) conn.execute( "UPDATE hedge_plan_legs SET status=?, close_reason=? WHERE id=?", ("hold_to_expiry", "hold_expiry_after_perp_target", opt["id"]), ) update_plan( conn, int(plan["id"]), # 计划保持 active,等期权到期收口 close_reason="perp_target_points", realized_pnl_perp=round(perp_pnl, 4), note="永续已按目标平仓,期权持有至到期", ) notify_hedge( cfg, build_hedge_alert_message( title="永续目标已平·期权持有至到期", plan_id=plan.get("id"), detail=f"指数 {idx:.2f} · 永续盈亏约 {perp_pnl:.2f}", ), ) return {"plan_id": plan["id"], "close_reason": reason, "perp_pnl": perp_pnl, "opt_hold": True} def _tick_po(cfg: dict[str, Any], conn: Any, plan: dict[str, Any], legs: list[dict[str, Any]]) -> Optional[dict[str, Any]]: perp = next((x for x in legs if x.get("leg_role") == "perp"), None) opt = next((x for x in legs if x.get("leg_role") == "option_hedge"), None) if not perp or perp.get("status") != "open": return None symbol = perp.get("symbol") or "" direction = (plan.get("perp_direction") or plan.get("direction") or "long").lower() live = _perp_live_contracts(cfg, symbol, direction) # API 失败 / 未注入 → 本轮跳过,绝不当「已平」 if live is None: return None # 仍有仓 → 未触达交易所 TP/SL if live > 0: return None # 开仓后宽限期:仓位同步延迟可误读为 0 if _within_open_grace(plan): return None entry = _sf(plan.get("entry_mark")) or _sf(perp.get("avg_open")) or 0 tp = _sf(plan.get("tp")) sl = _sf(plan.get("sl")) mark = None ex = cfg.get("exchange") if ex is not None and symbol: try: t = ex.fetch_ticker(symbol) mark = _sf((t.get("info") or {}).get("markPx")) or _sf(t.get("last")) except Exception: mark = None reason = _classify_po_flat_reason( direction=direction, entry=float(entry or 0), mark=mark, tp=tp, sl=sl ) # 上一轮止损强平未完成:粘滞为 SL,避免 mark 反弹误判 TP 跳过强平 pending_reason = str(plan.get("close_reason") or "") if pending_reason == "perp_sl_pending_opt": reason = "perp_sl" elif pending_reason == "perp_tp_pending_opt": reason = "perp_tp" # 未明确 TP/SL 时不收口,下轮再判 if reason == "perp_flat_unknown": return None premium = float(plan.get("premium_total") or 0) cs = float(cfg.get("default_contract_size") or 0.01) get_cs = cfg.get("get_contract_size") if callable(get_cs) and symbol: try: cs = float(get_cs(symbol) or cs) except Exception: pass size = float(perp.get("size") or 0) exit_px = mark or (tp if reason == "perp_tp" else sl) or entry coins = size * cs if direction == "short": perp_pnl = (entry - exit_px) * coins else: perp_pnl = (exit_px - entry) * coins opt_pnl = -premium if reason == "perp_sl" and opt and str(opt.get("status") or "") == "open": if _env_bool("HEDGE_PLAN_ON_PERP_SL_CLOSE_OPTIONS", True): close_r = _sell_option( cfg, inst_id=str(opt.get("inst_id") or ""), sheets=float(opt.get("size") or 1), ) if close_r.get("already_flat"): opt_pnl = _option_leg_pnl_after_close(cfg, opt, fallback=-premium) conn.execute( "UPDATE hedge_plan_legs SET status=?, close_reason=?, closed_at=?, realized_pnl=? WHERE id=?", ("closed", reason, _now(), opt_pnl, opt["id"]), ) elif not close_r.get("ok") or not close_r.get("fully_closed", True): notify_hedge( cfg, build_hedge_alert_message( title="永续止损后期权强制平仓失败(将重试)", plan_id=plan.get("id"), detail=str(close_r.get("msg") or close_r), ), ) update_plan(conn, int(plan["id"]), close_reason="perp_sl_pending_opt") return { "plan_id": plan["id"], "msg": "止损后期权未平完", "close": close_r, "retry": True, } else: bid = _sf(close_r.get("bid") or close_r.get("locked_bid_px")) 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) est = (bid - ask_open) * float(opt.get("size") or 1) * ct else: 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"]), ) else: conn.execute( "UPDATE hedge_plan_legs SET status=?, close_reason=? WHERE id=?", ("hold_to_expiry", "orphaned_after_sl", opt["id"]), ) opt_pnl = -premium elif reason == "perp_tp" and opt and str(opt.get("status") or "") == "open": if _env_bool("HEDGE_PLAN_ON_PERP_TP_CLOSE_OPTIONS", False): close_r = _sell_option( cfg, inst_id=str(opt.get("inst_id") or ""), sheets=float(opt.get("size") or 1) ) if close_r.get("already_flat"): opt_pnl = _option_leg_pnl_after_close(cfg, opt, fallback=-premium) conn.execute( "UPDATE hedge_plan_legs SET status=?, close_reason=?, closed_at=?, realized_pnl=? WHERE id=?", ("closed", reason, _now(), opt_pnl, opt["id"]), ) elif not close_r.get("ok") or not close_r.get("fully_closed", True): notify_hedge( cfg, build_hedge_alert_message( title="永续止盈后期权平仓失败(将重试)", plan_id=plan.get("id"), detail=str(close_r.get("msg") or close_r), ), ) update_plan(conn, int(plan["id"]), close_reason="perp_tp_pending_opt") return { "plan_id": plan["id"], "msg": "止盈后期权未平完", "close": close_r, "retry": True, } else: bid = _sf(close_r.get("bid") or close_r.get("locked_bid_px")) 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) est = (bid - ask_open) * float(opt.get("size") or 1) * ct else: 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"]), ) else: conn.execute( "UPDATE hedge_plan_legs SET status=?, close_reason=? WHERE id=?", ("hold_to_expiry", "orphaned_after_tp", opt["id"]), ) opt_pnl = -premium total = perp_pnl + opt_pnl conn.execute( "UPDATE hedge_plan_legs SET status=?, close_reason=?, closed_at=?, realized_pnl=? WHERE id=?", ("closed", reason, _now(), perp_pnl, perp["id"]), ) # partial → closed 也走同一收口 update_plan( conn, int(plan["id"]), status="closed", close_reason=reason, realized_pnl_perp=round(perp_pnl, 4), realized_pnl_options=round(opt_pnl, 4), realized_pnl_total=round(total, 4), stats_bucket="tp" if reason == "perp_tp" else "sl", closed_at=_now(), ) _notify_end_reload(cfg, conn, int(plan["id"])) 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) sheets = float(leg.get("size") or 1) ct = float(leg.get("ct_mult") or 0.01) if bid is not None and float(bid) > 0: return float(bid) * sheets * ct - premium if idx is None: return -premium strike = _sf(leg.get("strike")) or 0 o = (leg.get("opt_type") or "").upper() intrinsic = max(0.0, idx - strike) if o == "C" else max(0.0, strike - idx) return intrinsic * sheets * ct - premium def _tick_oo_close_rest( cfg: dict[str, Any], conn: Any, plan: dict[str, Any], legs: list[dict[str, Any]] ) -> Optional[dict[str, Any]]: """盈利腿已平后:残值平模式清亏损腿. 条件:买一回收 ≤ 本合约初始权利金×20%,且买一有流动性;失败或未达条件则下轮重试. """ from lib.hedge_plan.hedge_plan_option_primary_lib import option_bid_liquidity_ok if resolve_oo_rest_close_mode(plan) != "close_all": return None open_legs = _oo_option_legs(legs, statuses=("open",)) closed_legs = _oo_option_legs(legs, statuses=("closed",)) # 至少已平一条,且仍有残腿;避免双腿都还 open 时误清 if len(closed_legs) < 1 or len(open_legs) < 1: return None reason0 = str(plan.get("close_reason") or "") allowed_reasons = ( "target_win_leg", "target_up_win_leg", "target_down_win_leg", "profit_rr_win_leg", "oo_rest_closing", "", ) if reason0 not in allowed_reasons and not ( len(closed_legs) >= 1 and len(open_legs) == 1 ): return None idx = _index_px(cfg, str(plan.get("underlying") or "ETH")) acted = False waiting = False for leg in list(open_legs): inst_id = str(leg.get("inst_id") or "") sheets = float(leg.get("size") or 1) premium = float(leg.get("premium") or 0) bid, bid_sz = _oo_quote_bid(cfg, inst_id) value = _oo_leg_mark_value(leg, bid) # 残值门槛:相对本合约初始权利金,买一回收须 ≤ 20% if premium > 0: if value is None: waiting = True continue if value > premium * OO_LOSS_LEG_RESIDUAL_RATIO + 1e-12: waiting = True continue liq_ok, liq_msg = option_bid_liquidity_ok(bid, bid_sz, need_sheets=sheets) if not liq_ok: waiting = True update_plan(conn, int(plan["id"]), close_reason="oo_rest_closing") return { "plan_id": plan["id"], "msg": "残值平等待买一流动性", "detail": liq_msg, "waiting": True, } close_r = _sell_option(cfg, inst_id=inst_id, sheets=sheets) if not close_r.get("ok"): notify_hedge( cfg, build_hedge_alert_message( title="期期残值平·亏损腿平仓失败(将重试)", plan_id=plan.get("id"), detail=str(close_r.get("msg") or close_r), ), ) update_plan(conn, int(plan["id"]), close_reason="oo_rest_closing") return {"plan_id": plan["id"], "msg": "残腿平仓失败", "close": close_r, "retry": True} bid_fill = _sf(close_r.get("bid")) or bid est = _estimate_leg_close_pnl(leg, idx, bid_fill) 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"]), ) leg["status"] = "closed" leg["realized_pnl"] = round(pnl, 4) acted = True if not acted: if waiting: update_plan(conn, int(plan["id"]), close_reason="oo_rest_closing") return {"plan_id": plan["id"], "msg": "残值平等待本合约权利金≤20%", "waiting": True} return None legs2 = get_plan_legs(conn, int(plan["id"])) still_open = _oo_option_legs(legs2, statuses=("open", "hold_to_expiry")) if still_open: update_plan(conn, int(plan["id"]), close_reason="oo_rest_closing") return {"plan_id": plan["id"], "msg": "残腿部分已平,继续重试", "remaining": len(still_open)} return _finalize_oo_all_closed( cfg, conn, plan, legs2, reason="oo_rest_closed" ) def _after_oo_winner_closed( cfg: dict[str, Any], conn: Any, plan: dict[str, Any], open_legs: list[dict[str, Any]], best: dict[str, Any], *, reason: str, extra: Optional[dict[str, Any]] = None, ) -> dict[str, Any]: """盈利腿已平后:残值平同轮尝试 / 到期平标记 hold_to_expiry.""" rest_mode = resolve_oo_rest_close_mode(plan) update_plan(conn, int(plan["id"]), close_reason=reason) mid = dict(plan) mid["close_reason"] = reason mid["status"] = "active" mid["oo_close_mode"] = rest_mode notify_plan_end(cfg, conn, mid) out: dict[str, Any] = { "plan_id": plan["id"], "close_reason": reason, "closed_leg": best.get("id"), "oo_close_mode": rest_mode, } if extra: out.update(extra) if rest_mode == "close_all": legs2 = get_plan_legs(conn, int(plan["id"])) rest = _tick_oo_close_rest(cfg, conn, mid, legs2) if rest: out["rest"] = rest return out for leg in open_legs: if int(leg.get("id") or 0) == int(best.get("id") or 0): continue conn.execute( "UPDATE hedge_plan_legs SET status=? WHERE id=?", ("hold_to_expiry", leg["id"]), ) return out def _tick_oo_profit_rr( cfg: dict[str, Any], conn: Any, plan: dict[str, Any], legs: list[dict[str, Any]], *, rr_target: float ) -> Optional[dict[str, Any]]: """期期:任一开仓腿盈亏比(该腿盈利金额/总权利金)达目标 → 平盈利腿.""" if not _env_bool("HEDGE_PLAN_OO_CLOSE_WINNER_ONLY", True): return None open_legs = _oo_option_legs(legs, statuses=("open",)) if len(open_legs) < 2: return None total_prem = _oo_plan_premium_total(plan, legs) if total_prem <= 0: return None ranked: list[tuple[float, float, dict[str, Any]]] = [] for leg in open_legs: bid, _bid_sz = _oo_quote_bid(cfg, str(leg.get("inst_id") or "")) rr = _oo_leg_profit_rr(leg, bid, total_premium=total_prem) if rr is None: continue value = _oo_leg_mark_value(leg, bid) or 0.0 premium = float(leg.get("premium") or 0) pnl = value - premium ranked.append((rr, pnl, leg)) if not ranked: return None ranked.sort(key=lambda x: x[0], reverse=True) best_rr, best_pnl, best = ranked[0] if best_rr + 1e-12 < float(rr_target) or best_pnl <= 0: return None close_r = _sell_option( cfg, inst_id=str(best.get("inst_id") or ""), sheets=float(best.get("size") or 1) ) if not close_r.get("ok"): notify_hedge( cfg, build_hedge_alert_message( title="期期平盈利腿失败", plan_id=plan.get("id"), detail=str(close_r.get("msg") or close_r), ), ) return {"plan_id": plan["id"], "msg": "平盈利腿失败", "close": close_r} reason = "profit_rr_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(), closed_pnl, best["id"]), ) idx = _index_px(cfg, str(plan.get("underlying") or "ETH")) return _after_oo_winner_closed( cfg, conn, plan, open_legs, best, reason=reason, extra={ "profit_rr": best_rr, "rr_target": float(rr_target), "total_premium": total_prem, "index": idx, }, ) def _tick_oo_target( cfg: dict[str, Any], conn: Any, plan: dict[str, Any], legs: list[dict[str, Any]] ) -> Optional[dict[str, Any]]: """期期:优先按盈亏比平盈利腿;旧单无 profit_rr 时回退上/下破目标价.""" rr_target = _oo_resolve_profit_rr(plan) if rr_target is not None: return _tick_oo_profit_rr(cfg, conn, plan, legs, rr_target=rr_target) idx = _index_px(cfg, str(plan.get("underlying") or "ETH")) if idx is None: return None up = _sf(plan.get("target_price_up")) down = _sf(plan.get("target_price_down")) # 旧计划仅有单目标:两边都用它 legacy = _sf(plan.get("target_price")) if up is None and legacy is not None: up = legacy if down is None and legacy is not None: down = legacy if up is None and down is None: return None hit_side: Optional[str] = None if up is not None and idx >= up * 0.998: hit_side = "up" elif down is not None and idx <= down * 1.002: hit_side = "down" if not hit_side: return None if not _env_bool("HEDGE_PLAN_OO_CLOSE_WINNER_ONLY", True): return None open_legs = _oo_option_legs(legs, statuses=("open",)) if len(open_legs) < 2: return None winners = [] for leg in open_legs: strike = _sf(leg.get("strike")) or 0 o = (leg.get("opt_type") or "").upper() intrinsic = max(0.0, idx - strike) if o == "C" else max(0.0, strike - idx) premium = float(leg.get("premium") or 0) pnl = intrinsic * float(leg.get("size") or 1) * float(leg.get("ct_mult") or 0.01) - premium winners.append((pnl, leg)) winners.sort(key=lambda x: x[0], reverse=True) best_pnl, best = winners[0] if best_pnl <= 0: return None close_r = _sell_option(cfg, inst_id=str(best.get("inst_id") or ""), sheets=float(best.get("size") or 1)) if not close_r.get("ok"): notify_hedge( cfg, build_hedge_alert_message( title="期期平盈利腿失败", plan_id=plan.get("id"), detail=str(close_r.get("msg") or close_r), ), ) 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(), closed_pnl, best["id"]), ) return _after_oo_winner_closed( cfg, conn, plan, open_legs, best, reason=reason, extra={"hit_side": hit_side, "index": idx}, ) def _tick_oo_expiry( cfg: dict[str, Any], conn: Any, plan: dict[str, Any], legs: list[dict[str, Any]] ) -> Optional[dict[str, Any]]: """期期:剩余期权腿全部到期 → 结算合计并结束计划.""" pending = [ 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 pending: # 若腿已全部 closed 但计划仍 active(异常残留)则用腿合计收口 closed_opts = [ x for x in legs if str(x.get("leg_role") or "").startswith("option") and x.get("status") == "closed" ] if len(closed_opts) < 1: return None total_opts = sum(float(x.get("realized_pnl") or 0) for x in closed_opts) reason = "oo_expiry_loss" if total_opts <= 0 else "oo_expiry_win" update_plan( conn, int(plan["id"]), status="closed", close_reason=reason, realized_pnl_options=round(total_opts, 4), realized_pnl_total=round(total_opts, 4), stats_bucket=reason if reason == "oo_expiry_loss" else "oo_target", closed_at=_now(), ) _notify_end_reload(cfg, conn, int(plan["id"])) return {"plan_id": plan["id"], "close_reason": reason, "total": total_opts} if not all(leg_is_expired(x) for x in pending): return None spot = _index_px(cfg, str(plan.get("underlying") or "ETH")) if spot is None: return None settled_sum = 0.0 for leg in pending: 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=?", ("closed", "expiry", _now(), round(pnl, 4), leg["id"]), ) already = sum( float(x.get("realized_pnl") or 0) for x in legs if str(x.get("leg_role") or "").startswith("option") and x.get("status") == "closed" ) total = already + settled_sum reason = "oo_expiry_loss" if total <= 0 else "oo_expiry_win" bucket = "oo_expiry_loss" if reason == "oo_expiry_loss" else "oo_target" update_plan( conn, int(plan["id"]), status="closed", close_reason=reason, realized_pnl_options=round(total, 4), realized_pnl_total=round(total, 4), stats_bucket=bucket, closed_at=_now(), ) _notify_end_reload(cfg, conn, int(plan["id"])) return {"plan_id": plan["id"], "close_reason": reason, "total": total, "spot": spot} def _settle_orphaned_after_tp(cfg: dict[str, Any], conn: Any) -> list[dict[str, Any]]: """永期止盈后 hold_to_expiry 期权到期:只更新腿,不回写计划合计.""" rows = conn.execute( """ SELECT l.id AS leg_id, l.plan_id, l.inst_id, l.opt_type, l.strike, l.size, l.premium, l.status, p.underlying, p.status AS plan_status FROM hedge_plan_legs l JOIN hedge_plans p ON p.id = l.plan_id WHERE l.status = 'hold_to_expiry' AND l.close_reason = 'orphaned_after_tp' LIMIT 40 """ ).fetchall() acted: list[dict[str, Any]] = [] for row in rows: leg = dict(row) if not leg_is_expired(leg): continue spot = _index_px(cfg, str(leg.get("underlying") or "ETH")) if spot is None: continue 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"]), ) # 故意不 UPDATE hedge_plans.realized_pnl_* acted.append( { "plan_id": leg["plan_id"], "close_reason": "orphaned_option_expiry", "leg_id": leg["leg_id"], "leg_pnl": round(pnl, 4), "note": "不回写计划合计", } ) return acted