crypto_okx/lib/hedge_plan/hedge_plan_option_primary_lib.py

"""永期「以期权为主」:定仓、方向映射、目标位与净利口径(纯函数为主)."""
from __future__ import annotations

import math
import os
from typing import Any, Optional

PREMIUM_EXEC_FACTOR = 0.95
DEFAULT_MIN_HOURS = 36.0
DEFAULT_STRIKE_INTERVAL = 15.0
DEFAULT_PERP_LEVERAGE = 100
DEFAULT_OPT_LEVERAGE_ITM_ATM = 100.0
DEFAULT_OPT_LEVERAGE_OTM = 200.0
DEFAULT_RATIO_ITM_ATM = 2.0
DEFAULT_RATIO_OTM = 4.0
OTM_LEV_FLOOR = 180.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 is_option_primary(body_or_plan: dict[str, Any] | None) -> bool:
    if not body_or_plan:
        return False
    v = body_or_plan.get("option_primary")
    if v in (True, 1, "1", "true", "yes", "on"):
        return True
    try:
        return int(v or 0) == 1
    except (TypeError, ValueError):
        return False


def fee_rate() -> float:
    try:
        return max(0.0, float(os.getenv("HEDGE_PLAN_FEE_RATE") or os.getenv("OKX_TAKER_FEE") or "0.0005"))
    except (TypeError, ValueError):
        return 0.0005


def floor2(v: float) -> float:
    """ETH 数量向下取两位小数."""
    if v <= 0:
        return 0.0
    return math.floor(float(v) * 100.0 + 1e-12) / 100.0


def opt_type_for_view(direction: str) -> str:
    """看法做多→Call,做空→Put."""
    return "P" if str(direction or "").strip().lower() == "short" else "C"


def perp_direction_for_view(direction: str) -> str:
    """看法做多→永续空,做空→永续多."""
    return "long" if str(direction or "").strip().lower() == "short" else "short"


def default_opt_leverage(moneyness: str) -> float:
    m = (moneyness or "").strip().lower()
    return DEFAULT_OPT_LEVERAGE_OTM if m == "otm" else DEFAULT_OPT_LEVERAGE_ITM_ATM


def default_ratio(moneyness: str) -> float:
    m = (moneyness or "").strip().lower()
    return DEFAULT_RATIO_OTM if m == "otm" else DEFAULT_RATIO_ITM_ATM


def effective_min_opt_leverage(moneyness: str, configured: Any) -> float:
    cfg = _sf(configured)
    base = cfg if cfg is not None and cfg > 0 else default_opt_leverage(moneyness)
    if (moneyness or "").strip().lower() == "otm":
        return max(base, OTM_LEV_FLOOR)
    return base


def hours_to_expiry_from_ms(exp_ms: Any, *, now_ms: Optional[float] = None) -> Optional[float]:
    exp = _sf(exp_ms)
    if exp is None or exp <= 0:
        return None
    # OKX exp 多为毫秒
    if exp < 1e12:
        exp *= 1000.0
    now = now_ms if now_ms is not None else __import__("time").time() * 1000.0
    return (exp - now) / 3600000.0


def target_hit(*, view_side: str, index_px: float, strike: float, points: float) -> bool:
    """相对 K 的点数目标:做多 index≥K+N;做空 index≤K−N.点数须 >0."""
    n = float(points or 0)
    k = float(strike)
    s = float(index_px)
    if n <= 0 or k <= 0 or s <= 0:
        return False
    side = str(view_side or "").strip().lower()
    if side == "short":
        return s <= (k - n)
    return s >= (k + n)


def option_bid_liquidity_ok(bid: Any, bid_sz: Any, *, need_sheets: float = 0) -> tuple[bool, str]:
    b = _sf(bid)
    if b is None or b <= 0:
        return False, "暂无买一报价,无法平期权"
    sz = _sf(bid_sz)
    if sz is not None and sz <= 0:
        return False, "买一深度为 0,无法平期权"
    need = float(need_sheets or 0)
    if need > 0 and sz is not None and sz + 1e-12 < need:
        return False, f"买一深度不足(需 {need:g} 张,买一 {sz:g})"
    return True, ""


