dekun
204 lines
5.8 KiB
Python
204 lines
5.8 KiB
Python
"""期期对冲选约：振幅高低点匹配虚值 Call + Put。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from dataclasses import dataclass
|
||
from datetime import datetime
|
||
from typing import Any
|
||
|
||
from ..exchange.candles import AmplitudeHL, fetch_amplitude_hl_for_runtime
|
||
from .selection import (
|
||
_complete_by_expiry,
|
||
hours_until_ms,
|
||
list_eligible_expiry_ymds,
|
||
option_leverage,
|
||
)
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class OoLeg:
|
||
side: str # call|put
|
||
strike: float
|
||
inst_id: str
|
||
ask: float
|
||
leverage: float
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class OoPickCore:
|
||
expiry_ymd: str
|
||
expiry_ms: int
|
||
hours_left: float
|
||
underlying_px: float
|
||
amplitude: AmplitudeHL
|
||
call: OoLeg
|
||
put: OoLeg
|
||
detail: str = "ok"
|
||
|
||
|
||
def _within_ref_pct(strike: float, ref: float, max_dev_pct: float) -> bool:
|
||
"""|K−ref|/ref ≤ max_dev_pct%。"""
|
||
if ref <= 0 or max_dev_pct < 0:
|
||
return False
|
||
return abs(float(strike) - float(ref)) / float(ref) * 100.0 <= float(
|
||
max_dev_pct
|
||
) + 1e-12
|
||
|
||
|
||
def pick_otm_call_strike(
|
||
strikes: list[float],
|
||
*,
|
||
spot: float,
|
||
high: float,
|
||
max_dev_pct: float = 1.0,
|
||
) -> float | None:
|
||
"""虚值 Call：K > spot，贴近振幅高点，且 |K−高|/高 ≤ max_dev_pct%。"""
|
||
cands = [float(s) for s in strikes if float(s) > float(spot) + 1e-9]
|
||
if max_dev_pct >= 0 and high > 0:
|
||
cands = [s for s in cands if _within_ref_pct(s, high, max_dev_pct)]
|
||
if not cands:
|
||
return None
|
||
return min(cands, key=lambda s: (abs(s - float(high)), s))
|
||
|
||
|
||
def pick_otm_put_strike(
|
||
strikes: list[float],
|
||
*,
|
||
spot: float,
|
||
low: float,
|
||
max_dev_pct: float = 1.0,
|
||
) -> float | None:
|
||
"""虚值 Put：K < spot，贴近振幅低点，且 |K−低|/低 ≤ max_dev_pct%。"""
|
||
cands = [float(s) for s in strikes if float(s) < float(spot) - 1e-9]
|
||
if max_dev_pct >= 0 and low > 0:
|
||
cands = [s for s in cands if _within_ref_pct(s, low, max_dev_pct)]
|
||
if not cands:
|
||
return None
|
||
return min(cands, key=lambda s: (abs(s - float(low)), s))
|
||
|
||
|
||
def select_oo_pair(
|
||
contracts: list[dict[str, Any]],
|
||
*,
|
||
spot: float,
|
||
high: float,
|
||
low: float,
|
||
min_hours: float,
|
||
now: datetime | None = None,
|
||
skip_expiry_ymds: set[str] | None = None,
|
||
max_dev_pct: float = 1.0,
|
||
) -> tuple[str, int, float, float, str, str] | None:
|
||
"""
|
||
返回 (expiry_ymd, expiry_ms, call_strike, put_strike, call_inst, put_inst)。
|
||
Call/Put 可不同行权价；须同到期、均为虚值，且相对高低点偏离不超过 max_dev_pct%。
|
||
"""
|
||
if spot <= 0 or high <= 0 or low <= 0 or high < low:
|
||
return None
|
||
complete = _complete_by_expiry(contracts)
|
||
if not complete:
|
||
return None
|
||
skip = skip_expiry_ymds or set()
|
||
eligible = [
|
||
y
|
||
for y in list_eligible_expiry_ymds(contracts, min_hours=min_hours, now=now)
|
||
if y not in skip
|
||
]
|
||
for ymd in eligible:
|
||
ems, strikes_map = complete[ymd]
|
||
strikes = list(strikes_map.keys())
|
||
ck = pick_otm_call_strike(
|
||
strikes, spot=spot, high=high, max_dev_pct=max_dev_pct
|
||
)
|
||
pk = pick_otm_put_strike(
|
||
strikes, spot=spot, low=low, max_dev_pct=max_dev_pct
|
||
)
|
||
if ck is None or pk is None:
|
||
continue
|
||
call_inst = strikes_map[ck].get("C")
|
||
put_inst = strikes_map[pk].get("P")
|
||
if not call_inst or not put_inst:
|
||
continue
|
||
hours_left = hours_until_ms(ems, now)
|
||
return (
|
||
ymd,
|
||
int(ems),
|
||
float(ck),
|
||
float(pk),
|
||
str(call_inst),
|
||
str(put_inst),
|
||
)
|
||
return None
|
||
|
||
|
||
def build_oo_pick_core(
|
||
*,
|
||
contracts: list[dict[str, Any]],
|
||
spot: float,
|
||
call_ask: float,
|
||
put_ask: float,
|
||
min_hours: float,
|
||
min_leverage: float,
|
||
amplitude_hours: float,
|
||
amplitude_pct: float,
|
||
amplitude: AmplitudeHL | None = None,
|
||
skip_expiry_ymds: set[str] | None = None,
|
||
now: datetime | None = None,
|
||
max_dev_pct: float = 1.0,
|
||
amplitude_filter_enabled: bool = True,
|
||
) -> OoPickCore | None:
|
||
"""完整期期选约：振幅门（可关）+ 虚值双腿（贴高低≤max_dev%） + 杠杆。"""
|
||
amp = amplitude or fetch_amplitude_hl_for_runtime(amplitude_hours)
|
||
if amp is None:
|
||
return None
|
||
if amplitude_filter_enabled and float(amp.range_pct) > float(amplitude_pct) + 1e-12:
|
||
return None
|
||
if spot <= 0:
|
||
spot = float(amp.mid)
|
||
picked = select_oo_pair(
|
||
contracts,
|
||
spot=float(spot),
|
||
high=float(amp.high),
|
||
low=float(amp.low),
|
||
min_hours=float(min_hours),
|
||
now=now,
|
||
skip_expiry_ymds=skip_expiry_ymds,
|
||
max_dev_pct=float(max_dev_pct),
|
||
)
|
||
if picked is None:
|
||
return None
|
||
ymd, ems, ck, pk, call_inst, put_inst = picked
|
||
if call_ask <= 0 or put_ask <= 0:
|
||
return None
|
||
c_lev = option_leverage(float(spot), float(call_ask))
|
||
p_lev = option_leverage(float(spot), float(put_ask))
|
||
if c_lev is None or p_lev is None:
|
||
return None
|
||
if c_lev + 1e-12 < float(min_leverage) or p_lev + 1e-12 < float(min_leverage):
|
||
return None
|
||
hours_left = hours_until_ms(ems, now)
|
||
return OoPickCore(
|
||
expiry_ymd=ymd,
|
||
expiry_ms=int(ems),
|
||
hours_left=float(hours_left),
|
||
underlying_px=float(spot),
|
||
amplitude=amp,
|
||
call=OoLeg(
|
||
side="call",
|
||
strike=float(ck),
|
||
inst_id=call_inst,
|
||
ask=float(call_ask),
|
||
leverage=float(c_lev),
|
||
),
|
||
put=OoLeg(
|
||
side="put",
|
||
strike=float(pk),
|
||
inst_id=put_inst,
|
||
ask=float(put_ask),
|
||
leverage=float(p_lev),
|
||
),
|
||
detail=(
|
||
f"amp={amp.range_pct:.2f}% H={amp.high:.2f} L={amp.low:.2f} "
|
||
f"C@{ck:g} P@{pk:g}"
|
||
),
|
||
)
|