dekun
e6907dfbbe
Wire idempotent notify on open/close/partial fail, settle OO at expiry, and close orphaned TP option legs without rewriting plan totals. Co-authored-by: Cursor <cursoragent@cursor.com>
63 lines
1.8 KiB
Python
63 lines
1.8 KiB
Python
"""对冲计划结算辅助:到期内在价值与期权腿收口."""
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
from typing import Any, Optional
|
|
|
|
from lib.exchange.okx_options_lib import normalize_option_exp_ms
|
|
from lib.hedge_plan.hedge_plan_calc_lib import option_expiry_pnl
|
|
|
|
|
|
def _sf(v: Any) -> Optional[float]:
|
|
try:
|
|
if v is None or v == "":
|
|
return None
|
|
return float(v)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def leg_exp_ms(leg: dict[str, Any]) -> Optional[int]:
|
|
return normalize_option_exp_ms(leg.get("exp_time"), str(leg.get("inst_id") or ""))
|
|
|
|
|
|
def leg_is_expired(leg: dict[str, Any], *, now_ms: Optional[int] = None) -> bool:
|
|
exp = leg_exp_ms(leg)
|
|
if exp is None:
|
|
return False
|
|
now = int(now_ms if now_ms is not None else time.time() * 1000)
|
|
return now >= int(exp)
|
|
|
|
|
|
def settle_option_leg_at_spot(leg: dict[str, Any], spot: float) -> float:
|
|
"""按到期结算口径估算腿盈亏(USDC)."""
|
|
premium = float(leg.get("premium") or 0)
|
|
strike = _sf(leg.get("strike"))
|
|
if strike is None:
|
|
return -premium
|
|
sheets = float(leg.get("size") or 1)
|
|
# ct_mult 未入库时默认 0.01
|
|
ct = float(leg.get("ct_mult") or 0.01)
|
|
return float(
|
|
option_expiry_pnl(
|
|
opt_type=str(leg.get("opt_type") or "P"),
|
|
strike=float(strike),
|
|
spot=float(spot),
|
|
sheets=sheets,
|
|
ct_mult=ct,
|
|
premium_paid=premium,
|
|
)
|
|
)
|
|
|
|
|
|
def all_option_legs_expired(legs: list[dict[str, Any]], *, now_ms: Optional[int] = None) -> bool:
|
|
opts = [
|
|
x
|
|
for x in legs
|
|
if str(x.get("leg_role") or "").startswith("option")
|
|
and str(x.get("status") or "") in ("open", "hold_to_expiry")
|
|
]
|
|
if not opts:
|
|
return False
|
|
return all(leg_is_expired(x, now_ms=now_ms) for x in opts)
|