crypto_monitor/lib/options/options_pricing_lib.py

"""OKX USDⓈ 期权:张数与权利金计算."""
from __future__ import annotations

import math
from typing import Any


def ct_mult_from_meta(meta: dict[str, Any] | None) -> float:
    if not meta:
        return 0.01
    try:
        return float(meta.get("ctMult") or 0.01)
    except (TypeError, ValueError):
        return 0.01


def min_sz_from_meta(meta: dict[str, Any] | None) -> int:
    if not meta:
        return 1
    try:
        return max(1, int(float(meta.get("minSz") or 1)))
    except (TypeError, ValueError):
        return 1


def premium_per_sheet(quote_per_unit: float, ct_mult: float = 0.01) -> float:
    """报价为每 1 ETH/BTC;每张权利金 = 报价 × ctMult."""
    return float(quote_per_unit) * float(ct_mult)


def format_quote_liquidity(px: float | None, sz: float | None, *, px_decimals: int = 4) -> str | None:
    """盘口展示:价格/张数,如 17.2/150."""
    if px is None:
        return None
    try:
        price = f"{float(px):.{px_decimals}f}".rstrip("0").rstrip(".")
    except (TypeError, ValueError):
        return None
    if sz is None:
        return price
    try:
        s = float(sz)
        size = str(int(s)) if abs(s - int(s)) < 1e-9 else str(s).rstrip("0").rstrip(".")
    except (TypeError, ValueError):
        return price
    return f"{price}/{size}"


def total_premium(quote_per_unit: float, eth_amount: float, ct_mult: float = 0.01) -> float:
    return float(quote_per_unit) * float(eth_amount)


def estimate_close_by_bids(
    bids: list[dict[str, Any]] | None,
    sheets: int | float,
    *,
    ct_mult: float = 0.01,
    premium_paid: float | None = None,
) -> dict[str, Any]:
    """按买一到买N逐档估算限价卖出可收回金额."""
    target = max(0, int(float(sheets or 0)))
    remaining = target
    total_received = 0.0
    levels: list[dict[str, Any]] = []
    if target <= 0 or ct_mult <= 0:
        return {
            "levels": [],
            "covered_sheets": 0,
            "uncovered_sheets": target,
            "total_received": 0.0,
            "avg_px": None,
            "estimated_pnl": None,
        }
    for i, level in enumerate(bids or [], start=1):
        if remaining <= 0:
            break
        try:
            px = float(level.get("px"))
            sz = int(float(level.get("sz")))
        except (AttributeError, TypeError, ValueError):
            continue
        if px <= 0 or sz <= 0:
            continue
        take = min(remaining, sz)
        eth_amount = eth_amount_from_sheets(take, ct_mult)
        received = total_premium(px, eth_amount)
        levels.append(
            {
                "level": i,
                "px": px,
                "available_sheets": sz,
                "sheets": take,
                "eth_amount": eth_amount,
                "received": round(received, 4),
            }
        )
        total_received += received
        remaining -= take
    covered = target - remaining
    avg_px = (total_received / eth_amount_from_sheets(covered, ct_mult)) if covered > 0 else None
    estimated_pnl = None
    if premium_paid is not None and covered > 0:
        paid_basis = float(premium_paid) * (covered / target)
        estimated_pnl = round(total_received - paid_basis, 4)
    return {
        "levels": levels,
        "covered_sheets": covered,
        "uncovered_sheets": remaining,
        "total_received": round(total_received, 4),
        "avg_px": round(avg_px, 4) if avg_px is not None else None,
        "estimated_pnl": estimated_pnl,
    }


def sheets_from_eth_amount(eth_amount: float, ct_mult: float = 0.01) -> int:
    if eth_amount <= 0 or ct_mult <= 0:
        return 0
    return int(math.floor(eth_amount / ct_mult + 1e-12))


