eth_hedge_sim/backend/app/live/executor.py

"""实盘执行：OKX 真下单 + 本地账本/持仓记录（与 Matcher 同结构）。"""

from __future__ import annotations

import logging
import time

from ..config import get_settings
from ..env_store import live_ready
from ..exchange.runtime import load_runtime_settings
from ..models.db import get_db
from ..sim.liquidity import contracts_for_eth, eth_from_contracts
from ..sim.matcher import CloseResult, Matcher, OpenResult
from ..sim.pricing import option_intrinsic
from ..strategy.session import get_session
from .okx_trade import OkxTradeClient
from .reconcile import (
    assert_safe_to_open_live,
    claim_open_slot,
    exchange_option_abs_size,
    perp_close_contracts_okx,
    perp_open_contracts_okx,
    recover_stuck_opening,
    release_open_slot_if_opening,
    stamp_opening_intent,
)
from .symbols import live_settings, resolve_perp_inst_id

logger = logging.getLogger(__name__)


class OkxLiveExecutor(Matcher):
    """开平仓走 OKX 私有接口；浮盈/残留逻辑复用 Matcher。"""

    def __init__(self, db=None) -> None:
        super().__init__(db)
        self._trade: OkxTradeClient | None = None

    def _client(self) -> OkxTradeClient:
        if self._trade is None:
            self._trade = OkxTradeClient()
        return self._trade

    def _perp_margin_mode(self) -> str:
        """永续全仓/逐仓；期权始终 cash，不受此设置影响。"""
        s = get_settings()
        raw = str(
            self.ledger.get_setting_str("perp_margin_mode", s.perp_margin_mode)
            or s.perp_margin_mode
            or "cross"
        ).strip().lower()
        return "isolated" if raw == "isolated" else "cross"

    def _perp_margin_mode_for_group(self, group_id: str | None) -> str:
        """平仓用开仓时写入的保证金模式；缺省回退当前设置。"""
        if group_id:
            g = self.db.fetchone(
                "SELECT perp_margin_mode FROM groups WHERE group_id=?", (group_id,)
            )
            if g is not None:
                m = str(g["perp_margin_mode"] or "").strip().lower()
                if m in ("cross", "isolated"):
                    return m
        return self._perp_margin_mode()

    def _guard_live(self) -> str | None:
        ok, reason = live_ready()
        if not ok:
            return reason
        return None

    def unrealized(self) -> dict:
        base = super().unrealized()
        if not base.get("has_position"):
            return base
        from .live_pnl import enrich_live_unrealized

        gid = base.get("group_id")
        open_at = None
        perp_inst = resolve_perp_inst_id(
            self.db, group_id=str(gid) if gid else None
        )
        if gid:
            g = self.db.fetchone(
                "SELECT open_at_ms, perp_inst_id FROM groups WHERE group_id=?",
                (gid,),
            )
            if g:
                open_at = int(g["open_at_ms"] or 0) or None
                if g["perp_inst_id"]:
                    perp_inst = str(g["perp_inst_id"])
        try:
            client = self._client()
        except Exception:
            return base
        return enrich_live_unrealized(
            base=base,
            db=self.db,
            client=client,
            exchange="okx",
            perp_inst_id=perp_inst,
            perp_side=str(base.get("perp_side") or ""),
            open_at_ms=open_at,
        )

    def open_group(
        self,
        *,
        group_id: str,
        bias: str,
        option_side: str,
        perp_side: str,
        option_inst_id: str,
        entry_index_px: float,
        strike: float | None = None,
        expiry_ymd: str | None = None,
    ) -> OpenResult:
        err = self._guard_live()
        if err:
            return OpenResult(ok=False, detail=err)

        claimed, claim_msg = claim_open_slot(self.db)
        if not claimed:
            return OpenResult(ok=False, detail=claim_msg)

        safe, safe_msg = assert_safe_to_open_live(self)
        if not safe:
            release_open_slot_if_opening(self.db)
            return OpenResult(ok=False, detail=safe_msg)

        s = live_settings()
        client = self._client()
        perp_inst = resolve_perp_inst_id(self.db)
        perp_qty = self.ledger.get_setting_float("perp_qty_eth", s.perp_qty_eth)
        opt_qty = self.ledger.get_setting_float("option_qty_eth", s.option_qty_eth)
        ct_mult = self._ct_mult(option_inst_id)
        opt_contracts = contracts_for_eth(opt_qty, ct_mult)

        # 意图先落库：崩溃后仍可 recover（含 option_inst_id）
        stamp_opening_intent(
            self.db,
            group_id=group_id,
            option_inst_id=option_inst_id,
            option_side=option_side,
            perp_side=perp_side,
            option_qty_eth=opt_qty,
            option_qty_contracts=float(opt_contracts),
            entry_index_px=entry_index_px,
        )

        # 期权：买入，张数 = contracts
        try:
            opt_fill = client.place_market(
                inst_id=option_inst_id,
                side="buy",
                sz=str(int(round(opt_contracts))),
                td_mode="cash",  # OKX 期权常见 cash；若账户不同可再扩展
            )
        except Exception as e:
            logger.exception("live open option failed")
            msg = str(e)
            # 已拿到 ordId：可能已成交，禁止释放 opening 以免重复开仓
            if "ordId=" in msg:
                return OpenResult(
                    ok=False,
                    detail=f"实盘开期权未确认成交(保留 opening 防重复开，请核对交易所): {e}",
                )
            release_open_slot_if_opening(self.db)
            return OpenResult(ok=False, detail=f"实盘开期权失败: {e}")

        # 以交易所实际成交张数回写名义
        filled_opt_contracts = float(opt_fill.sz) if opt_fill.sz and opt_fill.sz > 0 else float(
            int(round(opt_contracts))
        )
        opt_contracts = filled_opt_contracts
        opt_qty = eth_from_contracts(opt_contracts, ct_mult)
        stamp_opening_intent(
            self.db,
            group_id=group_id,
            option_inst_id=option_inst_id,
            option_side=option_side,
            perp_side=perp_side,
            option_qty_eth=opt_qty,
            option_qty_contracts=float(opt_contracts),
            entry_index_px=entry_index_px,
            option_entry_px=float(opt_fill.avg_px),
        )

        # 永续市价：按产品假设，失败原因实质为保证金不足 → 必须回滚期权
        mgn = self._perp_margin_mode()
        try:
            ct_val = client.get_ct_val(perp_inst, inst_type="SWAP")
            perp_sz = perp_open_contracts_okx(perp_qty_eth=perp_qty, ct_val=ct_val)
            if perp_side == "long":
                side, pos_side = "buy", "long"
            else:
                side, pos_side = "sell", "short"
            leverage = self.ledger.get_setting_float("leverage", s.leverage)
            try:
                client.set_leverage(
                    perp_inst, leverage, mgn_mode=mgn, pos_side=pos_side
                )
            except Exception as e_lev:
                logger.warning("okx set_leverage failed: %s", e_lev)
            perp_fill_live = client.place_market(
                inst_id=perp_inst,
                side=side,
                sz=str(perp_sz),
                td_mode=mgn,
                pos_side=pos_side,
            )
        except Exception as e:
            logger.exception("live open perp failed (likely margin); rollback option")
            # 永续可能已成交：先查仓；查失败或有仓均不得回滚期权
            try:
                live_perp = client.get_perp_pos_sz(
                    perp_inst, pos_side=("long" if perp_side == "long" else "short")
                )
            except Exception:
                live_perp = None
            if live_perp is None or live_perp > 1e-8:
                return OpenResult(
                    ok=False,
                    detail=(
                        f"永续开仓未确认(保留 opening，禁止回滚期权): {e}; "
                        f"ex_perp={live_perp}"
                    ),
                )
            try:
                client.place_market(
                    inst_id=option_inst_id,
                    side="sell",
                    sz=str(int(round(opt_contracts))),
                    td_mode="cash",
                    reduce_only=True,
                )
            except Exception as e2:
                logger.exception("live option rollback failed: %s", e2)
                self._persist_half_open(
                    group_id=group_id,
                    bias=bias,
                    option_side=option_side,
                    perp_side=perp_side,
                    option_inst_id=option_inst_id,
                    entry_index_px=entry_index_px,
                    strike=strike,
                    expiry_ymd=expiry_ymd,
                    opt_qty=opt_qty,
                    opt_contracts=opt_contracts,
                    of_px=float(opt_fill.avg_px),
                    of_fee=float(opt_fill.fee),
                    detail=f"保证金开永续失败且期权回滚失败: {e} / {e2}",
                )
                return OpenResult(
                    ok=False,
                    group_id=group_id,
                    detail=f"永续开仓失败(保证金)且期权回滚失败，已标记 half_open: {e} / {e2}",
                )
            release_open_slot_if_opening(self.db)
            return OpenResult(
                ok=False,
                detail=f"永续开仓失败(多为保证金不足)，已回滚期权: {e}",
            )

        of_px = float(opt_fill.avg_px)
        pf_px = float(perp_fill_live.avg_px)
        of_fee = float(opt_fill.fee)
        pf_fee = float(perp_fill_live.fee)
        filled_perp_sz = float(perp_fill_live.sz) if perp_fill_live.sz and perp_fill_live.sz > 0 else float(perp_sz)
        perp_qty = filled_perp_sz * float(ct_val)
        initial_premium = of_px * opt_qty
        of_notional = of_px * opt_qty
        pf_notional = pf_px * perp_qty

        # LIVE：交易所已成交，本地账本允许透支镜像，禁止因账本拒记导致「交易所有仓、DB 空」
        self.ledger.apply_cash(
            -(of_notional + of_fee),
            kind="open_option",
            group_id=group_id,
            note=f"LIVE open option {group_id}",
            allow_negative=True,
        )
        self.ledger.apply_cash(
            -pf_fee,
            kind="open_perp_fee",
            group_id=group_id,
            note=f"LIVE open perp {group_id}",
            allow_negative=True,
        )

        now = int(time.time() * 1000)
        with self.db._lock:
            self.db._conn.execute(
                """INSERT INTO groups(
                    group_id, status, bias, option_side, perp_side, option_inst_id, perp_inst_id,
                    strike, expiry_ymd, entry_index_px, initial_premium, open_at_ms, fees, slip_cost,
                    exec_mode, perp_margin_mode
                ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
                (
                    group_id,
                    "open",
                    bias,
                    option_side,
                    perp_side,
                    option_inst_id,
                    perp_inst,
                    strike,
                    expiry_ymd,
                    entry_index_px,
                    initial_premium,
                    now,
                    of_fee + pf_fee,
                    0.0,
                    "LIVE",
                    mgn,
                ),
            )
            self.db._conn.execute(
                """INSERT INTO fills(group_id, leg, action, side, inst_id, qty_eth, qty_contracts,
                   base_px, fill_px, fee, slip, notional, ts_ms, exec_mode)
                   VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
                (
                    group_id,
                    "option",
                    "open",
                    "long",
                    option_inst_id,
                    opt_qty,
                    opt_contracts,
                    of_px,
                    of_px,
                    of_fee,
                    0.0,
                    of_notional,
                    now,
                    "LIVE",
                ),
            )
            self.db._conn.execute(
                """INSERT INTO fills(group_id, leg, action, side, inst_id, qty_eth, qty_contracts,
                   base_px, fill_px, fee, slip, notional, ts_ms, exec_mode)
                   VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
                (
                    group_id,
                    "perp",
                    "open",
                    perp_side,
                    perp_inst,
                    perp_qty,
                    None,
                    pf_px,
                    pf_px,
                    pf_fee,
                    0.0,
                    pf_notional,
                    now + 1,
                    "LIVE",
                ),
            )
            self.db._conn.execute(
                """UPDATE positions SET
                    group_id=?, perp_side=?, perp_qty_eth=?, perp_entry_px=?,
                    option_inst_id=?, option_side=?, option_qty_eth=?, option_qty_contracts=?,
                    option_entry_px=?, entry_index_px=?, initial_premium=?, status=?
                    WHERE id=1""",
                (
                    group_id,
                    perp_side,
                    perp_qty,
                    pf_px,
                    option_inst_id,
                    option_side,
                    opt_qty,
                    opt_contracts,
                    of_px,
                    entry_index_px,
                    initial_premium,
                    "open",
                ),
            )
            self.db._conn.commit()

        try:
            from ..strategy.exits import lock_trade_exit_target

            lock_trade_exit_target(
                self.db, group_id=group_id, initial_premium=initial_premium
            )
        except Exception:
            logger.exception("lock exit target failed group=%s", group_id)

        leverage = self.ledger.get_setting_float("leverage", get_settings().leverage)
        perp_margin = (
            abs(float(pf_px) * float(perp_qty)) / float(leverage)
            if leverage and float(leverage) > 0
            else None
        )
        return OpenResult(
            ok=True,
            group_id=group_id,
            detail="opened_live",
            data={
                "group_id": group_id,
                "exec_mode": "LIVE",
                "bias": bias,
                "option_side": option_side,
                "perp_side": perp_side,
                "option_inst_id": option_inst_id,
                "strike": strike,
                "expiry_ymd": expiry_ymd,
                "option_ord": opt_fill.ord_id,
                "perp_ord": perp_fill_live.ord_id,
                "perp_qty_eth": float(perp_qty),
                "option_qty_eth": float(opt_qty),
                "perp_entry_px": float(pf_px),
                "option_entry_px": float(of_px),
                "initial_premium": initial_premium,
                "perp_margin": perp_margin,
                "leverage": float(leverage) if leverage else None,
                "fees": of_fee + pf_fee,
            },
        )

    def _persist_half_open(
        self,
        *,
        group_id: str,
        bias: str,
        option_side: str,
        perp_side: str,
        option_inst_id: str,
        entry_index_px: float,
        strike: float | None,
        expiry_ymd: str | None,
        opt_qty: float,
        opt_contracts: float,
        of_px: float,
        of_fee: float,
        detail: str,
    ) -> None:
        """期权已成交、永续未开且回滚失败 → 落 half_open，禁止新开，待 repair。"""
        perp_inst = resolve_perp_inst_id(self.db, group_id=group_id)
        initial_premium = of_px * opt_qty
        # 幂等：崩溃重入时勿二次扣权利金
        prior_cash = self.db.fetchone(
            "SELECT id FROM ledger_entries WHERE group_id=? AND kind='open_option' LIMIT 1",
            (group_id,),
        )
        if prior_cash is None:
            self.ledger.apply_cash(
                -(of_px * opt_qty + of_fee),
                kind="open_option",
                group_id=group_id,
                note=f"LIVE half_open option {group_id}",
                allow_negative=True,
            )
        now = int(time.time() * 1000)
        with self.db._lock:
            existing = self.db._conn.execute(
                "SELECT group_id FROM groups WHERE group_id=?", (group_id,)
            ).fetchone()
            if existing is None:
                self.db._conn.execute(
                    """INSERT INTO groups(
                        group_id, status, bias, option_side, perp_side, option_inst_id, perp_inst_id,
                        strike, expiry_ymd, entry_index_px, initial_premium, open_at_ms, fees, slip_cost,
                        exec_mode, note
                    ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
                    (
                        group_id,
                        "half_open",
                        bias,
                        option_side,
                        perp_side,
                        option_inst_id,
                        perp_inst,
                        strike,
                        expiry_ymd,
                        entry_index_px,
                        initial_premium,
                        now,
                        of_fee,
                        0.0,
                        "LIVE",
                        detail[:200],
                    ),
                )
                self.db._conn.execute(
                    """INSERT INTO fills(group_id, leg, action, side, inst_id, qty_eth, qty_contracts,
                       base_px, fill_px, fee, slip, notional, ts_ms, exec_mode)
                       VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
                    (
                        group_id,
                        "option",
                        "open",
                        "long",
                        option_inst_id,
                        opt_qty,
                        opt_contracts,
                        of_px,
                        of_px,
                        of_fee,
                        0.0,
                        of_px * opt_qty,
                        now,
                        "LIVE",
                    ),
                )
            self.db._conn.execute(
                """UPDATE positions SET
                    group_id=?, perp_side=?, perp_qty_eth=0, perp_entry_px=NULL,
                    option_inst_id=?, option_side=?, option_qty_eth=?, option_qty_contracts=?,
                    option_entry_px=?, entry_index_px=?, initial_premium=?, status='half_open'
                    WHERE id=1""",
                (
                    group_id,
                    perp_side,
                    option_inst_id,
                    option_side,
                    opt_qty,
                    opt_contracts,
                    of_px,
                    entry_index_px,
                    initial_premium,
                ),
            )
            self.db._conn.commit()
        try:
            from ..strategy.exits import lock_trade_exit_target

            lock_trade_exit_target(
                self.db, group_id=group_id, initial_premium=initial_premium
            )
        except Exception:
            logger.exception("lock exit target failed half_open group=%s", group_id)

    def repair_half_open(self) -> CloseResult:
        """卖出 half_open 残留期权，清本地状态。"""
        err = self._guard_live()
        if err:
            return CloseResult(ok=False, detail=err)
        pos = self.current_position()
        if pos.get("status") != "half_open":
            return CloseResult(ok=False, detail="非 half_open 状态")
        group_id = str(pos.get("group_id") or "")
        option_inst_id = str(pos.get("option_inst_id") or "")
        opt_contracts = float(pos.get("option_qty_contracts") or 0)
        opt_qty = float(pos.get("option_qty_eth") or 0)
        if not option_inst_id:
            return CloseResult(ok=False, detail="half_open 缺期权合约信息")
        client = self._client()
        ex_sz = exchange_option_abs_size(client, option_inst_id)
        if ex_sz is None:
            return CloseResult(ok=False, detail="half_open：无法核对交易所期权仓位")
        if ex_sz <= 1e-8:
            of_px, of_fee, of_notional = 0.0, 0.0, 0.0
            opt_contracts = 0.0
            opt_qty = 0.0
            opt_entry = float(pos.get("option_entry_px") or 0)
            # 已空：不发明现金
        else:
            opt_contracts = float(ex_sz)
            opt_qty = eth_from_contracts(opt_contracts, self._ct_mult(option_inst_id))
            try:
                opt_live = client.place_market(
                    inst_id=option_inst_id,
                    side="sell",
                    sz=str(max(1, int(round(opt_contracts)))),
                    td_mode="cash",
                    reduce_only=True,
                )
            except Exception as e:
                return CloseResult(ok=False, detail=f"half_open 平期权失败: {e}")
            of_px = float(opt_live.avg_px)
            of_fee = float(opt_live.fee)
            filled = float(opt_live.sz) if opt_live.sz and float(opt_live.sz) > 0 else 0.0
            if filled > 0:
                opt_contracts = filled
                opt_qty = eth_from_contracts(opt_contracts, self._ct_mult(option_inst_id))
            of_notional = of_px * opt_qty
            ex_left = exchange_option_abs_size(client, option_inst_id)
            if ex_left is None or ex_left > 1e-8:
                return CloseResult(
                    ok=False,
                    detail=f"half_open：卖后仍有仓或无法核对 left={ex_left}",
                )
            opt_entry = float(pos.get("option_entry_px") or of_px)
            self.ledger.apply_cash(
                of_notional - of_fee,
                kind="close_option",
                group_id=group_id or None,
                note="LIVE repair half_open",
                allow_negative=True,
            )
        now = int(time.time() * 1000)
        with self.db._lock:
            if group_id:
                self.db._conn.execute(
                    """INSERT INTO fills(group_id, leg, action, side, inst_id, qty_eth, qty_contracts,
                       base_px, fill_px, fee, slip, notional, ts_ms, exec_mode)
                       VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
                    (
                        group_id,
                        "option",
                        "close",
                        "flat",
                        option_inst_id,
                        opt_qty,
                        opt_contracts,
                        of_px,
                        of_px,
                        of_fee,
                        0.0,
                        of_notional,
                        now,
                        "LIVE",
                    ),
                )
                opt_pnl = (of_px - opt_entry) * opt_qty - of_fee
                self.db._conn.execute(
                    """UPDATE groups SET status=?, close_at_ms=?, close_reason=?, realized_pnl=?, note=?
                       WHERE group_id=?""",
                    (
                        "closed",
                        now,
                        "half_open_repair",
                        float(opt_pnl),
                        "repaired half_open",
                        group_id,
                    ),
                )
            self.db._conn.execute(
                """UPDATE positions SET
                    group_id=NULL, perp_side=NULL, perp_qty_eth=0, perp_entry_px=NULL,
                    option_inst_id=NULL, option_side=NULL, option_qty_eth=0, option_qty_contracts=0,
                    option_entry_px=NULL, entry_index_px=NULL, initial_premium=0,
                    exit_target_usdt=NULL, status='flat'
                    WHERE id=1"""
            )
            self.db._conn.commit()
        return CloseResult(
            ok=True,
            detail="half_open_repaired",
            data={"group_id": group_id, "exec_mode": "LIVE"},
        )

    def recover_opening(self) -> CloseResult:
        """恢复 stuck opening（交易所对账后 half_open/open/清槽）。"""
        err = self._guard_live()
        if err:
            return CloseResult(ok=False, detail=err)
        r = recover_stuck_opening(self)
        if r is None:
            return CloseResult(ok=False, detail="非 opening 状态")
        return r

    def _promote_opening_to_open(
        self,
        *,
        pos: dict,
        perp_inst: str,
        opt_sz: float,
        perp_total: float,
    ) -> CloseResult:
        """opening + 交易所双边有仓 → 落本地 open（用 stamp/设置数量）。"""
        s = live_settings()
        group_id = str(pos.get("group_id") or f"RCV-{int(time.time())}")
        option_inst_id = str(pos.get("option_inst_id") or "")
        option_side = str(pos.get("option_side") or "call")
        perp_side = str(pos.get("perp_side") or "long")
        of_px = float(pos.get("option_entry_px") or 0) or 0.0
        # 数量以交易所为准
        opt_contracts = float(opt_sz) if opt_sz > 0 else float(
            pos.get("option_qty_contracts") or 0
        )
        opt_qty = (
            eth_from_contracts(opt_contracts, self._ct_mult(option_inst_id))
            if opt_contracts > 0
            else float(pos.get("option_qty_eth") or 0)
        )
        perp_qty = float(pos.get("perp_qty_eth") or 0) or float(
            self.ledger.get_setting_float("perp_qty_eth", s.perp_qty_eth)
        )
        if perp_total > 0:
            try:
                ct_val = self._client().get_ct_val(perp_inst, inst_type="SWAP")
                if ct_val > 0:
                    perp_qty = float(perp_total) * float(ct_val)
            except Exception:
                pass
        entry_index = float(pos.get("entry_index_px") or 0) or 0.0
        # 永续入场价未知时用指数近似（仅恢复镜像）
        pf_px = entry_index if entry_index > 0 else of_px
        initial_premium = of_px * opt_qty
        mgn = self._perp_margin_mode()
        # 幂等补记权利金（崩溃在成交后、账本前时）
        prior_cash = self.db.fetchone(
            "SELECT id FROM ledger_entries WHERE group_id=? AND kind='open_option' LIMIT 1",
            (group_id,),
        )
        if prior_cash is None and of_px > 0 and opt_qty > 0:
            self.ledger.apply_cash(
                -(of_px * opt_qty),
                kind="open_option",
                group_id=group_id,
                note=f"LIVE recover promote open_option {group_id}",
                allow_negative=True,
            )
        now = int(time.time() * 1000)
        with self.db._lock:
            existing = self.db._conn.execute(
                "SELECT group_id FROM groups WHERE group_id=?", (group_id,)
            ).fetchone()
            if existing is None:
                self.db._conn.execute(
                    """INSERT INTO groups(
                        group_id, status, bias, option_side, perp_side, option_inst_id, perp_inst_id,
                        strike, expiry_ymd, entry_index_px, initial_premium, open_at_ms, fees, slip_cost,
                        exec_mode, perp_margin_mode, note
                    ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
                    (
                        group_id,
                        "open",
                        "recover",
                        option_side,
                        perp_side,
                        option_inst_id,
                        perp_inst,
                        None,
                        None,
                        entry_index,
                        initial_premium,
                        now,
                        0.0,
                        0.0,
                        "LIVE",
                        mgn,
                        "recover_opening both legs",
                    ),
                )
            self.db._conn.execute(
                """UPDATE positions SET
                    group_id=?, perp_side=?, perp_qty_eth=?, perp_entry_px=?,
                    option_inst_id=?, option_side=?, option_qty_eth=?, option_qty_contracts=?,
                    option_entry_px=?, entry_index_px=?, initial_premium=?, status='open'
                    WHERE id=1""",
                (
                    group_id,
                    perp_side,
                    perp_qty,
                    pf_px,
                    option_inst_id,
                    option_side,
                    opt_qty,
                    opt_contracts,
                    of_px,
                    entry_index,
                    initial_premium,
                ),
            )
            self.db._conn.commit()
        try:
            from ..strategy.exits import lock_trade_exit_target

            lock_trade_exit_target(
                self.db, group_id=group_id, initial_premium=initial_premium
            )
        except Exception:
            logger.exception("lock exit target failed recover group=%s", group_id)
        return CloseResult(
            ok=True,
            detail="recover_opening: 已提升为 open",
            data={"group_id": group_id, "exec_mode": "LIVE"},
        )

    def open_oo_group(
        self,
        *,
        group_id: str,
        call_inst_id: str,
        put_inst_id: str,
        call_strike: float,
        put_strike: float,
        entry_index_px: float,
        expiry_ymd: str | None = None,
    ) -> OpenResult:
        """期期 LIVE：先买 Call 再买 Put。"""
        err = self._guard_live()
        if err:
            return OpenResult(ok=False, detail=err)
        claimed, claim_msg = claim_open_slot(self.db)
        if not claimed:
            return OpenResult(ok=False, detail=claim_msg)
        safe, safe_msg = assert_safe_to_open_live(self)
        if not safe:
            release_open_slot_if_opening(self.db)
            return OpenResult(ok=False, detail=safe_msg)

        s = live_settings()
        client = self._client()
        call_qty = self.ledger.get_setting_float("option_qty_eth", s.option_qty_eth)
        put_qty = self.ledger.get_setting_float("oo_put_qty_eth", call_qty)
        call_ct = self._ct_mult(call_inst_id)
        put_ct = self._ct_mult(put_inst_id)
        call_contracts = contracts_for_eth(call_qty, call_ct)
        put_contracts = contracts_for_eth(put_qty, put_ct)
        stamp_opening_intent(
            self.db,
            group_id=group_id,
            option_inst_id=call_inst_id,
            option_side="call",
            perp_side=f"oo_put:{put_inst_id}",
            option_qty_eth=call_qty,
            option_qty_contracts=float(call_contracts),
            entry_index_px=entry_index_px,
        )
        try:
            call_fill = client.place_market(
                inst_id=call_inst_id,
                side="buy",
                sz=str(int(round(call_contracts))),
                td_mode="cash",
            )
        except Exception as e:
            logger.exception("live oo open call failed")
            if "ordId=" not in str(e):
                release_open_slot_if_opening(self.db)
            return OpenResult(ok=False, detail=f"期期开 Call 失败: {e}")

        call_contracts = (
            float(call_fill.sz)
            if call_fill.sz and call_fill.sz > 0
            else float(int(round(call_contracts)))
        )
        call_qty = eth_from_contracts(call_contracts, call_ct)
        # Put 用独立定仓数量，不跟 Call 成交量对齐
        put_contracts = contracts_for_eth(put_qty, put_ct)
        try:
            put_fill = client.place_market(
                inst_id=put_inst_id,
                side="buy",
                sz=str(int(round(put_contracts))),
                td_mode="cash",
            )
        except Exception as e:
            logger.exception("live oo open put failed; rolling back call")
            try:
                client.place_market(
                    inst_id=call_inst_id,
                    side="sell",
                    sz=str(int(round(call_contracts))),
                    td_mode="cash",
                )
            except Exception as e2:
                logger.exception("oo call rollback failed: %s", e2)
                return OpenResult(
                    ok=False,
                    detail=f"期期 Put 失败且 Call 回滚未确认(保留 opening): {e}",
                )
            release_open_slot_if_opening(self.db)
            return OpenResult(ok=False, detail=f"期期开 Put 失败已回滚 Call: {e}")

        put_contracts = (
            float(put_fill.sz)
            if put_fill.sz and put_fill.sz > 0
            else float(int(round(put_contracts)))
        )
        of_px = float(call_fill.avg_px)
        pf_px = float(put_fill.avg_px)
        call_qty = eth_from_contracts(call_contracts, call_ct)
        put_qty = eth_from_contracts(put_contracts, put_ct)
        call_prem = of_px * call_qty
        put_prem = pf_px * put_qty
        call_fee = float(getattr(call_fill, "fee", 0) or 0)
        put_fee = float(getattr(put_fill, "fee", 0) or 0)
        # 与永期 LIVE 开仓一致：镜像扣权利金+手续费
        self.ledger.apply_cash(
            -(call_prem + call_fee),
            kind="open_option",
            group_id=group_id,
            note=f"LIVE oo open call {group_id}",
            allow_negative=True,
        )
        self.ledger.apply_cash(
            -(put_prem + put_fee),
            kind="open_option",
            group_id=group_id,
            note=f"LIVE oo open put {group_id}",
            allow_negative=True,
        )
        now = int(time.time() * 1000)
        with self.db._lock:
            self.db._conn.execute(
                """INSERT INTO groups(
                    group_id, status, bias, option_side, perp_side, option_inst_id, perp_inst_id,
                    strike, expiry_ymd, entry_index_px, initial_premium, open_at_ms, fees, slip_cost,
                    exec_mode, hedge_mode, option2_inst_id, option2_side, strike2, initial_premium2
                ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
                (
                    group_id,
                    "open",
                    "option_option",
                    "call",
                    None,
                    call_inst_id,
                    None,
                    float(call_strike),
                    expiry_ymd,
                    entry_index_px,
                    call_prem,
                    now,
                    call_fee + put_fee,
                    0.0,
                    "LIVE",
                    "option_option",
                    put_inst_id,
                    "put",
                    float(put_strike),
                    put_prem,
                ),
            )
            for leg, inst, contracts, fill_px, fee, ts, q in (
                (
                    "option",
                    call_inst_id,
                    call_contracts,
                    of_px,
                    getattr(call_fill, "fee", 0),
                    now,
                    call_qty,
                ),
                (
                    "option2",
                    put_inst_id,
                    put_contracts,
                    pf_px,
                    getattr(put_fill, "fee", 0),
                    now + 1,
                    put_qty,
                ),
            ):
                self.db._conn.execute(
                    """INSERT INTO fills(group_id, leg, action, side, inst_id, qty_eth, qty_contracts,
                       base_px, fill_px, fee, slip, notional, ts_ms, exec_mode)
                       VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
                    (
                        group_id,
                        leg,
                        "open",
                        "long",
                        inst,
                        q,
                        contracts,
                        fill_px,
                        fill_px,
                        float(fee or 0),
                        0.0,
                        float(fill_px) * float(q),
                        ts,
                        "LIVE",
                    ),
                )
            self.db._conn.execute(
                """UPDATE positions SET
                    group_id=?, perp_side=NULL, perp_qty_eth=0, perp_entry_px=NULL,
                    option_inst_id=?, option_side='call', option_qty_eth=?, option_qty_contracts=?,
                    option_entry_px=?, entry_index_px=?, initial_premium=?, status='open',
                    hedge_mode='option_option', option2_inst_id=?, option2_side='put',
                    option2_qty_eth=?, option2_qty_contracts=?, option2_entry_px=?,
                    strike2=?, initial_premium2=?
                    WHERE id=1""",
                (
                    group_id,
                    call_inst_id,
                    call_qty,
                    call_contracts,
                    of_px,
                    entry_index_px,
                    call_prem,
                    put_inst_id,
                    put_qty,
                    put_contracts,
                    pf_px,
                    float(put_strike),
                    put_prem,
                ),
            )
            self.db._conn.commit()
        try:
            from ..strategy.exits import lock_trade_exit_target

            lock_trade_exit_target(
                self.db, group_id=group_id, initial_premium=call_prem + put_prem
            )
        except Exception:
            logger.exception("lock exit oo live failed")
        return OpenResult(
            ok=True,
            group_id=group_id,
            detail="opened_oo_live",
            data={"hedge_mode": "option_option", "exec_mode": "LIVE"},
        )

    def live_sell_oo_both(
        self, *, bypass_liquidity: bool = False, reason: str = ""
    ) -> None:
        """紧急等：按交易所张数市价卖掉 Call+Put。到期不卖（交易所自动结算）。"""
        if str(reason or "") == "expiry":
            logger.info("live_sell_oo_both: skip on expiry (exchange auto-settle)")
            return
        pos = self.current_position()
        client = self._client()
        for inst, contracts in (
            (str(pos.get("option_inst_id") or ""), float(pos.get("option_qty_contracts") or 0)),
            (str(pos.get("option2_inst_id") or ""), float(pos.get("option2_qty_contracts") or 0)),
        ):
            if not inst:
                continue
            ex_sz = exchange_option_abs_size(client, inst)
            if ex_sz is None:
                if not bypass_liquidity:
                    raise RuntimeError(f"期期卖腿查仓失败: {inst}")
                continue
            if ex_sz <= 1e-8:
                continue
            sz = max(1, int(round(float(ex_sz))))
            try:
                client.place_market(
                    inst_id=inst,
                    side="sell",
                    sz=str(sz),
                    td_mode="cash",
                    reduce_only=True,
                )
            except Exception:
                logger.exception("live_sell_oo_both failed inst=%s", inst)
                if not bypass_liquidity:
                    raise

    def close_oo_full(
        self, *, reason: str = "expiry", bypass_liquidity: bool = False
    ) -> CloseResult:
        """期期 LIVE 全平：以交易所空仓为准；到期不卖期权，仅镜像已结算。"""
        err = self._guard_live()
        if err:
            return CloseResult(ok=False, detail=err)
        pos = self.current_position()
        st = str(pos.get("status") or "")
        if st not in ("open", "closing") or not pos.get("group_id"):
            return CloseResult(ok=False, detail="无期期持仓可平")
        if not (
            str(pos.get("hedge_mode") or "") == "option_option"
            or pos.get("option2_inst_id")
        ):
            return CloseResult(ok=False, detail="非期期持仓")
        group_id = str(pos["group_id"])
        client = self._client()
        legs = [
            ("option", str(pos.get("option_inst_id") or ""), float(pos.get("option_qty_eth") or 0), float(pos.get("option_qty_contracts") or 0)),
            ("option2", str(pos.get("option2_inst_id") or ""), float(pos.get("option2_qty_eth") or 0), float(pos.get("option2_qty_contracts") or 0)),
        ]
        if reason != "expiry":
            try:
                self.live_sell_oo_both(bypass_liquidity=bypass_liquidity, reason=reason)
            except Exception as e:
                return CloseResult(ok=False, detail=f"期期全平卖腿失败: {e}")
        # 必须以交易所两腿皆空才落本地 flat
        for _leg, inst, _qty, _c in legs:
            if not inst:
                continue
            ex_sz = exchange_option_abs_size(client, inst)
            if ex_sz is None:
                return CloseResult(
                    ok=False, detail=f"期期全平无法核对交易所仓位: {inst}"
                )
            if ex_sz > 1e-8:
                return CloseResult(
                    ok=False,
                    detail=(
                        f"期期全平等待交易所{'到期结算' if reason == 'expiry' else '成交'}"
                        f": {inst} 仍有 {ex_sz}"
                    ),
                )
        now = int(time.time() * 1000)
        for i, (leg, inst, qty, contracts) in enumerate(legs):
            if not inst:
                continue
            if reason == "expiry":
                px, fee, notional, cash = self._live_option_settlement_fill(
                    option_inst_id=inst, qty_eth=qty, group_id=group_id
                )
            else:
                px, fee, notional, cash = 0.0, 0.0, 0.0, 0.0
            if abs(cash) > 1e-12:
                self.ledger.apply_cash(
                    cash,
                    kind="close_option",
                    group_id=group_id,
                    note=f"LIVE oo settle {leg} {reason}",
                    allow_negative=True,
                )
            with self.db._lock:
                self.db._conn.execute(
                    """INSERT INTO fills(group_id, leg, action, side, inst_id, qty_eth, qty_contracts,
                       base_px, fill_px, fee, slip, notional, ts_ms, exec_mode)
                       VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
                    (
                        group_id,
                        leg,
                        "close",
                        "flat",
                        inst,
                        qty,
                        contracts,
                        px,
                        px,
                        fee,
                        0.0,
                        notional,
                        now + i,
                        "LIVE",
                    ),
                )
                self.db._conn.commit()
        from ..sim.pnl import summarize_fills_pnl

        fill_rows = self.db.fetchall(
            "SELECT * FROM fills WHERE group_id=? ORDER BY id ASC", (group_id,)
        )
        summary = summarize_fills_pnl(list(fill_rows))
        net = float(summary.get("net_pnl") or 0.0)
        with self.db._lock:
            self.db._conn.execute(
                """UPDATE groups SET status=?, close_at_ms=?, close_reason=?, realized_pnl=?,
                   note=? WHERE group_id=?""",
                (
                    "closed",
                    int(time.time() * 1000),
                    reason,
                    net,
                    f"oo full close {reason} exchange_flat_or_settle",
                    group_id,
                ),
            )
            self.db._conn.execute(
                """UPDATE positions SET
                    group_id=NULL, perp_side=NULL, perp_qty_eth=0, perp_entry_px=NULL,
                    option_inst_id=NULL, option_side=NULL, option_qty_eth=0,
                    option_qty_contracts=0, option_entry_px=NULL, entry_index_px=NULL,
                    initial_premium=0, exit_target_usdt=NULL, status='flat',
                    hedge_mode=NULL, option2_inst_id=NULL, option2_side=NULL,
                    option2_qty_eth=0, option2_qty_contracts=0, option2_entry_px=NULL,
                    strike2=NULL, initial_premium2=NULL
                    WHERE id=1"""
            )
            self.db._conn.commit()
        return CloseResult(
            ok=True,
            detail="oo_full_closed_live",
            data={"group_id": group_id, "reason": reason, "net": net},
        )

    def close_winning_oo_leave_residual(
        self, *, reason: str = "target_oo_win"
    ) -> CloseResult:
        """期期达标：先标记 closing，再按交易所张数卖掉盈利腿，验空后落库。"""
        err = self._guard_live()
        if err:
            return CloseResult(ok=False, detail=err)
        pos = self.current_position()
        st = str(pos.get("status") or "")
        # 防重入：已在 closing 则优先收尾（须在 open 判断之前）
        if st == "closing":
            return self._finish_oo_win_after_exchange(reason=reason, pos=pos)
        if st != "open" or not pos.get("option2_inst_id"):
            return CloseResult(ok=False, detail="无期期持仓")
        upl = self.unrealized()
        call_upl = float(upl.get("option_upl") or 0)
        put_upl = float(upl.get("option2_upl") or 0)
        if call_upl >= put_upl and call_upl > 0:
            win_leg = "option"
            win_id = str(pos["option_inst_id"])
        elif put_upl > 0:
            win_leg = "option2"
            win_id = str(pos["option2_inst_id"])
        else:
            return CloseResult(ok=False, detail="无明确盈利腿")
        client = self._client()
        ex_sz = exchange_option_abs_size(client, win_id)
        if ex_sz is None:
            return CloseResult(ok=False, detail="期期平盈利腿：无法核对交易所仓位")
        if ex_sz <= 1e-8:
            return self._finish_oo_win_after_exchange(
                reason=reason, pos=pos, win_leg=win_leg, fill_px=0.0, fill_fee=0.0, fill_c=0.0
            )
        with self.db._lock:
            self.db._conn.execute(
                "UPDATE positions SET status='closing' WHERE id=1 AND status='open'"
            )
            self.db._conn.commit()
        try:
            live = client.place_market(
                inst_id=win_id,
                side="sell",
                sz=str(max(1, int(round(float(ex_sz))))),
                td_mode="cash",
                reduce_only=True,
            )
        except Exception as e:
            with self.db._lock:
                self.db._conn.execute(
                    "UPDATE positions SET status='open' WHERE id=1 AND status='closing'"
                )
                self.db._conn.commit()
            return CloseResult(ok=False, detail=f"期期平盈利腿失败: {e}")
        fill_c = float(live.sz) if live.sz and float(live.sz) > 0 else 0.0
        return self._finish_oo_win_after_exchange(
            reason=reason,
            pos=self.current_position(),
            win_leg=win_leg,
            fill_px=float(live.avg_px),
            fill_fee=float(live.fee),
            fill_c=fill_c,
        )

    def _finish_oo_win_after_exchange(
        self,
        *,
        reason: str,
        pos: dict,
        win_leg: str | None = None,
        fill_px: float | None = None,
        fill_fee: float | None = None,
        fill_c: float | None = None,
    ) -> CloseResult:
        """盈利腿卖后：确认交易所已空，再用真实成交落库。"""
        client = self._client()
        call_id = str(pos.get("option_inst_id") or "")
        put_id = str(pos.get("option2_inst_id") or "")
        if not win_leg:
            # 崩溃恢复：哪条腿已空且另一条仍有仓 → 已空者为赢腿
            c_sz = exchange_option_abs_size(client, call_id) if call_id else None
            p_sz = exchange_option_abs_size(client, put_id) if put_id else None
            if c_sz is None or p_sz is None:
                return CloseResult(
                    ok=False, detail="closing 收尾：无法核对交易所两腿仓位"
                )
            if c_sz <= 1e-8 and p_sz > 1e-8:
                win_leg = "option"
            elif p_sz <= 1e-8 and c_sz > 1e-8:
                win_leg = "option2"
            elif c_sz <= 1e-8 and p_sz <= 1e-8:
                # 两腿皆空：走全平镜像
                return self.close_oo_full(reason=reason, bypass_liquidity=True)
            else:
                return CloseResult(
                    ok=False, detail="closing 收尾：盈利腿仍在交易所，请重试卖出"
                )
        win_id = call_id if win_leg == "option" else put_id
        ex_win = exchange_option_abs_size(client, win_id)
        if ex_win is None:
            return CloseResult(ok=False, detail="closing 收尾：无法核对盈利腿仓位")
        if ex_win > 1e-8:
            return CloseResult(
                ok=False,
                detail=f"closing 收尾：盈利腿仍有仓 {ex_win}，禁止本地清仓",
            )
        return super().close_winning_oo_leave_residual(
            reason=reason,
            skip_market=True,
            live_fill_px=0.0 if fill_px is None else float(fill_px),
            live_fill_fee=0.0 if fill_fee is None else float(fill_fee),
            live_fill_contracts=fill_c,
            live_win_leg=win_leg,
        )

    def close_group(self, *, reason: str, bypass_liquidity: bool = False) -> CloseResult:
        err = self._guard_live()
        if err:
            return CloseResult(ok=False, detail=err)

        s = live_settings()
        pos = self.current_position()
        st = str(pos.get("status") or "")
        if st == "opening":
            return self.recover_opening()
        if st == "half_open":
            return self.repair_half_open()
        if st == "closing":
            # 期期盈利腿收尾
            if pos.get("option2_inst_id") or str(pos.get("hedge_mode") or "") == "option_option":
                return self.close_winning_oo_leave_residual(reason=reason or "closing_retry")
            return CloseResult(ok=False, detail="closing 非期期状态，请人工核对")
        is_oo = (
            str(pos.get("hedge_mode") or "") == "option_option"
            or bool(pos.get("option2_inst_id"))
        )
        if is_oo and st == "open":
            return self.close_oo_full(reason=reason, bypass_liquidity=bypass_liquidity)
        if st not in ("open", "option_closed_perp_pending") or not pos.get("group_id"):
            return CloseResult(ok=False, detail="无持仓可平")

        group_id = str(pos["group_id"])
        option_inst_id = str(pos["option_inst_id"])
        option_side = str(pos["option_side"])
        perp_side = str(pos["perp_side"])
        opt_qty = float(pos["option_qty_eth"])
        perp_qty = float(pos["perp_qty_eth"])
        opt_contracts = float(pos["option_qty_contracts"] or 0)
        perp_inst = resolve_perp_inst_id(self.db, group_id=group_id)
        client = self._client()
        is_expiry = reason == "expiry"
        pending_perp_only = st == "option_closed_perp_pending"

        sess = get_session()
        snap = sess.snapshot()
        strike = self._group_strike(group_id, option_inst_id)
        spot = self._close_spot_px(snap)
        intrinsic = None
        if strike is not None and spot is not None:
            intrinsic = option_intrinsic(
                option_side=option_side, strike=float(strike), spot=float(spot)
            )

        of_px = 0.0
        of_fee = 0.0
        of_slip = 0.0
        of_notional = 0.0

        option_apply_cash = True
        perp_already_flat = False

        if pending_perp_only:
            # 期权已在上次处理；只读上次平期权 fill（到期可无 fill：交易所自动结算）
            prev = self.db.fetchone(
                """SELECT fill_px, fee, notional, slip FROM fills
                   WHERE group_id=? AND leg='option' AND action='close'
                   ORDER BY id DESC LIMIT 1""",
                (group_id,),
            )
            if prev is None and not is_expiry:
                return CloseResult(
                    ok=False,
                    detail="option_closed_perp_pending 缺期权平仓记录，请人工核对",
                )
            if prev is not None:
                of_px = float(prev["fill_px"])
                of_fee = float(prev["fee"] or 0)
                of_notional = float(prev["notional"] or (of_px * opt_qty))
            else:
                of_px = 0.0
                of_fee = 0.0
                of_notional = 0.0
                self._ensure_option_closed_perp_pending(
                    group_id=group_id,
                    option_inst_id=option_inst_id,
                    opt_qty=opt_qty,
                    opt_contracts=opt_contracts,
                    of_px=of_px,
                    of_fee=of_fee,
                    of_notional=of_notional,
                    of_slip=0.0,
                    reason=reason,
                    apply_cash=False,
                )
            of_slip = 0.0
            option_apply_cash = False
        elif is_expiry:
            # 到期：交易所自动结算期权，本地只平永续；交割现金从账单镜像（禁止 intrinsic）
            ex_opt = exchange_option_abs_size(client, option_inst_id)
            if ex_opt is not None and ex_opt > 1e-8:
                logger.warning(
                    "expiry: option still on exchange sz=%.4f group=%s; "
                    "skip option, close perp only",
                    ex_opt,
                    group_id,
                )
            of_px, of_fee, of_notional, settle_cash = self._live_option_settlement_fill(
                option_inst_id=option_inst_id,
                qty_eth=opt_qty,
                group_id=group_id,
            )
            of_slip = 0.0
            option_apply_cash = abs(settle_cash) > 1e-12
            logger.info(
                "expiry: skip option order, close perp only group=%s settle_cash=%.4f",
                group_id,
                settle_cash,
            )
            self._ensure_option_closed_perp_pending(
                group_id=group_id,
                option_inst_id=option_inst_id,
                opt_qty=opt_qty,
                opt_contracts=opt_contracts,
                of_px=of_px,
                of_fee=of_fee,
                of_notional=of_notional,
                of_slip=of_slip,
                reason=reason,
                apply_cash=option_apply_cash,
            )
            pending_perp_only = True
        else:
            # 非到期：按交易所仓位卖期权
            ex_opt_pre = exchange_option_abs_size(client, option_inst_id)
            if ex_opt_pre is not None and ex_opt_pre > 1e-8:
                opt_contracts = float(ex_opt_pre)
                opt_qty = eth_from_contracts(
                    opt_contracts, self._ct_mult(option_inst_id)
                )
            try:
                if ex_opt_pre is not None and ex_opt_pre <= 1e-8:
                    raise RuntimeError("option already flat on exchange")
                if opt_contracts <= 0:
                    return CloseResult(
                        ok=False, detail="实盘平期权失败: 无有效张数"
                    )
                opt_live = client.place_market(
                    inst_id=option_inst_id,
                    side="sell",
                    sz=str(max(1, int(round(opt_contracts)))),
                    td_mode="cash",
                    reduce_only=True,
                )
                of_px = float(opt_live.avg_px)
                of_fee = float(opt_live.fee)
                filled_c = (
                    float(opt_live.sz) if opt_live.sz and opt_live.sz > 0 else 0.0
                )
                if filled_c <= 1e-12:
                    ex_after = exchange_option_abs_size(client, option_inst_id)
                    if ex_after is None:
                        return CloseResult(
                            ok=False,
                            detail="实盘平期权失败: 成交张数未知且无法核对仓位",
                        )
                    if ex_after > 1e-8:
                        return CloseResult(
                            ok=False,
                            detail=f"实盘平期权失败: 未确认成交仍有仓 {ex_after}",
                        )
                    # 已空：零现金镜像
                    of_px = 0.0
                    of_fee = 0.0
                    of_notional = 0.0
                    option_apply_cash = False
                else:
                    opt_contracts = filled_c
                    opt_qty = eth_from_contracts(
                        opt_contracts, self._ct_mult(option_inst_id)
                    )
                    of_notional = of_px * opt_qty
            except Exception as e:
                ex_opt = exchange_option_abs_size(client, option_inst_id)
                if ex_opt is not None and ex_opt <= 1e-8:
                    prev = self.db.fetchone(
                        """SELECT fill_px, fee, notional, slip FROM fills
                           WHERE group_id=? AND leg='option' AND action='close'
                           ORDER BY id DESC LIMIT 1""",
                        (group_id,),
                    )
                    if prev is not None:
                        of_px = float(prev["fill_px"])
                        of_fee = float(prev["fee"] or 0)
                        of_notional = float(prev["notional"] or (of_px * opt_qty))
                        of_slip = float(prev["slip"] or 0)
                        option_apply_cash = False
                    else:
                        # 已空且无历史 fill：零现金同步，禁止发明成交
                        of_px = 0.0
                        of_fee = 0.0
                        of_notional = 0.0
                        of_slip = 0.0
                        option_apply_cash = False
                    logger.warning(
                        "option already flat on exchange; skip resell: %s", e
                    )
                elif not bypass_liquidity:
                    return CloseResult(
                        ok=False,
                        detail=f"实盘平期权失败: {e}",
                        liquidity_wait=True,
                    )
                else:
                    return CloseResult(ok=False, detail=f"实盘平期权失败: {e}")

            self._ensure_option_closed_perp_pending(
                group_id=group_id,
                option_inst_id=option_inst_id,
                opt_qty=opt_qty,
                opt_contracts=opt_contracts,
                of_px=of_px,
                of_fee=of_fee,
                of_notional=of_notional,
                of_slip=of_slip,
                reason=reason,
                apply_cash=option_apply_cash,
            )
            pending_perp_only = True

        try:
            ct_val = client.get_ct_val(perp_inst, inst_type="SWAP")
            # 实盘平仓数量一律以交易所为准，禁止 DB fallback
            perp_sz = perp_close_contracts_okx(
                client,
                perp_inst=perp_inst,
                perp_side=perp_side,
                perp_qty_eth=perp_qty,
                ct_val=ct_val,
                allow_db_fallback=False,
            )
            if perp_sz is None:
                return CloseResult(
                    ok=False,
                    detail="期权已平，永续待平(无法核对交易所仓位，禁止空仓 finalize)",
                )
            if perp_sz <= 0:
                pf_px = 0.0
                pf_fee = 0.0
                perp_already_flat = True
                logger.warning(
                    "perp already flat on exchange; finalize without order group=%s",
                    group_id,
                )
            else:
                if perp_side == "long":
                    side, pos_side = "sell", "long"
                else:
                    side, pos_side = "buy", "short"
                perp_live = client.place_market(
                    inst_id=perp_inst,
                    side=side,
                    sz=str(perp_sz),
                    td_mode=self._perp_margin_mode_for_group(group_id),
                    pos_side=pos_side,
                    reduce_only=True,
                )
                pf_px = float(perp_live.avg_px)
                pf_fee = float(perp_live.fee)
                try:
                    perp_qty = float(perp_sz) * float(ct_val)
                    pos = {**pos, "perp_qty_eth": perp_qty}
                except Exception:
                    pass
        except Exception as e:
            return CloseResult(
                ok=False,
                detail=f"期权已平，永续待平(option_closed_perp_pending): {e}",
            )

        return self._finalize_dual_close(
            pos=pos,
            group_id=group_id,
            option_inst_id=option_inst_id,
            opt_qty=opt_qty,
            opt_contracts=opt_contracts,
            of_px=of_px,
            of_fee=of_fee,
            of_slip=of_slip,
            of_notional=of_notional,
            pf_px=pf_px,
            pf_fee=pf_fee,
            reason=reason,
            option_fill_already_written=bool(pending_perp_only),
            skip_option_cash=True,  # 已在 pending 路径入账或到期不入账
            skip_perp_cash=bool(perp_already_flat),
            skip_perp_fill=bool(perp_already_flat),
            settle_index_px=spot,
        )

    def _live_option_settlement_fill(
        self,
        *,
        option_inst_id: str,
        qty_eth: float,
        group_id: str | None = None,
        begin_ms: int | None = None,
    ) -> tuple[float, float, float, float]:
        """返回 (fill_px, fee, notional, cash_to_apply)。查不到交割则全 0。"""
        from .option_settle import fetch_option_settlement, settlement_to_fill

        open_ms = begin_ms
        if open_ms is None and group_id:
            g = self.db.fetchone(
                "SELECT open_at_ms FROM groups WHERE group_id=?", (group_id,)
            )
            if g and g["open_at_ms"]:
                open_ms = int(g["open_at_ms"])
        try:
            ex_name = load_runtime_settings().exchange
        except Exception:
            ex_name = get_settings().exchange
        st = fetch_option_settlement(
            self._client(),
            exchange=str(ex_name or "okx"),
            option_inst_id=option_inst_id,
            qty_eth=float(qty_eth),
            begin_ms=open_ms,
        )
        px, fee, notional = settlement_to_fill(st, qty_eth=float(qty_eth))
        cash = float(st.cash) if st.found else 0.0
        if st.found:
            logger.info(
                "option settlement %s source=%s px=%.6f fee=%.6f cash=%.6f (%s)",
                option_inst_id,
                st.source,
                px,
                fee,
                cash,
                st.detail,
            )
        return px, fee, notional, cash

    def _ensure_option_closed_perp_pending(
        self,
        *,
        group_id: str,
        option_inst_id: str,
        opt_qty: float,
        opt_contracts: float,
        of_px: float,
        of_fee: float,
        of_notional: float,
        of_slip: float,
        reason: str,
        apply_cash: bool = True,
    ) -> None:
        """幂等：落 option_closed_perp_pending；已有 close fill 则只改状态、不二次入账。"""
        st_now = str(self.current_position().get("status") or "")
        prev_close = self.db.fetchone(
            """SELECT id FROM fills
               WHERE group_id=? AND leg='option' AND action='close'
               ORDER BY id DESC LIMIT 1""",
            (group_id,),
        )
        if st_now == "option_closed_perp_pending" or prev_close is not None:
            if st_now != "option_closed_perp_pending":
                with self.db._lock:
                    self.db._conn.execute(
                        "UPDATE positions SET status='option_closed_perp_pending' WHERE id=1"
                    )
                    self.db._conn.commit()
            return
        self._mark_option_closed_perp_pending(
            group_id=group_id,
            option_inst_id=option_inst_id,
            opt_qty=opt_qty,
            opt_contracts=opt_contracts,
            of_px=of_px,
            of_fee=of_fee,
            of_notional=of_notional,
            of_slip=of_slip,
            reason=reason,
            apply_cash=apply_cash,
        )

    def _mark_option_closed_perp_pending(
        self,
        *,
        group_id: str,
        option_inst_id: str,
        opt_qty: float,
        opt_contracts: float,
        of_px: float,
        of_fee: float,
        of_notional: float,
        of_slip: float,
        reason: str,
        apply_cash: bool = True,
    ) -> None:
        if apply_cash:
            self.ledger.apply_cash(
                of_notional - of_fee,
                kind="close_option",
                group_id=group_id,
                note=f"LIVE close option pending perp {reason}",
                allow_negative=True,
            )
        now = int(time.time() * 1000)
        note = (
            f"option_closed_perp_pending:{reason}"
            if apply_cash
            else f"option_closed_perp_pending:{reason}:no_cash_exchange_sot"
        )
        with self.db._lock:
            self.db._conn.execute(
                """INSERT INTO fills(group_id, leg, action, side, inst_id, qty_eth, qty_contracts,
                   base_px, fill_px, fee, slip, notional, ts_ms, exec_mode)
                   VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
                (
                    group_id,
                    "option",
                    "close",
                    "flat",
                    option_inst_id,
                    opt_qty,
                    opt_contracts,
                    of_px,
                    of_px,
                    of_fee,
                    of_slip,
                    of_notional,
                    now,
                    "LIVE",
                ),
            )
            self.db._conn.execute(
                "UPDATE positions SET status='option_closed_perp_pending' WHERE id=1"
            )
            self.db._conn.execute(
                "UPDATE groups SET fees=COALESCE(fees,0)+?, note=? WHERE group_id=?",
                (of_fee if apply_cash else 0.0, note, group_id),
            )
            self.db._conn.commit()

    def _finalize_dual_close(
        self,
        *,
        pos: dict,
        group_id: str,
        option_inst_id: str,
        opt_qty: float,
        opt_contracts: float,
        of_px: float,
        of_fee: float,
        of_slip: float,
        of_notional: float,
        pf_px: float,
        pf_fee: float,
        reason: str,
        option_fill_already_written: bool,
        skip_option_cash: bool,
        skip_perp_cash: bool = False,
        skip_perp_fill: bool = False,
        settle_index_px: float | None = None,
    ) -> CloseResult:
        s = live_settings()
        perp_inst = resolve_perp_inst_id(self.db, group_id=group_id)
        perp_side = str(pos["perp_side"])
        perp_qty = float(pos["perp_qty_eth"])
        opt_entry = float(pos["option_entry_px"] or 0)
        perp_entry = float(pos["perp_entry_px"] or pf_px or 0)
        opt_pnl = (of_px - opt_entry) * opt_qty
        if perp_side == "long":
            perp_pnl = (pf_px - perp_entry) * perp_qty
        else:
            perp_pnl = (perp_entry - pf_px) * perp_qty

        if not skip_option_cash:
            self.ledger.apply_cash(
                of_notional - of_fee,
                kind="close_option",
                group_id=group_id,
                note=f"LIVE close option {reason}",
                allow_negative=True,
            )
        if not skip_perp_cash:
            self.ledger.apply_cash(
                perp_pnl - pf_fee,
                kind="close_perp",
                group_id=group_id,
                note=f"LIVE close perp {reason}",
                allow_negative=True,
            )

        now = int(time.time() * 1000)
        g = self.db.fetchone("SELECT * FROM groups WHERE group_id=?", (group_id,))
        base_fees = float((g["fees"] if g else 0) or 0)
        fees = (
            base_fees
            + (0.0 if skip_option_cash else of_fee)
            + (0.0 if skip_perp_cash else pf_fee)
        )
        # LIVE：真实成交价已含盘口冲击，不另计/不计模拟滑点
        of_slip = 0.0
        slip = 0.0
        from ..sim.pnl import summarize_fills_pnl

        with self.db._lock:
            if not option_fill_already_written:
                self.db._conn.execute(
                    """INSERT INTO fills(group_id, leg, action, side, inst_id, qty_eth, qty_contracts,
                       base_px, fill_px, fee, slip, notional, ts_ms, exec_mode)
                       VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
                    (
                        group_id,
                        "option",
                        "close",
                        "flat",
                        option_inst_id,
                        opt_qty,
                        opt_contracts,
                        of_px,
                        of_px,
                        of_fee,
                        of_slip,
                        of_notional,
                        now,
                        "LIVE",
                    ),
                )
            if not skip_perp_fill:
                self.db._conn.execute(
                    """INSERT INTO fills(group_id, leg, action, side, inst_id, qty_eth, qty_contracts,
                       base_px, fill_px, fee, slip, notional, ts_ms, exec_mode)
                       VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
                    (
                        group_id,
                        "perp",
                        "close",
                        "flat",
                        perp_inst,
                        perp_qty,
                        None,
                        pf_px,
                        pf_px,
                        pf_fee,
                        0.0,
                        pf_px * perp_qty,
                        now + 1,
                        "LIVE",
                    ),
                )
            fills = self.db._conn.execute(
                "SELECT * FROM fills WHERE group_id=? ORDER BY id ASC", (group_id,)
            ).fetchall()
            summary = summarize_fills_pnl(list(fills))
            net = summary.get("net_pnl")
            if net is None:
                net = opt_pnl + perp_pnl - of_fee - pf_fee
            close_index = (
                float(settle_index_px) if settle_index_px is not None else None
            )
            self.db._conn.execute(
                """UPDATE groups SET status=?, close_at_ms=?, close_reason=?, realized_pnl=?,
                   fees=?, slip_cost=?, settle_index_px=COALESCE(?, settle_index_px) WHERE group_id=?""",
                ("closed", now, reason, float(net), fees, slip, close_index, group_id),
            )
            self.db._conn.execute(
                """UPDATE positions SET
                    group_id=NULL, perp_side=NULL, perp_qty_eth=0, perp_entry_px=NULL,
                    option_inst_id=NULL, option_side=NULL, option_qty_eth=0, option_qty_contracts=0,
                    option_entry_px=NULL, entry_index_px=NULL, initial_premium=0,
                    exit_target_usdt=NULL, status='flat'
                    WHERE id=1"""
            )
            self.db._conn.commit()

        from .live_pnl import reconcile_closed_group_pnl

        g2 = self.db.fetchone(
            "SELECT open_at_ms, perp_inst_id FROM groups WHERE group_id=?",
            (group_id,),
        )
        net = reconcile_closed_group_pnl(
            db=self.db,
            client=self._client(),
            exchange="okx",
            group_id=group_id,
            perp_inst_id=str((g2["perp_inst_id"] if g2 else None) or perp_inst),
            open_at_ms=int(g2["open_at_ms"]) if g2 and g2["open_at_ms"] else None,
            local_net=float(net) if net is not None else None,
        )

        fills_summary = None
        try:
            from ..sim.pnl import summarize_fills_pnl

            fill_rows = self.db.fetchall(
                "SELECT * FROM fills WHERE group_id=? ORDER BY id ASC", (group_id,)
            )
            fills_summary = summarize_fills_pnl(list(fill_rows))
        except Exception:
            fills_summary = None
        return CloseResult(
            ok=True,
            detail="closed_live",
            data={
                "group_id": group_id,
                "reason": reason,
                "perp_pnl": (
                    fills_summary.get("perp_pnl")
                    if fills_summary
                    else None
                ),
                "option_pnl": (
                    fills_summary.get("option_pnl")
                    if fills_summary
                    else None
                ),
                "net": net,
                "net_pnl": net,
                "fees": fills_summary.get("fees_total") if fills_summary else None,
                "exec_mode": "LIVE",
                "pnl_source": "live_exchange",
            },
        )

    def _sync_residual_contracts_with_exchange(self, row: dict) -> dict | None:
        """按交易所持仓修正本地残留数量；已空仓则直接结清。返回待卖 row 或 None(已处理/跳过)。"""
        option_inst_id = str(row.get("option_inst_id") or "")
        group_id = str(row.get("group_id") or "")
        client = self._client()
        ex_sz = exchange_option_abs_size(client, option_inst_id)
        if ex_sz is None:
            logger.warning(
                "residual %s: exchange size unknown, skip until query ok", group_id
            )
            return None
        ct = self._ct_mult(option_inst_id)
        local_c = float(row.get("option_qty_contracts") or 0)
        if local_c <= 0:
            local_c = float(
                contracts_for_eth(float(row.get("option_qty_eth") or 0), ct) or 0
            )
        if ex_sz <= 1e-8:
            now_ms = int(time.time() * 1000)
            qty_eth = float(row.get("option_qty_eth") or 0)
            begin = None
            try:
                begin = int(row.get("created_at_ms") or 0) or None
            except Exception:
                begin = None
            px, fee, notional, _cash = self._live_option_settlement_fill(
                option_inst_id=option_inst_id,
                qty_eth=qty_eth,
                group_id=group_id,
                begin_ms=begin,
            )
            booked = self._book_residual_market_close(
                row,
                fill_px=px,
                fee=fee,
                notional=notional,
                slip=0.0,
                now_ms=now_ms,
                note=(
                    f"LIVE residual already flat; settlement px={px}"
                    if notional > 0 or fee > 0
                    else "LIVE residual already flat on exchange"
                ),
                exec_mode="LIVE",
                filled_contracts=float(row.get("option_qty_contracts") or 0) if (notional > 0 or fee > 0) else 0.0,
                remaining_contracts=0.0,
                close_reason="residual_premium_close",
            )
            logger.warning(
                "residual %s already flat on exchange; local settled=%s",
                group_id,
                booked is not None,
            )
            return None
        # 交易所数量为准：本地偏离则同步（含本地偏少）
        if abs(local_c - ex_sz) > 1e-8:
            rem_eth = eth_from_contracts(float(ex_sz), ct)
            init = float(row.get("initial_premium") or 0)
            local_eth = float(row.get("option_qty_eth") or 0)
            if local_eth > 1e-12:
                init = init * (rem_eth / local_eth)
            with self.db._lock:
                self.db._conn.execute(
                    """UPDATE residual_options SET
                       option_qty_eth=?, option_qty_contracts=?, initial_premium=?
                       WHERE group_id=? AND status='pending'""",
                    (rem_eth, float(ex_sz), init, group_id),
                )
                self.db._conn.commit()
            row = {
                **row,
                "option_qty_eth": rem_eth,
                "option_qty_contracts": float(ex_sz),
                "initial_premium": init,
            }
            logger.info(
                "residual %s sync contracts local=%.4f → ex=%.4f",
                group_id,
                local_c,
                ex_sz,
            )
        return row

    def try_close_one_residual(
        self, row: dict, *, skip_premium_ratio: bool = False
    ) -> dict | None:
        """LIVE：流动性通过后按最新买一 IOC 限价卖；自动路径另要求权利金比例。"""
        err = self._guard_live()
        if err:
            logger.warning("residual close blocked: %s", err)
            return None
        synced = self._sync_residual_contracts_with_exchange(row)
        if synced is None:
            return None
        row = synced
        require_ratio = not skip_premium_ratio
        skip, close_bid, _oq = self._evaluate_residual_premium_close(
            row, require_premium_ratio=require_ratio
        )
        if skip or close_bid is None:
            if skip:
                logger.debug(
                    "residual close skip %s: %s",
                    row.get("group_id"),
                    skip,
                )
            return None
        option_inst_id = str(row.get("option_inst_id") or "")
        opt_contracts = float(row.get("option_qty_contracts") or 0)
        opt_qty = float(row.get("option_qty_eth") or 0)
        if opt_contracts <= 0:
            ct = self._ct_mult(option_inst_id)
            opt_contracts = float(contracts_for_eth(opt_qty, ct) or 0)
        if opt_contracts <= 0:
            logger.warning(
                "residual close skip %s: bad contracts", row.get("group_id")
            )
            return None
        oq2 = self._quote_held_option(option_inst_id)
        if oq2 is None or oq2.bid is None:
            return None
        bid_px = float(oq2.bid)
        skip2 = self._residual_bid_gate(
            row, bid=bid_px, oq=oq2, require_premium_ratio=require_ratio
        )
        if skip2:
            logger.debug(
                "residual close recheck skip %s: %s",
                row.get("group_id"),
                skip2,
            )
            return None
        client = self._client()
        try:
            opt_live = client.place_ioc(
                inst_id=option_inst_id,
                side="sell",
                sz=str(max(1, int(round(opt_contracts)))),
                px=bid_px,
                td_mode="cash",
                reduce_only=True,
            )
        except Exception as e:
            logger.warning(
                "residual close bid-ioc sell failed %s: %s",
                row.get("group_id"),
                e,
            )
            return None
        of_px = float(opt_live.avg_px)
        of_fee = float(opt_live.fee)
        filled_c = float(opt_live.sz) if opt_live.sz and float(opt_live.sz) > 0 else 0.0
        if filled_c <= 1e-12:
            return None
        ex_left = exchange_option_abs_size(client, option_inst_id)
        if ex_left is None:
            logger.warning(
                "residual close %s: filled but remaining size unknown; leave pending",
                row.get("group_id"),
            )
            return None
        remaining = max(0.0, float(ex_left))
        fill_eth = eth_from_contracts(filled_c, self._ct_mult(option_inst_id))
        now_ms = int(time.time() * 1000)
        tag = "manual" if skip_premium_ratio else "mid"
        return self._book_residual_market_close(
            row,
            fill_px=of_px,
            fee=of_fee,
            notional=of_px * fill_eth,
            slip=0.0,
            now_ms=now_ms,
            note=f"LIVE residual {tag}-close at bid IOC px={bid_px}",
            exec_mode="LIVE",
            filled_contracts=filled_c,
            remaining_contracts=remaining,
            close_reason=(
                "residual_manual_close"
                if skip_premium_ratio
                else "residual_premium_close"
            ),
        )

    def _try_exchange_flatten_residual(
        self, row: dict, *, force: bool = False
    ) -> dict | None:
        """到期/紧急：优先交易所卖掉残留；失败返回 None（LIVE 禁止本地发明结算）。"""
        err = self._guard_live()
        if err:
            return None
        option_inst_id = str(row.get("option_inst_id") or "")
        client = self._client()
        ex_sz = exchange_option_abs_size(client, option_inst_id)
        if ex_sz is None:
            logger.warning(
                "residual flatten %s: exchange size unknown", row.get("group_id")
            )
            return None
        if ex_sz <= 1e-8:
            qty_eth = float(row.get("option_qty_eth") or 0)
            begin = None
            try:
                begin = int(row.get("created_at_ms") or 0) or None
            except Exception:
                begin = None
            gid = str(row.get("group_id") or "")
            px, fee, notional, _cash = self._live_option_settlement_fill(
                option_inst_id=option_inst_id,
                qty_eth=qty_eth,
                group_id=gid or None,
                begin_ms=begin,
            )
            return {
                "fill_px": px,
                "fee": fee,
                "notional": notional,
                "slip": 0.0,
                "filled_contracts": float(row.get("option_qty_contracts") or 0),
                "remaining_contracts": 0.0,
                "note": (
                    f"LIVE residual flat on exchange; settlement px={px}"
                    if notional > 0 or fee > 0
                    else "LIVE residual flat on exchange before settle"
                ),
                "exec_mode": "LIVE",
                "close_reason": "emergency" if force else "expiry",
            }
        if not force:
            # 到期：交易所自动结算期权，本地只镜像已空；仍有仓则等下次对账
            logger.info(
                "residual expiry %s: exchange still holds %.4f, wait auto-settle",
                row.get("group_id"),
                ex_sz,
            )
            return None
        opt_contracts = float(ex_sz)
        if opt_contracts <= 0:
            return None
        oq = self._quote_held_option(option_inst_id)
        bid_px = float(oq.bid) if oq is not None and oq.bid is not None else 0.0
        try:
            if bid_px > 0 and not force:
                opt_live = client.place_ioc(
                    inst_id=option_inst_id,
                    side="sell",
                    sz=str(max(1, int(round(opt_contracts)))),
                    px=bid_px,
                    td_mode="cash",
                    reduce_only=True,
                )
            else:
                opt_live = client.place_market(
                    inst_id=option_inst_id,
                    side="sell",
                    sz=str(max(1, int(round(opt_contracts)))),
                    td_mode="cash",
                    reduce_only=True,
                )
        except Exception as e:
            logger.warning(
                "residual exchange flatten failed %s force=%s: %s",
                row.get("group_id"),
                force,
                e,
            )
            return None
        filled_c = float(opt_live.sz) if opt_live.sz and float(opt_live.sz) > 0 else 0.0
        if filled_c <= 1e-12 and force:
            # 紧急：再试市价
            try:
                opt_live = client.place_market(
                    inst_id=option_inst_id,
                    side="sell",
                    sz=str(max(1, int(round(opt_contracts)))),
                    td_mode="cash",
                    reduce_only=True,
                )
                filled_c = (
                    float(opt_live.sz) if opt_live.sz and float(opt_live.sz) > 0 else 0.0
                )
            except Exception as e:
                logger.warning("residual emergency market sell failed: %s", e)
                return None
        if filled_c <= 1e-12:
            return None
        of_px = float(opt_live.avg_px)
        of_fee = float(opt_live.fee)
        fill_eth = eth_from_contracts(filled_c, self._ct_mult(option_inst_id))
        ex_left = exchange_option_abs_size(client, option_inst_id)
        remaining = max(0.0, float(ex_left)) if ex_left is not None else 0.0
        # 到期/紧急要求尽量结清：若仍有剩余且 force，不在此硬结（返回 None 让内在价值兜底会重复）
        # 有成交则先入账已成交部分；剩余留 pending 由下次处理，除非交易所已空
        return {
            "fill_px": of_px,
            "fee": of_fee,
            "notional": of_px * fill_eth,
            "slip": 0.0,
            "filled_contracts": filled_c,
            "remaining_contracts": remaining,
            "note": f"LIVE residual exchange flatten force={force}",
            "exec_mode": "LIVE",
            "close_reason": "emergency" if force else "expiry",
        }

    def close_perp_abandon_option(
        self, *, reason: str = "target_perp_only", require_deep_otm: bool = True
    ) -> CloseResult:
        err = self._guard_live()
        if err:
            return CloseResult(ok=False, detail=err)

        pos = self.current_position()
        st = str(pos.get("status") or "")
        if st not in ("open", "option_closed_perp_pending") or not pos.get("group_id"):
            return CloseResult(ok=False, detail="无持仓可平")
        # 若期权已平只剩永续，走 close_group 续平即可
        if st == "option_closed_perp_pending":
            return self.close_group(reason=reason, bypass_liquidity=True)

        # 优先尝试双腿全平（含交易所卖期权）
        dual = self.close_group(reason=reason, bypass_liquidity=True)
        if dual.ok:
            return dual

        # close_group 可能已卖掉期权并落 option_closed_perp_pending；勿再记 residual，只续平永续
        pos_after = self.current_position()
        if str(pos_after.get("status") or "") == "option_closed_perp_pending":
            return self.close_group(reason=reason, bypass_liquidity=True)

        if require_deep_otm and not self.option_is_deep_otm():
            return CloseResult(
                ok=False,
                detail=f"期权非远虚且双腿全平失败，应人工处理: {dual.detail}",
            )

        s = live_settings()
        group_id = str(pos["group_id"])
        perp_inst = resolve_perp_inst_id(self.db, group_id=group_id)
        perp_side = str(pos["perp_side"])
        perp_qty = float(pos["perp_qty_eth"])
        perp_entry = float(pos["perp_entry_px"])
        client = self._client()
        try:
            ct_val = client.get_ct_val(perp_inst, inst_type="SWAP")
            perp_sz = perp_close_contracts_okx(
                client,
                perp_inst=perp_inst,
                perp_side=perp_side,
                perp_qty_eth=perp_qty,
                ct_val=ct_val,
                allow_db_fallback=False,
            )
            if perp_sz is None:
                return CloseResult(
                    ok=False,
                    detail="弃期权平永续失败: 无法核对交易所永续仓位",
                )
            if perp_sz > 0:
                if perp_side == "long":
                    side, pos_side = "sell", "long"
                else:
                    side, pos_side = "buy", "short"
                perp_live = client.place_market(
                    inst_id=perp_inst,
                    side=side,
                    sz=str(perp_sz),
                    td_mode=self._perp_margin_mode_for_group(group_id),
                    pos_side=pos_side,
                    reduce_only=True,
                )
                pf_px = float(perp_live.avg_px)
                pf_fee = float(perp_live.fee)
                try:
                    perp_qty = float(perp_sz) * float(ct_val)
                except Exception:
                    pass
            else:
                # 永续已空：仍归档期权 residual
                pf_px = 0.0
                pf_fee = 0.0
                logger.warning(
                    "abandon: perp already flat on exchange; archive option residual group=%s",
                    group_id,
                )
        except Exception as e:
            return CloseResult(ok=False, detail=f"实盘平永续失败: {e}")

        if perp_side == "long":
            perp_pnl = (pf_px - perp_entry) * perp_qty if pf_px else 0.0
        else:
            perp_pnl = (perp_entry - pf_px) * perp_qty if pf_px else 0.0

        if abs(perp_pnl) + abs(pf_fee) > 1e-12:
            self.ledger.apply_cash(
                perp_pnl - pf_fee,
                kind="close_perp",
                group_id=group_id,
                note=f"LIVE close perp abandon option {reason}",
                allow_negative=True,
            )

        # 复用父类归档写入：临时改 fill 路径太重，直接调用父类会再平一次本地假价。
        # 因此把实盘价写入后走父类结构——这里内联父类 abandon 的 DB 段。
        option_inst_id = str(pos["option_inst_id"])
        option_side = str(pos["option_side"])
        strike = self._group_strike(group_id, option_inst_id)
        g = self.db.fetchone("SELECT * FROM groups WHERE group_id=?", (group_id,))
        expiry_ymd = str(g["expiry_ymd"]) if g and g["expiry_ymd"] else None
        expiry_ms = None
        if expiry_ymd:
            try:
                from ..exchange.expiry import expiry_ms_from_ymd

                expiry_ms = int(expiry_ms_from_ymd(expiry_ymd))
            except Exception:
                expiry_ms = None

        now = int(time.time() * 1000)
        open_fees = float((g["fees"] if g else 0) or 0)
        fees = open_fees + pf_fee
        slip = float((g["slip_cost"] if g else 0) or 0)
        interim_net = perp_pnl - open_fees - pf_fee
        spot = self._close_spot_px(get_session().snapshot())

        with self.db._lock:
            self.db._conn.execute(
                """INSERT INTO fills(group_id, leg, action, side, inst_id, qty_eth, qty_contracts,
                   base_px, fill_px, fee, slip, notional, ts_ms, exec_mode)
                   VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
                (
                    group_id,
                    "perp",
                    "close",
                    "flat",
                    perp_inst,
                    perp_qty,
                    None,
                    pf_px,
                    pf_px,
                    pf_fee,
                    0.0,
                    pf_px * perp_qty,
                    now,
                    "LIVE",
                ),
            )
            self.db._conn.execute(
                """INSERT INTO residual_options(
                    group_id, option_inst_id, option_side, option_qty_eth, option_qty_contracts,
                    option_entry_px, strike, expiry_ymd, expiry_ms, entry_index_px,
                    initial_premium, status, created_at_ms, note
                ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
                (
                    group_id,
                    option_inst_id,
                    option_side,
                    float(pos["option_qty_eth"]),
                    float(pos["option_qty_contracts"] or 0),
                    float(pos["option_entry_px"]),
                    float(strike) if strike is not None else None,
                    expiry_ymd,
                    expiry_ms,
                    float(pos["entry_index_px"] or 0),
                    float(pos["initial_premium"] or 0),
                    "pending",
                    now,
                    f"LIVE abandoned after {reason}; spot={spot}",
                ),
            )
            self.db._conn.execute(
                """UPDATE groups SET status=?, close_at_ms=?, close_reason=?, realized_pnl=?,
                   fees=?, slip_cost=?, note=?, exec_mode=? WHERE group_id=?""",
                (
                    "option_residual",
                    now,
                    reason,
                    interim_net,
                    fees,
                    slip,
                    "LIVE perp_closed; option residual until expiry",
                    "LIVE",
                    group_id,
                ),
            )
            self.db._conn.execute(
                """UPDATE positions SET
                    group_id=NULL, perp_side=NULL, perp_qty_eth=0, perp_entry_px=NULL,
                    option_inst_id=NULL, option_side=NULL, option_qty_eth=0, option_qty_contracts=0,
                    option_entry_px=NULL, entry_index_px=NULL, initial_premium=0,
                    exit_target_usdt=NULL, status='flat'
                    WHERE id=1"""
            )
            self.db._conn.commit()

        return CloseResult(
            ok=True,
            detail="perp_closed_option_residual_live",
            data={
                "group_id": group_id,
                "reason": reason,
                "mode": "target_perp_only",
                "perp_pnl": perp_pnl,
                "option_pnl": None,
                "interim_net": interim_net,
                "net": interim_net,
                "net_pnl": interim_net,
                "option_abandoned": True,
                "exec_mode": "LIVE",
            },
        )


def get_executor(db=None) -> Matcher:
    """按 MODE + 交易所返回执行器。"""
    from ..models.db import get_db

    database = db or get_db()
    s = get_settings()
    if s.is_sim:
        return Matcher(database)
    ex = load_runtime_settings().exchange
    if ex == "binance":
        from .binance_executor import BinanceLiveExecutor

        return BinanceLiveExecutor(database)
    return OkxLiveExecutor(database)