300 lines
13 KiB
Python
300 lines
13 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, DEFAULT_CONTRACT_SIZE
|
|
from services.strategy_templates import generate_all, strikes_for
|
|
|
|
MAX_SEEDS_FOR_RESIDUAL_SEARCH = 40
|
|
RESIDUAL_ITERATIONS_PER_SEED = 8
|
|
RESIDUAL_MAX_EVALS = 400
|
|
|
|
GREEK_KEYS = ("delta", "gamma", "theta", "vega", "rho")
|
|
|
|
# Neutral band and "strong" percentile threshold, both self-calibrated against the actual
|
|
# candidate pool (see _greek_target_match) rather than a hardcoded absolute number — there's
|
|
# no single "big gamma" that means the same thing for EURUSD and for GOLD, but "top third of
|
|
# what's achievable for this instrument under this scenario" means the same thing for both.
|
|
_TOLERANCE_NEUTRAL_FRACTION = {"etroite": 0.05, "normale": 0.15, "large": 0.30}
|
|
_TOLERANCE_STRONG_PERCENTILE = {"etroite": 0.75, "normale": 0.66, "large": 0.50}
|
|
|
|
|
|
def _percentile_rank(sorted_vals: List[float], v: float) -> float:
|
|
if not sorted_vals:
|
|
return 0.0
|
|
import bisect
|
|
return bisect.bisect_left(sorted_vals, v) / len(sorted_vals)
|
|
|
|
|
|
def _greek_target_match(value: float, state: str, tolerance: str, sorted_abs_pool: List[float]):
|
|
"""Returns (match_score in [0,1], hard_fail: bool) for one candidate's Greek value
|
|
against one target. hard_fail only ever fires under "etroite" tolerance on a sign/neutral
|
|
violation — everything else is a soft score, blended in by optimize()."""
|
|
if state == "free":
|
|
return 1.0, False
|
|
max_abs = sorted_abs_pool[-1] if sorted_abs_pool else 0.0
|
|
neutral_band = max_abs * _TOLERANCE_NEUTRAL_FRACTION.get(tolerance, 0.15)
|
|
if state == "neutral":
|
|
ok = abs(value) <= max(neutral_band, 1e-9)
|
|
return (1.0 if ok else 0.0), (tolerance == "etroite" and not ok)
|
|
desired_sign = -1 if "negative" in state else 1
|
|
actual_sign = 1 if value > 1e-9 else (-1 if value < -1e-9 else 0)
|
|
if actual_sign != desired_sign:
|
|
return 0.0, (tolerance == "etroite")
|
|
if state in ("strong_negative", "strong_positive"):
|
|
pct = _percentile_rank(sorted_abs_pool, abs(value))
|
|
threshold = _TOLERANCE_STRONG_PERCENTILE.get(tolerance, 0.66)
|
|
return (1.0 if pct >= threshold else 0.55), False
|
|
return 1.0, False # plain "positive"/"negative": correct sign is enough
|
|
|
|
|
|
def _apply_greek_profile(scored: List[Dict[str, Any]], greek_profile: Optional[Dict[str, Any]]) -> List[Dict[str, Any]]:
|
|
"""Post-hoc re-ranking of an already-evaluated candidate pool against the requested
|
|
Greek behavior profile — see routers/strategy_builder.GreekProfileIn and project memory
|
|
(Strategy Builder Greeks plan, Phase 2). Scoped to entry-state greeks_now (a candidate's
|
|
immediate nature), not the residual search's own hill-climbing objective — the search
|
|
still climbs toward the base objective (net_pnl/return_on_risk/prob_weighted); this only
|
|
re-orders the resulting pool, it doesn't steer the search itself (a Phase 2.x refinement
|
|
if the un-guided pool turns out too shallow in practice)."""
|
|
if not scored:
|
|
return scored
|
|
active = {
|
|
k: t for k, t in (greek_profile or {}).items()
|
|
if k in GREEK_KEYS and t.get("state", "free") != "free"
|
|
}
|
|
if not active:
|
|
scored.sort(key=lambda c: c["score"], reverse=True)
|
|
return scored
|
|
|
|
# For "strong" states, the percentile pool must be restricted to candidates that already
|
|
# share the desired sign — otherwise a large WRONG-signed value (e.g. a deep-negative
|
|
# delta candidate) inflates the pool's max and makes a genuinely strong correctly-signed
|
|
# candidate look merely average by comparison.
|
|
def _sign_of(v: float) -> int:
|
|
return 1 if v > 1e-9 else (-1 if v < -1e-9 else 0)
|
|
|
|
pool_abs: Dict[str, List[float]] = {}
|
|
for k, t in active.items():
|
|
state = t.get("state", "free")
|
|
if state in ("strong_negative", "strong_positive"):
|
|
desired_sign = -1 if "negative" in state else 1
|
|
pool_abs[k] = sorted(
|
|
abs(c["greeks_now"][k]) for c in scored if _sign_of(c["greeks_now"][k]) == desired_sign
|
|
)
|
|
else:
|
|
pool_abs[k] = sorted(abs(c["greeks_now"][k]) for c in scored)
|
|
kept: List[Dict[str, Any]] = []
|
|
for c in scored:
|
|
hard_fail = False
|
|
match_scores, weights = [], []
|
|
for k, t in active.items():
|
|
v = c["greeks_now"].get(k, 0.0)
|
|
m, fail = _greek_target_match(v, t.get("state", "free"), t.get("tolerance", "normale"), pool_abs[k])
|
|
if fail:
|
|
hard_fail = True
|
|
break
|
|
match_scores.append(m)
|
|
weights.append(max(0.0, min(100.0, t.get("weight", 50.0))) / 100.0)
|
|
if hard_fail:
|
|
continue
|
|
avg_weight = sum(weights) / len(weights) if weights else 0.0
|
|
greek_match = sum(m * w for m, w in zip(match_scores, weights)) / sum(weights) if weights else 1.0
|
|
c["greek_match_score"] = round(greek_match, 3)
|
|
c["greek_weight"] = round(avg_weight, 3)
|
|
kept.append(c)
|
|
|
|
if not kept:
|
|
# Every candidate hard-failed an "étroite" target — fall back to the un-filtered
|
|
# pool (ranked on the base objective alone) rather than returning nothing, since an
|
|
# empty result reads as "no strategies exist" instead of "no strategy matches this
|
|
# strict a Greek target".
|
|
scored.sort(key=lambda c: c["score"], reverse=True)
|
|
for c in scored:
|
|
c["greek_match_score"] = 0.0
|
|
c["greek_weight"] = 0.0
|
|
return scored
|
|
|
|
# Linear blend of the two percentile ranks, NOT a product — a candidate with a perfect
|
|
# Greek match but the pool's worst raw score has base_pct=0, and multiplying would zero
|
|
# it out regardless of how much weight the user put on the Greek profile.
|
|
base_scores = sorted(c["score"] for c in kept)
|
|
for c in kept:
|
|
base_pct = _percentile_rank(base_scores, c["score"])
|
|
aw, gm = c["greek_weight"], c["greek_match_score"]
|
|
c["final_rank_score"] = round((1 - aw) * base_pct + aw * gm, 4)
|
|
kept.sort(key=lambda c: c["final_rank_score"], reverse=True)
|
|
return kept
|
|
|
|
|
|
def _score(priced: Dict[str, Any], legs: List[Dict[str, Any]], objective: str, surface_scenario: ScenarioSurface, horizon_days: int, r: float, contract_size: 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"], contract_size=contract_size)
|
|
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
|
|
delta_threshold = constraints.get("delta_threshold")
|
|
if delta_threshold is not None and abs(priced["net_delta_now"]) > 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,
|
|
contract_size: float = DEFAULT_CONTRACT_SIZE,
|
|
) -> Optional[Dict[str, Any]]:
|
|
if len(legs) > constraints["max_legs"] or len(legs) == 0:
|
|
return None
|
|
try:
|
|
# 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):
|
|
return None
|
|
score = _score(priced, legs, objective, surface_scenario, horizon_days, r, contract_size)
|
|
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,
|
|
contract_size: float = DEFAULT_CONTRACT_SIZE,
|
|
) -> 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, contract_size,
|
|
)
|
|
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_slope_shift: float,
|
|
manual_grid: Optional[List[Dict[str, Any]]],
|
|
n_expiries: int,
|
|
rate: float,
|
|
constraints: Dict[str, Any],
|
|
objective: str,
|
|
top_n: int = 20,
|
|
contract_size: float = DEFAULT_CONTRACT_SIZE,
|
|
rate_shock_bps: float = 0.0,
|
|
dte_min: Optional[int] = None,
|
|
dte_max: Optional[int] = None,
|
|
greek_profile: Optional[Dict[str, Any]] = None,
|
|
) -> List[Dict[str, Any]]:
|
|
r = rate + rate_shock_bps / 10000.0
|
|
chain_slice = get_chain_slice(symbol, horizon_days, n_expiries, dte_min=dte_min, dte_max=dte_max)
|
|
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_slope_shift=term_slope_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, r, constraints, objective, contract_size)
|
|
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, r, constraints, objective, contract_size)
|
|
scored.extend(refined)
|
|
|
|
scored = _apply_greek_profile(scored, greek_profile)
|
|
return to_native(_dedup_top_n(scored, top_n))
|