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-19 10:23:02 +02:00
parent 0cf6a68d2d
commit 5d8bfa3371
3 changed files with 68 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

75
backend/services/strategy_engine.py
View File

@@ -10,6 +10,7 @@ import math
from typing import Any, Dict, List, Optional
import numpy as np
from scipy.optimize import minimize_scalar
from services.options_pricer import black_scholes
from services.option_chain import find_quote
@@ -116,6 +117,7 @@ def price_combo(
horizon_days: int,
r: float = 0.05,
contract_size: float = DEFAULT_CONTRACT_SIZE,
precise: bool = True,
) -> Dict[str, Any]:
spot_now = chain_slice["spot"]
spot_scenario = surface_scenario.spot
@@ -156,7 +158,7 @@ def price_combo(
# "today's vol" into a number sitting next to net_pnl (which uses the scenario's
# shocked vol), producing a max_gain that could be below net_pnl. Pricing both with
# the same scenario vol view keeps them consistent.
bounded = check_bounded_risk(legs, entry_ref, surface_scenario, spot_now, r, contract_size)
bounded = check_bounded_risk(legs, entry_ref, surface_scenario, spot_now, r, contract_size, precise)
delta_now = greeks_at(legs, spot_now, 0, surface_now, r)["delta"]
delta_scenario = greeks_at(legs, spot_scenario, horizon_days, surface_scenario, r)["delta"]
@@ -177,7 +179,10 @@ def price_combo(
})
def check_bounded_risk(legs: List[Dict[str, Any]], entry_ref: float, surface: Any, spot: float, r: float = 0.05, contract_size: float = DEFAULT_CONTRACT_SIZE) -> Dict[str, Any]:
def check_bounded_risk(
legs: List[Dict[str, Any]], entry_ref: float, surface: Any, spot: float, r: float = 0.05,
contract_size: float = DEFAULT_CONTRACT_SIZE, precise: bool = True,
) -> Dict[str, Any]:
"""
Scan a wide log-spaced spot range at expiry and inspect both tails independently for LOSS
vs GAIN direction. "Bounded risk" only requires the loss side to be capped — a long
@@ -191,13 +196,21 @@ def check_bounded_risk(legs: List[Dict[str, Any]], entry_ref: float, surface: An
every leg simultaneously (pure intrinsic, no vol assumption involved at all). For a
calendar/diagonal spread it's only the near leg — the far leg is still alive and needs
`surface` to be priced, so max_gain/max_loss there is only as good as that vol input.
`precise=False` skips the dense near-strike grid and the 1-D refinement below (bounded-
ness itself is unaffected — it only needs the tail behavior). The optimizer's bulk scan
(hundreds of candidates) uses this fast path since ranking only needs relative ordering;
the single-candidate /price call uses the full precise path.
"""
eval_days = min(l["days_to_expiry"] for l in legs)
def f(s: float) -> float:
return value_at(legs, s, eval_days, surface, r, contract_size) - entry_ref
# Wide, log-spaced tail grid — used only to detect whether the payoff flattens out
# (bounded) toward either extreme, or keeps moving further away.
tail_grid = np.geomspace(spot * 0.05, spot * 20, 300)
tail_values = [value_at(legs, float(p), eval_days, surface, r, contract_size) - entry_ref for p in tail_grid]
tail_values = [f(float(p)) for p in tail_grid]
tail_n = max(3, len(tail_values) // 30)
tol = max(abs(entry_ref), 1.0) * 0.01
@@ -207,23 +220,54 @@ def check_bounded_risk(legs: List[Dict[str, Any]], entry_ref: float, surface: An
loss_bounded = (lo_edge >= lo_in - tol) and (hi_edge >= hi_in - tol)
gain_bounded = (lo_edge <= lo_in + tol) and (hi_edge <= hi_in + tol)
if not precise:
return {
"bounded": loss_bounded,
"max_loss": round(min(tail_values), 2) if loss_bounded else None,
"max_gain": round(max(tail_values), 2) if gain_bounded else None,
}
# A calendar spread's (or ratio spread's) real best/worst case is a sharp peak right at
# a strike, not out in the tails — over a log-spaced 0.05x-20x sweep the two nearest
# samples can straddle right over it, missing the true extremum entirely (confirmed:
# for a real calendar spread the tail grid reported max_gain=-0.05 while the payoff at
# the strike itself was +0.22 — the grid simply never sampled that point). Add a dense
# linear sweep across the legs' own strikes to capture it.
# a strike, not out in the tails. Any FIXED grid — however dense — is a different finite
# sampling of the same continuous curve than whatever grid a chart or another caller
# uses, so two "close but not identical" readings of the same peak are pretty much
# guaranteed (confirmed: this grid vs the payoff chart's own grid gave two different
# peak heights for the same trade). Add a dense linear sweep across the legs' own
# strikes to locate the right neighborhood...
strikes = [l["strike"] for l in legs]
lo_k, hi_k = min(strikes) * 0.7, max(strikes) * 1.3
near_grid = np.linspace(max(lo_k, spot * 0.05), min(hi_k, spot * 20), 400)
near_values = [value_at(legs, float(p), eval_days, surface, r, contract_size) - entry_ref for p in near_grid]
near_values = [f(float(p)) for p in near_grid]
all_values = tail_values + near_values
grid_all = np.concatenate([tail_grid, near_grid])
values_all = np.concatenate([tail_values, near_values])
order = np.argsort(grid_all)
grid_sorted, values_sorted = grid_all[order], values_all[order]
# ...then refine with a bounded 1-D optimizer in a NARROW bracket around that grid
# point (the payoff at a fixed date/vol is smooth in spot, built from Black-Scholes).
# A wide bracket is actively harmful here: tested directly on a real calendar spread,
# minimize_scalar given a wide bound (±50% of the strike) converged on a flat false
# optimum far from the true spike (-65 instead of +1170) because Brent's method isn't
# guaranteed to find the global optimum over a non-unimodal interval. Anchoring the
# bracket tightly around the grid's own best point keeps the search unimodal, and
# taking max()/min() against the grid value means refinement can never do worse.
def refine(is_max: bool) -> float:
idx = int(np.argmax(values_sorted)) if is_max else int(np.argmin(values_sorted))
grid_value = float(values_sorted[idx])
step = grid_sorted[min(idx + 1, len(grid_sorted) - 1)] - grid_sorted[max(idx - 1, 0)]
step = float(step) if step > 0 else spot * 0.001
lo_b, hi_b = grid_sorted[idx] - 3 * step, grid_sorted[idx] + 3 * step
if lo_b >= hi_b:
return grid_value
res = minimize_scalar((lambda s: -f(s)) if is_max else f, bounds=(float(lo_b), float(hi_b)), method="bounded")
refined = -res.fun if is_max else res.fun
return max(refined, grid_value) if is_max else min(refined, grid_value)
return {
"bounded": loss_bounded,
"max_loss": round(min(all_values), 2) if loss_bounded else None,
"max_gain": round(max(all_values), 2) if gain_bounded else None,
"max_loss": round(refine(False), 2) if loss_bounded else None,
"max_gain": round(refine(True), 2) if gain_bounded else None,
}
@@ -289,11 +333,16 @@ def payoff_curves(
# uniform sweep over the full 0.6x-1.4x range (n points) can straddle right over that
# peak without ever sampling it (same issue fixed in check_bounded_risk). Blend a
# coarse baseline (overall shape) with a dense window around the legs' own strikes.
# The exact spot/scenario spot are forced in as sample points too — otherwise hovering
# right on the "Spot"/"Scénario" reference line reads the nearest grid point, which can
# sit meaningfully off the true value on a peak this steep, and disagree with the
# entry_cost/net_pnl tiles (computed at the exact spot, not a grid sample).
baseline = np.linspace(lo, hi, n)
strikes = [l["strike"] for l in legs]
lo_k, hi_k = max(min(strikes) * 0.9, lo), min(max(strikes) * 1.1, hi)
near_strikes = np.linspace(lo_k, hi_k, n * 3)
prices = np.unique(np.concatenate([baseline, near_strikes]))
exact_points = np.array([spot, surface_scenario.spot] + strikes)
prices = np.unique(np.concatenate([baseline, near_strikes, exact_points]))
prices.sort()
eval_days_expiry = min(l["days_to_expiry"] for l in legs)

5
backend/services/strategy_optimizer.py
View File

@@ -51,7 +51,10 @@ def _evaluate(
if len(legs) > constraints["max_legs"] or len(legs) == 0:
return None
try:
priced = price_combo(legs, chain_slice, surface_now, surface_scenario, horizon_days, r, contract_size)
# precise=False: skips the dense near-strike refinement (see check_bounded_risk) —
# ranking/filtering hundreds of candidates only needs relative ordering, not the
# exact peak height. The single loaded candidate gets refined precision via /price.
priced = price_combo(legs, chain_slice, surface_now, surface_scenario, horizon_days, r, contract_size, precise=False)
except Exception:
return None
if not _passes_constraints(legs, priced, constraints):