def eth_amount_from_sheets(sheets: int, ct_mult: float = 0.01) -> float:
    return round(int(sheets) * float(ct_mult), 8)


def calc_order_size(
    *,
    quote_per_unit: float,
    ct_mult: float,
    min_sz: int,
    budget_usdc: float | None = None,
    budget_buffer: float = 0.95,
    eth_amount: float | None = None,
    sheets: int | None = None,
    budget_cap: float | None = None,
) -> dict[str, Any]:
    """
    返回 sheets, eth_amount, total_premium.
    mode: budget_full / eth_amount / sheets.
    """
    if quote_per_unit <= 0:
        return {"ok": False, "msg": "卖一价无效", "sheets": 0, "eth_amount": 0.0, "total_premium": 0.0}

    if sheets is not None and int(sheets) > 0:
        sheets = int(sheets)
    elif eth_amount is not None and eth_amount > 0:
        sheets = sheets_from_eth_amount(eth_amount, ct_mult)
    elif budget_usdc is not None and budget_usdc > 0:
        eff = float(budget_usdc) * float(budget_buffer)
        per_sheet = premium_per_sheet(quote_per_unit, ct_mult)
        if per_sheet <= 0:
            return {"ok": False, "msg": "无法计算单张权利金", "sheets": 0, "eth_amount": 0.0, "total_premium": 0.0}
        sheets = int(math.floor(eff / per_sheet))
    else:
        return {"ok": False, "msg": "请指定预算,币数量或张数", "sheets": 0, "eth_amount": 0.0, "total_premium": 0.0}

    if sheets < min_sz:
        per = premium_per_sheet(quote_per_unit, ct_mult)
        return {
            "ok": False,
            "msg": f"预算不足,无法买入 {min_sz} 张(单张约 {per:.4f} USDC)",
            "sheets": sheets,
            "eth_amount": eth_amount_from_sheets(sheets, ct_mult),
            "total_premium": total_premium(quote_per_unit, eth_amount_from_sheets(sheets, ct_mult)),
        }

    eth = eth_amount_from_sheets(sheets, ct_mult)
    prem = total_premium(quote_per_unit, eth)
    if budget_cap is not None and prem > float(budget_cap) + 1e-9:
        return {
            "ok": False,
            "msg": f"权利金 {prem:.4f} 超过单笔上限 {budget_cap} USDC",
            "sheets": sheets,
            "eth_amount": eth,
            "total_premium": prem,
        }
    return {"ok": True, "msg": "", "sheets": sheets, "eth_amount": eth, "total_premium": prem}


def is_shallow_itm(
    *,
    opt_type: str,
    strike: float,
    index_px: float,
    max_dist_usd: float,
) -> bool:
    o = (opt_type or "").upper()
    if o == "C":
        if strike >= index_px:
            return False
        return (index_px - strike) <= max_dist_usd
    if o == "P":
        if strike <= index_px:
            return False
        return (strike - index_px) <= max_dist_usd
    return False


def option_moneyness(*, opt_type: str, strike: float, index_px: float) -> str:
    """返回 itm / otm / atm."""
    o = (opt_type or "").upper()
    if strike is None or index_px is None or index_px <= 0:
        return "unknown"
    atm_band = max(index_px * 0.002, 2.0)
    if abs(strike - index_px) <= atm_band:
        return "atm"
    if o == "C":
        return "itm" if strike < index_px else "otm"
    if o == "P":
        return "itm" if strike > index_px else "otm"
    return "unknown"


def option_moneyness_label(moneyness: str) -> str:
    return {"itm": "实值", "otm": "虚值", "atm": "平值"}.get((moneyness or "").lower(), "")


def expiry_breakeven_from_ask(
    *,
    opt_type: str,
    strike: float | None,
    ask_px: float | None,
    mark_px: float | None = None,
) -> float | None:
    """买入前预估到期平衡:权利金按卖一;无卖一时回退标记价."""
    prem = ask_px if ask_px is not None and ask_px > 0 else mark_px
    return expiry_breakeven_px(opt_type=opt_type, strike=strike, avg_px=prem)


