crypto_okx/lib/sim/pricing_lib.py

"""成交价与手续费.

永续:滑点=1×fee_rate,手续费=名义×fee_rate.
期权(币本位):滑点仍按权利金价×fee_rate;手续费对齐 OKX=
  Min(fee_rate×标的币数量, 权利金×7%).
"""

from __future__ import annotations

import os
from dataclasses import asdict, dataclass


@dataclass(slots=True)
class PriceResult:
    base_px: float
    fill_px: float
    fee: float
    slip: float
    notional: float

    def to_dict(self) -> dict[str, float]:
        return asdict(self)


def sim_fee_rate(override: float | None = None) -> float:
    if override is not None:
        return float(override)
    try:
        return float(os.getenv("SIM_FEE_RATE") or "0.0005")
    except (TypeError, ValueError):
        return 0.0005


def perp_fill(
    *,
    side: str,
    action: str,
    bid: float,
    ask: float,
    qty: float,
    fee_rate: float,
) -> PriceResult:
    """
    side: long|short
    action: open|close
    开多/平空: 吃卖一 ×(1+f)
    开空/平多: 吃买一 ×(1-f)
    qty: 标的数量(合约张数 × 合约面值)
    """
    f = float(fee_rate)
    buying = (action == "open" and side == "long") or (action == "close" and side == "short")
    if buying:
        base = float(ask)
        fill = base * (1.0 + f)
    else:
        base = float(bid)
        fill = base * (1.0 - f)
    notional = abs(fill * float(qty))
    fee = notional * f
    slip = abs(fill - base) * float(qty)
    return PriceResult(base_px=base, fill_px=fill, fee=fee, slip=slip, notional=notional)


def option_fee_coin(
    *,
    underlying_coin: float,
    premium_coin: float,
    fee_rate: float | None = None,
    premium_fee_cap: float = 0.07,
) -> float:
    """OKX 币本位期权手续费(权利金币种).

    Min(费率 × 张数对应标的币数量, 权利金×封顶比例).
    注意:费率作用在标的币面值上,不是权利金名义上.
    """
    f = sim_fee_rate(fee_rate)
    qty = abs(float(underlying_coin or 0.0))
    prem = abs(float(premium_coin or 0.0))
    if qty <= 0 and prem <= 0:
        return 0.0
    by_rate = qty * f
    cap_ratio = abs(float(premium_fee_cap or 0.07))
    if cap_ratio <= 0 or prem <= 0:
        return round(by_rate, 8)
    return round(min(by_rate, prem * cap_ratio), 8)


def option_fill(
    *,
    action: str,
    bid: float,
    ask: float,
    qty: float,
    fee_rate: float,
) -> PriceResult:
    """开仓买入吃卖一; 平仓卖出吃买一.

    qty = sheets × ct_mult(标的币数量).
    滑点仍按权利金价×费率;手续费按 OKX 币本位公式(标的币×费率,权利金7%封顶).
    """
    f = float(fee_rate)
    if action == "open":
        base = float(ask)
        fill = base * (1.0 + f)
    else:
        base = float(bid)
        fill = base * (1.0 - f)
    qty_f = abs(float(qty))
    notional = abs(fill * qty_f)
    fee = option_fee_coin(
        underlying_coin=qty_f,
        premium_coin=notional,
        fee_rate=f,
    )
    slip = abs(fill - base) * qty_f
    return PriceResult(base_px=base, fill_px=fill, fee=fee, slip=slip, notional=notional)


@dataclass(slots=True)
class SpotConvertResult:
    """USDC/USDT 现货兑换: price = USDT per USDC."""

    direction: str
    from_ccy: str
    to_ccy: str
    from_amount: float
    to_amount: float
    base_px: float
    fill_px: float
    fee: float


def spot_usdc_usdt_fill(
    *,
    direction: str,
    amount: float,
    bid: float,
    ask: float,
    fee_rate: float,
) -> SpotConvertResult:
    """
    对齐实盘 USDC-USDT 现货市价:
    - usdt_to_usdc: 用 USDT 买 USDC, 吃卖一 ×(1+f)
    - usdc_to_usdt: 卖 USDC 换 USDT, 吃买一 ×(1-f)
    amount 为付出币种数量.
    """
    f = float(fee_rate)
    amt = float(amount)
    d = (direction or "").strip().lower()
    if d == "usdt_to_usdc":
        base = float(ask)
        fill = base * (1.0 + f)
        if fill <= 0:
            raise ValueError("无效卖一价")
        to_amt = amt / fill
        fee = amt * f
        return SpotConvertResult(
            direction=d,
            from_ccy="USDT",
            to_ccy="USDC",
            from_amount=amt,
            to_amount=to_amt,
            base_px=base,
            fill_px=fill,
            fee=fee,
        )
    if d == "usdc_to_usdt":
        base = float(bid)
        fill = base * (1.0 - f)
        if fill <= 0:
            raise ValueError("无效买一价")
        to_amt = amt * fill
        fee = to_amt * f
        return SpotConvertResult(
            direction=d,
            from_ccy="USDC",
            to_ccy="USDT",
            from_amount=amt,
            to_amount=to_amt,
            base_px=base,
            fill_px=fill,
            fee=fee,
        )
    raise ValueError("direction 须为 usdt_to_usdc 或 usdc_to_usdt")


def spot_coin_usdt_fill(
    *,
    direction: str,
    amount: float,
    bid: float,
    ask: float,
    fee_rate: float,
    coin: str = "ETH",
) -> SpotConvertResult:
    """
    对齐实盘 ETH-USDT / BTC-USDT 现货市价:
    - usdt_to_coin: 用 USDT 买币, 吃卖一 ×(1+f); amount=USDT
    - coin_to_usdt: 卖币换 USDT, 吃买一 ×(1-f); amount=币数量
    """
    f = float(fee_rate)
    amt = float(amount)
    ccy = (coin or "ETH").strip().upper() or "ETH"
    d = (direction or "").strip().lower()
    if d in ("usdt_to_coin", "usdt_to_eth", "usdt_to_btc"):
        base = float(ask)
        fill = base * (1.0 + f)
        if fill <= 0:
            raise ValueError("无效卖一价")
        to_amt = amt / fill
        fee = amt * f
        return SpotConvertResult(
            direction="usdt_to_coin",
            from_ccy="USDT",
            to_ccy=ccy,
            from_amount=amt,
            to_amount=to_amt,
            base_px=base,
            fill_px=fill,
            fee=fee,
        )
    if d in ("coin_to_usdt", "eth_to_usdt", "btc_to_usdt"):
        base = float(bid)
        fill = base * (1.0 - f)
        if fill <= 0:
            raise ValueError("无效买一价")
        to_amt = amt * fill
        fee = to_amt * f
        return SpotConvertResult(
            direction="coin_to_usdt",
            from_ccy=ccy,
            to_ccy="USDT",
            from_amount=amt,
            to_amount=to_amt,
            base_px=base,
            fill_px=fill,
            fee=fee,
        )
    raise ValueError("direction 须为 usdt_to_coin 或 coin_to_usdt")