crypto_okx/lib/options/options_close_gate_lib.py

"""期权按买盘平仓门控:权利金倍数或净盈亏(USDT)达标并持续 hold 秒后才允许目标平仓."""
from __future__ import annotations

import os
import threading
import time
from typing import Any


def _env_float(key: str, default: float) -> float:
    try:
        return float(os.getenv(key, str(default)))
    except (TypeError, ValueError):
        return default


def _close_gate_mode() -> str:
    raw = (os.getenv("OKX_OPTIONS_CLOSE_GATE_MODE") or "premium").strip().lower()
    return raw if raw in ("premium", "net_pnl") else "premium"


def _close_recycle_mult(premium_ccy: str) -> float:
    ccy = (premium_ccy or "USDC").strip().upper()
    if ccy in ("ETH", "BTC"):
        return _env_float("OKX_OPTIONS_CLOSE_RECYCLE_MULT_COIN", 1.05)
    return _env_float("OKX_OPTIONS_CLOSE_RECYCLE_MULT", 2.0)


def _close_hold_seconds() -> float:
    return _env_float("OKX_OPTIONS_CLOSE_HOLD_SECONDS", 120.0)


def _close_net_pnl_min_usdt() -> float:
    return _env_float("OKX_OPTIONS_CLOSE_NET_PNL_MIN_USDT", 0.0)


# 兼容旧引用
CLOSE_RECYCLE_MIN_MULT = _env_float("OKX_OPTIONS_CLOSE_RECYCLE_MULT", 2.0)
CLOSE_RECYCLE_HOLD_SECONDS = _env_float("OKX_OPTIONS_CLOSE_HOLD_SECONDS", 120.0)

_lock = threading.Lock()
# inst_id -> gate state
_gates: dict[str, dict[str, Any]] = {}


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 clear_close_gate(inst_id: str | None = None) -> None:
    with _lock:
        if inst_id:
            _gates.pop(str(inst_id).strip(), None)
        else:
            _gates.clear()


def mark_close_gate_passed(inst_id: str) -> None:
    """标记同仓已通过门控,续批平仓只验流动性."""
    inst = (inst_id or "").strip()
    if not inst:
        return
    with _lock:
        st = _gates.get(inst) or {}
        st["passed"] = True
        st["updated"] = time.time()
        _gates[inst] = st


def is_close_gate_passed(inst_id: str) -> bool:
    inst = (inst_id or "").strip()
    if not inst:
        return False
    with _lock:
        return bool((_gates.get(inst) or {}).get("passed"))


def _fmt_gate_amt(v: float, *, ccy: str) -> str:
    unit = (ccy or "USDC").strip().upper() or "USDC"
    if unit in ("ETH", "BTC"):
        txt = f"{float(v):.8f}".rstrip("0").rstrip(".")
        return txt or "0"
    return f"{float(v):.4f}"


def _fmt_usdt(v: float) -> str:
    sign = "+" if v >= 0 else ""
    return f"{sign}{float(v):.2f}"


def _net_pnl_usdt(
    *,
    recv: float | None,
    prem: float | None,
    premium_ccy: str,
    index_px: float | None,
) -> float | None:
    if recv is None or prem is None:
        return None
    net = float(recv) - float(prem)
    ccy = (premium_ccy or "USDC").strip().upper() or "USDC"
    if ccy in ("ETH", "BTC"):
        idx = _safe_float(index_px)
        if idx is None or idx <= 0:
            return None
        return net * float(idx)
    return net


def update_close_gate(
    inst_id: str,
    *,
    recycle_usdc: float | None,
    premium_paid: float | None,
    now: float | None = None,
    min_mult: float | None = None,
    hold_seconds: float | None = None,
    premium_ccy: str | None = None,
    index_px: float | None = None,
    gate_mode: str | None = None,
    net_pnl_min_usdt: float | None = None,
) -> dict[str, Any]:
    """
    根据当前买盘可回收金额刷新门控.
    premium: 可回收 ≥ 权利金×倍数(币本位默认×1.05,USDC 默认×2)
    net_pnl: 净盈亏(买一可回收−权利金)折 USDT ≥ 阈值(默认 0U)
    条件不满足时重置计时;满足时从首次满足起累计持续时间.
    """
    inst = (inst_id or "").strip()
    if not inst:
        return {
            "ok": False,
            "ready": False,
            "recycle_ok": False,
            "msg": "缺少合约",
        }
    ts = float(now if now is not None else time.time())
    mode = (gate_mode or _close_gate_mode()).strip().lower()
    if mode not in ("premium", "net_pnl"):
        mode = "premium"
    hold = float(hold_seconds if hold_seconds is not None else _close_hold_seconds())
    if hold < 0:
        hold = 0.0

    with _lock:
        prev = _gates.get(inst) or {}
        prev_ccy = prev.get("premium_ccy")
    ccy = (premium_ccy or prev_ccy or "USDC").strip().upper() or "USDC"
    if ccy not in ("ETH", "BTC", "USDC"):
        ccy = "USDC"
    need_decimals = 8 if ccy in ("ETH", "BTC") else 4

    prem = _safe_float(premium_paid)
    recv = _safe_float(recycle_usdc)
    mult = float(min_mult if min_mult is not None else _close_recycle_mult(ccy))
    if mult <= 0:
        mult = _close_recycle_mult(ccy)
    min_pnl_u = float(net_pnl_min_usdt if net_pnl_min_usdt is not None else _close_net_pnl_min_usdt())
    idx = _safe_float(index_px)
    if idx is None:
        idx = _safe_float(prev.get("index_px"))

    need = round(prem * mult, need_decimals) if prem is not None and prem > 0 else None
    net_pnl_u = _net_pnl_usdt(recv=recv, prem=prem, premium_ccy=ccy, index_px=idx)

    if mode == "net_pnl":
        if prem is None or prem <= 0:
            condition_ok = False
        elif recv is None:
            condition_ok = False
        elif net_pnl_u is None:
            condition_ok = False
        else:
            condition_ok = bool(net_pnl_u + 1e-9 >= min_pnl_u)
    else:
        condition_ok = bool(
            prem is not None and prem > 0 and recv is not None and need is not None and recv + 1e-12 >= need
        )

    with _lock:
        prev = _gates.get(inst) or {}
        prev_mode = prev.get("gate_mode")
        ok_since = prev.get("ok_since")
        if prev_mode != mode:
            ok_since = None
        if condition_ok:
            if ok_since is None:
                ok_since = ts
        else:
            ok_since = None
        held = (ts - float(ok_since)) if ok_since is not None else 0.0
        ready = bool(condition_ok and held + 1e-9 >= hold)
        prev_passed = bool(prev.get("passed"))
        passed = prev_passed or ready
        state = {
            "ok_since": ok_since,
            "recycle": recv,
            "premium": prem,
            "need": need,
            "updated": ts,
            "min_mult": mult,
            "hold_seconds": hold,
            "passed": passed,
            "premium_ccy": ccy,
            "gate_mode": mode,
            "net_pnl_usdt": net_pnl_u,
            "net_pnl_min_usdt": min_pnl_u,
            "index_px": idx,
        }
        _gates[inst] = state

    remain = max(0.0, hold - held) if condition_ok and not ready else None
    if mode == "net_pnl":
        if prem is None or prem <= 0:
            msg = "缺少权利金,无法校验平仓门控"
        elif recv is None:
            msg = "暂无有效买盘可回收金额"
        elif net_pnl_u is None:
            msg = "缺少指数价,无法将币本位净盈亏折算为 USDT"
        elif not condition_ok:
            msg = (
                f"净盈亏 {_fmt_usdt(net_pnl_u)} U(估) < {min_pnl_u:g} U,"
                f"目标平仓门控未过"
            )
        elif not ready:
            msg = (
                f"净盈亏 {_fmt_usdt(net_pnl_u)} U(估) ≥ {min_pnl_u:g} U,"
                f"需再持续 {remain:.0f}s(已 {held:.0f}/{hold:.0f}s)门控才通过"
            )
        else:
            msg = f"净盈亏 ≥ {min_pnl_u:g} U 且持续≥{hold:.0f}s,目标触达后可按买一平仓"
    elif prem is None or prem <= 0:
        msg = "缺少权利金,无法校验平仓门控"
    elif recv is None:
        msg = "暂无有效买盘可回收金额"
    elif not condition_ok:
        msg = (
            f"可回收 {_fmt_gate_amt(recv, ccy=ccy)} {ccy} < 权利金×{mult:g}"
            f"({_fmt_gate_amt(need, ccy=ccy)}),目标平仓门控未过"
        )
    elif not ready:
        msg = (
            f"可回收已达×{mult:g}({_fmt_gate_amt(recv, ccy=ccy)}/{_fmt_gate_amt(need, ccy=ccy)} {ccy}),"
            f"需再持续 {remain:.0f}s(已 {held:.0f}/{hold:.0f}s)门控才通过"
        )
    else:
        msg = f"可回收已达×{mult:g}且持续≥{hold:.0f}s,目标触达后可按买一平仓"

    auto_blocked = not (ready or passed)
    return {
        "ok": True,
        "ready": ready,
        "passed": passed,
        "recycle_ok": condition_ok,
        "recycle_usdc": recv,
        "premium_paid": prem,
        "need_recycle_usdc": need,
        "premium_ccy": ccy,
        "min_mult": mult,
        "hold_seconds": hold,
        "held_seconds": round(held, 1) if condition_ok else 0.0,
        "remain_seconds": round(remain, 1) if remain is not None else None,
        "ok_since": ok_since,
        "msg": msg,
        "auto_close_blocked": auto_blocked,
        "close_gate_blocked": auto_blocked,
        "gate_mode": mode,
        "net_pnl_usdt": net_pnl_u,
        "net_pnl_min_usdt": min_pnl_u,
        "index_px": idx,
    }


