172 lines
6.4 KiB
Python
172 lines
6.4 KiB
Python
"""
|
|
Optimizer: template generation + bounded hill-climbing residual search + scoring/filtering.
|
|
|
|
Scans hundreds-to-thousands of candidate 1-4 leg structures (templates.generate_all,
|
|
plus off-template perturbations) and returns the top-N ranked by the user's chosen
|
|
objective, restricted to non-directional / bounded-risk candidates.
|
|
"""
|
|
import random
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from services.option_chain import get_chain_slice
|
|
from services.vol_surface import Surface, ScenarioSurface, build_surface, apply_scenario
|
|
from services.strategy_engine import price_combo, expected_pnl_scenario, to_native
|
|
from services.strategy_templates import generate_all, strikes_for
|
|
|
|
MAX_SEEDS_FOR_RESIDUAL_SEARCH = 40
|
|
RESIDUAL_ITERATIONS_PER_SEED = 8
|
|
RESIDUAL_MAX_EVALS = 400
|
|
|
|
|
|
def _score(priced: Dict[str, Any], legs: List[Dict[str, Any]], objective: str, surface_scenario: ScenarioSurface, horizon_days: int, r: float) -> Optional[float]:
|
|
if objective == "net_pnl":
|
|
return priced["net_pnl"]
|
|
if objective == "return_on_risk":
|
|
if not priced["max_loss"]:
|
|
return None
|
|
return priced["net_pnl"] / abs(priced["max_loss"])
|
|
if objective == "prob_weighted":
|
|
return expected_pnl_scenario(legs, surface_scenario, horizon_days, r, priced["entry_cost"])
|
|
raise ValueError(f"Objectif inconnu: {objective}")
|
|
|
|
|
|
def _passes_constraints(legs: List[Dict[str, Any]], priced: Dict[str, Any], constraints: Dict[str, Any]) -> bool:
|
|
if len(legs) > constraints["max_legs"]:
|
|
return False
|
|
if abs(priced["net_delta_now"]) > constraints["delta_threshold"]:
|
|
return False
|
|
if not priced["bounded_risk"]:
|
|
return False
|
|
cap = constraints.get("max_loss_cap")
|
|
if cap is not None and priced["max_loss"] is not None and abs(priced["max_loss"]) > cap:
|
|
return False
|
|
return True
|
|
|
|
|
|
def _evaluate(
|
|
name: str, legs: List[Dict[str, Any]], chain_slice: Dict[str, Any], surface_now: Surface,
|
|
surface_scenario: ScenarioSurface, horizon_days: int, r: float, constraints: Dict[str, Any], objective: str,
|
|
) -> Optional[Dict[str, Any]]:
|
|
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)
|
|
except Exception:
|
|
return None
|
|
if not _passes_constraints(legs, priced, constraints):
|
|
return None
|
|
score = _score(priced, legs, objective, surface_scenario, horizon_days, r)
|
|
if score is None:
|
|
return None
|
|
return {"template_name": name, "legs": legs, "score": round(score, 2), "objective": objective, **priced}
|
|
|
|
|
|
def _perturb(legs: List[Dict[str, Any]], strikes_by_expiry: Dict[Any, List[float]]) -> List[Dict[str, Any]]:
|
|
new_legs = [dict(l) for l in legs]
|
|
idx = random.randrange(len(new_legs))
|
|
leg = new_legs[idx]
|
|
kind = random.choice(["strike", "strike", "quantity"])
|
|
|
|
if kind == "strike":
|
|
strikes = strikes_by_expiry.get((leg["expiry_date"], leg["option_type"]), [])
|
|
if not strikes:
|
|
return new_legs
|
|
try:
|
|
cur_idx = strikes.index(leg["strike"])
|
|
except ValueError:
|
|
cur_idx = min(range(len(strikes)), key=lambda i: abs(strikes[i] - leg["strike"]))
|
|
step = random.choice([-2, -1, 1, 2])
|
|
new_idx = max(0, min(len(strikes) - 1, cur_idx + step))
|
|
leg["strike"] = strikes[new_idx]
|
|
else:
|
|
leg["quantity"] = max(1, min(3, leg["quantity"] + random.choice([-1, 1])))
|
|
|
|
return new_legs
|
|
|
|
|
|
def _residual_search(
|
|
seeds: List[Dict[str, Any]], chain_slice: Dict[str, Any], surface_now: Surface, surface_scenario: ScenarioSurface,
|
|
horizon_days: int, r: float, constraints: Dict[str, Any], objective: str,
|
|
) -> List[Dict[str, Any]]:
|
|
strikes_by_expiry = {
|
|
(exp["expiry_date"], opt_type): strikes_for(exp, opt_type)
|
|
for exp in chain_slice["expiries"] for opt_type in ("call", "put")
|
|
}
|
|
found: List[Dict[str, Any]] = []
|
|
evals = 0
|
|
|
|
for seed in seeds:
|
|
current = seed
|
|
for _ in range(RESIDUAL_ITERATIONS_PER_SEED):
|
|
if evals >= RESIDUAL_MAX_EVALS:
|
|
break
|
|
candidate_legs = _perturb(current["legs"], strikes_by_expiry)
|
|
evals += 1
|
|
evaluated = _evaluate(
|
|
f"{seed['template_name']} (variante)", candidate_legs, chain_slice, surface_now,
|
|
surface_scenario, horizon_days, r, constraints, objective,
|
|
)
|
|
if evaluated and evaluated["score"] > current["score"]:
|
|
current = evaluated
|
|
found.append(evaluated)
|
|
if evals >= RESIDUAL_MAX_EVALS:
|
|
break
|
|
|
|
return found
|
|
|
|
|
|
def _dedup_top_n(scored: List[Dict[str, Any]], top_n: int) -> List[Dict[str, Any]]:
|
|
seen = set()
|
|
out = []
|
|
for c in scored:
|
|
sig = (
|
|
c["template_name"].replace(" (variante)", ""),
|
|
tuple(sorted(round(l["strike"]) for l in c["legs"])),
|
|
tuple(sorted(l["expiry_date"] for l in c["legs"])),
|
|
)
|
|
if sig in seen:
|
|
continue
|
|
seen.add(sig)
|
|
out.append(c)
|
|
if len(out) >= top_n:
|
|
break
|
|
return out
|
|
|
|
|
|
def optimize(
|
|
symbol: str,
|
|
horizon_days: int,
|
|
spot_shock_pct: float,
|
|
iv_level_shift: float,
|
|
skew_tilt: float,
|
|
term_shift: float,
|
|
manual_grid: Optional[List[Dict[str, Any]]],
|
|
n_expiries: int,
|
|
rate: float,
|
|
constraints: Dict[str, Any],
|
|
objective: str,
|
|
top_n: int = 20,
|
|
) -> List[Dict[str, Any]]:
|
|
chain_slice = get_chain_slice(symbol, horizon_days, n_expiries)
|
|
surface_now = build_surface(chain_slice)
|
|
surface_scenario = apply_scenario(
|
|
surface_now, spot_shock_pct=spot_shock_pct, iv_level_shift=iv_level_shift,
|
|
skew_tilt=skew_tilt, term_shift=term_shift, manual_grid=manual_grid,
|
|
)
|
|
|
|
candidates = generate_all(chain_slice)
|
|
scored: List[Dict[str, Any]] = []
|
|
for name, legs in candidates:
|
|
evaluated = _evaluate(name, legs, chain_slice, surface_now, surface_scenario, horizon_days, rate, constraints, objective)
|
|
if evaluated:
|
|
scored.append(evaluated)
|
|
|
|
scored.sort(key=lambda c: c["score"], reverse=True)
|
|
seeds = scored[:MAX_SEEDS_FOR_RESIDUAL_SEARCH]
|
|
|
|
refined = _residual_search(seeds, chain_slice, surface_now, surface_scenario, horizon_days, rate, constraints, objective)
|
|
scored.extend(refined)
|
|
|
|
scored.sort(key=lambda c: c["score"], reverse=True)
|
|
return to_native(_dedup_top_n(scored, top_n))
|