"""Simulation portfolio risk analysis — mirrors IBKR risk module for logged trades.""" from __future__ import annotations from typing import Any, Dict, List, Optional from services.database import get_conn _BEARISH_KEYWORDS = {"bear", "put", "short", "sell", "vente", "baissier"} # Ticker → asset_class mapping for common instruments (fallback when pattern not in DB) _TICKER_AC: Dict[str, str] = { # Energy "CL=F": "energy", "BZ=F": "energy", "NG=F": "energy", "RB=F": "energy", "HO=F": "energy", "USO": "energy", "XLE": "energy", "XOP": "energy", "OIL": "energy", # Metals "GC=F": "metals", "SI=F": "metals", "HG=F": "metals", "PL=F": "metals", "PA=F": "metals", "GLD": "metals", "SLV": "metals", "GDX": "metals", "GDXJ": "metals", # Agriculture "ZW=F": "agriculture", "ZC=F": "agriculture", "ZS=F": "agriculture", "CT=F": "agriculture", "KC=F": "agriculture", "SB=F": "agriculture", "CC=F": "agriculture", "WEAT": "agriculture", "CORN": "agriculture", "SOYB": "agriculture", # Equity indices "^GSPC": "indices", "^DJI": "indices", "^NDX": "indices", "^RUT": "indices", "^VIX": "indices", "^FTSE": "indices", "^GDAXI": "indices", "^FCHI": "indices", "^N225": "indices", "^HSI": "indices", "^NSEI": "indices", "^BSESN": "indices", "^STOXX50E": "indices", "^IBEX": "indices", "SPY": "indices", "QQQ": "indices", "IWM": "indices", "DIA": "indices", "VXX": "indices", "UVXY": "indices", "SVXY": "indices", # Forex "EURUSD=X": "forex", "GBPUSD=X": "forex", "USDJPY=X": "forex", "AUDUSD=X": "forex", "USDCAD=X": "forex", "USDCHF=X": "forex", "NZDUSD=X": "forex", "EURGBP=X": "forex", "EURJPY=X": "forex", "GBPJPY=X": "forex", "USDCNH=X": "forex", "FXE": "forex", "UUP": "forex", "FXB": "forex", "FXY": "forex", # Rates "ZB=F": "rates", "ZN=F": "rates", "ZF=F": "rates", "ZT=F": "rates", "TLT": "rates", "IEF": "rates", "SHY": "rates", "HYG": "rates", "LQD": "rates", "EMB": "rates", } def _infer_asset_class(ticker: str) -> str: """Derive asset_class from ticker when not stored in DB.""" t = (ticker or "").upper().strip() if t in _TICKER_AC: return _TICKER_AC[t] if ":" in t: # NSE:RELIANCE, BSE:TCS, etc. return "equities" if t.endswith("=X") and len(t) >= 7: # forex pairs like EURUSD=X return "forex" if t.endswith("=F"): # generic futures return "energy" # most unknown futures are commodities if t.startswith("^"): # index return "indices" if t.isalpha() and len(t) <= 5: # short alpha = equity return "equities" return "unknown" def _direction(strategy: str) -> str: s = (strategy or "").lower() return "bearish" if any(kw in s for kw in _BEARISH_KEYWORDS) else "bullish" def get_open_simulation_trades() -> List[Dict[str, Any]]: """Open trades enriched with asset_class — via stored column, JOIN fallback, then ticker inference.""" conn = get_conn() rows = conn.execute(""" SELECT tep.*, COALESCE(tep.asset_class, cp.asset_class) AS asset_class FROM trade_entry_prices tep LEFT JOIN custom_patterns cp ON cp.id = tep.pattern_id WHERE (tep.status IS NULL OR tep.status = 'open') ORDER BY tep.entry_date DESC """).fetchall() conn.close() result = [] for r in rows: d = dict(r) if not d.get("asset_class"): d["asset_class"] = _infer_asset_class(d.get("underlying", "")) result.append(d) return result def analyze_simulation_portfolio() -> Dict[str, Any]: """Full risk breakdown of the open simulation portfolio.""" trades = get_open_simulation_trades() if not trades: return { "open_count": 0, "conflicts": [], "concentration": {}, "by_underlying": {}, "alerts": [], "direction_exposure": {}, } total = len(trades) # Group by asset_class by_class: Dict[str, List] = {} for t in trades: ac = (t.get("asset_class") or "unknown").lower() by_class.setdefault(ac, []).append(t) concentration = { ac: { "count": len(items), "pct": round(len(items) / total * 100, 1), "tickers": sorted({t["underlying"] for t in items}), "bullish": sum(1 for t in items if _direction(t.get("strategy", "")) == "bullish"), "bearish": sum(1 for t in items if _direction(t.get("strategy", "")) == "bearish"), } for ac, items in by_class.items() } # Group by underlying by_underlying: Dict[str, List] = {} for t in trades: u = (t.get("underlying") or "").upper() if u: by_underlying.setdefault(u, []).append(t) # Detect directional conflicts conflicts = [] for underlying, group in by_underlying.items(): dirs = [_direction(t.get("strategy", "")) for t in group] if "bullish" in dirs and "bearish" in dirs: conflicts.append({ "underlying": underlying, "trades": [ { "id": t["id"], "strategy": t.get("strategy", ""), "direction": _direction(t.get("strategy", "")), "entry_date": t.get("entry_date", ""), "pattern_name": t.get("pattern_name", ""), "score_at_entry": t.get("score_at_entry", 0), "asset_class": (t.get("asset_class") or "").lower(), } for t in group ], }) # Direction exposure summary per class direction_exposure = { ac: { "bullish": data["bullish"], "bearish": data["bearish"], "net": data["bullish"] - data["bearish"], "bias": "bullish" if data["bullish"] > data["bearish"] else "bearish" if data["bearish"] > data["bullish"] else "neutral", } for ac, data in concentration.items() } # Build alerts — sorted by severity alerts: List[Dict] = [] for c in conflicts: alerts.append({ "type": "conflict", "level": "danger", "underlying": c["underlying"], "message": f"Positions contradictoires sur {c['underlying']} — {len(c['trades'])} trades opposés", }) for ac, data in concentration.items(): if data["pct"] >= 35: alerts.append({ "type": "concentration", "level": "warning", "asset_class": ac, "pct": data["pct"], "message": f"{ac} = {data['pct']}% du portefeuille simulé ({data['count']} trades)", }) for underlying, group in by_underlying.items(): if len(group) >= 3: alerts.append({ "type": "overweight", "level": "warning", "underlying": underlying, "count": len(group), "message": f"{len(group)} trades sur {underlying} — sur-exposition", }) alerts.sort(key=lambda a: {"danger": 0, "warning": 1}.get(a["level"], 2)) return { "open_count": total, "conflicts": conflicts, "concentration": concentration, "by_underlying": {u: len(g) for u, g in by_underlying.items()}, "direction_exposure": direction_exposure, "alerts": alerts, } def check_new_trade(underlying: str, strategy: str, asset_class: str) -> Dict[str, Any]: """Pre-entry check: would this new trade create conflicts or concentration issues?""" open_trades = get_open_simulation_trades() total = len(open_trades) new_dir = _direction(strategy) underlying_upper = underlying.upper() warnings: List[Dict] = [] same_underlying = [t for t in open_trades if (t.get("underlying") or "").upper() == underlying_upper] for t in same_underlying: existing_dir = _direction(t.get("strategy", "")) if existing_dir != new_dir: warnings.append({ "type": "conflict", "level": "danger", "trade_id": t["id"], "existing_direction": existing_dir, "message": f"Conflit : trade #{t['id']} est {existing_dir} sur {underlying} ({t.get('strategy', '')})", }) else: warnings.append({ "type": "accumulation", "level": "info", "trade_id": t["id"], "message": f"Accumulation {new_dir} sur {underlying} (trade #{t['id']} déjà en position)", }) # Forecast concentration after adding this trade total_after = total + 1 ac_key = (asset_class or "").lower() current_ac = sum( 1 for t in open_trades if (t.get("asset_class") or "").lower() == ac_key ) future_pct = round((current_ac + 1) / total_after * 100, 1) if total_after > 0 else 100 if future_pct >= 35 and ac_key: warnings.append({ "type": "concentration", "level": "warning", "asset_class": ac_key, "future_pct": future_pct, "message": f"Ce trade porterait {ac_key} à {future_pct}% du portefeuille simulé", }) return { "warnings": warnings, "ok": not any(w["level"] == "danger" for w in warnings), "underlying": underlying, "strategy": strategy, "direction": new_dir, "total_open_after": total_after, } def build_monitor_context(risk: Dict[str, Any]) -> str: """Build a compact text summary of portfolio risk for GPT-4o.""" lines = [ f"PORTEFEUILLE SIMULÉ — {risk['open_count']} positions ouvertes", "", "RÉPARTITION PAR CLASSE D'ACTIF:", ] for ac, data in sorted(risk["concentration"].items(), key=lambda x: -x[1]["pct"]): exp = risk["direction_exposure"].get(ac, {}) lines.append( f" {ac}: {data['pct']}% ({data['count']} trades) — " f"bull={exp.get('bullish', 0)} bear={exp.get('bearish', 0)} — " f"tickers: {', '.join(data['tickers'][:5])}" ) if risk["conflicts"]: lines += ["", "CONFLITS DIRECTIONNELS (DANGER):"] for c in risk["conflicts"]: trades_desc = "; ".join( f"#{t['id']} {t['direction']} ({t['strategy']}) depuis {t['entry_date']}" for t in c["trades"] ) lines.append(f" {c['underlying']}: {trades_desc}") if risk["alerts"]: lines += ["", "ALERTES:"] for a in risk["alerts"]: lines.append(f" [{a['level'].upper()}] {a['message']}") return "\n".join(lines)