def check_close_gate(
    inst_id: str,
    *,
    recycle_usdc: float | None = None,
    premium_paid: float | None = None,
    premium_ccy: str | None = None,
    index_px: float | None = None,
    refresh: bool = True,
) -> dict[str, Any]:
    """检查是否允许平仓;默认先用最新回收/权利金刷新."""
    inst = (inst_id or "").strip()
    if refresh:
        if recycle_usdc is None or premium_paid is None or premium_ccy is None or index_px is None:
            with _lock:
                prev = _gates.get(inst) or {}
            if recycle_usdc is None:
                recycle_usdc = prev.get("recycle")
            if premium_paid is None:
                premium_paid = prev.get("premium")
            if premium_ccy is None:
                premium_ccy = prev.get("premium_ccy")
            if index_px is None:
                index_px = prev.get("index_px")
        return update_close_gate(
            inst,
            recycle_usdc=recycle_usdc,
            premium_paid=premium_paid,
            premium_ccy=premium_ccy,
            index_px=index_px,
        )
    with _lock:
        prev = _gates.get(inst)
    if not prev:
        return update_close_gate(
            inst,
            recycle_usdc=recycle_usdc,
            premium_paid=premium_paid,
            premium_ccy=premium_ccy,
            index_px=index_px,
        )
    return update_close_gate(
        inst,
        recycle_usdc=recycle_usdc if recycle_usdc is not None else prev.get("recycle"),
        premium_paid=premium_paid if premium_paid is not None else prev.get("premium"),
        premium_ccy=premium_ccy if premium_ccy is not None else prev.get("premium_ccy"),
        index_px=index_px if index_px is not None else prev.get("index_px"),
    )