dekun
154 lines
4.8 KiB
Python
154 lines
4.8 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 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)
|