314 lines
15 KiB
Python
314 lines
15 KiB
Python
"""
|
|
Multi-leg strategy catalog for the Backtest page.
|
|
|
|
Backtest simulates years of history (2022-2024 etc.) via a single trailing-realized-vol
|
|
Black-Scholes price per leg (see routers/backtest.py) — there's no real option chain to
|
|
draw strikes from that far back (accumulated Saxo history only covers the last few weeks,
|
|
see services/option_chain.py's own docstring). The 14 template-based strategies below
|
|
therefore reuse services.strategy_templates's offset-based generators (built for Strategy
|
|
Builder against a REAL chain) against a SYNTHETIC strike grid centered on spot instead —
|
|
same leg-selection logic, just fed a fabricated but structurally identical "expiry" dict.
|
|
|
|
The 6 single/vertical strategies use `strike_offset_pct` directly (S * (1 +/- pct)),
|
|
matching the original single-leg backtest's behavior exactly rather than going through
|
|
the grid, since they don't need a strike LIST to pick from.
|
|
|
|
Vertical spreads (bull/bear call/put) aren't in strategy_templates.py — Strategy Builder's
|
|
own residual search finds them without needing a template — so they're defined locally
|
|
here rather than added to that shared module, to avoid changing Strategy Builder's and
|
|
Portfolio's already-shipped optimizer behavior as a side effect of this feature.
|
|
"""
|
|
from typing import Any, Dict, List, Optional, Tuple
|
|
|
|
from services import strategy_templates as tmpl
|
|
|
|
Leg = Dict[str, Any]
|
|
|
|
# STRATEGIES entries are (key, label, n_legs) — n_legs is purely informational (frontend
|
|
# leg-count badge), the actual leg count comes from what build_legs() returns.
|
|
STRATEGIES: List[Tuple[str, str, int]] = [
|
|
("long_call", "Long Call", 1),
|
|
("long_put", "Long Put", 1),
|
|
("bull_call_spread", "Bull Call Spread", 2),
|
|
("bear_put_spread", "Bear Put Spread", 2),
|
|
("bear_call_spread", "Bear Call Spread", 2),
|
|
("bull_put_spread", "Bull Put Spread", 2),
|
|
("long_straddle", "Long Straddle", 2),
|
|
("short_straddle", "Short Straddle", 2),
|
|
("long_strangle", "Long Strangle", 2),
|
|
("short_strangle", "Short Strangle", 2),
|
|
("call_ratio_spread", "Call Ratio Spread", 2),
|
|
("put_ratio_spread", "Put Ratio Spread", 2),
|
|
("calendar_spread", "Calendar Spread", 2),
|
|
("diagonal_spread", "Diagonal Spread", 2),
|
|
("call_butterfly", "Call Butterfly", 3),
|
|
("put_butterfly", "Put Butterfly", 3),
|
|
("call_condor", "Call Condor", 4),
|
|
("put_condor", "Put Condor", 4),
|
|
("iron_condor", "Iron Condor", 4),
|
|
("iron_butterfly", "Iron Butterfly", 4),
|
|
("broken_wing_butterfly", "Broken Wing Butterfly", 3),
|
|
("ratio_backspread", "Ratio Backspread", 2),
|
|
("jade_lizard", "Jade Lizard", 3),
|
|
("risk_reversal", "Risk Reversal", 2),
|
|
("box_spread", "Box Spread", 4),
|
|
("covered_call", "Covered Call (Buy-Write)", 2),
|
|
("protective_put", "Protective Put", 2),
|
|
("collar", "Collar (tunnel)", 3),
|
|
]
|
|
|
|
# Everything past the 6 direct (single/vertical) strategies is built from a strike LIST
|
|
# rather than a plain spot*(1+/-pct) formula.
|
|
_TEMPLATE_STRATEGY_KEYS = {s[0] for s in STRATEGIES[6:]}
|
|
_STOCK_LEG_STRATEGY_KEYS = {"covered_call", "protective_put", "collar"}
|
|
|
|
_GRID_STEP_PCT = 0.02
|
|
_GRID_HALF_WIDTH = 25 # strikes from -50% to +50% of spot in 2% steps — enough room for
|
|
# strategy_templates' OFFSETS/WIDTHS (max reach ~10 steps either side)
|
|
|
|
|
|
def synthetic_expiry(expiry_date: str, days_to_expiry: int, spot: float) -> Dict[str, Any]:
|
|
"""A fabricated 'expiry' shaped exactly like services.option_chain.get_chain_slice's
|
|
real output (expiry_date/days_to_expiry/calls/puts with {strike} rows) — strategy_templates'
|
|
generators only ever read strike lists off it, so they work unmodified against this."""
|
|
strikes = [round(spot * (1 + i * _GRID_STEP_PCT), 4) for i in range(-_GRID_HALF_WIDTH, _GRID_HALF_WIDTH + 1)]
|
|
return {
|
|
"expiry_date": expiry_date, "days_to_expiry": days_to_expiry,
|
|
"calls": [{"strike": k} for k in strikes], "puts": [{"strike": k} for k in strikes],
|
|
}
|
|
|
|
|
|
def _atm_index(strikes: List[float], spot: float) -> int:
|
|
return min(range(len(strikes)), key=lambda i: abs(strikes[i] - spot))
|
|
|
|
|
|
def _at(strikes: List[float], idx: int) -> Optional[float]:
|
|
return strikes[idx] if 0 <= idx < len(strikes) else None
|
|
|
|
|
|
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 _first_by_name(candidates: List[Tuple[str, List[Leg]]], name: str) -> Optional[List[Leg]]:
|
|
return next((legs for n, legs in candidates if n == name), None)
|
|
|
|
|
|
def _stock_leg(expiry: Dict[str, Any], spot: float, position: str, quantity: int = 1) -> Leg:
|
|
"""A position in the underlying itself (Covered Call/Protective Put/Collar), not an
|
|
option — see services.strategy_engine's option_type=="stock" handling. strike/expiry
|
|
are placeholders never read for pricing: strike=spot is harmless in check_bounded_risk's
|
|
strike-range hint, and days_to_expiry is set far out so this leg never wins a
|
|
min(l["days_to_expiry"] for l in legs) used elsewhere to pick the position's eval date."""
|
|
return {
|
|
"expiry_date": expiry["expiry_date"], "days_to_expiry": 36_500,
|
|
"strike": round(spot, 4), "option_type": "stock", "position": position, "quantity": quantity,
|
|
}
|
|
|
|
|
|
def broken_wing_butterfly(expiry: Dict[str, Any], spot: float) -> List[Leg]:
|
|
"""Like a call butterfly but the far wing is pulled wider than the near one — the
|
|
resulting asymmetry removes risk on one side entirely (funded by the wider wing's
|
|
lower cost) at the expense of a small max loss on the other."""
|
|
calls = tmpl.call_strikes(expiry)
|
|
if not calls:
|
|
return []
|
|
atm = _atm_index(calls, spot)
|
|
near_k, mid_k, far_k = _at(calls, atm + 1), _at(calls, atm + 3), _at(calls, atm + 8)
|
|
if None in (near_k, mid_k, far_k):
|
|
return []
|
|
return [_leg(expiry, near_k, "call", "long"), _leg(expiry, mid_k, "call", "short", 2), _leg(expiry, far_k, "call", "long")]
|
|
|
|
|
|
def ratio_backspread(expiry: Dict[str, Any], spot: float) -> List[Leg]:
|
|
"""Opposite of strategy_templates.ratio_spread: sell the near strike, buy 2x the
|
|
farther one — net long gamma/vega, unlimited gain if the underlying makes a big move
|
|
past the long strikes, bounded loss in the flat middle zone."""
|
|
calls = tmpl.call_strikes(expiry)
|
|
if not calls:
|
|
return []
|
|
atm = _atm_index(calls, spot)
|
|
near_k, far_k = _at(calls, atm + 1), _at(calls, atm + 4)
|
|
if None in (near_k, far_k):
|
|
return []
|
|
return [_leg(expiry, near_k, "call", "short"), _leg(expiry, far_k, "call", "long", 2)]
|
|
|
|
|
|
def jade_lizard(expiry: Dict[str, Any], spot: float) -> List[Leg]:
|
|
"""Short put + short call spread, sized so the call side's width is fully covered by
|
|
the combined credit — no upside risk by construction, only downside (below the short
|
|
put) and a capped zone in between."""
|
|
puts, calls = tmpl.put_strikes(expiry), tmpl.call_strikes(expiry)
|
|
if not puts or not calls:
|
|
return []
|
|
put_k = _at(puts, _atm_index(puts, spot) - 2)
|
|
call_k, far_call_k = _at(calls, _atm_index(calls, spot) + 1), _at(calls, _atm_index(calls, spot) + 3)
|
|
if None in (put_k, call_k, far_call_k):
|
|
return []
|
|
return [_leg(expiry, put_k, "put", "short"), _leg(expiry, call_k, "call", "short"), _leg(expiry, far_call_k, "call", "long")]
|
|
|
|
|
|
def risk_reversal(expiry: Dict[str, Any], spot: float) -> List[Leg]:
|
|
"""Sell an OTM put, buy an OTM call — a low-cost (often near-zero) directional bet,
|
|
funded by giving up protection below the short put strike."""
|
|
puts, calls = tmpl.put_strikes(expiry), tmpl.call_strikes(expiry)
|
|
if not puts or not calls:
|
|
return []
|
|
put_k = _at(puts, _atm_index(puts, spot) - 2)
|
|
call_k = _at(calls, _atm_index(calls, spot) + 2)
|
|
if None in (put_k, call_k):
|
|
return []
|
|
return [_leg(expiry, put_k, "put", "short"), _leg(expiry, call_k, "call", "long")]
|
|
|
|
|
|
def box_spread(expiry: Dict[str, Any], spot: float) -> List[Leg]:
|
|
"""Synthetic long (long call + short put) at K1 combined with a synthetic short at
|
|
K2 — a pure financing structure (locks in the strike-width discounted at the risk-free
|
|
rate) whose payoff is independent of the underlying, not a market bet."""
|
|
puts, calls = tmpl.put_strikes(expiry), tmpl.call_strikes(expiry)
|
|
common = sorted(set(puts) & set(calls))
|
|
if len(common) < 2:
|
|
return []
|
|
atm = _atm_index(common, spot)
|
|
k1 = _at(common, atm)
|
|
k2 = _at(common, min(atm + 3, len(common) - 1))
|
|
if k1 is None or k2 is None or k1 == k2:
|
|
return []
|
|
return [
|
|
_leg(expiry, k1, "call", "long"), _leg(expiry, k2, "call", "short"),
|
|
_leg(expiry, k2, "put", "long"), _leg(expiry, k1, "put", "short"),
|
|
]
|
|
|
|
|
|
def covered_call(expiry: Dict[str, Any], spot: float) -> List[Leg]:
|
|
calls = tmpl.call_strikes(expiry)
|
|
call_k = _at(calls, _atm_index(calls, spot) + 2) if calls else None
|
|
if call_k is None:
|
|
return []
|
|
return [_stock_leg(expiry, spot, "long"), _leg(expiry, call_k, "call", "short")]
|
|
|
|
|
|
def protective_put(expiry: Dict[str, Any], spot: float) -> List[Leg]:
|
|
puts = tmpl.put_strikes(expiry)
|
|
put_k = _at(puts, _atm_index(puts, spot) - 3) if puts else None
|
|
if put_k is None:
|
|
return []
|
|
return [_stock_leg(expiry, spot, "long"), _leg(expiry, put_k, "put", "long")]
|
|
|
|
|
|
def collar(expiry: Dict[str, Any], spot: float) -> List[Leg]:
|
|
puts, calls = tmpl.put_strikes(expiry), tmpl.call_strikes(expiry)
|
|
if not puts or not calls:
|
|
return []
|
|
put_k = _at(puts, _atm_index(puts, spot) - 3)
|
|
call_k = _at(calls, _atm_index(calls, spot) + 3)
|
|
if None in (put_k, call_k):
|
|
return []
|
|
return [_stock_leg(expiry, spot, "long"), _leg(expiry, put_k, "put", "long"), _leg(expiry, call_k, "call", "short")]
|
|
|
|
|
|
def _vertical(expiry: Dict[str, Any], spot: float, offset_pct: float, option_type: str, buy_near: bool) -> List[Leg]:
|
|
"""2-leg vertical, same expiry/type: one leg at spot*(1+/-offset_pct), the other at
|
|
3x that offset. buy_near=True -> debit spread (long the closer strike, short the
|
|
farther); False -> credit spread (short the closer, long the farther)."""
|
|
sign = 1 if option_type == "call" else -1
|
|
near_k = round(spot * (1 + sign * offset_pct), 4)
|
|
far_k = round(spot * (1 + sign * offset_pct * 3), 4)
|
|
near_pos, far_pos = ("long", "short") if buy_near else ("short", "long")
|
|
return [_leg(expiry, near_k, option_type, near_pos), _leg(expiry, far_k, option_type, far_pos)]
|
|
|
|
|
|
def build_legs(
|
|
strategy_key: str, spot: float, strike_offset_pct: float,
|
|
near_expiry: Dict[str, Any], far_expiry: Optional[Dict[str, Any]],
|
|
) -> List[Leg]:
|
|
"""Returns the leg list for one of STRATEGIES' keys, or [] if it can't be built
|
|
(calendar/diagonal with no far_expiry, or the synthetic grid came up short)."""
|
|
if strategy_key == "long_call":
|
|
return [_leg(near_expiry, round(spot * (1 + strike_offset_pct), 4), "call", "long")]
|
|
if strategy_key == "long_put":
|
|
return [_leg(near_expiry, round(spot * (1 - strike_offset_pct), 4), "put", "long")]
|
|
if strategy_key == "bull_call_spread":
|
|
return _vertical(near_expiry, spot, strike_offset_pct, "call", buy_near=True)
|
|
if strategy_key == "bear_put_spread":
|
|
return _vertical(near_expiry, spot, strike_offset_pct, "put", buy_near=True)
|
|
if strategy_key == "bear_call_spread":
|
|
return _vertical(near_expiry, spot, strike_offset_pct, "call", buy_near=False)
|
|
if strategy_key == "bull_put_spread":
|
|
return _vertical(near_expiry, spot, strike_offset_pct, "put", buy_near=False)
|
|
|
|
if strategy_key not in _TEMPLATE_STRATEGY_KEYS:
|
|
return []
|
|
|
|
if strategy_key in ("long_straddle", "short_straddle", "long_strangle", "short_strangle"):
|
|
name = {"long_straddle": "Long Straddle", "short_straddle": "Short Straddle",
|
|
"long_strangle": "Long Strangle", "short_strangle": "Short Strangle"}[strategy_key]
|
|
return _first_by_name(list(tmpl.straddle_strangle(near_expiry, spot)), name) or []
|
|
if strategy_key in ("call_butterfly", "put_butterfly"):
|
|
name = "Call Butterfly" if strategy_key == "call_butterfly" else "Put Butterfly"
|
|
return _first_by_name(list(tmpl.butterfly(near_expiry, spot)), name) or []
|
|
if strategy_key == "iron_butterfly":
|
|
return _first_by_name(list(tmpl.iron_butterfly(near_expiry, spot)), "Iron Butterfly") or []
|
|
if strategy_key in ("call_condor", "put_condor"):
|
|
name = "Call Condor" if strategy_key == "call_condor" else "Put Condor"
|
|
return _first_by_name(list(tmpl.condor(near_expiry, spot)), name) or []
|
|
if strategy_key == "iron_condor":
|
|
return _first_by_name(list(tmpl.iron_condor(near_expiry, spot)), "Iron Condor") or []
|
|
if strategy_key in ("call_ratio_spread", "put_ratio_spread"):
|
|
name = "Call Ratio Spread" if strategy_key == "call_ratio_spread" else "Put Ratio Spread"
|
|
return _first_by_name(list(tmpl.ratio_spread(near_expiry, spot)), name) or []
|
|
if strategy_key == "calendar_spread":
|
|
if far_expiry is None:
|
|
return []
|
|
return _first_by_name(list(tmpl.calendar_spread(near_expiry, far_expiry, spot)), "Calendar Spread") or []
|
|
if strategy_key == "diagonal_spread":
|
|
if far_expiry is None:
|
|
return []
|
|
return _first_by_name(list(tmpl.diagonal_spread(near_expiry, far_expiry, spot)), "Diagonal Spread") or []
|
|
if strategy_key == "broken_wing_butterfly":
|
|
return broken_wing_butterfly(near_expiry, spot)
|
|
if strategy_key == "ratio_backspread":
|
|
return ratio_backspread(near_expiry, spot)
|
|
if strategy_key == "jade_lizard":
|
|
return jade_lizard(near_expiry, spot)
|
|
if strategy_key == "risk_reversal":
|
|
return risk_reversal(near_expiry, spot)
|
|
if strategy_key == "box_spread":
|
|
return box_spread(near_expiry, spot)
|
|
if strategy_key == "covered_call":
|
|
return covered_call(near_expiry, spot)
|
|
if strategy_key == "protective_put":
|
|
return protective_put(near_expiry, spot)
|
|
if strategy_key == "collar":
|
|
return collar(near_expiry, spot)
|
|
return []
|
|
|
|
|
|
_NOMINAL_SPOT = 100.0
|
|
_NOMINAL_NEAR_DAYS = 90
|
|
_NOMINAL_FAR_DAYS = 180
|
|
|
|
|
|
def default_legs_pct(strategy_key: str, strike_offset_pct: float = 0.05) -> List[Dict[str, Any]]:
|
|
"""A preset's legs expressed relative to spot (strike_pct = strike/spot, e.g. 1.05 =
|
|
5% OTM call) instead of the absolute strikes build_legs() returns — this is what
|
|
seeds the frontend's editable leg editor when a preset is picked. Computed once at a
|
|
nominal spot=100, not per simulated date (routers/backtest.py's /run instead takes
|
|
the user-edited legs directly and reapplies strike_pct * spot at each entry date)."""
|
|
near = synthetic_expiry("near", _NOMINAL_NEAR_DAYS, _NOMINAL_SPOT)
|
|
far = synthetic_expiry("far", _NOMINAL_FAR_DAYS, _NOMINAL_SPOT)
|
|
legs = build_legs(strategy_key, _NOMINAL_SPOT, strike_offset_pct, near, far)
|
|
return [
|
|
{
|
|
"option_type": leg["option_type"], "position": leg["position"], "quantity": leg["quantity"],
|
|
"strike_pct": round(leg["strike"] / _NOMINAL_SPOT, 4),
|
|
# A stock leg's placeholder days_to_expiry (36500, "never expires") would
|
|
# otherwise misclassify it as "far" here — it's always effectively "near".
|
|
"expiry": "near" if leg["option_type"] == "stock" or leg["days_to_expiry"] == _NOMINAL_NEAR_DAYS else "far",
|
|
}
|
|
for leg in legs
|
|
]
|