crypto_okx/lib/hedge_plan/hedge_plan_moneyness_lib.py

"""对冲计划虚实值选约与校验.

永期(perp_options):期权腿仅允许实值或平值(禁虚值).
期期(options_options):两腿仅允许平值或虚值(禁实值).
"""
from __future__ import annotations

import os
from typing import Any, Optional


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


def itm_max_dist_usd() -> float:
    """过深实值上限(USD).优先对冲专用,否则回退期权页."""
    raw = (os.getenv("HEDGE_PLAN_ITM_MAX_DIST_USD") or "").strip()
    if raw:
        try:
            return max(0.0, float(raw))
        except ValueError:
            pass
    return max(0.0, _env_float("OKX_OPTIONS_ITM_MAX_DIST_USD", 30.0))


def min_option_hours() -> float:
    return max(0.0, _env_float("HEDGE_PLAN_MIN_OPTION_HOURS", 8.0))


def min_option_leverage() -> float:
    """指数/卖一 最低杠杆门槛;0=不启用."""
    return max(0.0, _env_float("HEDGE_PLAN_MIN_OPTION_LEVERAGE", 0.0))


def _sf(v: Any) -> Optional[float]:
    if v is None or v == "":
        return None
    try:
        return float(v)
    except (TypeError, ValueError):
        return None


def normalize_opt_type(opt_type: Any, inst_id: str = "") -> str:
    o = str(opt_type or "").strip().upper()
    if o in ("C", "CALL"):
        return "C"
    if o in ("P", "PUT"):
        return "P"
    inst = str(inst_id or "").upper()
    if inst.endswith("-C") or inst.endswith("-CALL"):
        return "C"
    if inst.endswith("-P") or inst.endswith("-PUT"):
        return "P"
    return ""


def classify_moneyness(*, opt_type: str, strike: float, index_px: float) -> str:
    """itm / atm / otm / unknown.与 options_pricing_lib.option_moneyness 同口径."""
    from lib.options.options_pricing_lib import option_moneyness

    return option_moneyness(opt_type=opt_type, strike=strike, index_px=index_px)


def is_itm_or_atm(*, opt_type: str, strike: float, index_px: float) -> bool:
    """Call: K<=S(+atm 带);Put: K>=S.用 classify 结果含 atm/itm."""
    m = classify_moneyness(opt_type=opt_type, strike=strike, index_px=index_px)
    if m in ("itm", "atm"):
        return True
    # 几何兜底(与 eth_hedge_sim 一致),避免 atm 带边界漏判
    o = normalize_opt_type(opt_type)
    k = float(strike)
    s = float(index_px)
    if o == "C":
        return k <= s + 1e-9
    if o == "P":
        return k >= s - 1e-9
    return False


def is_atm_or_otm(*, opt_type: str, strike: float, index_px: float) -> bool:
    m = classify_moneyness(opt_type=opt_type, strike=strike, index_px=index_px)
    if m in ("atm", "otm"):
        return True
    o = normalize_opt_type(opt_type)
    k = float(strike)
    s = float(index_px)
    if o == "C":
        return k >= s - 1e-9  # 平值带内或虚值
    if o == "P":
        return k <= s + 1e-9
    return False


def itm_depth_usd(*, opt_type: str, strike: float, index_px: float) -> float:
    o = normalize_opt_type(opt_type)
    k = float(strike)
    s = float(index_px)
    if o == "C" and k < s:
        return s - k
    if o == "P" and k > s:
        return k - s
    return 0.0


def parse_strike_from_inst(inst_id: str) -> Optional[float]:
    """从 OKX 合约名解析行权价: ETH-USD-260731-1800-P."""
    parts = str(inst_id or "").strip().upper().split("-")
    if len(parts) < 5:
        return None
    return _sf(parts[-2])


def pick_itm_or_atm_contract(
    contracts: list[dict[str, Any]],
    *,
    opt_type: str,
    index_px: float,
    itm_max_dist: Optional[float] = None,
) -> Optional[dict[str, Any]]:
    """在合约列表中选距标的最近的实值/平值腿."""
    want = normalize_opt_type(opt_type)
    if not want or index_px <= 0:
        return None
    max_dist = itm_max_dist if itm_max_dist is not None else itm_max_dist_usd()
    cands: list[tuple[float, float, dict[str, Any]]] = []
    for c in contracts or []:
        if normalize_opt_type(c.get("opt_type"), str(c.get("inst_id") or "")) != want:
            continue
        k = _sf(c.get("strike"))
        if k is None:
            continue
        if not is_itm_or_atm(opt_type=want, strike=k, index_px=index_px):
            continue
        depth = itm_depth_usd(opt_type=want, strike=k, index_px=index_px)
        if max_dist > 0 and depth > max_dist:
            continue
        cands.append((abs(k - index_px), k, c))
    if not cands:
        return None
    cands.sort(key=lambda x: (x[0], x[1]))
    return cands[0][2]


