3ee39d5f08
- New options_technical_agent.py: rule engine (IVR, skew, term structure, flow)
+ GPT-4o narrative per trade; verdict OK/WARN/ALERT + fit_score
- options_trade_assessments table in DB for Journal badge persistence
- auto_cycle.py step 5.2: assess newly logged trades after log_trade_entries;
results embedded in cycle report
- suggest_patterns_from_market_context: +iv_context param + explicit IV→strategy
rules in prompt (IVR<30%→Long, 30-60%→Spread, >60%→no naked long, >80%→short)
- Pre-fetch iv_context at step 1.9 so suggestion step gets strategy rules
- reports.py: /api/reports/assessments/latest + /assessments/{run_id} endpoints
- RapportIA.tsx: "Validation Technique Options" section with per-trade IVBar,
VerdictBadge, issues list, GPT-4o analysis, optimal strategy suggestion
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
465 lines
21 KiB
Python
465 lines
21 KiB
Python
"""
|
||
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
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# ── 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 ──────────────────────────────────────────────────
|
||
if iv_rank is not None:
|
||
if is_long_vol:
|
||
if iv_rank >= 80:
|
||
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"
|
||
)
|
||
score -= 45
|
||
elif iv_rank >= 60:
|
||
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 >= 65:
|
||
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 >= 50:
|
||
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 ──────────────────────────────────────────────────────────
|
||
if skew_pct is not None:
|
||
if skew_pct > 8 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 > 5:
|
||
issues.append(f"Skew put positif ({skew_pct:+.1f}pts) — demande de protection élevée, marché anxieux")
|
||
score -= 8
|
||
elif skew_pct < -5 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 |
|
||
| 30–60% (vol moderate) | Bull Call Spread, Bear Put Spread | Long Straddle, Long Strangle naked |
|
||
| 60–80% (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é
|
||
"""
|