dekun
135 lines
4.4 KiB
Python
135 lines
4.4 KiB
Python
"""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 total_premium(quote_per_unit: float, eth_amount: float, ct_mult: float = 0.01) -> float:
|
||
return float(quote_per_unit) * float(eth_amount)
|
||
|
||
|
||
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(), "")
|