Files
OpenFin/backend/services/options_technical_agent.py
OpenSquared c7ccf237d7 feat: IV gate — block ALERT trades before logging + configurable thresholds
- auto_cycle.py: pre-fetch IV snapshots at step 1.9; _apply_iv_gate() runs
  before log_trade_entries, removes ALERT-verdict trades (not just reports)
- options_technical_agent.py: _IVR_HIGH/_IVR_EXTREME/_SKEW_THRESH as
  module-level vars; straddle/strangle penalty -60 (vs -56 naked) at extreme IVR
  so Long Straddle at IVR ≥ 80% → ALERT; thresholds respected in rule engine
- database.py: seed 4 iv_gate config keys (iv_gate_enabled, iv_gate_ivr_high=60,
  iv_gate_ivr_extreme=80, iv_gate_skew_threshold=8) — editable from Config page
- Blocked trades logged as skipped_trades with [IV_GATE] detail + optimal strategy

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-20 10:15:45 +02:00

479 lines
22 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""
Options Technical Agent — validates each logged trade from a pure options pricing perspective.
Checks performed per trade:
1. IV Rank / IV Percentile → is vol cheap or expensive?
2. IV vs Historical Vol ratio → is options premium justified?
3. Skew (put/call) → what is the market hedging against?
4. Term Structure (contango/back) → front-month stress or calm?
5. Options flow (call/put ratio) → smart money direction?
6. Strategy / IV fit → is the strategy appropriate for current vol regime?
Verdict: OK | WARN | ALERT
"""
import logging
from typing import Any, Dict, List, Optional, Tuple
logger = logging.getLogger(__name__)
# ── Configurable thresholds ────────────────────────────────────────────────────
# These can be overridden by auto_cycle._apply_iv_gate() using values from
# the app_config DB table (keys: iv_gate_ivr_high, iv_gate_ivr_extreme,
# iv_gate_skew_threshold). Edit them via the Config page in the UI.
_IVR_HIGH: float = 60.0 # IVR above this → no naked long vol
_IVR_EXTREME: float = 80.0 # IVR above this → sell vol only
_SKEW_THRESH: float = 8.0 # Put skew above this → puts very expensive
# ── IV regime → strategy mapping ───────────────────────────────────────────────
IV_STRATEGY_RULES = {
"long_vol": ["long call", "long put", "long straddle", "long strangle"],
"short_vol": ["short call", "short put", "short strangle", "iron condor",
"covered call", "cash secured put", "cash-secured put"],
"spread": ["bull call spread", "bear put spread", "bull put spread",
"bear call spread", "call spread", "put spread"],
}
STRATEGY_REGIME = {
# IVR < 30 → buy vol (cheap)
"low": {"preferred": ["Long Call", "Long Put", "Long Straddle"],
"avoid": ["Iron Condor", "Short Strangle"]},
# IVR 30-60 → spreads (moderate cost)
"mid": {"preferred": ["Bull Call Spread", "Bear Put Spread", "Bull Put Spread", "Bear Call Spread"],
"avoid": ["Long Straddle", "Long Strangle"]},
# IVR > 60 → sell vol / use spreads (expensive)
"high": {"preferred": ["Bull Call Spread", "Bear Put Spread", "Cash-Secured Put",
"Iron Condor", "Covered Call", "Short Strangle"],
"avoid": ["Long Call", "Long Put", "Long Straddle", "Long Strangle"]},
}
def _iv_regime(iv_rank: Optional[float]) -> str:
if iv_rank is None:
return "unknown"
if iv_rank < 30:
return "low"
if iv_rank < 60:
return "mid"
return "high"
def _infer_direction(strategy: str) -> str:
s = strategy.lower()
if any(k in s for k in ["call", "bull", "haussier"]):
return "bullish"
if any(k in s for k in ["put", "bear", "baissier"]):
return "bearish"
if any(k in s for k in ["straddle", "strangle", "iron condor"]):
return "neutral"
return "bullish"
def _optimal_strategy(direction: str, iv_rank: Optional[float]) -> str:
regime = _iv_regime(iv_rank)
if direction == "bullish":
return {"low": "Long Call", "mid": "Bull Call Spread", "high": "Cash-Secured Put ou Bull Put Spread"}.get(regime, "Bull Call Spread")
elif direction == "bearish":
return {"low": "Long Put", "mid": "Bear Put Spread", "high": "Covered Call ou Bear Call Spread"}.get(regime, "Bear Put Spread")
else:
return {"low": "Long Straddle", "mid": "Iron Condor (wing buy)", "high": "Iron Condor ou Short Strangle"}.get(regime, "Iron Condor")
def assess_strategy_fit(
strategy: str,
iv_rank: Optional[float],
iv_current_pct: Optional[float],
iv_min_52w: Optional[float],
iv_max_52w: Optional[float],
skew_pct: Optional[float],
term_structure: Optional[str],
flow_bias: Optional[str],
) -> Dict[str, Any]:
"""Rule-based assessment of strategy vs options pricing environment."""
issues: List[str] = []
positives: List[str] = []
score = 100
s = strategy.lower()
is_long_vol = any(k in s for k in IV_STRATEGY_RULES["long_vol"])
is_short_vol = any(k in s for k in IV_STRATEGY_RULES["short_vol"])
is_spread = any(k in s for k in IV_STRATEGY_RULES["spread"])
direction = _infer_direction(strategy)
regime = _iv_regime(iv_rank)
# ── Rule 1: IV Rank fit (uses configurable thresholds _IVR_HIGH / _IVR_EXTREME) ──
ivr_high = _IVR_HIGH # default 60, configurable via app_config
ivr_extreme = _IVR_EXTREME # default 80
if iv_rank is not None:
if is_long_vol:
if iv_rank >= ivr_extreme:
is_double_sided = any(k in s for k in ["straddle", "strangle"])
issues.append(
f"IVR {iv_rank:.0f}% — achat de vol au pic annuel (IV crush quasi-certain). "
f"Min 52s={iv_min_52w:.1f}% | Max={iv_max_52w:.1f}%"
if iv_min_52w and iv_max_52w else
f"IVR {iv_rank:.0f}% — achat de vol au pic annuel, IV crush probable"
)
# Straddles/strangles buy BOTH sides at peak IV → double vega exposure
score -= 60 if is_double_sided else 56
elif iv_rank >= ivr_high:
issues.append(f"IVR {iv_rank:.0f}% — vol chère, un spread débiteur réduirait le coût de vega de ~40-60%")
score -= 25
elif iv_rank <= 20:
positives.append(f"IVR {iv_rank:.0f}% — vol bon marché, timing idéal pour acheter des options")
score += 5
elif iv_rank <= 35:
positives.append(f"IVR {iv_rank:.0f}% — vol modérément bon marché, stratégie long vol pertinente")
elif is_short_vol:
if iv_rank <= 25:
issues.append(f"IVR {iv_rank:.0f}% — prime collectée faible, risque/rendement défavorable pour vendeur")
score -= 25
elif iv_rank >= ivr_high + 5:
positives.append(f"IVR {iv_rank:.0f}% — vol chère, timing favorable pour la vente de prime")
score += 10
elif is_spread:
if iv_rank >= ivr_high - 10:
positives.append(f"IVR {iv_rank:.0f}% — spread adapté : coût vega réduit, convient au régime de vol élevée")
elif iv_rank <= 20:
issues.append(f"IVR {iv_rank:.0f}% — vol bon marché, option pure plus efficace qu'un spread (gain plafonné inutilement)")
score -= 10
# ── Rule 2: IV vs Historical Vol ratio ──────────────────────────────────
if iv_current_pct and iv_min_52w and iv_max_52w:
iv_range = iv_max_52w - iv_min_52w
if iv_range > 0:
iv_normalized = (iv_current_pct - iv_min_52w) / iv_range * 100
# This is essentially IVR recalculated — use for extra context
if is_long_vol and iv_normalized >= 90:
issues.append(
f"IV actuelle {iv_current_pct:.1f}% vs range 52s [{iv_min_52w:.1f}%{iv_max_52w:.1f}%] "
f"— dans le top 10% du range annuel"
)
# ── Rule 3: Skew (uses _SKEW_THRESH) ─────────────────────────────────────
skew_thresh = _SKEW_THRESH # default 8
if skew_pct is not None:
if skew_pct > skew_thresh and "put" in s and is_long_vol:
issues.append(f"Skew put élevé ({skew_pct:+.1f}pts) — protection déjà très chère, marché en mode hedge")
score -= 15
elif skew_pct > skew_thresh * 0.6:
issues.append(f"Skew put positif ({skew_pct:+.1f}pts) — demande de protection élevée, marché anxieux")
score -= 8
elif skew_pct < -skew_thresh * 0.6 and "call" in s and is_long_vol:
issues.append(f"Skew call négatif ({skew_pct:+.1f}pts) — demande de calls élevée, options call chères relativement")
score -= 8
elif abs(skew_pct) <= 3:
positives.append(f"Skew neutre ({skew_pct:+.1f}pts) — pas de biais de protection excessif dans le marché")
# ── Rule 4: Term structure ────────────────────────────────────────────────
if term_structure:
if term_structure == "backwardation" and is_short_vol:
issues.append("Term structure en backwardation — vol front-month > back, vendre la vol est risqué en période de stress")
score -= 20
elif term_structure == "backwardation" and is_long_vol and "straddle" in s:
positives.append("Backwardation + long straddle : vol front-month élevée, catalyseur événementiel probable")
score += 8
elif term_structure == "contango" and is_short_vol:
positives.append("Contango : vol croît avec le temps, calendar spread ou vente de vol front-month avantageuse")
score += 5
elif term_structure == "contango" and is_long_vol:
issues.append("Contango : vol augmente avec l'échéance, payer le temps est coûteux pour les options longues")
score -= 5
# ── Rule 5: Options flow ──────────────────────────────────────────────────
if flow_bias:
if flow_bias == "bearish" and direction == "bullish":
issues.append("Flow options bearish (plus de puts achetés) alors que la stratégie est haussière — signal contra")
score -= 10
elif flow_bias == "bullish" and direction == "bearish":
issues.append("Flow options bullish (plus de calls achetés) alors que la stratégie est baissière — signal contra")
score -= 10
elif flow_bias == direction:
positives.append(f"Flow options aligné sur la direction ({flow_bias}) — confirmation par le smart money")
score += 5
score = max(0, min(100, score))
verdict = "OK" if score >= 70 else "WARN" if score >= 45 else "ALERT"
return {
"fit_score": score,
"verdict": verdict,
"issues": issues,
"positives": positives,
"direction_inferred": direction,
"iv_regime": regime,
"optimal_strategy": _optimal_strategy(direction, iv_rank),
}
# ── Main agent function ────────────────────────────────────────────────────────
def assess_logged_trades(
scoring_run_id: str,
scored: List[Dict],
ai_key: str,
) -> Optional[Dict]:
"""
For each trade logged in this cycle:
1. Fetch IV snapshot (IVR, skew, term structure, flow)
2. Run rule-based assessment
3. GPT-4o narrative for all trades together
Returns dict with per-trade assessments + global assessment.
"""
import os
import json as _json
os.environ["OPENAI_API_KEY"] = ai_key
# ── Get newly logged trades ───────────────────────────────────────────────
try:
from services.database import get_conn
conn = get_conn()
rows = conn.execute(
"""SELECT id, underlying, strategy, entry_date, score_at_entry, pattern_name, capital_invested
FROM trade_entry_prices WHERE run_id=? ORDER BY entry_date DESC""",
(scoring_run_id,),
).fetchall()
conn.close()
trades = [dict(r) for r in rows]
except Exception as e:
logger.warning(f"[OptionsTech] Failed to fetch logged trades: {e}")
return None
if not trades:
logger.info("[OptionsTech] No trades logged this cycle — skipping assessment")
return {"assessments": [], "global_assessment": "", "global_score": None}
# ── Fetch IV snapshots per unique ticker ──────────────────────────────────
from services.iv_engine import get_full_iv_snapshot
iv_snapshots: Dict[str, Dict] = {}
unique_tickers = list({t["underlying"] for t in trades if t.get("underlying")})
for ticker in unique_tickers:
try:
iv_snapshots[ticker] = get_full_iv_snapshot(ticker)
logger.debug(f"[OptionsTech] IV snapshot fetched for {ticker}")
except Exception as e:
logger.debug(f"[OptionsTech] IV snapshot failed for {ticker}: {e}")
iv_snapshots[ticker] = {}
# ── Rule-based assessment per trade ──────────────────────────────────────
assessments: List[Dict] = []
for t in trades:
ticker = t.get("underlying") or ""
strategy = t.get("strategy") or "Long Call"
snap = iv_snapshots.get(ticker, {})
iv_rank = snap.get("iv_rank")
iv_current_pct = snap.get("iv_current_pct")
iv_min_52w = snap.get("iv_min_52w_pct")
iv_max_52w = snap.get("iv_max_52w_pct")
skew = snap.get("skew", {})
skew_pct = skew.get("skew_pct")
term = snap.get("term_structure", {})
term_structure = term.get("structure")
flow = snap.get("options_flow", {})
flow_bias = flow.get("flow_bias")
call_put_ratio = flow.get("call_put_ratio")
iv_source = snap.get("iv_source", "none")
fit = assess_strategy_fit(
strategy=strategy,
iv_rank=iv_rank,
iv_current_pct=iv_current_pct,
iv_min_52w=iv_min_52w,
iv_max_52w=iv_max_52w,
skew_pct=skew_pct,
term_structure=term_structure,
flow_bias=flow_bias,
)
assessments.append({
"trade_id": t.get("id"),
"ticker": ticker,
"strategy": strategy,
"pattern_name": t.get("pattern_name", ""),
# IV data
"iv_rank": iv_rank,
"iv_current_pct": iv_current_pct,
"iv_min_52w_pct": iv_min_52w,
"iv_max_52w_pct": iv_max_52w,
"iv_source": iv_source,
"skew_pct": skew_pct,
"skew_interpretation": skew.get("interpretation"),
"term_structure": term_structure,
"term_iv_30d": term.get("iv_30d"),
"term_iv_90d": term.get("iv_90d"),
"flow_bias": flow_bias,
"call_put_ratio": call_put_ratio,
# Rule-based verdict
"fit_score": fit["fit_score"],
"verdict": fit["verdict"],
"issues": fit["issues"],
"positives": fit["positives"],
"iv_regime": fit["iv_regime"],
"direction_inferred": fit["direction_inferred"],
"optimal_strategy": fit["optimal_strategy"],
# GPT-4o analysis filled below
"analysis": "",
"when_to_enter": "",
})
# ── GPT-4o narrative (all trades together) ────────────────────────────────
global_assessment = ""
global_score = None
try:
from services.ai_analyzer import _chat
trades_for_prompt = []
for a in assessments:
entry = {
"ticker": a["ticker"],
"strategy": a["strategy"],
"pattern": a["pattern_name"],
"iv_rank_pct": a["iv_rank"],
"iv_current_pct": a["iv_current_pct"],
"iv_range_52w": f"{a['iv_min_52w_pct']:.1f}%{a['iv_max_52w_pct']:.1f}%"
if a["iv_min_52w_pct"] and a["iv_max_52w_pct"] else None,
"skew_pts": a["skew_pct"],
"skew_interp": a["skew_interpretation"],
"term_structure": a["term_structure"],
"iv_30d": a["term_iv_30d"],
"iv_90d": a["term_iv_90d"],
"flow_bias": a["flow_bias"],
"call_put_ratio": a["call_put_ratio"],
"rule_verdict": a["verdict"],
"rule_issues": a["issues"],
"rule_positives": a["positives"],
"optimal_strategy_suggested": a["optimal_strategy"],
}
trades_for_prompt.append(entry)
prompt = f"""Tu es un trader d'options senior avec 20 ans d'expérience en market making et volatilité.
Analyse les trades ci-dessous qui viennent d'être loggés dans notre système.
Pour chaque trade, fournis une analyse technique options rigoureuse en utilisant TOUS les indicateurs disponibles.
TRADES À ANALYSER:
{_json.dumps(trades_for_prompt, ensure_ascii=False, indent=2)}
Pour chaque trade, analyse:
1. Le TIMING de vol (IVR, range 52s, IV/HV implicite) — est-ce le bon moment pour cette stratégie ?
2. La STRUCTURE (skew, term structure, flow) — que dit le marché options lui-même ?
3. La STRATÉGIE choisie — est-elle optimale pour ce régime de vol ?
4. Le PRICE d'ENTRÉE OPTIMAL — quelle condition améliorerait le timing ?
Rappel règles d'or:
- IVR > 70% + achat d'option naked = payer la prime maximale = IV crush probable
- Skew put élevé = marché en mode protection, puts chers
- Backwardation = stress, ne pas vendre la vol
- Iron Condor / Short Strangle = uniquement IVR > 65%
- Long Straddle = uniquement si catalyseur + IVR < 30%
Réponds en JSON EXACT:
{{
"trade_assessments": [
{{
"ticker": "<ticker>",
"strategy": "<stratégie>",
"technical_score": <0-100>,
"analysis": "<3-4 phrases d'analyse technique précise: timing vol, skew, structure>",
"when_to_enter": "<condition précise et mesurable pour un meilleur timing: ex 'Attendre IVR < 40%', 'Après event X'>"
}}
],
"global_assessment": "<2-3 phrases sur la qualité technique globale des entrées de ce cycle>",
"global_score": <0-100, note technique globale du cycle d'entrées>
}}"""
result = _chat(
"Tu es un expert options/volatilité. Analyse technique précise en JSON.",
prompt,
model="gpt-4o",
json_mode=True,
max_tokens=1200,
)
if result:
global_assessment = result.get("global_assessment", "")
global_score = result.get("global_score")
ai_assessments = {a["ticker"]: a for a in result.get("trade_assessments", [])}
for a in assessments:
ai = ai_assessments.get(a["ticker"], {})
a["analysis"] = ai.get("analysis", "")
a["when_to_enter"] = ai.get("when_to_enter", "")
if ai.get("technical_score") is not None:
a["fit_score"] = int(ai["technical_score"])
a["verdict"] = "OK" if a["fit_score"] >= 70 else "WARN" if a["fit_score"] >= 45 else "ALERT"
logger.info(f"[OptionsTech] GPT-4o assessment complete — {len(assessments)} trades, global_score={global_score}")
except Exception as e:
logger.warning(f"[OptionsTech] GPT-4o narrative failed: {e}")
return {
"run_id": scoring_run_id,
"assessments": assessments,
"global_assessment": global_assessment,
"global_score": global_score,
"n_alert": sum(1 for a in assessments if a["verdict"] == "ALERT"),
"n_warn": sum(1 for a in assessments if a["verdict"] == "WARN"),
"n_ok": sum(1 for a in assessments if a["verdict"] == "OK"),
}
def save_assessments_to_db(assessments: List[Dict], run_id: str) -> None:
"""Persist per-trade assessments for Journal badge display."""
import json as _json
try:
from services.database import get_conn
conn = get_conn()
for a in assessments:
conn.execute(
"""INSERT OR REPLACE INTO options_trade_assessments
(run_id, trade_id, ticker, strategy, assessed_at,
iv_rank, iv_current_pct, skew_pct, term_structure,
fit_score, verdict, issues_json, optimal_strategy, analysis, when_to_enter)
VALUES (?, ?, ?, ?, datetime('now'), ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
(
run_id,
a.get("trade_id"),
a.get("ticker"),
a.get("strategy"),
a.get("iv_rank"),
a.get("iv_current_pct"),
a.get("skew_pct"),
a.get("term_structure"),
a.get("fit_score"),
a.get("verdict"),
_json.dumps(a.get("issues", []) + (["" + p for p in a.get("positives", [])]), ensure_ascii=False),
a.get("optimal_strategy"),
a.get("analysis", ""),
a.get("when_to_enter", ""),
),
)
conn.commit()
conn.close()
except Exception as e:
logger.warning(f"[OptionsTech] DB save failed: {e}")
# ── IV-strategy injection string for AI prompts ────────────────────────────────
OPTIONS_STRATEGY_RULES = """
## ⚠️ RÈGLES STRICTES IV → STRATÉGIE (à appliquer pour chaque trade suggéré)
Le choix de stratégie doit tenir compte du COÛT de la volatilité implicite (IVR = IV Rank 52 semaines):
| IVR | Stratégie AUTORISÉE | Stratégie INTERDITE |
|-----|---------------------|---------------------|
| < 30% (vol cheap) | Long Call, Long Put, Long Straddle | Iron Condor, Short Strangle |
| 3060% (vol moderate) | Bull Call Spread, Bear Put Spread | Long Straddle, Long Strangle naked |
| 6080% (vol chère) | Spreads débiteurs, Cash-Secured Put | Long Call naked, Long Put naked |
| > 80% (vol très chère)| Iron Condor, Short Strangle, Covered Call, Cash-Secured Put | TOUTE option long naked |
Règles supplémentaires:
- Skew put élevé (> 5 pts) → éviter Long Put (trop cher), préférer Bear Put Spread
- Term structure backwardation → ne pas vendre de vol (vendeur piégé si vol monte encore)
- Si IVR inconnu → utiliser Bull Call Spread / Bear Put Spread par défaut (neutre au coût de vol)
- Long Straddle uniquement si: IVR < 25% ET catalyseur événementiel clairement identifié
"""