From c7ccf237d71761af009f04c97b27da409e03f41f Mon Sep 17 00:00:00 2001 From: OpenSquared Date: Sat, 20 Jun 2026 10:15:45 +0200 Subject: [PATCH] =?UTF-8?q?feat:=20IV=20gate=20=E2=80=94=20block=20ALERT?= =?UTF-8?q?=20trades=20before=20logging=20+=20configurable=20thresholds?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 --- backend/services/auto_cycle.py | 162 +++++++++++++++++++- backend/services/database.py | 5 + backend/services/options_technical_agent.py | 36 +++-- 3 files changed, 189 insertions(+), 14 deletions(-) diff --git a/backend/services/auto_cycle.py b/backend/services/auto_cycle.py index ae4eef7..9bbecda 100644 --- a/backend/services/auto_cycle.py +++ b/backend/services/auto_cycle.py @@ -300,10 +300,11 @@ def run_cycle_once(trigger: str = "auto") -> Dict[str, Any]: dominant = scenarios.get("dominant", "incertain") summary["dominant_regime"] = dominant - # ── Step 1.9: Pre-fetch IV context for strategy suggestion rules ────── + # ── Step 1.9: Pre-fetch IV context + snapshots (suggestion rules + gate) ─ iv_context = "" + _iv_snapshots: Dict[str, Dict] = {} try: - from services.iv_engine import get_iv_context_for_prompt, IV_WATCHLIST + from services.iv_engine import get_iv_context_for_prompt, get_full_iv_snapshot, IV_WATCHLIST from services.database import get_mtm_trades_with_traces _mtm_pre = get_mtm_trades_with_traces(days=90) _trade_tickers_pre = list({ @@ -313,8 +314,17 @@ def run_cycle_once(trigger: str = "auto") -> Dict[str, Any]: }) _iv_tickers_pre = (_trade_tickers_pre + IV_WATCHLIST[:6])[:10] iv_context = get_iv_context_for_prompt(_iv_tickers_pre) + # Also collect structured snapshots — used by IV gate before log_trade_entries + for _tkr in _iv_tickers_pre: + try: + _iv_snapshots[_tkr] = get_full_iv_snapshot(_tkr) + except Exception: + pass if iv_context: - logger.info(f"[Cycle {run_id[:16]}] IV context pre-fetched for {len(_iv_tickers_pre)} tickers (suggestion step)") + logger.info( + f"[Cycle {run_id[:16]}] IV context pre-fetched: {len(_iv_tickers_pre)} tickers, " + f"{len(_iv_snapshots)} snapshots" + ) except Exception as _e: logger.warning(f"[Cycle] IV context pre-fetch failed (non-blocking): {_e}") @@ -540,6 +550,55 @@ def run_cycle_once(trigger: str = "auto") -> Dict[str, Any]: news_count=len(news), run_id=scoring_run_id) _options_assessment = None + + # ── IV Gate: block ALERT-verdict trades before logging ──────────────── + _iv_gate_blocked: List[Dict] = [] + try: + from services.database import get_config as _gc + if (_gc("iv_gate_enabled") or "true").lower() == "true": + # Augment _iv_snapshots with any new tickers from scored that weren't in the pre-fetch + from services.iv_engine import get_full_iv_snapshot as _gfiv + _scored_tickers = { + (tr.get("underlying") or "").upper() + for sp in scored + for tr in (sp.get("trade_rankings") or sp.get("suggested_trades") or []) + if tr.get("underlying") + } + for _stk in _scored_tickers - set(_iv_snapshots.keys()): + try: + _iv_snapshots[_stk] = _gfiv(_stk) + except Exception: + pass + + scored, _iv_gate_blocked = _apply_iv_gate(scored, _iv_snapshots) + if _iv_gate_blocked: + summary["iv_gate_blocked"] = len(_iv_gate_blocked) + logger.warning( + f"[IVGate] {len(_iv_gate_blocked)} trade(s) blocked by IV rules: " + + ", ".join(f"{b['ticker']} {b['strategy']}" for b in _iv_gate_blocked) + ) + for _blk in _iv_gate_blocked: + try: + from services.database import log_skipped_trade + log_skipped_trade( + run_id=scoring_run_id, + pattern_id=_blk.get("pattern_id", ""), + pattern_name=_blk.get("pattern_name", ""), + underlying=_blk.get("ticker", ""), + strategy=_blk.get("strategy", ""), + score=_blk.get("score", 0), + expected_move_pct=0, + skip_detail=( + f"[IV_GATE] ALERT — {'; '.join(_blk.get('iv_issues', [])[:2])}. " + f"Stratégie optimale: {_blk.get('optimal_strategy', '?')}" + ), + asset_class=_blk.get("asset_class", ""), + ) + except Exception: + pass + except Exception as _ige: + logger.warning(f"[IVGate] Failed (non-blocking): {_ige}") + log_trade_entries(run_id=scoring_run_id, scored_patterns=scored, quotes=quotes) # ── Step 5.2: Options Technical Agent — validate newly logged trades ── @@ -808,6 +867,103 @@ Réponds UNIQUEMENT en JSON: {{"commentary": "", "key_ris return None +# ── IV Gate ─────────────────────────────────────────────────────────────────── + +def _apply_iv_gate( + scored: List[Dict], + iv_snapshots: Dict[str, Dict], +) -> tuple: + """ + Filter ALERT-verdict trades from each scored pattern's suggested_trades/trade_rankings + before they reach log_trade_entries. + + Returns (filtered_scored, blocked_trades_list). + IVR thresholds are read from app_config (iv_gate_ivr_high, iv_gate_ivr_extreme, + iv_gate_skew_threshold) so they can be tuned from the Config page. + """ + from services.options_technical_agent import assess_strategy_fit + + # Read thresholds from config (fall back to hardcoded defaults) + try: + from services.database import get_config as _gc + _ivr_high = float(_gc("iv_gate_ivr_high") or 60) + _ivr_extreme = float(_gc("iv_gate_ivr_extreme") or 80) + _skew_thresh = float(_gc("iv_gate_skew_threshold") or 8) + except Exception: + _ivr_high, _ivr_extreme, _skew_thresh = 60.0, 80.0, 8.0 + + # Monkey-patch thresholds into the rule engine for this call + import services.options_technical_agent as _ota_mod + _orig_ivr_high = getattr(_ota_mod, "_IVR_HIGH", None) + try: + _ota_mod._IVR_HIGH = _ivr_high + _ota_mod._IVR_EXTREME = _ivr_extreme + _ota_mod._SKEW_THRESH = _skew_thresh + except Exception: + pass + + all_blocked: List[Dict] = [] + + for sp in scored: + trade_key = "trade_rankings" if "trade_rankings" in sp else "suggested_trades" + trades = sp.get(trade_key) or [] + allowed = [] + + for trade in trades: + ticker = (trade.get("underlying") or trade.get("ticker") or "").upper() + strategy = trade.get("strategy") or "Long Call" + snap = iv_snapshots.get(ticker, {}) + + if not snap: + # No IV data → can't block; allow but don't judge + allowed.append(trade) + continue + + 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_pct = (snap.get("skew") or {}).get("skew_pct") + term_structure = (snap.get("term_structure") or {}).get("structure") + flow_bias = (snap.get("options_flow") or {}).get("flow_bias") + + result = 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, + ) + + if result["verdict"] == "ALERT": + blocked_entry = { + "ticker": ticker, + "strategy": strategy, + "pattern_id": sp.get("pattern_id"), + "pattern_name": sp.get("geo_trigger") or sp.get("pattern_name") or "", + "score": sp.get("score", 0), + "asset_class": trade.get("asset_class") or sp.get("asset_class") or "", + "iv_rank": iv_rank, + "iv_issues": result["issues"], + "optimal_strategy": result["optimal_strategy"], + "fit_score": result["fit_score"], + } + all_blocked.append(blocked_entry) + logger.warning( + f"[IVGate] BLOCKED {ticker} {strategy} (IVR={iv_rank}, score={result['fit_score']}): " + f"{result['issues'][0][:80] if result['issues'] else 'ALERT'}" + ) + else: + allowed.append(trade) + + sp[trade_key] = allowed + + return scored, all_blocked + + # ── Cycle report generation ─────────────────────────────────────────────────── def _generate_cycle_report( diff --git a/backend/services/database.py b/backend/services/database.py index 2d28183..39808ea 100644 --- a/backend/services/database.py +++ b/backend/services/database.py @@ -499,6 +499,11 @@ def init_db(): for _key, _val in [ ("journal_retention_days", "90"), ("maturity_threshold_pct", "35"), + # IV Gate — blocks ALERT trades before they are logged + ("iv_gate_enabled", "true"), # enable/disable the IV gate entirely + ("iv_gate_ivr_high", "60"), # IVR above this = "high vol" (no naked long) + ("iv_gate_ivr_extreme", "80"), # IVR above this = "extreme vol" (sell vol only) + ("iv_gate_skew_threshold", "8"), # put skew above this = protect is expensive ]: existing = c.execute("SELECT value FROM config WHERE key=?", (_key,)).fetchone() if not existing: diff --git a/backend/services/options_technical_agent.py b/backend/services/options_technical_agent.py index d93f4c4..73414a8 100644 --- a/backend/services/options_technical_agent.py +++ b/backend/services/options_technical_agent.py @@ -12,10 +12,19 @@ Checks performed per trade: Verdict: OK | WARN | ALERT """ import logging -from typing import Any, Dict, List, Optional +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 = { @@ -93,18 +102,22 @@ def assess_strategy_fit( direction = _infer_direction(strategy) regime = _iv_regime(iv_rank) - # ── Rule 1: IV Rank fit ────────────────────────────────────────────────── + # ── 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 >= 80: + 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" ) - score -= 45 - elif iv_rank >= 60: + # 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: @@ -116,11 +129,11 @@ def assess_strategy_fit( 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: + 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 >= 50: + 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)") @@ -138,15 +151,16 @@ def assess_strategy_fit( f"— dans le top 10% du range annuel" ) - # ── Rule 3: Skew ────────────────────────────────────────────────────────── + # ── Rule 3: Skew (uses _SKEW_THRESH) ───────────────────────────────────── + skew_thresh = _SKEW_THRESH # default 8 if skew_pct is not None: - if skew_pct > 8 and "put" in s and is_long_vol: + 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 > 5: + 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 < -5 and "call" in s and is_long_vol: + 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: