eth_hedge_sim/backend/app/live/executor.py

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

from __future__ import annotations

import logging
import time
from typing import Any

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
from ..sim.matcher import CloseResult, Matcher, OpenResult
from ..sim.pricing import option_expiry_settle, option_intrinsic
from ..strategy.session import get_session
from .okx_trade import OkxTradeClient

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 _guard_live(self) -> str | None:
        ok, reason = live_ready()
        if not ok:
            return reason
        return None

    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)

        s = get_settings()
        pos = self.current_position()
        if pos.get("status") == "open" and pos.get("group_id"):
            return OpenResult(ok=False, detail="已有持仓组，请先平仓")

        client = self._client()
        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)

        # 期权：买入，张数 = 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")
            return OpenResult(ok=False, detail=f"实盘开期权失败: {e}")

        # 永续：按仓位方向
        try:
            ct_val = client.get_ct_val(s.perp_inst_id, inst_type="SWAP")
            perp_sz = max(1, int(round(perp_qty / ct_val)))
            if perp_side == "long":
                side, pos_side = "buy", "long"
            else:
                side, pos_side = "sell", "short"
            perp_fill_live = client.place_market(
                inst_id=s.perp_inst_id,
                side=side,
                sz=str(perp_sz),
                td_mode="cross",
                pos_side=pos_side,
            )
        except Exception as e:
            logger.exception("live open perp failed; attempting option close")
            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)
                return OpenResult(
                    ok=False,
                    detail=f"永续开仓失败且期权回滚失败: {e} / {e2}",
                )
            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)
        initial_premium = of_px * opt_qty
        of_notional = of_px * opt_qty
        pf_notional = pf_px * perp_qty

        try:
            self.ledger.apply_cash(
                -(of_notional + of_fee),
                kind="open_option",
                group_id=group_id,
                note=f"LIVE open option {group_id}",
            )
            self.ledger.apply_cash(
                -pf_fee,
                kind="open_perp_fee",
                group_id=group_id,
                note=f"LIVE open perp {group_id}",
            )
        except RuntimeError as e:
            return OpenResult(ok=False, detail=str(e))

        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
                ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
                (
                    group_id,
                    "open",
                    bias,
                    option_side,
                    perp_side,
                    option_inst_id,
                    s.perp_inst_id,
                    strike,
                    expiry_ymd,
                    entry_index_px,
                    initial_premium,
                    now,
                    of_fee + pf_fee,
                    0.0,
                    "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,
                    "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,
                    s.perp_inst_id,
                    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()

        return OpenResult(
            ok=True,
            group_id=group_id,
            detail="opened_live",
            data={
                "group_id": group_id,
                "exec_mode": "LIVE",
                "option_ord": opt_fill.ord_id,
                "perp_ord": perp_fill_live.ord_id,
                "initial_premium": initial_premium,
                "fees": of_fee + pf_fee,
            },
        )

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

        s = get_settings()
        pos = self.current_position()
        if pos.get("status") != "open" 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)
        client = self._client()
        is_expiry = reason == "expiry"
        fee_rate = self._fee_rate()

        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

        if is_expiry:
            if intrinsic is None:
                return CloseResult(ok=False, detail="到期结算失败：缺行权价或标的价")
            of = option_expiry_settle(
                intrinsic=float(intrinsic), qty_eth=opt_qty, fee_rate=fee_rate
            )
            of_px, of_fee, of_slip, of_notional = of.fill_px, of.fee, of.slip, of.notional
        else:
            try:
                opt_live = client.place_market(
                    inst_id=option_inst_id,
                    side="sell",
                    sz=str(int(round(opt_contracts))),
                    td_mode="cash",
                    reduce_only=True,
                )
                of_px = float(opt_live.avg_px)
                of_fee = float(opt_live.fee)
                of_notional = of_px * opt_qty
            except Exception as e:
                if not bypass_liquidity:
                    return CloseResult(
                        ok=False,
                        detail=f"实盘平期权失败: {e}",
                        liquidity_wait=True,
                    )
                return CloseResult(ok=False, detail=f"实盘平期权失败: {e}")

        try:
            ct_val = client.get_ct_val(s.perp_inst_id, inst_type="SWAP")
            perp_sz = max(1, int(round(perp_qty / ct_val)))
            if perp_side == "long":
                side, pos_side = "sell", "long"
            else:
                side, pos_side = "buy", "short"
            perp_live = client.place_market(
                inst_id=s.perp_inst_id,
                side=side,
                sz=str(perp_sz),
                td_mode="cross",
                pos_side=pos_side,
                reduce_only=True,
            )
            pf_px = float(perp_live.avg_px)
            pf_fee = float(perp_live.fee)
        except Exception as e:
            return CloseResult(ok=False, detail=f"期权已平但永续平仓失败: {e}")

        opt_entry = float(pos["option_entry_px"])
        perp_entry = float(pos["perp_entry_px"])
        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

        self.ledger.apply_cash(
            of_notional - of_fee,
            kind="close_option",
            group_id=group_id,
            note=f"LIVE close option {reason}",
        )
        self.ledger.apply_cash(
            perp_pnl - pf_fee,
            kind="close_perp",
            group_id=group_id,
            note=f"LIVE close perp {reason}",
        )

        now = int(time.time() * 1000)
        g = self.db.fetchone("SELECT * FROM groups WHERE group_id=?", (group_id,))
        fees = float((g["fees"] if g else 0) or 0) + of_fee + pf_fee
        slip = float((g["slip_cost"] if g else 0) or 0) + of_slip
        from ..sim.pnl import summarize_fills_pnl

        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(
                """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",
                    s.perp_inst_id,
                    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
            self.db._conn.execute(
                """UPDATE groups SET status=?, close_at_ms=?, close_reason=?, realized_pnl=?,
                   fees=?, slip_cost=? WHERE group_id=?""",
                ("closed", now, reason, float(net), fees, slip, 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, status='flat'
                    WHERE id=1"""
            )
            self.db._conn.commit()

        return CloseResult(
            ok=True,
            detail="closed_live",
            data={"group_id": group_id, "reason": reason, "net_pnl": net, "exec_mode": "LIVE"},
        )

    def close_perp_abandon_option(self, *, reason: str = "target_perp_only") -> CloseResult:
        err = self._guard_live()
        if err:
            return CloseResult(ok=False, detail=err)
        # 先校验远虚，再实盘只平永续，其余写入复用父类逻辑的简化版：
        if not self.option_is_deep_otm():
            return CloseResult(ok=False, detail="期权非远虚，应走双腿全平")

        s = get_settings()
        pos = self.current_position()
        if pos.get("status") != "open" or not pos.get("group_id"):
            return CloseResult(ok=False, detail="无持仓可平")

        group_id = str(pos["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(s.perp_inst_id, inst_type="SWAP")
            perp_sz = max(1, int(round(perp_qty / ct_val)))
            if perp_side == "long":
                side, pos_side = "sell", "long"
            else:
                side, pos_side = "buy", "short"
            perp_live = client.place_market(
                inst_id=s.perp_inst_id,
                side=side,
                sz=str(perp_sz),
                td_mode="cross",
                pos_side=pos_side,
                reduce_only=True,
            )
        except Exception as e:
            return CloseResult(ok=False, detail=f"实盘平永续失败: {e}")

        pf_px = float(perp_live.avg_px)
        pf_fee = float(perp_live.fee)
        if perp_side == "long":
            perp_pnl = (pf_px - perp_entry) * perp_qty
        else:
            perp_pnl = (perp_entry - pf_px) * perp_qty

        self.ledger.apply_cash(
            perp_pnl - pf_fee,
            kind="close_perp",
            group_id=group_id,
            note=f"LIVE close perp abandon option {reason}",
        )

        # 复用父类归档写入：临时改 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",
                    s.perp_inst_id,
                    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_reason=?, realized_pnl=?,
                   fees=?, slip_cost=?, note=?, exec_mode=? WHERE group_id=?""",
                (
                    "option_residual",
                    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, 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", "exec_mode": "LIVE"},
        )


class BinanceLiveStub(Matcher):
    def open_group(self, **kwargs: Any) -> OpenResult:  # type: ignore[override]
        return OpenResult(ok=False, detail="币安实盘下单尚未接入，请使用 OKX 或切回 SIM")

    def close_group(self, **kwargs: Any) -> CloseResult:  # type: ignore[override]
        return CloseResult(ok=False, detail="币安实盘下单尚未接入，请使用 OKX 或切回 SIM")

    def close_perp_abandon_option(self, **kwargs: Any) -> CloseResult:  # type: ignore[override]
        return CloseResult(ok=False, detail="币安实盘下单尚未接入，请使用 OKX 或切回 SIM")


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":
        return BinanceLiveStub(database)
    return OkxLiveExecutor(database)