def expiry_breakeven_px(
    *,
    opt_type: str,
    strike: float | None,
    avg_px: float | None,
    be_px_api: float | None = None,
) -> float | None:
    """到期平衡点:持有至到期时标的指数盈亏为 0 的价格.优先 OKX bePx."""
    if be_px_api is not None and be_px_api > 0:
        return round(float(be_px_api), 2)
    if strike is None or avg_px is None:
        return None
    o = (opt_type or "").upper()
    if o == "C":
        return round(strike + avg_px, 2)
    if o == "P":
        return round(strike - avg_px, 2)
    return None


def close_breakeven_idx(
    *,
    opt_type: str,
    idx_px: float | None,
    mark_px: float | None,
    avg_px: float | None,
    delta_pa: float | None = None,
    pos: float = 0,
    ct_mult: float = 0.01,
) -> float | None:
    """
    平掉回本:标的指数达到该价位时,按标记价平仓近似盈亏为 0.
    优先用 deltaPA 线性外推,否则用时间价值近似(适合短期轻度实值).
    """
    if idx_px is None or mark_px is None or avg_px is None:
        return None
    eth_amt = abs(float(pos)) * float(ct_mult)
    if eth_amt > 1e-12 and delta_pa is not None and abs(float(delta_pa)) > 1e-12:
        slope = float(delta_pa) / eth_amt
        return round(float(idx_px) + (float(avg_px) - float(mark_px)) / slope, 2)
    o = (opt_type or "").upper()
    if o == "C":
        return round(float(idx_px) + float(avg_px) - float(mark_px), 2)
    if o == "P":
        return round(float(idx_px) + float(mark_px) - float(avg_px), 2)
    return None


def idx_distance_to_be(idx_px: float | None, be_px: float | None) -> float | None:
    """指数距平衡点(正=指数需上涨才到平衡点)."""
    if idx_px is None or be_px is None:
        return None
    return round(float(be_px) - float(idx_px), 2)


def format_options_breakeven_line(
    *,
    expiry_be_px: float | None,
    close_be_px: float | None,
    idx_px: float | None = None,
) -> str:
    """持仓摘要行:到期平衡 / 平掉回本."""
    parts: list[str] = []
    if expiry_be_px is not None:
        parts.append(f"到期平衡{expiry_be_px:.0f}")
    if close_be_px is not None:
        parts.append(f"平掉回本{close_be_px:.0f}")
    if idx_px is not None and parts:
        return " ".join(parts) + f"(指数{idx_px:.0f})"
    return " ".join(parts)


def estimate_expiry_profit_at_index(
    *,
    opt_type: str,
    strike: float | None,
    target_idx: float | None,
    entry_px: float | None,
    eth_amount: float | None,
) -> float | None:
    """到期测算:目标指数价下按内在价值减权利金(每 1 币报价 × 币量)."""
    if strike is None or target_idx is None or entry_px is None or eth_amount is None:
        return None
    if eth_amount <= 0:
        return None
    o = (opt_type or "").upper()
    if o == "C":
        intrinsic = max(0.0, float(target_idx) - float(strike))
    elif o == "P":
        intrinsic = max(0.0, float(strike) - float(target_idx))
    else:
        return None
    return round((intrinsic - float(entry_px)) * float(eth_amount), 4)


def equivalent_contract_leverage(
    *,
    index_px: float | None,
    eth_amount: float | None,
    total_premium: float | None,
) -> float | None:
    """名义价值 / 权利金,近似相当于永续合约杠杆倍数(测算用)."""
    if index_px is None or eth_amount is None or total_premium is None:
        return None
    if eth_amount <= 0 or total_premium <= 0:
        return None
    return round(float(index_px) * float(eth_amount) / float(total_premium), 1)