dekun
aa1a2da2b7
Call 为 BE−K、Put 为 K−BE；链/下单/持仓统一；顺带修复 format_position_row 中 row_mode 引用顺序。 Co-authored-by: Cursor <cursoragent@cursor.com>
701 lines
22 KiB
Python
701 lines
22 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 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)
|
||
|
||
|
||
# 买一相对标记价/内在价值低于该比例 → 视为残档,禁止按买盘自动/多档平仓
|
||
BID_CLOSE_MIN_RATIO = 0.3
|
||
|
||
|
||
def _safe_px(v: Any) -> float | None:
|
||
if v is None or v == "":
|
||
return None
|
||
try:
|
||
x = float(v)
|
||
except (TypeError, ValueError):
|
||
return None
|
||
return x if x > 0 else None
|
||
|
||
|
||
def _quote_in_coin_from_context(
|
||
*,
|
||
quote_in_coin: bool | None = None,
|
||
inst_id: str | None = None,
|
||
margin_mode: str | None = None,
|
||
) -> bool:
|
||
"""币本位(ETH-USD/BTC-USD)权利金按币报价;USDC(USD_UM)按美元点差."""
|
||
if quote_in_coin is not None:
|
||
return bool(quote_in_coin)
|
||
if inst_id:
|
||
try:
|
||
from lib.options.options_margin_mode_lib import MODE_COIN, margin_mode_from_inst_id
|
||
|
||
return margin_mode_from_inst_id(inst_id) == MODE_COIN
|
||
except Exception:
|
||
pass
|
||
if margin_mode is not None:
|
||
try:
|
||
from lib.options.options_margin_mode_lib import is_coin_margin_mode
|
||
|
||
return is_coin_margin_mode(margin_mode)
|
||
except Exception:
|
||
return str(margin_mode).strip().lower() in ("coin", "coin_margin", "crypto")
|
||
return False
|
||
|
||
|
||
def intrinsic_px_per_unit(
|
||
opt_type: str | None,
|
||
strike: float | None,
|
||
index_px: float | None,
|
||
*,
|
||
quote_in_coin: bool | None = None,
|
||
inst_id: str | None = None,
|
||
margin_mode: str | None = None,
|
||
) -> float | None:
|
||
"""
|
||
与盘口同单位的内在价值(每 1 标的).
|
||
- USDC / USD_UM: 美元点差 max(0, S−K) / max(0, K−S)
|
||
- 币本位 ETH-USD / BTC-USD: 币报价 max(0, S−K)/S / max(0, K−S)/S
|
||
"""
|
||
o = (opt_type or "").strip().upper()
|
||
if strike is None or index_px is None:
|
||
return None
|
||
try:
|
||
k = float(strike)
|
||
idx = float(index_px)
|
||
except (TypeError, ValueError):
|
||
return None
|
||
if o == "C" and idx > k:
|
||
points = idx - k
|
||
elif o == "P" and idx < k:
|
||
points = k - idx
|
||
else:
|
||
return None
|
||
if _quote_in_coin_from_context(
|
||
quote_in_coin=quote_in_coin, inst_id=inst_id, margin_mode=margin_mode
|
||
):
|
||
if idx <= 0:
|
||
return None
|
||
return points / idx
|
||
return points
|
||
|
||
|
||
def is_stub_bid_px(
|
||
bid_px: float | None,
|
||
*,
|
||
mark_px: float | None = None,
|
||
intrinsic_px: float | None = None,
|
||
min_ratio: float = BID_CLOSE_MIN_RATIO,
|
||
) -> tuple[bool, str]:
|
||
"""
|
||
判断买一是否为无效残档(如标记 42、买一 0.2).
|
||
返回 (is_stub, reason).
|
||
"""
|
||
bid = _safe_px(bid_px)
|
||
if bid is None:
|
||
return True, "无买一"
|
||
ref = _safe_px(mark_px)
|
||
ref_name = "标记价"
|
||
intrinsic = _safe_px(intrinsic_px)
|
||
if intrinsic is not None and (ref is None or intrinsic > ref):
|
||
ref = intrinsic
|
||
ref_name = "内在价值"
|
||
if ref is None:
|
||
return False, ""
|
||
ratio = float(min_ratio) if min_ratio and min_ratio > 0 else BID_CLOSE_MIN_RATIO
|
||
if bid < ref * ratio:
|
||
return True, f"买一{bid:g}远低于{ref_name}{ref:g},属无效残档,禁止按买盘自动平仓"
|
||
return False, ""
|
||
|
||
|
||
def fetch_option_mark_px(ex: Any, inst_id: str) -> float | None:
|
||
"""优先 mark-price 接口,失败则 None."""
|
||
inst_id = (inst_id or "").strip()
|
||
if not inst_id or ex is None:
|
||
return None
|
||
try:
|
||
rows = ex.public_get_public_mark_price({"instType": "OPTION", "instId": inst_id}).get("data") or []
|
||
if rows:
|
||
return _safe_px(rows[0].get("markPx"))
|
||
except Exception:
|
||
pass
|
||
return None
|
||
|
||
|
||
def close_ref_prices(
|
||
*,
|
||
mark_px: float | None = None,
|
||
opt_type: str | None = None,
|
||
strike: float | None = None,
|
||
index_px: float | None = None,
|
||
quote_in_coin: bool | None = None,
|
||
inst_id: str | None = None,
|
||
margin_mode: str | None = None,
|
||
) -> tuple[float | None, float | None]:
|
||
"""返回 (mark_px, intrinsic_px) 供残档判断;intrinsic 与盘口同单位."""
|
||
return _safe_px(mark_px), intrinsic_px_per_unit(
|
||
opt_type,
|
||
strike,
|
||
index_px,
|
||
quote_in_coin=quote_in_coin,
|
||
inst_id=inst_id,
|
||
margin_mode=margin_mode,
|
||
)
|
||
|
||
|
||
def filter_bids_for_close(
|
||
bids: list[dict[str, Any]] | None,
|
||
*,
|
||
mark_px: float | None = None,
|
||
intrinsic_px: float | None = None,
|
||
min_ratio: float = BID_CLOSE_MIN_RATIO,
|
||
) -> tuple[list[dict[str, Any]], bool, str]:
|
||
"""过滤不可用于平仓的残档买盘.返回 (usable_bids, had_stub_only, reason)."""
|
||
raw = list(bids or [])
|
||
usable: list[dict[str, Any]] = []
|
||
stub_reason = ""
|
||
for level in raw:
|
||
px = _safe_px(level.get("px") if isinstance(level, dict) else None)
|
||
stub, reason = is_stub_bid_px(px, mark_px=mark_px, intrinsic_px=intrinsic_px, min_ratio=min_ratio)
|
||
if stub:
|
||
if not stub_reason:
|
||
stub_reason = reason or "买一无效"
|
||
continue
|
||
usable.append(level)
|
||
if raw and not usable:
|
||
return [], True, stub_reason or "暂无有效买盘"
|
||
return usable, False, ""
|
||
|
||
|
||
def estimate_close_by_bids(
|
||
bids: list[dict[str, Any]] | None,
|
||
sheets: int | float,
|
||
*,
|
||
ct_mult: float = 0.01,
|
||
premium_paid: float | None = None,
|
||
mark_px: float | None = None,
|
||
intrinsic_px: float | None = None,
|
||
min_bid_ratio: float = BID_CLOSE_MIN_RATIO,
|
||
max_levels: int = 1,
|
||
) -> dict[str, Any]:
|
||
"""按买盘估算限价卖出可收回金额;默认只估算买一(与实盘平仓一致);残档不参与."""
|
||
target = max(0, int(float(sheets or 0)))
|
||
remaining = target
|
||
total_received = 0.0
|
||
levels: list[dict[str, Any]] = []
|
||
max_lv = max(1, int(max_levels or 1))
|
||
empty = {
|
||
"levels": [],
|
||
"covered_sheets": 0,
|
||
"uncovered_sheets": target,
|
||
"total_received": 0.0,
|
||
"avg_px": None,
|
||
"estimated_pnl": None,
|
||
"estimated_pnl_ratio_pct": None,
|
||
"bid_invalid": False,
|
||
"bid_invalid_reason": None,
|
||
"auto_close_blocked": False,
|
||
"max_levels": max_lv,
|
||
}
|
||
if target <= 0 or ct_mult <= 0:
|
||
return empty
|
||
usable, stub_only, stub_reason = filter_bids_for_close(
|
||
bids, mark_px=mark_px, intrinsic_px=intrinsic_px, min_ratio=min_bid_ratio
|
||
)
|
||
if stub_only:
|
||
out = dict(empty)
|
||
out["bid_invalid"] = True
|
||
out["bid_invalid_reason"] = stub_reason
|
||
out["auto_close_blocked"] = True
|
||
out["raw_bid_px"] = _safe_px((bids or [{}])[0].get("px")) if bids else None
|
||
return out
|
||
for i, level in enumerate(usable[:max_lv], 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
|
||
# 净盈亏 = 本轮买盘可回收 − 全部权利金(买一不够时剩余张数计入 uncovered)
|
||
estimated_pnl = None
|
||
estimated_pnl_ratio_pct = None
|
||
if premium_paid is not None and covered > 0:
|
||
paid = float(premium_paid)
|
||
estimated_pnl = round(total_received - paid, 4)
|
||
if paid > 0:
|
||
estimated_pnl_ratio_pct = round(estimated_pnl / paid * 100.0, 2)
|
||
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,
|
||
"estimated_pnl_ratio_pct": estimated_pnl_ratio_pct,
|
||
"bid_invalid": False,
|
||
"bid_invalid_reason": None,
|
||
"auto_close_blocked": False,
|
||
"max_levels": max_lv,
|
||
}
|
||
|
||
|
||
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 resolve_budget_full_usdc(trading_usdc: float, trade_budget_usdc: float) -> float:
|
||
"""按可用余额打满:余额大于预算用预算,否则用余额."""
|
||
return min(float(trading_usdc), float(trade_budget_usdc))
|
||
|
||
|
||
def resolve_compound_full_usdc(
|
||
trading_usdc: float,
|
||
*,
|
||
cap_enabled: bool = False,
|
||
cap_usdc: float | None = None,
|
||
) -> float:
|
||
"""全仓复利:默认用期权交易户全部可用;上限开关开启时再封顶."""
|
||
bal = max(0.0, float(trading_usdc or 0))
|
||
if not cap_enabled:
|
||
return bal
|
||
try:
|
||
cap = float(cap_usdc) if cap_usdc is not None else 0.0
|
||
except (TypeError, ValueError):
|
||
cap = 0.0
|
||
if cap <= 0:
|
||
return bal
|
||
return min(bal, cap)
|
||
|
||
|
||
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,
|
||
quote_in_coin: bool | None = None,
|
||
inst_id: str | None = None,
|
||
margin_mode: str | 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,
|
||
quote_in_coin=quote_in_coin,
|
||
inst_id=inst_id,
|
||
margin_mode=margin_mode,
|
||
)
|
||
|
||
|
||
def expiry_breakeven_px(
|
||
*,
|
||
opt_type: str,
|
||
strike: float | None,
|
||
avg_px: float | None,
|
||
be_px_api: float | None = None,
|
||
quote_in_coin: bool | None = None,
|
||
inst_id: str | None = None,
|
||
margin_mode: str | 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
|
||
try:
|
||
k = float(strike)
|
||
p = float(avg_px)
|
||
except (TypeError, ValueError):
|
||
return None
|
||
if p <= 0:
|
||
return None
|
||
o = (opt_type or "").upper()
|
||
coin = _quote_in_coin_from_context(
|
||
quote_in_coin=quote_in_coin, inst_id=inst_id, margin_mode=margin_mode
|
||
)
|
||
if coin:
|
||
# 币本位:权利金为币报价;到期结算 payoff 亦为币 → K/(1±p)
|
||
if o == "C":
|
||
if p >= 1:
|
||
return None
|
||
return round(k / (1 - p), 2)
|
||
if o == "P":
|
||
return round(k / (1 + p), 2)
|
||
return None
|
||
if o == "C":
|
||
return round(k + p, 2)
|
||
if o == "P":
|
||
return round(k - p, 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 strike_distance_to_be(
|
||
strike: float | None,
|
||
be_px: float | None,
|
||
*,
|
||
opt_type: str | None = None,
|
||
) -> float | None:
|
||
"""行权价到到期平衡价的价差(Call:BE−K, Put:K−BE)."""
|
||
if strike is None or be_px is None:
|
||
return None
|
||
k = float(strike)
|
||
be = float(be_px)
|
||
o = (opt_type or "").upper()
|
||
if o == "C":
|
||
return round(be - k, 2)
|
||
if o == "P":
|
||
return round(k - be, 2)
|
||
return round(abs(be - k), 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_value_at_index(
|
||
*,
|
||
opt_type: str,
|
||
strike: float | None,
|
||
target_idx: float | None,
|
||
eth_amount: float | None,
|
||
) -> float | None:
|
||
"""到期测算:目标指数价下期权内在价值总额(不含已付权利金)."""
|
||
if strike is None or target_idx 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(eth_amount), 2)
|
||
|
||
|
||
def estimate_expiry_profit_at_index(
|
||
*,
|
||
opt_type: str,
|
||
strike: float | None,
|
||
target_idx: float | None,
|
||
entry_px: float | None,
|
||
eth_amount: float | None,
|
||
total_premium: float | None = None,
|
||
) -> float | None:
|
||
"""到期测算:目标指数价下净盈利 = 预计价值 − 权利金."""
|
||
value = estimate_expiry_value_at_index(
|
||
opt_type=opt_type,
|
||
strike=strike,
|
||
target_idx=target_idx,
|
||
eth_amount=eth_amount,
|
||
)
|
||
if value is None:
|
||
return None
|
||
prem = total_premium
|
||
if prem is None and entry_px is not None and eth_amount is not None:
|
||
prem = float(entry_px) * float(eth_amount)
|
||
if prem is None:
|
||
return None
|
||
return round(float(value) - float(prem), 2)
|
||
|
||
|
||
def equivalent_contract_leverage(
|
||
*,
|
||
index_px: float | None,
|
||
eth_amount: float | None,
|
||
total_premium: float | None,
|
||
margin_mode: str | None = None,
|
||
) -> float | None:
|
||
"""名义价值 / 权利金,近似相当于永续合约杠杆倍数(测算用).
|
||
|
||
USDC: 权利金为美元 → index×eth/premium.
|
||
币本位: 权利金为币 → eth/premium(=1/ask 当 premium=ask×eth).
|
||
"""
|
||
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
|
||
try:
|
||
from lib.options.options_margin_mode_lib import normalize_options_margin_mode
|
||
|
||
mode = normalize_options_margin_mode(margin_mode)
|
||
except Exception:
|
||
mode = (str(margin_mode or "usdc").strip().lower() or "usdc")
|
||
if mode == "coin":
|
||
return round(float(eth_amount) / float(total_premium), 1)
|
||
return round(float(index_px) * float(eth_amount) / float(total_premium), 1)
|
||
|
||
|
||
def straddle_ask_per_unit(
|
||
call_ask: float | None,
|
||
put_ask: float | None,
|
||
) -> float | None:
|
||
"""跨式双买:每 1 标的币的卖一报价之和."""
|
||
if call_ask is None or put_ask is None:
|
||
return None
|
||
if float(call_ask) <= 0 or float(put_ask) <= 0:
|
||
return None
|
||
return round(float(call_ask) + float(put_ask), 4)
|
||
|
||
|
||
def straddle_premium_total(
|
||
call_ask: float | None,
|
||
put_ask: float | None,
|
||
eth_amount: float | None,
|
||
) -> float | None:
|
||
"""跨式双买权利金总额(USDC)."""
|
||
per = straddle_ask_per_unit(call_ask, put_ask)
|
||
if per is None or eth_amount is None or float(eth_amount) <= 0:
|
||
return None
|
||
return round(per * float(eth_amount), 2)
|
||
|
||
|
||
def straddle_breakeven_band(
|
||
strike: float | None,
|
||
combined_ask_per_unit: float | None = None,
|
||
*,
|
||
call_ask: float | None = None,
|
||
put_ask: float | None = None,
|
||
quote_in_coin: bool = False,
|
||
) -> tuple[float | None, float | None]:
|
||
"""跨式到期平衡带:下平衡 ~ 上平衡."""
|
||
if strike is None:
|
||
return None, None
|
||
k = float(strike)
|
||
if quote_in_coin:
|
||
pc = _safe_px(call_ask)
|
||
pp = _safe_px(put_ask)
|
||
if pc is None or pp is None or pc <= 0 or pp <= 0 or pc >= 1:
|
||
return None, None
|
||
return round(k / (1 + pp), 2), round(k / (1 - pc), 2)
|
||
if combined_ask_per_unit is None:
|
||
return None, None
|
||
d = float(combined_ask_per_unit)
|
||
return round(k - d, 2), round(k + d, 2)
|
||
|
||
|
||
def format_straddle_band(
|
||
strike: float | None,
|
||
combined_ask_per_unit: float | None,
|
||
) -> str:
|
||
lo, hi = straddle_breakeven_band(strike, combined_ask_per_unit)
|
||
if lo is None or hi is None:
|
||
return ""
|
||
return f"{lo:.0f} ~ {hi:.0f}"
|