223 lines
9.5 KiB
Python
223 lines
9.5 KiB
Python
"""
|
|
Parametric generators for canonical non-directional / defined-risk option structures.
|
|
|
|
Each generator sweeps a small, bounded grid of strike offsets from ATM (not raw brute
|
|
force over every strike) so the candidate count stays in the hundreds-to-low-thousands
|
|
per expiry, not a combinatorial explosion. Every generator yields (template_name, legs).
|
|
|
|
Real chains are often asymmetric — a strike can be listed for puts but not for calls
|
|
(illiquid/untraded contract). Every generator therefore draws strikes from the
|
|
type-specific list (calls_strikes/put_strikes) for whichever leg it's building, never
|
|
from a call+put union — picking an unlisted strike would silently fall back to a
|
|
theoretical smile price instead of a real tradeable quote.
|
|
"""
|
|
from typing import Any, Dict, Iterator, List, Tuple
|
|
|
|
Leg = Dict[str, Any]
|
|
|
|
OFFSETS = [1, 2, 3, 4, 5, 6]
|
|
WIDTHS = [1, 2, 3, 4]
|
|
|
|
|
|
def call_strikes(expiry: Dict[str, Any]) -> List[float]:
|
|
return sorted({r["strike"] for r in expiry["calls"]})
|
|
|
|
|
|
def put_strikes(expiry: Dict[str, Any]) -> List[float]:
|
|
return sorted({r["strike"] for r in expiry["puts"]})
|
|
|
|
|
|
def strikes_for(expiry: Dict[str, Any], option_type: str) -> List[float]:
|
|
return call_strikes(expiry) if option_type == "call" else put_strikes(expiry)
|
|
|
|
|
|
def _atm_index(strikes: List[float], spot: float) -> int:
|
|
return min(range(len(strikes)), key=lambda i: abs(strikes[i] - spot))
|
|
|
|
|
|
def _leg(expiry: Dict[str, Any], strike: float, option_type: str, position: str, quantity: int = 1) -> Leg:
|
|
return {
|
|
"expiry_date": expiry["expiry_date"],
|
|
"days_to_expiry": expiry["days_to_expiry"],
|
|
"strike": strike,
|
|
"option_type": option_type,
|
|
"position": position,
|
|
"quantity": quantity,
|
|
}
|
|
|
|
|
|
def _at(strikes: List[float], idx: int) -> float | None:
|
|
return strikes[idx] if 0 <= idx < len(strikes) else None
|
|
|
|
|
|
def iron_condor(expiry: Dict[str, Any], spot: float) -> Iterator[Tuple[str, List[Leg]]]:
|
|
puts, calls = put_strikes(expiry), call_strikes(expiry)
|
|
if not puts or not calls:
|
|
return
|
|
atm_p, atm_c = _atm_index(puts, spot), _atm_index(calls, spot)
|
|
for po in OFFSETS[1:]:
|
|
for co in OFFSETS[1:]:
|
|
for w in WIDTHS[:3]:
|
|
sp, lp = _at(puts, atm_p - po), _at(puts, atm_p - po - w)
|
|
sc, lc = _at(calls, atm_c + co), _at(calls, atm_c + co + w)
|
|
if None in (sp, lp, sc, lc):
|
|
continue
|
|
yield "Iron Condor", [
|
|
_leg(expiry, sp, "put", "short"), _leg(expiry, lp, "put", "long"),
|
|
_leg(expiry, sc, "call", "short"), _leg(expiry, lc, "call", "long"),
|
|
]
|
|
|
|
|
|
def iron_butterfly(expiry: Dict[str, Any], spot: float) -> Iterator[Tuple[str, List[Leg]]]:
|
|
puts, calls = put_strikes(expiry), call_strikes(expiry)
|
|
common = sorted(set(puts) & set(calls))
|
|
if not common:
|
|
return
|
|
atm = _atm_index(common, spot)
|
|
for center in (0, 1):
|
|
center_strike = _at(common, atm + center)
|
|
if center_strike is None:
|
|
continue
|
|
p_idx, c_idx = puts.index(center_strike), calls.index(center_strike)
|
|
for w in OFFSETS:
|
|
lp, lc = _at(puts, p_idx - w), _at(calls, c_idx + w)
|
|
if None in (lp, lc):
|
|
continue
|
|
yield "Iron Butterfly", [
|
|
_leg(expiry, center_strike, "put", "short"), _leg(expiry, center_strike, "call", "short"),
|
|
_leg(expiry, lp, "put", "long"), _leg(expiry, lc, "call", "long"),
|
|
]
|
|
|
|
|
|
def butterfly(expiry: Dict[str, Any], spot: float) -> Iterator[Tuple[str, List[Leg]]]:
|
|
"""Call or put butterfly: long 1 low, short 2 mid, long 1 high (all same type, debit)."""
|
|
for opt_type in ("call", "put"):
|
|
strikes = strikes_for(expiry, opt_type)
|
|
if not strikes:
|
|
continue
|
|
atm = _atm_index(strikes, spot)
|
|
for center in (-1, 0, 1):
|
|
for w in OFFSETS:
|
|
mid, lo, hi = _at(strikes, atm + center), _at(strikes, atm + center - w), _at(strikes, atm + center + w)
|
|
if None in (mid, lo, hi):
|
|
continue
|
|
yield f"{opt_type.capitalize()} Butterfly", [
|
|
_leg(expiry, lo, opt_type, "long"), _leg(expiry, mid, opt_type, "short", 2),
|
|
_leg(expiry, hi, opt_type, "long"),
|
|
]
|
|
|
|
|
|
def condor(expiry: Dict[str, Any], spot: float) -> Iterator[Tuple[str, List[Leg]]]:
|
|
"""Call or put condor: long low, short mid-low, short mid-high, long high (same type)."""
|
|
for opt_type in ("call", "put"):
|
|
strikes = strikes_for(expiry, opt_type)
|
|
if not strikes:
|
|
continue
|
|
atm = _atm_index(strikes, spot)
|
|
for inner in (1, 2, 3):
|
|
for w in WIDTHS:
|
|
lo, mid_lo = _at(strikes, atm - inner - w), _at(strikes, atm - inner)
|
|
mid_hi, hi = _at(strikes, atm + inner), _at(strikes, atm + inner + w)
|
|
if None in (lo, mid_lo, mid_hi, hi):
|
|
continue
|
|
yield f"{opt_type.capitalize()} Condor", [
|
|
_leg(expiry, lo, opt_type, "long"), _leg(expiry, mid_lo, opt_type, "short"),
|
|
_leg(expiry, mid_hi, opt_type, "short"), _leg(expiry, hi, opt_type, "long"),
|
|
]
|
|
|
|
|
|
def straddle_strangle(expiry: Dict[str, Any], spot: float) -> Iterator[Tuple[str, List[Leg]]]:
|
|
puts, calls = put_strikes(expiry), call_strikes(expiry)
|
|
common = sorted(set(puts) & set(calls))
|
|
if common:
|
|
atm_strike = _at(common, _atm_index(common, spot))
|
|
if atm_strike is not None:
|
|
yield "Long Straddle", [_leg(expiry, atm_strike, "call", "long"), _leg(expiry, atm_strike, "put", "long")]
|
|
yield "Short Straddle", [_leg(expiry, atm_strike, "call", "short"), _leg(expiry, atm_strike, "put", "short")]
|
|
|
|
if not puts or not calls:
|
|
return
|
|
atm_p, atm_c = _atm_index(puts, spot), _atm_index(calls, spot)
|
|
for w in OFFSETS:
|
|
put_k, call_k = _at(puts, atm_p - w), _at(calls, atm_c + w)
|
|
if None in (put_k, call_k):
|
|
continue
|
|
yield "Long Strangle", [_leg(expiry, call_k, "call", "long"), _leg(expiry, put_k, "put", "long")]
|
|
yield "Short Strangle", [_leg(expiry, call_k, "call", "short"), _leg(expiry, put_k, "put", "short")]
|
|
|
|
|
|
def ratio_spread(expiry: Dict[str, Any], spot: float) -> Iterator[Tuple[str, List[Leg]]]:
|
|
for opt_type in ("call", "put"):
|
|
strikes = strikes_for(expiry, opt_type)
|
|
if not strikes:
|
|
continue
|
|
atm = _atm_index(strikes, spot)
|
|
sign = 1 if opt_type == "call" else -1
|
|
for near in (1, 2, 3):
|
|
for far in (2, 3, 4, 5):
|
|
if far <= near:
|
|
continue
|
|
near_k = _at(strikes, atm + sign * near)
|
|
far_k = _at(strikes, atm + sign * far)
|
|
if None in (near_k, far_k):
|
|
continue
|
|
yield f"{opt_type.capitalize()} Ratio Spread", [
|
|
_leg(expiry, near_k, opt_type, "long"), _leg(expiry, far_k, opt_type, "short", 2),
|
|
]
|
|
|
|
|
|
def calendar_spread(near_expiry: Dict[str, Any], far_expiry: Dict[str, Any], spot: float) -> Iterator[Tuple[str, List[Leg]]]:
|
|
for opt_type in ("call", "put"):
|
|
near_strikes = strikes_for(near_expiry, opt_type)
|
|
far_set = set(strikes_for(far_expiry, opt_type))
|
|
if not near_strikes or not far_set:
|
|
continue
|
|
atm = _atm_index(near_strikes, spot)
|
|
for offset in (-1, 0, 1):
|
|
k = _at(near_strikes, atm + offset)
|
|
if k is None or k not in far_set:
|
|
continue
|
|
yield "Calendar Spread", [
|
|
_leg(near_expiry, k, opt_type, "short"), _leg(far_expiry, k, opt_type, "long"),
|
|
]
|
|
|
|
|
|
def diagonal_spread(near_expiry: Dict[str, Any], far_expiry: Dict[str, Any], spot: float) -> Iterator[Tuple[str, List[Leg]]]:
|
|
for opt_type in ("call", "put"):
|
|
near_strikes = strikes_for(near_expiry, opt_type)
|
|
far_strikes = strikes_for(far_expiry, opt_type)
|
|
if not near_strikes or not far_strikes:
|
|
continue
|
|
atm_near, atm_far = _atm_index(near_strikes, spot), _atm_index(far_strikes, spot)
|
|
sign = 1 if opt_type == "call" else -1
|
|
for near_off in (1, 2, 3):
|
|
for far_off in (0, 1, 2):
|
|
near_k = _at(near_strikes, atm_near + sign * near_off)
|
|
far_k = _at(far_strikes, atm_far + sign * far_off)
|
|
if None in (near_k, far_k) or near_k == far_k:
|
|
continue
|
|
yield "Diagonal Spread", [
|
|
_leg(near_expiry, near_k, opt_type, "short"), _leg(far_expiry, far_k, opt_type, "long"),
|
|
]
|
|
|
|
|
|
def generate_all(chain_slice: Dict[str, Any]) -> List[Tuple[str, List[Leg]]]:
|
|
"""All template candidates across the fetched expiries. Single-expiry templates run per
|
|
expiry; calendar/diagonal templates pair the two nearest expiries."""
|
|
spot = chain_slice["spot"]
|
|
expiries = chain_slice["expiries"]
|
|
candidates: List[Tuple[str, List[Leg]]] = []
|
|
|
|
for exp in expiries:
|
|
for gen in (iron_condor, iron_butterfly, butterfly, condor, straddle_strangle, ratio_spread):
|
|
candidates.extend(gen(exp, spot))
|
|
|
|
by_date = sorted(expiries, key=lambda e: e["days_to_expiry"])
|
|
if len(by_date) >= 2:
|
|
near, far = by_date[0], by_date[1]
|
|
if far["days_to_expiry"] > near["days_to_expiry"]:
|
|
candidates.extend(calendar_spread(near, far, spot))
|
|
candidates.extend(diagonal_spread(near, far, spot))
|
|
|
|
return candidates
|