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-08-03 09:36:13 +02:00
parent 7e640a09d0
commit 663e7eaa74
6 changed files with 440 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

62
backend/routers/strategy_builder.py
View File

@@ -15,6 +15,15 @@ from services.database import (
router = APIRouter(prefix="/api/strategy-builder", tags=["strategy-builder"])
class PathPointIn(BaseModel):
"""One anchor point of a scenario time-path: `value` is in the same unit as the
scalar field it overrides (spot_shock_pct: %, iv_level_shift: vol pts, skew_tilt/
term_slope_shift: same units as their scalar counterparts). `day` is elapsed days
from entry (0 = today)."""
day: float
value: float
class LegIn(BaseModel):
expiry_date: str
days_to_expiry: int
@@ -33,6 +42,19 @@ class ScenarioIn(BaseModel):
term_slope_shift: float = 0.0 # term-structure slope, per 30 days (0 at days=0)
rate_shock_bps: float = 0.0
manual_grid: Optional[List[Dict[str, Any]]] = None
# Optional time-paths: when given, the /price payoff table prices each day-row against
# the path's own interpolated value at that day (see services.scenario_path and
# payoff_heatmap's surface_at_day) instead of the single terminal shock applied
# uniformly across every day. spot_shock_pct/iv_level_shift/skew_tilt/term_slope_shift
# remain the fallback for days outside the path (and the only inputs when no path is
# given at all) and are also what /optimize and /suggested-profile still read — those
# endpoints price a single scenario point, not a full trajectory, and are unaffected
# by these fields. See services/lib/scenarioPath.ts for how shapes (bell/oscillation/
# exponential/step/custom) turn into these plain anchor-point lists.
spot_path: Optional[List[PathPointIn]] = None
iv_path: Optional[List[PathPointIn]] = None
skew_path: Optional[List[PathPointIn]] = None
term_path: Optional[List[PathPointIn]] = None
rate: float = 0.05
n_expiries: int = 3
contract_size: float = DEFAULT_CONTRACT_SIZE
@@ -131,6 +153,24 @@ class StrategySaveRequest(BaseModel):
source: str = "synthetic" # "synthetic" (Construire) | "historical" (Analyse période historique)
def _resolve_terminal_shocks(scenario: "ScenarioIn"):
"""The single point-in-time shock at horizon_days — from the path's own interpolated
value there when a path is given, otherwise the plain scalar (unchanged behavior).
This is what /optimize, /suggested-profile, and the entry/scenario cost figures use;
the day-by-day payoff table (payoff_heatmap) reads the full path directly instead."""
from services.scenario_path import interpolate_path
spot_pts = [p.model_dump() for p in scenario.spot_path] if scenario.spot_path else None
iv_pts = [p.model_dump() for p in scenario.iv_path] if scenario.iv_path else None
skew_pts = [p.model_dump() for p in scenario.skew_path] if scenario.skew_path else None
term_pts = [p.model_dump() for p in scenario.term_path] if scenario.term_path else None
return (
interpolate_path(spot_pts, scenario.horizon_days, scenario.spot_shock_pct),
interpolate_path(iv_pts, scenario.horizon_days, scenario.iv_level_shift),
interpolate_path(skew_pts, scenario.horizon_days, scenario.skew_tilt),
interpolate_path(term_pts, scenario.horizon_days, scenario.term_slope_shift),
)
def _build_surfaces(scenario: ScenarioIn):
chain_slice = get_chain_slice(
scenario.symbol, scenario.horizon_days, scenario.n_expiries,
@@ -146,12 +186,13 @@ def _build_surfaces(scenario: ScenarioIn):
)
surface_scenario = build_surface(checkpoint_chain)
else:
spot_shock, iv_shift, skew_tilt, term_slope = _resolve_terminal_shocks(scenario)
surface_scenario = apply_scenario(
surface_now,
spot_shock_pct=scenario.spot_shock_pct,
iv_level_shift=scenario.iv_level_shift,
skew_tilt=scenario.skew_tilt,
term_slope_shift=scenario.term_slope_shift,
spot_shock_pct=spot_shock,
iv_level_shift=iv_shift,
skew_tilt=skew_tilt,
term_slope_shift=term_slope,
manual_grid=scenario.manual_grid,
)
return chain_slice, surface_now, surface_scenario
@@ -220,10 +261,23 @@ def price(req: PriceRequest):
raise HTTPException(status_code=404, detail=str(e))
legs = [leg.model_dump() for leg in req.legs]
# Paths only drive the day-by-day payoff table, and only make sense for the synthetic
# parametric scenario — "Analyse période historique" (checkpoint_as_of) prices against
# a real remembered chain instead, which has no notion of a hypothesized path.
use_paths = not req.scenario.checkpoint_as_of
result = payoff_curves(
legs, chain_slice, surface_now, surface_scenario,
req.scenario.horizon_days, req.scenario.shocked_rate,
contract_size=req.scenario.contract_size,
spot_path=([p.model_dump() for p in req.scenario.spot_path] if use_paths and req.scenario.spot_path else None),
iv_path=([p.model_dump() for p in req.scenario.iv_path] if use_paths and req.scenario.iv_path else None),
skew_path=([p.model_dump() for p in req.scenario.skew_path] if use_paths and req.scenario.skew_path else None),
term_path=([p.model_dump() for p in req.scenario.term_path] if use_paths and req.scenario.term_path else None),
base_spot_shock_pct=req.scenario.spot_shock_pct,
base_iv_level_shift=req.scenario.iv_level_shift,
base_skew_tilt=req.scenario.skew_tilt,
base_term_slope_shift=req.scenario.term_slope_shift,
manual_grid=req.scenario.manual_grid,
)
result["spot"] = chain_slice["spot"]
result["scenario_spot"] = surface_scenario.spot

31
backend/services/scenario_path.py Normal file
View File

@@ -0,0 +1,31 @@
"""
Time-path scenario support: lets a scenario describe an evolving trajectory (bell, range/
oscillation, exponential, step, custom points...) for spot shock / IV level / skew tilt /
term slope across the days between entry and the scenario horizon, instead of only a single
point-in-time shock. The frontend is responsible for turning a shape+params choice into a
plain list of {day, value} anchor points (services/lib/scenarioPath.ts) — this module only
interpolates whatever anchor points it's given, so it has no notion of "bell" or
"oscillation" itself and stays reusable across spot/IV/skew/term alike.
A path is optional everywhere it's accepted: when None/empty, every call site here falls
back to the scalar shock value it already had (unchanged behavior from before paths existed).
"""
from typing import Any, Dict, List, Optional
def interpolate_path(path: Optional[List[Dict[str, Any]]], day: float, default: float) -> float:
"""Linear interpolation between anchor points {day, value}, clamped flat beyond the
first/last anchor. Falls back to `default` when no path is given at all."""
if not path:
return default
pts = sorted(path, key=lambda p: p["day"])
if day <= pts[0]["day"]:
return pts[0]["value"]
if day >= pts[-1]["day"]:
return pts[-1]["value"]
for p0, p1 in zip(pts, pts[1:]):
if p0["day"] <= day <= p1["day"]:
span = p1["day"] - p0["day"]
w = (day - p0["day"]) / span if span > 1e-9 else 0.0
return p0["value"] + (p1["value"] - p0["value"]) * w
return pts[-1]["value"]

51
backend/services/strategy_engine.py
View File

@@ -7,14 +7,15 @@ A "leg" dict: {expiry_date, days_to_expiry, strike, option_type ("call"/"put"),
position ("long"/"short"), quantity}
"""
import math
from typing import Any, Dict, List, Optional
from typing import Any, Callable, 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
from services.vol_surface import Surface, ScenarioSurface
from services.vol_surface import Surface, ScenarioSurface, apply_scenario
from services.scenario_path import interpolate_path
DEFAULT_SPREAD_PCT = 0.05 # fallback relative bid/ask spread when no live quote is found
DEFAULT_CONTRACT_SIZE = 100_000 # notional per 1 contract/lot (e.g. a standard FX lot); "quantity" on a leg is the number of these
@@ -408,8 +409,8 @@ def _find_breakevens(
def payoff_heatmap(
legs: List[Dict[str, Any]], surface: Any, eval_days_expiry: float, r: float, spot: float,
entry_ref: float, contract_size: float = DEFAULT_CONTRACT_SIZE, n_prices: int = 17, n_days: int = 7,
legs: List[Dict[str, Any]], surface_at_day: Callable[[float], Any], eval_days_expiry: float, r: float,
spot: float, entry_ref: float, contract_size: float = DEFAULT_CONTRACT_SIZE, n_prices: int = 17, n_days: int = 7,
) -> Dict[str, Any]:
"""Price x days-to-expiry grid of P&L — rows are elapsed-day checkpoints from today down
to expiry (top-to-bottom reading matches watching the position age). Columns are
@@ -418,12 +419,20 @@ def payoff_heatmap(
window left more than half the grid flat at max loss/gain for a near-the-money position,
wasting resolution nowhere near where the P&L actually transitions. The exact expiry
breakeven(s) are pinned in as extra columns (breakeven_prices in the response) instead
of only ever landing near one by luck of the price sampling."""
of only ever landing near one by luck of the price sampling.
`surface_at_day` is a function of elapsed days -> a Surface-like object (.iv_at), called
ONCE per row rather than per cell. When the scenario has no time-path, the caller just
passes a constant `lambda d: surface_scenario` (today's single-shock behavior, unchanged);
with a path, each row gets its own interpolated spot/IV/skew/term shock — e.g. a vol pop
described for day 4 onward actually raises the extrinsic value in that row (and every
later one), not just at whatever single instant the old single-point scenario evaluated."""
strikes = [l["strike"] for l in legs if l["option_type"] != "stock"]
half_width = max(max(abs(spot - k) for k in strikes) * 1.4, spot * 0.03) if strikes else spot * 0.15
lo, hi = max(spot - half_width, spot * 0.01), spot + half_width
breakevens = _find_breakevens(legs, surface, eval_days_expiry, r, entry_ref, spot, contract_size)
surface_at_expiry = surface_at_day(eval_days_expiry)
breakevens = _find_breakevens(legs, surface_at_expiry, eval_days_expiry, r, entry_ref, spot, contract_size)
near_breakevens = sorted((p for p in breakevens if lo <= p <= hi), key=lambda p: abs(p - spot))[:2]
price_points = np.unique(np.concatenate([np.linspace(lo, hi, n_prices), np.array(near_breakevens)]))
@@ -437,11 +446,12 @@ def payoff_heatmap(
rows = []
for d in day_points:
d = float(d)
surf = surface_at_day(d)
pnl_row, delta_row, gamma_row, theta_row, vega_row, rho_row = [], [], [], [], [], []
for p in price_points:
p = float(p)
pnl_row.append(round(float(value_at(legs, p, d, surface, r, contract_size) - entry_ref), 2))
g = greeks_at(legs, p, d, surface, r)
pnl_row.append(round(float(value_at(legs, p, d, surf, r, contract_size) - entry_ref), 2))
g = greeks_at(legs, p, d, surf, r)
# greeks_at's own round() leaves numpy float64 as numpy float64 (round() doesn't
# coerce to native Python) — black_scholes is scipy-backed, and FastAPI's default
# JSON encoder can't serialize a bare numpy scalar (unlike pnl_row above, which
@@ -471,12 +481,35 @@ def payoff_curves(
horizon_days: int,
r: float = 0.05,
contract_size: float = DEFAULT_CONTRACT_SIZE,
spot_path: Optional[List[Dict[str, Any]]] = None,
iv_path: Optional[List[Dict[str, Any]]] = None,
skew_path: Optional[List[Dict[str, Any]]] = None,
term_path: Optional[List[Dict[str, Any]]] = None,
base_spot_shock_pct: float = 0.0,
base_iv_level_shift: float = 0.0,
base_skew_tilt: float = 0.0,
base_term_slope_shift: float = 0.0,
manual_grid: Optional[List[Dict[str, Any]]] = None,
) -> Dict[str, Any]:
spot = chain_slice["spot"]
priced = price_combo(legs, chain_slice, surface_now, surface_scenario, horizon_days, r, contract_size)
entry_ref = priced["entry_cost"]
eval_days_expiry = min(l["days_to_expiry"] for l in legs)
heatmap = payoff_heatmap(legs, surface_scenario, eval_days_expiry, r, spot, entry_ref, contract_size)
if spot_path or iv_path or skew_path or term_path:
def surface_at_day(d: float):
return apply_scenario(
surface_now,
spot_shock_pct=interpolate_path(spot_path, d, base_spot_shock_pct),
iv_level_shift=interpolate_path(iv_path, d, base_iv_level_shift),
skew_tilt=interpolate_path(skew_path, d, base_skew_tilt),
term_slope_shift=interpolate_path(term_path, d, base_term_slope_shift),
manual_grid=manual_grid,
)
else:
def surface_at_day(d: float):
return surface_scenario
heatmap = payoff_heatmap(legs, surface_at_day, eval_days_expiry, r, spot, entry_ref, contract_size)
return {"heatmap": heatmap, **priced}