crypto_okx/lib/hedge_plan/hedge_plan_monitor_lib.py

"""对冲计划监控:永期 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