admin/OpenFin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: strategy builder

Browse Source
This commit is contained in:
OpenSquared
2026-07-30 13:28:03 +02:00
parent 1c4d8013c4
commit 81165581d7
8 changed files with 361 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files

146
backend/services/backtest_strategies.py
View File

@@ -47,9 +47,20 @@ STRATEGIES: List[Tuple[str, str, int]] = [
("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),
]
_TEMPLATE_STRATEGY_KEYS = {s[0] for s in STRATEGIES[6:]} # everything past the 6 direct ones
# 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
@@ -86,6 +97,119 @@ def _first_by_name(candidates: List[Tuple[str, List[Leg]]], name: str) -> Option
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
@@ -144,6 +268,22 @@ def build_legs(
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 []
@@ -165,7 +305,9 @@ def default_legs_pct(strategy_key: str, strike_offset_pct: float = 0.05) -> List
{
"option_type": leg["option_type"], "position": leg["position"], "quantity": leg["quantity"],
"strike_pct": round(leg["strike"] / _NOMINAL_SPOT, 4),
"expiry": "near" if leg["days_to_expiry"] == _NOMINAL_NEAR_DAYS else "far",
# 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
]

44
backend/services/strategy_engine.py
View File

@@ -56,7 +56,14 @@ def _quote_spread_pct(quote: Optional[Dict[str, Any]]) -> float:
def entry_price(leg: Dict[str, Any], chain_slice: Dict[str, Any], surface_now: Surface, r: float) -> Dict[str, float]:
"""Real execution price (crossing the spread) + theoretical mid, for one leg today."""
"""Real execution price (crossing the spread) + theoretical mid, for one leg today.
A "stock" leg (Covered Call/Protective Put/Collar's underlying position, not an
option) has no strike/vol at all — its price IS the spot, no spread modeled (the
underlying's own spread is typically far tighter than any option on it, and this
tool doesn't have a live quote for it the way find_quote does for options)."""
if leg["option_type"] == "stock":
S = chain_slice["spot"]
return {"exec_price": S, "mid": S, "spread_pct": 0.0}
quote = find_quote(chain_slice, leg["expiry_date"], leg["strike"], leg["option_type"])
T = max(leg["days_to_expiry"], 0.001) / 365
sigma = surface_now.iv_at(leg["strike"], leg["days_to_expiry"])
@@ -87,14 +94,17 @@ def value_at(
computed and discarded on every one of those calls."""
total = 0.0
for leg in legs:
remaining = leg["days_to_expiry"] - eval_days_from_now
qty = leg.get("quantity", 1)
sign = _sign(leg)
if remaining <= 0:
price = _intrinsic(S, leg["strike"], leg["option_type"])
if leg["option_type"] == "stock":
price = S # a share/lot of the underlying is worth exactly the spot, always
else:
sigma = surface.iv_at(leg["strike"], remaining)
price = black_scholes(S, leg["strike"], remaining / 365, r, sigma, leg["option_type"], include_second_order=False)["price"]
remaining = leg["days_to_expiry"] - eval_days_from_now
if remaining <= 0:
price = _intrinsic(S, leg["strike"], leg["option_type"])
else:
sigma = surface.iv_at(leg["strike"], remaining)
price = black_scholes(S, leg["strike"], remaining / 365, r, sigma, leg["option_type"], include_second_order=False)["price"]
total += sign * price * qty * contract_size
return total
@@ -105,9 +115,14 @@ def greeks_at(legs: List[Dict[str, Any]], S: float, eval_days_from_now: float, s
"vanna": 0.0, "charm": 0.0, "vomma": 0.0, "veta": 0.0, "speed": 0.0, "color": 0.0, "zomma": 0.0,
}
for leg in legs:
remaining = max(leg["days_to_expiry"] - eval_days_from_now, 0.001)
qty = leg.get("quantity", 1)
sign = _sign(leg)
if leg["option_type"] == "stock":
# d(spot)/d(spot) = 1, and every other Greek (gamma, theta, vega, rho, the
# second-order ones) is exactly zero for a position in the underlying itself.
net["delta"] += sign * qty
continue
remaining = max(leg["days_to_expiry"] - eval_days_from_now, 0.001)
sigma = surface.iv_at(leg["strike"], remaining)
g = black_scholes(S, leg["strike"], remaining / 365, r, sigma, leg["option_type"])
for k in net:
@@ -169,14 +184,17 @@ def price_combo(
scenario_mid = 0.0
scenario_exec = 0.0
for leg in legs:
remaining = max(leg["days_to_expiry"] - horizon_days, 0.001)
qty = leg.get("quantity", 1)
sign = _sign(leg)
sigma = surface_scenario.iv_at(leg["strike"], remaining)
theo = black_scholes(spot_scenario, leg["strike"], remaining / 365, r, sigma, leg["option_type"])["price"]
quote = find_quote(chain_slice, leg["expiry_date"], leg["strike"], leg["option_type"])
spread_pct = _quote_spread_pct(quote)
exec_price = theo * (1 - spread_pct / 2) if leg.get("position", "long") == "long" else theo * (1 + spread_pct / 2)
if leg["option_type"] == "stock":
theo = exec_price = spot_scenario # no spread modeled for the underlying itself
else:
remaining = max(leg["days_to_expiry"] - horizon_days, 0.001)
sigma = surface_scenario.iv_at(leg["strike"], remaining)
theo = black_scholes(spot_scenario, leg["strike"], remaining / 365, r, sigma, leg["option_type"])["price"]
quote = find_quote(chain_slice, leg["expiry_date"], leg["strike"], leg["option_type"])
spread_pct = _quote_spread_pct(quote)
exec_price = theo * (1 - spread_pct / 2) if leg.get("position", "long") == "long" else theo * (1 + spread_pct / 2)
scenario_mid += sign * theo * qty * contract_size
scenario_exec += sign * exec_price * qty * contract_size

12
backend/services/strategy_replay.py
View File

@@ -36,7 +36,11 @@ def replay_position(
saxo_symbol = get_saxo_option_symbol_for_ticker(symbol) or symbol.upper()
signed_qty = [(1 if leg["position"] == "long" else -1) * leg.get("quantity", 1) for leg in legs]
avg_days = sum(leg.get("days_to_expiry", 30) for leg in legs) / len(legs)
# A "stock" leg's placeholder days_to_expiry (~effectively infinite, see
# backtest_strategies._stock_leg) would otherwise skew this — it's not a real option
# expiry and never should influence which expiries the chain fetch favors.
option_legs = [leg for leg in legs if leg["option_type"] != "stock"]
avg_days = sum(leg.get("days_to_expiry", 30) for leg in option_legs) / len(option_legs) if option_legs else 30
points: List[Dict[str, Any]] = []
entry_value = None
@@ -55,6 +59,12 @@ def replay_position(
value = 0.0 # dollar value of the whole position, contract_size already applied
complete = True
for leg, sq in zip(legs, signed_qty):
if leg["option_type"] == "stock":
if chain.get("spot") is None:
complete = False
break
value += sq * chain["spot"] * contract_size
continue
q = find_quote(chain, leg["expiry_date"], leg["strike"], leg["option_type"])
if not q or q["mid"] <= 0:
complete = False