Files
OpenFin/backend/services/scenario_profile.py
2026-07-27 18:58:02 +02:00

140 lines
6.3 KiB
Python

"""
Phase 4 of the Strategy Builder Greeks plan (see project memory) — Mode 1 ("scenario only")
and the contradiction-detection layer from the user's spec, section 12.
`infer_natural_greek_profile` answers "what Greek behavior does this scenario already imply,
before the user sets any explicit target?" — a deterministic, rule-based reading of the
spec's own lookup tables (2.1 spot / 2.2 IV), NOT a fitted or learned model. Thresholds are
judgment calls, documented inline, meant as a starting suggestion the Phase 2 profile panel
can be pre-filled with and the user can freely override — not an authoritative answer.
`detect_greek_contradictions` answers "did the user just ask for something that's hard to
get on a single option structure?" — static checks on the requested profile alone (no need
to run the optimizer), returned as non-blocking warnings, never filtering the request.
"""
from typing import Any, Dict, List, Optional
_POSITIVE_STATES = {"positive", "strong_positive"}
_STRONG_STATES = {"strong_positive", "strong_negative"}
def infer_natural_greek_profile(spot_shock_pct: float, iv_level_shift: float, horizon_days: int) -> Dict[str, Any]:
horizon_days = max(horizon_days, 1)
speed = abs(spot_shock_pct) / horizon_days # %/day intensity of the anticipated move
if abs(spot_shock_pct) < 1.0:
spot_dir = "stable"
elif spot_shock_pct > 0:
spot_dir = "hausse"
else:
spot_dir = "baisse"
# Thresholds are a judgment call, not calibrated against real move distributions —
# ~0.8%/day is "a few percent in a few days" (fast), ~0.15%/day is "a percent or two
# over a couple weeks" (progressive), below that reads as effectively directionless drift.
if speed >= 0.8:
spot_speed = "rapide"
elif speed >= 0.15:
spot_speed = "moderee"
else:
spot_speed = "lente"
if iv_level_shift >= 0.05:
iv_bucket = "forte_hausse"
elif iv_level_shift >= 0.02:
iv_bucket = "hausse_moderee"
elif iv_level_shift <= -0.02:
iv_bucket = "baisse"
else:
iv_bucket = "faible"
delta = gamma = theta = "free"
rationale: List[str] = []
# Spot -> delta/gamma/theta, spec section 2.1's table
if spot_dir == "stable":
delta, theta = "neutral", "positive"
rationale.append("Spot quasi stable → Delta proche de zéro, Theta plutôt positif (collecte de temps).")
elif spot_dir == "hausse" and spot_speed == "rapide":
delta, gamma = "strong_positive", "positive"
rationale.append("Hausse forte et rapide → Delta et Gamma positifs, la vitesse du mouvement compte autant que le niveau.")
elif spot_dir == "hausse":
delta = "positive"
theta = "positive" if spot_speed == "lente" else "neutral"
rationale.append("Hausse modérée/progressive → Delta positif, Theta plutôt positif si le mouvement reste lent.")
elif spot_dir == "baisse" and spot_speed == "rapide":
delta, gamma = "strong_negative", "positive"
rationale.append("Baisse forte et rapide → Delta négatif et Gamma positif, la vitesse compte plus que le niveau.")
else: # baisse, lente/modérée
delta = "negative"
theta = "positive" if spot_speed == "lente" else "neutral"
rationale.append("Baisse modérée ou stagnation baissière → Delta négatif faible, Theta plutôt positif.")
# IV -> vega, spec section 2.2's table — can nuance the theta read above when IV dominates
if iv_bucket == "forte_hausse":
vega = "strong_positive"
rationale.append("Forte hausse d'IV anticipée → Vega positif, idéalement avec de la convexité de vol (Vomma).")
elif iv_bucket == "hausse_moderee":
vega = "positive"
rationale.append("Hausse modérée d'IV → Vega positif, sans excès.")
elif iv_bucket == "baisse":
vega = "negative"
if theta == "free":
theta = "positive"
rationale.append("Baisse d'IV attendue (normalisation) → Vega négatif, Theta plutôt positif.")
else:
vega = "free"
return {
"delta": delta, "gamma": gamma, "theta": theta, "vega": vega, "rho": "free",
"rationale": rationale,
"reading": {"spot_direction": spot_dir, "spot_speed": spot_speed, "iv_bucket": iv_bucket},
}
def detect_greek_contradictions(
greek_profile: Optional[Dict[str, Any]], n_expiries: int,
dte_min: Optional[int], dte_max: Optional[int],
) -> List[str]:
if not greek_profile:
return []
def state_of(key: str) -> str:
return (greek_profile.get(key) or {}).get("state", "free")
def weight_of(key: str) -> float:
return (greek_profile.get(key) or {}).get("weight", 50.0)
single_expiry = (n_expiries or 1) <= 1 or (
dte_min is not None and dte_max is not None and dte_max - dte_min <= 5
)
warnings: List[str] = []
gamma_state, theta_state, delta_state, vega_state = (
state_of("gamma"), state_of("theta"), state_of("delta"), state_of("vega"),
)
if (gamma_state in _POSITIVE_STATES and theta_state in _POSITIVE_STATES
and weight_of("gamma") >= 30 and weight_of("theta") >= 30 and single_expiry):
warnings.append(
"Gamma positif et Theta positif en même temps sont difficiles à obtenir sur une seule "
"échéance. Solutions : élargir la fenêtre DTE (calendars/diagonales), réduire l'exigence "
"sur l'un des deux, ou n'exiger un Theta positif qu'autour du scénario central."
)
if delta_state == "neutral" and gamma_state in _STRONG_STATES and weight_of("delta") >= 30 and weight_of("gamma") >= 30:
warnings.append(
"Delta neutre et Gamma fortement positif se contredisent dans la durée : un Gamma élevé "
"fait bouger le Delta dès que le marché évolue — il ne restera « neutre » qu'au voisinage "
"immédiat du scénario central."
)
if vega_state == "strong_positive" and theta_state == "strong_positive" and weight_of("vega") >= 30 and weight_of("theta") >= 30:
warnings.append(
"Vega fortement positif et Theta fortement positif combinent rarement bien : la convexité "
"de volatilité coûte généralement du portage — vérifiez que le crédit net visé reste "
"cohérent avec cet objectif."
)
return warnings