def size_from_premium(
    *,
    premium_budget: float,
    ask: float,
    ct_mult: float,
    ratio: float,
    contract_size: float,
    exec_factor: float = PREMIUM_EXEC_FACTOR,
) -> dict[str, Any]:
    """权利金×0.95 → ETH 两位小数 → 期权张 → 永续跟比例."""
    budget = float(premium_budget or 0)
    a = float(ask or 0)
    ct = float(ct_mult or 0.01)
    r = float(ratio or 0)
    cs = float(contract_size or 0.01)
    usable = budget * float(exec_factor or PREMIUM_EXEC_FACTOR)
    if budget <= 0 or a <= 0 or ct <= 0 or r <= 0 or cs <= 0:
        return {
            "ok": False,
            "msg": "定仓参数无效",
            "usable_premium": round(usable, 4),
            "eth_qty": 0.0,
            "sheets": 0.0,
            "perp_eth": 0.0,
            "contracts": 0.0,
        }
    # ask 为每 1 币权利金;ETH 数量 = usable / ask
    eth_qty = floor2(usable / a)
    if eth_qty <= 0:
        return {
            "ok": False,
            "msg": "权利金不足以买入 0.01 ETH 名义期权",
            "usable_premium": round(usable, 4),
            "eth_qty": 0.0,
            "sheets": 0.0,
            "perp_eth": 0.0,
            "contracts": 0.0,
        }
    sheets = eth_qty / ct
    # 张数向下取整到整数张(OKX 期权常见整张)
    sheets_i = float(math.floor(sheets + 1e-12))
    if sheets_i <= 0:
        return {
            "ok": False,
            "msg": "换算期权张数不足 1 张",
            "usable_premium": round(usable, 4),
            "eth_qty": eth_qty,
            "sheets": 0.0,
            "perp_eth": 0.0,
            "contracts": 0.0,
        }
    # 用整张回写 ETH,保持与下单一致
    eth_qty = round(sheets_i * ct, 2)
    perp_eth = eth_qty / r
    contracts = perp_eth / cs
    premium_est = a * sheets_i * ct
    return {
        "ok": True,
        "msg": "",
        "usable_premium": round(usable, 4),
        "eth_qty": eth_qty,
        "sheets": sheets_i,
        "perp_eth": round(perp_eth, 6),
        "contracts": contracts,
        "premium_est": round(premium_est, 4),
        "ratio": r,
        "exec_factor": float(exec_factor or PREMIUM_EXEC_FACTOR),
    }


def estimate_combo_net_pnl(
    *,
    view_side: str,
    strike: float,
    index_px: float,
    ask_open: float,
    bid: float,
    sheets: float,
    ct_mult: float,
    perp_direction: str,
    perp_entry: float,
    perp_mark: float,
    contracts: float,
    contract_size: float,
    fee: Optional[float] = None,
) -> dict[str, Any]:
    """组合净利(扣费);平仓/卖出手续费按买入费率估算."""
    fr = fee if fee is not None else fee_rate()
    ct = float(ct_mult or 0.01)
    sh = float(sheets or 0)
    a = float(ask_open or 0)
    b = float(bid or 0)
    premium = a * sh * ct
    opt_proceeds = b * sh * ct
    opt_open_fee = premium * fr
    opt_close_fee = opt_proceeds * fr  # 卖出费用按买入费率
    opt_net = opt_proceeds - premium - opt_open_fee - opt_close_fee

    coins = float(contracts or 0) * float(contract_size or 0.01)
    entry = float(perp_entry or 0)
    mark = float(perp_mark or 0)
    pd = str(perp_direction or "").strip().lower()
    if pd == "short":
        perp_gross = (entry - mark) * coins
    else:
        perp_gross = (mark - entry) * coins
    perp_notional_open = abs(entry * coins)
    perp_notional_close = abs(mark * coins)
    perp_open_fee = perp_notional_open * fr
    perp_close_fee = perp_notional_close * fr
    perp_net = perp_gross - perp_open_fee - perp_close_fee
    total = opt_net + perp_net
    return {
        "opt_net": round(opt_net, 4),
        "perp_net": round(perp_net, 4),
        "net": round(total, 4),
        "fee_rate": fr,
        "premium": round(premium, 4),
        "opt_proceeds": round(opt_proceeds, 4),
    }