def pick_atm_or_otm_contract(
    contracts: list[dict[str, Any]],
    *,
    opt_type: str,
    index_px: float,
    prefer: str = "atm",
) -> Optional[dict[str, Any]]:
    """选平值或虚值腿.prefer=atm 取距标的最近;prefer=otm 取最近虚值(不含实值)."""
    want = normalize_opt_type(opt_type)
    if not want or index_px <= 0:
        return None
    prefer_l = (prefer or "atm").strip().lower()
    cands: list[tuple[float, float, dict[str, Any]]] = []
    for c in contracts or []:
        if normalize_opt_type(c.get("opt_type"), str(c.get("inst_id") or "")) != want:
            continue
        k = _sf(c.get("strike"))
        if k is None:
            continue
        if not is_atm_or_otm(opt_type=want, strike=k, index_px=index_px):
            continue
        m = classify_moneyness(opt_type=want, strike=k, index_px=index_px)
        if prefer_l == "otm" and m != "otm":
            continue
        if prefer_l == "atm" and m == "otm":
            # 仍可入选,但排序靠后(先 atm)
            cands.append((1_000_000 + abs(k - index_px), k, c))
        else:
            cands.append((abs(k - index_px), k, c))
    if not cands:
        return None
    cands.sort(key=lambda x: (x[0], x[1]))
    return cands[0][2]


def recommend_oo_legs(
    contracts: list[dict[str, Any]],
    *,
    index_px: float,
    template: str = "atm_straddle",
) -> Optional[tuple[dict[str, Any], dict[str, Any]]]:
    """期期推荐两腿.atm_straddle=最近平值 Call+Put;double_otm=最近虚值 Call+Put."""
    tpl = (template or "atm_straddle").strip().lower()
    prefer = "otm" if tpl in ("double_otm", "otm_otm", "otm") else "atm"
    call = pick_atm_or_otm_contract(
        contracts, opt_type="C", index_px=index_px, prefer=prefer
    )
    put = pick_atm_or_otm_contract(
        contracts, opt_type="P", index_px=index_px, prefer=prefer
    )
    if not call or not put:
        return None
    if str(call.get("inst_id") or "") == str(put.get("inst_id") or ""):
        return None
    return call, put


def validate_po_option_moneyness(
    *,
    opt_type: str,
    strike: Any,
    index_px: Any,
    ask: Any = None,
    hours_to_expiry: Any = None,
) -> Optional[str]:
    """永期保险腿校验;返回错误文案或 None."""
    o = normalize_opt_type(opt_type)
    k = _sf(strike)
    s = _sf(index_px)
    if o not in ("C", "P"):
        return "期权类型无效"
    if k is None or s is None or s <= 0:
        return "行权价或指数无效,无法校验虚实值"
    if not is_itm_or_atm(opt_type=o, strike=k, index_px=s):
        return "永期保险腿须为实值或平值,不可选虚值"
    max_dist = itm_max_dist_usd()
    depth = itm_depth_usd(opt_type=o, strike=k, index_px=s)
    if max_dist > 0 and depth > max_dist:
        return f"实值过深(距现价 {depth:.1f}U > {max_dist:.0f}U),请换更接近平值的档"
    min_h = min_option_hours()
    h = _sf(hours_to_expiry)
    if min_h > 0 and h is not None and h < min_h:
        return f"剩余到期约 {h:.1f}h,低于最低 {min_h:.0f}h"
    min_lev = min_option_leverage()
    a = _sf(ask)
    if min_lev > 0 and a is not None and a > 0:
        lev = s / a
        if lev < min_lev:
            return f"期权杠杆 S/ask≈{lev:.0f} 低于门槛 {min_lev:.0f}"
    return None


def validate_oo_leg_moneyness(
    *,
    opt_type: str,
    strike: Any,
    index_px: Any,
    role: str = "腿",
) -> Optional[str]:
    o = normalize_opt_type(opt_type)
    k = _sf(strike)
    s = _sf(index_px)
    if o not in ("C", "P"):
        return f"{role}期权类型无效"
    if k is None or s is None or s <= 0:
        return f"{role}行权价或指数无效,无法校验虚实值"
    m = classify_moneyness(opt_type=o, strike=k, index_px=s)
    if m == "itm":
        return f"{role}须为平值或虚值,不可选实值"
    if not is_atm_or_otm(opt_type=o, strike=k, index_px=s):
        return f"{role}须为平值或虚值"
    return None


def validate_oo_legs_moneyness(
    leg_a: dict[str, Any],
    leg_b: dict[str, Any],
    *,
    index_px: Any,
) -> Optional[str]:
    err = validate_oo_leg_moneyness(
        opt_type=leg_a.get("opt_type"),
        strike=leg_a.get("strike"),
        index_px=index_px,
        role="腿A",
    )
    if err:
        return err
    err = validate_oo_leg_moneyness(
        opt_type=leg_b.get("opt_type"),
        strike=leg_b.get("strike"),
        index_px=index_px,
        role="腿B",
    )
    if err:
        return err
    return None