crypto_monitor/lib/options/options_close_gate_lib.py

"""期权按买盘平仓门控:可回收需 ≥ N×权利金,并持续持有一段时间后才允许平仓."""
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


# 可回收 ≥ 权利金 × 倍数,且该状态持续满 hold_seconds 才允许按买盘平仓
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 -> {"ok_since": float|None, "recycle": float, "premium": float, "updated": float}
_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:
    """标记同仓已通过 2× 门控,续批平仓只验流动性."""
    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 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,
) -> dict[str, Any]:
    """
    根据当前买盘可回收金额刷新门控.
    条件不满足时重置计时;满足时从首次满足起累计持续时间.
    """
    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())
    mult = float(min_mult if min_mult is not None else CLOSE_RECYCLE_MIN_MULT)
    hold = float(hold_seconds if hold_seconds is not None else CLOSE_RECYCLE_HOLD_SECONDS)
    if mult <= 0:
        mult = 2.0
    if hold < 0:
        hold = 0.0

    prem = _safe_float(premium_paid)
    recv = _safe_float(recycle_usdc)
    need = round(prem * mult, 4) if prem is not None and prem > 0 else None
    recycle_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 {}
        ok_since = prev.get("ok_since")
        if recycle_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(recycle_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,
        }
        _gates[inst] = state

    remain = max(0.0, hold - held) if recycle_ok and not ready else None
    if prem is None or prem <= 0:
        msg = "缺少权利金,无法校验平仓门控"
    elif recv is None:
        msg = "暂无有效买盘可回收金额"
    elif not recycle_ok:
        msg = f"可回收 {recv:.4f} USDC < 权利金×{mult:g}({need:.4f}),目标平仓门控未过"
    elif not ready:
        msg = (
            f"可回收已达×{mult:g}({recv:.4f}/{need:.4f}),"
            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": recycle_ok,
        "recycle_usdc": recv,
        "premium_paid": prem,
        "need_recycle_usdc": need,
        "min_mult": mult,
        "hold_seconds": hold,
        "held_seconds": round(held, 1) if recycle_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,
    }


def check_close_gate(
    inst_id: str,
    *,
    recycle_usdc: float | None = None,
    premium_paid: 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:
            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")
        return update_close_gate(inst, recycle_usdc=recycle_usdc, premium_paid=premium_paid)
    with _lock:
        prev = _gates.get(inst)
    if not prev:
        return update_close_gate(inst, recycle_usdc=recycle_usdc, premium_paid=premium_paid)
    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"),
    )