crypto_monitor/lib/options/options_close_exec_lib.py

"""期权平仓执行:只锁买一限价卖出;永不市价."""
from __future__ import annotations

import time
from typing import Any

from lib.options.options_close_gate_lib import (
    clear_close_gate,
    is_close_gate_passed,
    mark_close_gate_passed,
    update_close_gate,
)
from lib.options.options_pricing_lib import (
    estimate_close_by_bids,
    fetch_option_mark_px,
    is_stub_bid_px,
    total_premium,
)


def _safe_float(v: Any) -> float | None:
    if v is None or v == "":
        return None
    try:
        return float(v)
    except (TypeError, ValueError):
        return None


def _open_premium_paid(cfg: dict[str, Any], inst_id: str) -> float | None:
    try:
        conn = cfg["get_db"]()
        try:
            from lib.options.options_db import init_options_tables, sum_open_premium_paid

            init_options_tables(conn)
            return sum_open_premium_paid(conn, inst_id)
        finally:
            conn.close()
    except Exception:
        pass
    return None


def _pos_close_refs(ex: Any, pos: dict[str, Any], quote: dict[str, Any] | None = None) -> tuple[float | None, float | None]:
    from lib.exchange.okx_options_lib import option_fields_from_inst_id
    from lib.options.options_pricing_lib import close_ref_prices

    inst_id = str(pos.get("instId") or pos.get("inst_id") or "")
    mark = _safe_float(pos.get("markPx")) or _safe_float((quote or {}).get("mark_px") or (quote or {}).get("mark"))
    if mark is None:
        mark = fetch_option_mark_px(ex, inst_id)
    opt_type = pos.get("optType") or (quote or {}).get("opt_type")
    strike = _safe_float(pos.get("stk")) or _safe_float((quote or {}).get("strike"))
    if not opt_type or strike is None:
        pt, ps = option_fields_from_inst_id(inst_id)
        opt_type = opt_type or pt
        if strike is None:
            strike = ps
    idx = _safe_float(pos.get("idxPx")) or _safe_float((quote or {}).get("index_px"))
    return close_ref_prices(
        mark_px=mark,
        opt_type=str(opt_type or ""),
        strike=strike,
        index_px=idx,
        inst_id=inst_id,
    )


def _avail_sheets(pos: dict[str, Any]) -> int:
    avail = _safe_float(pos.get("availPos"))
    if avail is None or avail <= 0:
        avail = abs(_safe_float(pos.get("pos")) or 0)
    return max(0, int(avail or 0))


def _cancel_sell_pending(ex: Any, inst_id: str) -> None:
    try:
        pending = ex.private_get_trade_orders_pending({"instType": "OPTION", "instId": inst_id}) or {}
        for o in pending.get("data") or []:
            if str(o.get("side") or "").lower() != "sell":
                continue
            oid = o.get("ordId")
            if not oid:
                continue
            try:
                ex.private_post_trade_cancel_order({"instId": inst_id, "ordId": oid})
            except Exception:
                pass
    except Exception:
        pass