def validate_option_primary_moneyness(
    *,
    opt_type: str,
    strike: Any,
    index_px: Any,
    ask: Any = None,
    moneyness: str = "atm",
    strike_interval: Any = DEFAULT_STRIKE_INTERVAL,
    min_hours: Any = DEFAULT_MIN_HOURS,
    hours_to_expiry: Any = None,
    min_opt_leverage: Any = None,
) -> Optional[str]:
    from lib.hedge_plan.hedge_plan_moneyness_lib import (
        classify_moneyness,
        is_atm_or_otm,
        is_itm_or_atm,
        normalize_opt_type,
    )

    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 "行权价或指数无效"
    m_want = (moneyness or "atm").strip().lower()
    m_got = classify_moneyness(opt_type=o, strike=k, index_px=s)
    if m_want == "itm":
        if not is_itm_or_atm(opt_type=o, strike=k, index_px=s):
            return "所选须为实值或平值"
    elif m_want == "atm":
        # 平值:距指数在间隔内即可(不强制 classify==atm)
        pass
    elif m_want == "otm":
        if m_got == "itm":
            return "虚值模式不可选实值"
        if not is_atm_or_otm(opt_type=o, strike=k, index_px=s):
            return "虚值模式须选虚值或平值档"
    else:
        return "期权类型(实/平/虚)无效"

    interval = float(_sf(strike_interval) or DEFAULT_STRIKE_INTERVAL)
    if interval > 0 and abs(k - s) > interval + 1e-9:
        return f"行权价偏离指数 {abs(k - s):.1f} > 间隔 {interval:.0f}"

    min_h = float(_sf(min_hours) or DEFAULT_MIN_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"

    a = _sf(ask)
    min_lev = effective_min_opt_leverage(m_want if m_want != "atm" else m_got or "atm", min_opt_leverage)
    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_option_primary_watch(body: dict[str, Any]) -> Optional[str]:
    """盯盘启动校验:只要参数,不要求已选具体合约."""
    need = (
        "direction",
        "exchange_symbol",
        "premium_budget",
        "option_target_points",
        "perp_target_points",
        "option_perp_ratio",
        "option_leverage",
    )
    for k in need:
        if body.get(k) in (None, ""):
            return f"缺少字段: {k}"
    try:
        if float(body["premium_budget"]) <= 0:
            return "权利金须大于 0"
        if float(body["option_target_points"]) <= 0 or float(body["perp_target_points"]) <= 0:
            return "目标位点数须大于 0"
        if float(body["option_perp_ratio"]) <= 0:
            return "期权永续比例须大于 0"
        if float(body["option_leverage"]) <= 0:
            return "期权杠杆须大于 0"
        lev_perp = _sf(body.get("leverage"))
        if lev_perp is not None and lev_perp <= 0:
            return "永续杠杆须大于 0"
    except (TypeError, ValueError):
        return "数值字段无效"
    direction = str(body.get("direction") or "").strip().lower()
    if direction not in ("long", "short"):
        return "方向须为 long 或 short"
    moneyness = str(body.get("moneyness") or body.get("option_moneyness") or "otm").strip().lower()
    if moneyness not in ("itm", "atm", "otm"):
        return "期权类型(实/平/虚)无效"
    return None


def validate_option_primary_start(body: dict[str, Any]) -> Optional[str]:
    need = (
        "direction",
        "contracts",
        "opt_inst_id",
        "sheets",
        "exchange_symbol",
        "premium_budget",
        "option_target_points",
        "perp_target_points",
        "option_perp_ratio",
    )
    for k in need:
        if body.get(k) in (None, ""):
            return f"缺少字段: {k}"
    try:
        if float(body["contracts"]) <= 0 or float(body["sheets"]) <= 0:
            return "张数必须大于 0"
        if float(body["premium_budget"]) <= 0:
            return "权利金须大于 0"
        if float(body["option_target_points"]) <= 0 or float(body["perp_target_points"]) <= 0:
            return "目标位点数须大于 0"
        if float(body["option_perp_ratio"]) <= 0:
            return "期权永续比例须大于 0"
    except (TypeError, ValueError):
        return "数值字段无效"
    direction = str(body.get("direction") or "").strip().lower()
    if direction not in ("long", "short"):
        return "方向须为 long 或 short"
    opt_type = str(body.get("opt_type") or "").strip().upper()
    if not opt_type:
        inst = str(body.get("opt_inst_id") or "")
        if inst.upper().endswith("-P"):
            opt_type = "P"
        elif inst.upper().endswith("-C"):
            opt_type = "C"
    want = opt_type_for_view(direction)
    if opt_type != want:
        return f"以期权为主时做{'多' if direction == 'long' else '空'}须用 {'Call' if want == 'C' else 'Put'}"
    moneyness = str(body.get("moneyness") or body.get("option_moneyness") or "otm").strip().lower()
    from lib.hedge_plan.hedge_plan_moneyness_lib import parse_strike_from_inst

    strike = body.get("strike")
    if strike in (None, ""):
        strike = parse_strike_from_inst(str(body.get("opt_inst_id") or ""))
    index_px = body.get("index_px") or body.get("entry")
    return validate_option_primary_moneyness(
        opt_type=opt_type,
        strike=strike,
        index_px=index_px,
        ask=body.get("ask"),
        moneyness=moneyness,
        strike_interval=body.get("strike_interval", DEFAULT_STRIKE_INTERVAL),
        min_hours=body.get("min_option_hours", DEFAULT_MIN_HOURS),
        hours_to_expiry=body.get("hours_to_expiry"),
        min_opt_leverage=body.get("option_leverage") or body.get("min_opt_leverage"),
    )


def pick_option_primary_candidate(
    chain: dict[str, Any],
    *,
    direction: str,
    moneyness: str = "otm",
    strike_interval: Any = DEFAULT_STRIKE_INTERVAL,
    min_hours: Any = DEFAULT_MIN_HOURS,
    min_opt_leverage: Any = None,
) -> Optional[dict[str, Any]]:
    """从期权链挑最近达标合约(间隔+虚实值+杠杆门)."""
    from lib.hedge_plan.hedge_plan_moneyness_lib import classify_moneyness

    want = opt_type_for_view(direction)
    m_want = (moneyness or "otm").strip().lower()
    interval = float(_sf(strike_interval) or DEFAULT_STRIKE_INTERVAL)
    min_h = float(_sf(min_hours) or DEFAULT_MIN_HOURS)
    try:
        idx = float(chain.get("index_px") or 0)
    except (TypeError, ValueError):
        idx = 0.0
    if idx <= 0:
        return None

    best: Optional[dict[str, Any]] = None
    best_dist: Optional[float] = None
    for exp in chain.get("expiries") or []:
        h = hours_to_expiry_from_ms(exp.get("exp_time"))
        if min_h > 0 and h is not None and h < min_h:
            continue
        for c in exp.get("contracts") or []:
            if str(c.get("opt_type") or "").upper() != want:
                continue
            try:
                k = float(c.get("strike") or 0)
                ask = float(c.get("ask") or 0)
            except (TypeError, ValueError):
                continue
            if k <= 0 or ask <= 0:
                continue
            if interval > 0 and abs(k - idx) > interval + 1e-9:
                continue
            m_got = classify_moneyness(opt_type=want, strike=k, index_px=idx)
            if m_want == "itm" and m_got not in ("itm", "atm"):
                continue
            if m_want == "atm" and m_got != "atm":
                continue
            if m_want == "otm" and m_got == "itm":
                continue
            min_lev = effective_min_opt_leverage(m_want if m_want != "atm" else (m_got or "atm"), min_opt_leverage)
            if min_lev > 0 and idx / ask < min_lev - 1e-9:
                continue
            dist = abs(k - idx)
            if best is None or best_dist is None or dist < best_dist:
                best = {
                    **dict(c),
                    "hours_to_expiry": h,
                    "exp_time": exp.get("exp_time"),
                    "moneyness": m_got,
                    "index_px": idx,
                    "leverage": round(idx / ask, 1),
                }
                best_dist = dist
    return best


def build_option_primary_preview(body: dict[str, Any]) -> dict[str, Any]:
    """情景:期权目标 / 永续目标粗估净利."""
    view = str(body.get("direction") or "long").lower()
    strike = float(body["strike"])
    n = float(body.get("option_target_points") or 0)
    m = float(body.get("perp_target_points") or 0)
    ask = float(body.get("ask") or 0)
    sheets = float(body.get("sheets") or 0)
    ct = float(body.get("ct_mult") or 0.01)
    contracts = float(body.get("contracts") or 0)
    cs = float(body.get("contract_size") or 0.01)
    entry = float(body.get("entry") or body.get("index_px") or 0)
    perp_dir = perp_direction_for_view(view)
    # 粗估到点时期权卖价:按内在价值近似(下限 0)
    def intrinsic(spot: float) -> float:
        o = opt_type_for_view(view)
        if o == "C":
            return max(0.0, spot - strike)
        return max(0.0, strike - spot)

    scenarios = []
    for label, pts, reason in (
        ("期权目标", n, "opt_target_points"),
        ("永续目标", m, "perp_target_points"),
    ):
        spot = strike + pts if view != "short" else strike - pts
        bid_est = max(ask * 0.5, intrinsic(spot) * 0.85)  # 保守估价
        net = estimate_combo_net_pnl(
            view_side=view,
            strike=strike,
            index_px=spot,
            ask_open=ask,
            bid=bid_est,
            sheets=sheets,
            ct_mult=ct,
            perp_direction=perp_dir,
            perp_entry=entry,
            perp_mark=spot,
            contracts=contracts,
            contract_size=cs,
        )
        scenarios.append(
            {
                "label": label,
                "reason": reason,
                "index": spot,
                "perp_pnl": net["perp_net"],
                "options_pnl": net["opt_net"],
                "total": net["net"],
                "note": "扣费净利估价;平仓费按买入费率",
            }
        )
    premium = ask * sheets * ct
    return {
        "plan_type": "perp_options",
        "option_primary": True,
        "summary": {
            "premium_paid": round(premium, 4),
            "usable_premium": round(float(body.get("premium_budget") or 0) * PREMIUM_EXEC_FACTOR, 4),
            "opt_target_total": scenarios[0]["total"] if scenarios else None,
            "perp_target_total": scenarios[1]["total"] if len(scenarios) > 1 else None,
            "perp_direction": perp_dir,
            "opt_type": opt_type_for_view(view),
        },
        "scenarios": scenarios,
    }