dekun
174 lines
5.3 KiB
Python
174 lines
5.3 KiB
Python
"""策略选约：剩余时长 + ATM 平值 + 期权杠杆（交易所无关）。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from datetime import datetime, timedelta
|
||
from typing import Any
|
||
from zoneinfo import ZoneInfo
|
||
|
||
from ..exchange.types import OptionPair
|
||
|
||
_SH = ZoneInfo("Asia/Shanghai")
|
||
|
||
|
||
def hours_until_ms(expiry_ms: int, now: datetime | None = None) -> float:
|
||
n = (now or datetime.now(tz=_SH)).astimezone(_SH)
|
||
return (int(expiry_ms) - int(n.timestamp() * 1000)) / 3_600_000.0
|
||
|
||
|
||
def hours_until_expiry(
|
||
ymd: str,
|
||
now: datetime | None = None,
|
||
*,
|
||
expiry_ms: int | None = None,
|
||
) -> float:
|
||
if expiry_ms is not None:
|
||
return hours_until_ms(expiry_ms, now)
|
||
# 兼容测试：无 ms 时按 OKX 惯例（UTC 08:00）推算
|
||
from ..exchange.expiry import expiry_ms_from_ymd
|
||
|
||
return hours_until_ms(expiry_ms_from_ymd(ymd), now)
|
||
|
||
|
||
def next_session_expiry_ymd(now: datetime | None = None) -> str:
|
||
now_sh = (now or datetime.now(tz=_SH)).astimezone(_SH)
|
||
open_today = now_sh.replace(hour=16, minute=0, second=0, microsecond=0)
|
||
if now_sh >= open_today:
|
||
target = now_sh.date() + timedelta(days=1)
|
||
else:
|
||
target = now_sh.date()
|
||
return target.strftime("%y%m%d")
|
||
|
||
|
||
def pick_atm_strike(strikes: list[float], mark_px: float) -> float | None:
|
||
if not strikes or mark_px <= 0:
|
||
return None
|
||
return min(strikes, key=lambda s: (abs(s - mark_px), s))
|
||
|
||
|
||
def atm_open_offset(strike: float, mark_px: float) -> float:
|
||
"""开仓用：ATM 行权价相对标的的绝对点差。"""
|
||
return abs(float(strike) - float(mark_px))
|
||
|
||
|
||
def atm_allows_open(
|
||
strike: float,
|
||
mark_px: float,
|
||
*,
|
||
max_offset: float,
|
||
enabled: bool = False,
|
||
) -> bool:
|
||
"""开启限制时：|strike − mark| ≤ max_offset 才允许开仓；关闭则始终允许。"""
|
||
if not enabled:
|
||
return True
|
||
if mark_px <= 0 or max_offset < 0:
|
||
return False
|
||
return atm_open_offset(strike, mark_px) <= float(max_offset) + 1e-9
|
||
|
||
|
||
def option_leverage(underlying_px: float, premium_ask: float) -> float | None:
|
||
if underlying_px <= 0 or premium_ask is None or premium_ask <= 0:
|
||
return None
|
||
return float(underlying_px) / float(premium_ask)
|
||
|
||
|
||
def _complete_by_expiry(
|
||
contracts: list[dict[str, Any]],
|
||
) -> dict[str, tuple[int, dict[float, dict[str, str]]]]:
|
||
"""ymd -> (expiry_ms, strike -> {C|P: instId})"""
|
||
by_exp: dict[str, dict[float, dict[str, str]]] = {}
|
||
ms_map: dict[str, int] = {}
|
||
for c in contracts:
|
||
y = str(c.get("expiry_ymd") or "")
|
||
stk = c.get("strike")
|
||
opt = str(c.get("side") or "").upper()
|
||
inst_id = str(c.get("inst_id") or "")
|
||
if not y or stk is None or opt not in ("C", "P") or not inst_id:
|
||
continue
|
||
by_exp.setdefault(y, {}).setdefault(float(stk), {})[opt] = inst_id
|
||
if c.get("expiry_ms") is not None:
|
||
ms_map[y] = int(c["expiry_ms"])
|
||
out: dict[str, tuple[int, dict[float, dict[str, str]]]] = {}
|
||
for ymd, strikes in by_exp.items():
|
||
complete = {s: v for s, v in strikes.items() if "C" in v and "P" in v}
|
||
if not complete:
|
||
continue
|
||
if ymd in ms_map:
|
||
ems = ms_map[ymd]
|
||
else:
|
||
from ..exchange.expiry import expiry_ms_from_ymd
|
||
|
||
ems = expiry_ms_from_ymd(ymd)
|
||
out[ymd] = (ems, complete)
|
||
return out
|
||
|
||
|
||
def list_eligible_expiry_ymds(
|
||
contracts: list[dict[str, Any]],
|
||
*,
|
||
min_hours: float,
|
||
now: datetime | None = None,
|
||
) -> list[str]:
|
||
complete = _complete_by_expiry(contracts)
|
||
eligible = [
|
||
ymd
|
||
for ymd, (ems, _) in complete.items()
|
||
if hours_until_ms(ems, now) + 1e-9 >= float(min_hours)
|
||
]
|
||
return sorted(eligible, key=lambda y: complete[y][0])
|
||
|
||
|
||
def select_option_pair(
|
||
contracts: list[dict[str, Any]],
|
||
*,
|
||
mark_px: float,
|
||
expiry_ymd: str | None = None,
|
||
min_hours: float | None = None,
|
||
now: datetime | None = None,
|
||
) -> OptionPair | None:
|
||
complete = _complete_by_expiry(contracts)
|
||
if not complete:
|
||
return None
|
||
|
||
if expiry_ymd:
|
||
ymd = expiry_ymd
|
||
if ymd not in complete:
|
||
return None
|
||
elif min_hours is not None:
|
||
eligible = list_eligible_expiry_ymds(contracts, min_hours=min_hours, now=now)
|
||
if not eligible:
|
||
return None
|
||
ymd = eligible[0]
|
||
else:
|
||
ymd = next_session_expiry_ymd(now)
|
||
if ymd not in complete:
|
||
eligible = list_eligible_expiry_ymds(contracts, min_hours=0, now=now)
|
||
if not eligible:
|
||
return None
|
||
ymd = eligible[0]
|
||
|
||
ems, strikes_map = complete[ymd]
|
||
atm = pick_atm_strike(list(strikes_map.keys()), mark_px)
|
||
if atm is None:
|
||
return None
|
||
legs = strikes_map[atm]
|
||
return OptionPair(
|
||
expiry_ymd=ymd,
|
||
expiry_ms=ems,
|
||
strike=atm,
|
||
call_inst_id=legs["C"],
|
||
put_inst_id=legs["P"],
|
||
)
|
||
|
||
|
||
def normalize_contracts(contracts_or_rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||
"""若已是中性结构则原样返回；否则按 OKX 原始行解析（测试兼容）。"""
|
||
if not contracts_or_rows:
|
||
return []
|
||
sample = contracts_or_rows[0]
|
||
if "inst_id" in sample and "expiry_ymd" in sample:
|
||
return contracts_or_rows
|
||
from ..exchange.okx.parse import rows_to_option_contracts
|
||
|
||
return rows_to_option_contracts(contracts_or_rows)
|