feat: cockpit

backend/services/portfolio_risk.py

@@ -184,6 +184,123 @@ def analyze_simulation_portfolio() -> Dict[str, Any]:
     }
 
 
+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_portfolio_risk_radar() -> Dict[str, Any]:
+    """5-axis risk radar for the Cockpit's Risk card (replaces the old asset-class donut,
+    which now lives separately as the allocation breakdown). Axes, each scaled 0-100:
+      - 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: trade_entry_prices has no notional/contract-size
+        column to compute real leverage from, so this measures "how spread thin" instead.
+      - Drawdown: max peak-to-trough drop in the simulated portfolio's total P&L %,
+        from services.var_service's snapshot history.
+    """
+    from services.data_fetcher import get_quote_with_volatility
+    from services.var_service import get_pnl_snapshots
+
+    trades = get_open_simulation_trades()
+    open_count = len(trades)
+    if not trades:
+        return {"axes": [], "open_count": 0}
+
+    weights = [max(t.get("capital_invested") or t.get("entry_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)
+
+    drawdown_pct = _compute_max_drawdown_pct(get_pnl_snapshots(200))
+
+    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 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()