admin/OpenFin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: IV gate — block ALERT trades before logging + configurable thresholds

Browse Source 
- 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>
This commit is contained in:
OpenSquared
2026-06-20 10:15:45 +02:00
parent 3ee39d5f08
commit c7ccf237d7
3 changed files with 189 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

162
backend/services/auto_cycle.py
View File

@@ -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": "<ton texte 4-6 phrases>", "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(

5
backend/services/database.py
View File

@@ -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:

36
backend/services/options_technical_agent.py
View File

@@ -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: