"""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 _compute_avg_pairwise_correlation(underlyings: List[str], days: int = 90) -> Optional[float]: """Average pairwise correlation of daily returns across the given tickers, using whichever of them yfinance actually resolves (Saxo-only underlyings without a yfinance equivalent are silently dropped, not treated as an error).""" import numpy as np import pandas as pd import yfinance as yf if len(underlyings) < 2: return None try: raw = yf.download(underlyings, period=f"{days}d", interval="1d", progress=False, auto_adjust=True) closes = raw["Close"] if isinstance(raw.columns, pd.MultiIndex) else raw[["Close"]] except Exception: return None closes = closes.dropna(axis=1, how="all") if closes.shape[1] < 2: return None returns = closes.pct_change().dropna(how="all") corr = returns.corr().to_numpy() n = corr.shape[0] if n < 2: return None off_diag = [corr[i, j] for i in range(n) for j in range(n) if i != j and not np.isnan(corr[i, j])] if not off_diag: return None return float(np.mean(off_diag)) def _compute_max_drawdown_pct(snapshots: List[Dict[str, Any]]) -> Optional[float]: """Max peak-to-trough drop in total_pnl_pct across the P&L snapshot history (services.var_service.get_pnl_snapshots — DESC order, so reverse to oldest-first).""" if not snapshots: return None ordered = list(reversed(snapshots)) # oldest -> newest peak = ordered[0].get("total_pnl_pct") if peak is None: return None max_dd = 0.0 for snap in ordered: v = snap.get("total_pnl_pct") if v is None: continue peak = max(peak, v) max_dd = max(max_dd, peak - v) return round(max_dd, 2) def _compute_max_drawdown_eur(curve: List[Dict[str, Any]]) -> Optional[float]: """Max peak-to-trough drop (in €) across an ascending cumulative-realized-PnL curve (e.g. the closed-positions equity curve from services.database.get_positions('closed')).""" if len(curve) < 2: return None peak = curve[0].get("cumulative", 0.0) or 0.0 max_dd = 0.0 for pt in curve: v = pt.get("cumulative", 0.0) or 0.0 peak = max(peak, v) max_dd = max(max_dd, peak - v) return round(max_dd, 2) def _build_risk_radar_axes(trades: List[Dict[str, Any]], drawdown_pct: Optional[float]) -> Dict[str, Any]: """Shared 5-axis computation (Concentration/Volatility/Correlation/Exposure/Drawdown), each scaled 0-100, given a list of open trades (needs 'underlying' + a capital/price field) and a pre-computed drawdown_pct. Used by both the simulated-portfolio and real-portfolio radars — only the trade source and the drawdown source differ. - Concentration: capital-weighted share of the single largest underlying. - Volatility: capital-weighted average 20d realized vol of open positions. - Correlation: average pairwise return correlation across open positions' underlyings (only positive correlation counts as risk — negative correlation is diversification, not danger). - Exposure: open position count against a soft target of 10 concurrent trades — a proxy, NOT true margin leverage (no notional/contract-size column to compute real leverage from), so this measures "how spread thin" instead. - Drawdown: max peak-to-trough drop, pre-computed by the caller from whichever equity-curve source applies to that portfolio. """ from services.data_fetcher import get_quote_with_volatility open_count = len(trades) if not trades: return {"axes": [], "open_count": 0} weights = [max(t.get("capital_invested") or t.get("entry_price") or t.get("entry_underlying_price") or 0, 0) for t in trades] total_w = sum(weights) or 1.0 by_underlying_w: Dict[str, float] = {} for t, w in zip(trades, weights): u = (t.get("underlying") or "").upper() if u: by_underlying_w[u] = by_underlying_w.get(u, 0) + w concentration_pct = (max(by_underlying_w.values()) / total_w * 100) if by_underlying_w else 0.0 vol_cache: Dict[str, Optional[float]] = {} weighted_vol_sum, vol_weight_total = 0.0, 0.0 for t, w in zip(trades, weights): u = (t.get("underlying") or "").upper() if not u: continue if u not in vol_cache: try: q = get_quote_with_volatility(u) vol_cache[u] = q.get("volatility_pct") if q else None except Exception: vol_cache[u] = None v = vol_cache[u] if v is not None: weighted_vol_sum += v * w vol_weight_total += w avg_vol_pct = (weighted_vol_sum / vol_weight_total) if vol_weight_total else None avg_corr = _compute_avg_pairwise_correlation(sorted(by_underlying_w.keys())) exposure_score = min(100.0, open_count / 10 * 100) def _scale(v: Optional[float], cap: float) -> Optional[float]: return round(min(100.0, max(0.0, v / cap * 100)), 1) if v is not None else None axes = [ {"axis": "Concentration", "value": round(concentration_pct, 1), "detail": f"{concentration_pct:.0f}% in top position"}, {"axis": "Volatility", "value": _scale(avg_vol_pct, 60), "detail": f"{avg_vol_pct:.0f}% avg 20d vol" if avg_vol_pct is not None else "n/a"}, {"axis": "Correlation", "value": round(max(0.0, avg_corr) * 100, 1) if avg_corr is not None else None, "detail": f"{avg_corr:+.2f} avg correlation" if avg_corr is not None else "n/a"}, {"axis": "Exposure", "value": round(exposure_score, 1), "detail": f"{open_count} open position{'s' if open_count != 1 else ''}"}, {"axis": "Drawdown", "value": _scale(drawdown_pct, 20) if drawdown_pct is not None else None, "detail": f"{drawdown_pct:.1f}pt from peak" if drawdown_pct is not None else "n/a"}, ] return {"axes": axes, "open_count": open_count} def compute_portfolio_risk_radar() -> Dict[str, Any]: """5-axis risk radar for the simulated trade log (trade_entry_prices). Drawdown comes from services.var_service's periodic P&L snapshot history.""" from services.var_service import get_pnl_snapshots trades = get_open_simulation_trades() drawdown_pct = _compute_max_drawdown_pct(get_pnl_snapshots(200)) return _build_risk_radar_axes(trades, drawdown_pct) def compute_real_portfolio_risk_radar() -> Dict[str, Any]: """5-axis risk radar for the REAL portfolio (services.database `portfolio` table). Same axes/scaling as compute_portfolio_risk_radar(), but the Drawdown axis comes from the realized equity curve of closed positions (the real portfolio has no periodic unrealized-PnL snapshot yet), normalized to a % of currently invested capital. """ from services.database import get_positions trades = get_positions("open") closed = sorted(get_positions("closed"), key=lambda p: p.get("close_date") or "") cumulative = 0.0 curve: List[Dict[str, Any]] = [] for p in closed: if p.get("close_value") is not None: pnl = (p["close_value"] - p["capital_invested"] - (p.get("ib_fees_entry") or 0) - (p.get("ib_fees_exit") or 0)) cumulative += pnl curve.append({"cumulative": cumulative}) max_dd_eur = _compute_max_drawdown_eur(curve) total_capital = sum(max(t.get("capital_invested") or 0, 0) for t in trades) drawdown_pct = round(max_dd_eur / total_capital * 100, 2) if max_dd_eur is not None and total_capital > 0 else None return _build_risk_radar_axes(trades, drawdown_pct) 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)