feat: cockpit

backend/routers/risk.py

@@ -1,3 +1,4 @@
+import math
 from fastapi import APIRouter, Query
 from services.database import (
     get_portfolio_exposure,
@@ -11,6 +12,17 @@ from services.database import (
 router = APIRouter(prefix="/api/risk", tags=["risk"])
 
 
+def _sanitize(obj):
+    """Replace NaN/Inf with None recursively for JSON compliance."""
+    if isinstance(obj, dict):
+        return {k: _sanitize(v) for k, v in obj.items()}
+    if isinstance(obj, list):
+        return [_sanitize(v) for v in obj]
+    if isinstance(obj, float) and (math.isnan(obj) or math.isinf(obj)):
+        return None
+    return obj
+
+
 @router.get("/exposure")
 def portfolio_exposure():
     """Exposure by asset class + risk factor for open positions."""
@@ -49,3 +61,10 @@ def kelly_sizing(
 def risk_dashboard():
     """Full portfolio risk snapshot: concentration, diversification, expected drawdown, recommendation."""
     return get_risk_dashboard()
+
+
+@router.get("/radar")
+def risk_radar():
+    """5-axis risk radar (Concentration/Volatility/Correlation/Exposure/Drawdown) for the real portfolio."""
+    from services.portfolio_risk import compute_real_portfolio_risk_radar
+    return _sanitize(compute_real_portfolio_risk_radar())

backend/services/portfolio_risk.py

@@ -232,29 +232,43 @@ def _compute_max_drawdown_pct(snapshots: List[Dict[str, Any]]) -> Optional[float
     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:
+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: 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.
+        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
-    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]
+    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] = {}
@@ -286,8 +300,6 @@ def compute_portfolio_risk_radar() -> Dict[str, Any]:
 
     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
 
@@ -301,6 +313,42 @@ def compute_portfolio_risk_radar() -> Dict[str, Any]:
     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()