def close_option_by_bid1(
    cfg: dict[str, Any],
    ex: Any,
    inst_id: str,
    *,
    sheets: int | None = None,
    require_recycle_gate: bool = False,
    signal_note: str | None = None,
) -> dict[str, Any]:
    """
    本轮只吃买一深度:
    - 本批张数 = min(请求张数, 持仓, 买一深度)
    - 限价 = 校验通过时锁定的买一价
    - 永不市价
    - 始终校验有效流动性(残档买一禁止)
    - require_recycle_gate=True 时:首次还需目标门控(权利金×倍数或净盈亏,U 口径)并持续 hold 秒;
      一旦通过后对同仓续批只验流动性
    """
    from lib.exchange.okx_options_lib import (
        _pos_side_from_position,
        invalidate_option_positions_cache,
    )

    inst_id = (inst_id or "").strip()
    if not inst_id:
        return {"ok": False, "msg": "缺少 inst_id"}

    q = cfg["quote_option_contract"](ex, inst_id)
    if not q.get("ok"):
        return {"ok": False, "msg": q.get("msg") or "报价失败"}
    tick_sz = q.get("tick_sz")
    ct_mult = float(q.get("ct_mult") or 0.01)

    raw_positions = cfg["fetch_option_positions"](ex)
    if raw_positions is None:
        return {"ok": False, "msg": "获取期权持仓失败"}
    pos = next((p for p in raw_positions if str(p.get("instId")) == inst_id), None)
    if not pos:
        clear_close_gate(inst_id)
        return {"ok": False, "msg": "未找到持仓", "already_flat": True}

    avail = _avail_sheets(pos)
    want = int(sheets) if sheets else avail
    want = min(want, avail)
    if want < 1:
        clear_close_gate(inst_id)
        return {"ok": False, "msg": "可平张数不足", "already_flat": True}

    td_mode = str(pos.get("mgnMode") or cfg.get("td_mode") or "isolated")
    pos_side = _pos_side_from_position(pos) or "net"
    mark_px, intrinsic_px = _pos_close_refs(ex, pos, q)
    premium_paid = _open_premium_paid(cfg, inst_id)
    if premium_paid is None:
        premium_paid = _safe_float(pos.get("premium_paid"))
    premium_ccy = str(q.get("premium_ccy") or pos.get("premium_ccy") or "USDC").strip().upper() or "USDC"
    index_px = _safe_float(pos.get("idxPx")) or _safe_float(q.get("index_px"))

    # 已有未成交卖平单:等成交,不撤不重挂
    try:
        pending = ex.private_get_trade_orders_pending({"instType": "OPTION", "instId": inst_id}) or {}
        sell_pending = [
            o
            for o in (pending.get("data") or [])
            if str(o.get("side") or "").lower() == "sell" and o.get("ordId")
        ]
        if sell_pending:
            time.sleep(0.5)
            invalidate_option_positions_cache()
            raw_positions = cfg["fetch_option_positions"](ex)
            if raw_positions is None:
                return {"ok": False, "msg": "获取期权持仓失败"}
            pos = next((p for p in raw_positions if str(p.get("instId")) == inst_id), None)
            if not pos or _avail_sheets(pos) < 1:
                clear_close_gate(inst_id)
                return {
                    "ok": True,
                    "already_flat": True,
                    "msg": "已有限价卖单成交",
                    "close_ord_id": ",".join(str(o.get("ordId")) for o in sell_pending),
                    "fully_closed": True,
                    "submitted_sheets": want,
                    "remaining_sheets": 0,
                    "mode": "bid1",
                }
            return {
                "ok": False,
                "msg": "等待已有买一限价卖单成交",
                "stopped_reason": "pending_close_order",
                "close_ord_id": ",".join(str(o.get("ordId")) for o in sell_pending),
            }
    except Exception:
        pass

    book = cfg["fetch_option_book_depth"](ex, inst_id, 1)
    preview = estimate_close_by_bids(
        book.get("bids") or [],
        want,
        ct_mult=ct_mult,
        premium_paid=premium_paid,
        mark_px=mark_px,
        intrinsic_px=intrinsic_px,
        max_levels=1,
    )
    if preview.get("bid_invalid") or preview.get("auto_close_blocked"):
        # 不撤他人挂单:仅拒绝本轮下单
        update_close_gate(
            inst_id,
            recycle_usdc=None,
            premium_paid=premium_paid,
            premium_ccy=premium_ccy,
            index_px=index_px,
        )
        return {
            "ok": False,
            "msg": preview.get("bid_invalid_reason") or "暂无有效买盘,禁止平仓",
            "stopped_reason": "stub_bid",
            "auto_close_blocked": True,
            "liquidity_blocked": True,
        }

    levels = preview.get("levels") or []
    if not levels:
        bid_px = _safe_float(q.get("bid"))
        stub, stub_reason = is_stub_bid_px(bid_px, mark_px=mark_px, intrinsic_px=intrinsic_px)
        if stub or bid_px is None or bid_px <= 0:
            update_close_gate(
                inst_id,
                recycle_usdc=None,
                premium_paid=premium_paid,
                premium_ccy=premium_ccy,
                index_px=index_px,
            )
            return {
                "ok": False,
                "msg": stub_reason or "暂无买一,无法限价平仓",
                "stopped_reason": "stub_bid" if stub else "no_bid",
                "auto_close_blocked": True,
                "liquidity_blocked": True,
            }
        return {
            "ok": False,
            "msg": "暂无买一深度,无法平仓",
            "stopped_reason": "no_bid_depth",
            "liquidity_blocked": True,
        }

    level = levels[0]
    level_sheets = int(level.get("sheets") or 0)
    level_px = float(level.get("px") or 0)
    if level_sheets <= 0 or level_px <= 0:
        return {"ok": False, "msg": "买一深度无效", "stopped_reason": "invalid_bid_depth"}

    stub_lv, stub_lv_reason = is_stub_bid_px(level_px, mark_px=mark_px, intrinsic_px=intrinsic_px)
    if stub_lv:
        update_close_gate(
            inst_id,
            recycle_usdc=None,
            premium_paid=premium_paid,
            premium_ccy=premium_ccy,
            index_px=index_px,
        )
        return {
            "ok": False,
            "msg": stub_lv_reason or "暂无有效买盘,禁止平仓",
            "stopped_reason": "stub_bid",
            "auto_close_blocked": True,
            "liquidity_blocked": True,
        }

    # 自动平仓:2×权利金门控(首次);通过后同仓续批只验流动性
    gate = update_close_gate(
        inst_id,
        recycle_usdc=_safe_float(preview.get("total_received")),
        premium_paid=premium_paid,
        premium_ccy=premium_ccy,
        index_px=index_px,
    )
    if require_recycle_gate and not is_close_gate_passed(inst_id) and not gate.get("ready"):
        return {
            "ok": False,
            "msg": gate.get("msg") or "平仓门控未就绪(需可回收≥2×权利金并持续一段时间)",
            "stopped_reason": "close_gate",
            "auto_close_blocked": True,
            "close_gate": gate,
        }

    locked_bid_px = level_px
    before_avail = avail
    order = cfg["place_option_limit_order"](
        ex,
        inst_id=inst_id,
        side="sell",
        sheets=level_sheets,
        price=locked_bid_px,
        td_mode=td_mode,
        tick_sz=tick_sz,
        reduce_only=True,
        pos_side=pos_side,
    )
    if not order.get("ok"):
        return {
            "ok": False,
            "msg": order.get("msg") or "买一限价平仓失败",
            "stopped_reason": "order_failed",
            "locked_bid_px": locked_bid_px,
            "batch_sheets": level_sheets,
        }
    # 仅下单被接受后才记门控已通过,避免下单失败却跳过后续 2× 等待
    if require_recycle_gate and gate.get("ready"):
        mark_close_gate_passed(inst_id)

    px = float(order.get("px", locked_bid_px))
    oid = str((order.get("data") or {}).get("ordId") or "")
    prem_recv = round(total_premium(px, level_sheets * ct_mult), 4)
    time.sleep(0.6)
    invalidate_option_positions_cache()
    raw2 = cfg["fetch_option_positions"](ex)
    if raw2 is None:
        return {
            "ok": False,
            "msg": "下单后获取持仓失败,未确认是否成交",
            "stopped_reason": "position_fetch_failed",
            "locked_bid_px": locked_bid_px,
            "batch_sheets": level_sheets,
            "close_ord_id": oid or None,
            "fully_closed": False,
        }
    after_pos = next((p for p in raw2 if str(p.get("instId")) == inst_id), None)
    after_avail = _avail_sheets(after_pos) if after_pos else 0
    reduced = max(0, before_avail - after_avail)
    remaining_pos = after_avail
    fully_closed = remaining_pos < 1

    if fully_closed:
        clear_close_gate(inst_id)
        conn = cfg["get_db"]()
        try:
            from lib.options.options_db import init_options_tables

            init_options_tables(conn)
            open_rows = conn.execute(
                """
                SELECT id, premium_paid FROM options_trades
                WHERE inst_id = ? AND status = 'open'
                ORDER BY id ASC
                """,
                (inst_id,),
            ).fetchall()
            total_paid = sum(float(r["premium_paid"] or 0) for r in open_rows)
            allocated = 0.0
            for i, row in enumerate(open_rows):
                paid = float(row["premium_paid"] or 0)
                if i == len(open_rows) - 1:
                    recv = round(prem_recv - allocated, 4)
                elif total_paid > 0:
                    recv = round(prem_recv * (paid / total_paid), 4)
                    allocated += recv
                else:
                    recv = round(prem_recv / len(open_rows), 4)
                    allocated += recv
                pnl = round(recv - paid, 4)
                note_sql = ""
                params: list[Any] = [px, recv, pnl, oid or None]
                if signal_note and i == len(open_rows) - 1:
                    note_sql = """,
                    signal_note = CASE
                        WHEN signal_note IS NULL OR TRIM(signal_note) = '' THEN ?
                        ELSE signal_note
                    END"""
                    params.append(signal_note)
                params.append(int(row["id"]))
                conn.execute(
                    f"""
                    UPDATE options_trades
                    SET status = 'closed', close_quote = ?, premium_received = ?,
                        realized_pnl = ?, close_ord_id = ?, closed_at = CURRENT_TIMESTAMP
                        {note_sql}
                    WHERE id = ?
                    """,
                    tuple(params),
                )
            conn.commit()
        finally:
            conn.close()
    elif require_recycle_gate:
        # 自动平已挂过单:同仓续批只验流动性
        mark_close_gate_passed(inst_id)

    out = {
        "ok": True,
        "mode": "bid1",
        "orders": [{"order": order, "px": px, "sheets": level_sheets}],
        "bid": px,
        "locked_bid_px": locked_bid_px,
        "submitted_sheets": level_sheets,
        "filled_or_reduced_sheets": min(reduced, level_sheets) if reduced else 0,
        "remaining_sheets": remaining_pos,
        "premium_received": prem_recv,
        "stopped_reason": None if fully_closed else ("partial_bid1" if reduced > 0 else "order_not_filled"),
        "close_ord_id": oid or None,
        "fully_closed": fully_closed,
        "msg": (
            f"已按买一 {locked_bid_px:g} 提交 {level_sheets} 张"
            + ("" if fully_closed else f",剩余 {remaining_pos} 张待下次平仓")
        ),
    }
    if fully_closed:
        try:
            from lib.options.options_coin_open_lib import maybe_sell_spot_after_close

            spot_sell = maybe_sell_spot_after_close(cfg, ex, inst_id=inst_id, close_result=out)
            if spot_sell is not None:
                out["spot_sell"] = spot_sell
                if spot_sell.get("bridge_status") == "pending_sell_spot":
                    out["msg"] = str(out.get("msg") or "") + ";卖回 USDT 失败,请重试卖回"
        except Exception as e:
            out["spot_sell"] = {"ok": False, "msg": str(e)}
    return out


# 兼容旧名
def close_option_by_bid_depth(
    cfg: dict[str, Any],
    ex: Any,
    inst_id: str,
    *,
    sheets: int | None = None,
) -> dict[str, Any]:
    return close_option_by_bid1(
        cfg,
        ex,
        inst_id,
        sheets=sheets,
        require_recycle_gate=True,
        signal_note="目标位平仓",
    )