From 09a0203494f34cf92d9e184a0b50ee450aa1f808 Mon Sep 17 00:00:00 2001 From: OpenSquared Date: Sun, 26 Jul 2026 15:46:07 +0200 Subject: [PATCH] feat: risk --- backend/services/portfolio_scenarios.py | 253 +++++++++++++++++++----- frontend/src/pages/RiskDashboard.tsx | 99 ++++++++-- 2 files changed, 286 insertions(+), 66 deletions(-) diff --git a/backend/services/portfolio_scenarios.py b/backend/services/portfolio_scenarios.py index 6c5df4a..95dda35 100644 --- a/backend/services/portfolio_scenarios.py +++ b/backend/services/portfolio_scenarios.py @@ -1,7 +1,7 @@ """ Portfolio scenario-exposure — answers "which of our 8 macro scenarios is my ACTUAL book of open positions really a bet on, and how many differently-named positions are secretly the -same bet?" Deliberately reuses the SAME 8 scenarios as the global macro regime +same bet?" Reuses the SAME 8 scenarios as the global macro regime (services.data_fetcher.SCENARIO_META / SCENARIO_ASSET_BIAS — goldilocks, desinflation, soft_landing, reflation, stagflation, inflation_shock, recession, crise_liquidite) as its single source of truth for labels/colors/emoji and directional bias, so this tool can never @@ -14,15 +14,37 @@ Distinct from two other pre-existing, coarser tools: Same 8 buckets, opposite direction of inference. - services.database.get_risk_dashboard()/get_risk_clusters(): buckets capital by asset_class and by a geopolitical-trigger keyword match — blind to whether a position - is long or short its underlying, so a bullish and a bearish position on the same - ticker land in the same bucket. Kept as a separate, complementary lens (thematic - capital exposure) rather than merged with this one (directional scenario alignment). + is long or short its underlying. Kept as a separate, complementary lens. -The evaluation itself is deliberately NOT an LLM call: it reprices each position's REAL legs -(same Saxo-first pricing as services.portfolio_pricing, used by mark-to-market and the -payoff chart) under each scenario's spot/vol shock via Black-Scholes. Same positions in, -same percentages out, every time — auditable and tied to real strikes/greeks, matching how -Curve Regime and the payoff diagram already work elsewhere in this app. +v2 methodology (v1 assigned each position to a single "best" scenario via argmax, which +collapsed multi-dimensional option payoffs into one arbitrary bucket and let ties resolve +by scenario list order rather than economics): + - A position "aligns" with a scenario only if repricing it under that scenario's shock + actually produces a POSITIVE P&L — not merely "the least bad of 8", so a structurally + one-sided axis (e.g. metals is bullish/neutral in all 8 scenarios, never bearish) no + longer forces a bearish gold position into a fake "best" scenario; it shows up as + genuinely unaligned instead. + - A position's weight is split PROPORTIONALLY across every scenario it aligns with + (weighted by how much it gains in each), instead of winner-take-all — so a position + that profits comparably in two scenarios (e.g. Recession and Crise de liquidité often + carry the same "bearish+" bias for indices) contributes to both instead of an arbitrary + single pick. + - Weight itself is the position's actual capital-at-risk (worst point on its real + at-expiry payoff curve, from services.portfolio_pricing.compute_payoff, already used by + the payoff-diagram feature) rather than the user-entered capital_invested field, which + isn't guaranteed to equal true max loss for spreads. + - Redundant correlated bets (e.g. Crude & Brent short call spreads) are surfaced via an + "effective N" independent-bets count, reusing services.portfolio_risk's existing + pairwise-correlation helper — the same diversification math already used by the risk + radar's Correlation axis. + - Each position is annotated with its real dollar delta/vega (from + services.options_pricer.black_scholes, which already computes them on every call — just + not extracted before now) and its current services.curve_regime classification, so the + UI can show the position's actual nature instead of guessing from its strategy name. + +The evaluation itself is still NOT an LLM call: same positions in, same numbers out, every +time — auditable and tied to real strikes/greeks, matching how Curve Regime and the payoff +diagram already work elsewhere in this app. """ from typing import Any, Dict, List, Optional, Tuple @@ -63,6 +85,10 @@ _FOREX_DEFENSIVE_SHOCK: Tuple[float, float] = (0.04, 0.02) _DIRECT_ASSET_CLASSES = {"energy", "metals", "indices", "equities", "agriculture"} +# Same futures/ETF proxy mapping as frontend/src/pages/Dashboard.tsx's PROXY_TICKER_ALIASES +# — BNO (Brent ETF, used as the options proxy) isn't itself a Watchlist ticker, BRENT is. +_CURVE_REGIME_TICKER_ALIASES = {"BNO": "BRENT"} + def _dimension_shock(asset_class: str, scenario_key: str) -> Tuple[float, float]: """Spot/vol shock for one asset_class under one of the 8 canonical scenarios, derived @@ -92,21 +118,15 @@ def _fx_dollar_sign(ticker: str) -> int: return 0 -def _reprice_position(pos: Dict[str, Any], spot_shock_pct: float, vol_shock_abs: float) -> Optional[float]: - """Real Black-Scholes reprice of this position's legs at a shocked spot/vol, mirroring - services.portfolio_pricing.compute_payoff's methodology but at ONE target spot instead - of a curve (no time decay applied — a "if this happened right now" snapshot). Returns - estimated P&L in currency units, or None if the position has no legs to price.""" +def _resolve_position_market(pos: Dict[str, Any]): + """Shared market-data resolution (Saxo chain if linked, else yfinance fallback) used by + both the scenario reprice and the greeks snapshot, so both read the exact same spot/vol + the payoff chart and mark-to-market already use.""" from datetime import date, datetime - from services.portfolio_pricing import resolve_saxo_chain, price_leg - from services.options_pricer import black_scholes + from services.portfolio_pricing import resolve_saxo_chain from services.data_fetcher import get_quote, compute_historical_iv underlying = pos["underlying"] - legs = pos.get("legs", []) - if not legs: - return None - expiry_date = pos.get("expiry_date") or "" if expiry_date: try: @@ -118,7 +138,6 @@ def _reprice_position(pos: Dict[str, Any], spot_shock_pct: float, vol_shock_abs: entry = datetime.strptime(pos["entry_date"][:10], "%Y-%m-%d").date() days_remaining = max(0, pos.get("expiry_days", 90) - (date.today() - entry).days) - r = 0.05 chain, surface = resolve_saxo_chain(underlying, target_days=max(days_remaining, 1)) fallback_spot = pos.get("entry_underlying_price") or 100.0 fallback_sigma = 0.20 @@ -127,6 +146,21 @@ def _reprice_position(pos: Dict[str, Any], spot_shock_pct: float, vol_shock_abs: fallback_spot = (q.get("price") if q else None) or fallback_spot fallback_sigma = compute_historical_iv(underlying) S = chain["spot"] if chain else fallback_spot + return chain, surface, S, days_remaining, expiry_date, fallback_spot, fallback_sigma + + +def _reprice_position(pos: Dict[str, Any], spot_shock_pct: float, vol_shock_abs: float) -> Optional[float]: + """Real Black-Scholes reprice of this position's legs at a shocked spot/vol (no time + decay applied — a "if this happened right now" snapshot). Returns estimated P&L in + currency units, or None if the position has no legs to price.""" + from services.portfolio_pricing import price_leg + from services.options_pricer import black_scholes + + legs = pos.get("legs", []) + if not legs: + return None + chain, surface, S, days_remaining, expiry_date, fallback_spot, fallback_sigma = _resolve_position_market(pos) + r = 0.05 S_shocked = S * (1 + spot_shock_pct) T_remaining = days_remaining / 365 @@ -149,23 +183,98 @@ def _reprice_position(pos: Dict[str, Any], spot_shock_pct: float, vol_shock_abs: return pnl +def _position_greeks(pos: Dict[str, Any]) -> Dict[str, Optional[float]]: + """Real position-level dollar delta/vega at the CURRENT (unshocked) market — from + services.options_pricer.black_scholes, which computes them on every pricing call + already. Used purely to annotate each position's actual nature (directional vs. + volatility play), not fed back into the scenario P&L math above.""" + from services.options_pricer import black_scholes + + legs = pos.get("legs", []) + if not legs: + return {"delta_dollars": None, "vega_dollars": None, "nature": None} + _, _, S, days_remaining, _, _, fallback_sigma = _resolve_position_market(pos) + r = 0.05 + T_remaining = max(days_remaining, 1) / 365 + delta_shares = 0.0 + vega_dollars = 0.0 + for leg in legs: + K = leg.get("strike") or S + opt_type = leg.get("option_type", "call") + qty = leg.get("quantity", 1) + sign = 1 if leg.get("position", "long") == "long" else -1 + g = black_scholes(S, K, T_remaining, r, fallback_sigma, opt_type) + mult = sign * qty * 100 + delta_shares += mult * g["delta"] + vega_dollars += mult * g["vega"] + delta_dollars = delta_shares * S / 100 # P&L per 1% move in the underlying + capital = max(pos.get("capital_invested") or 1000.0, 1.0) + d_impact = abs(delta_dollars) # P&L for a 1% move + v_impact = abs(vega_dollars) * 3 # P&L for a typical 3-vol-point move + if v_impact > d_impact * 1.5: + nature = "Volatilité longue" if vega_dollars > 0 else "Volatilité courte" + elif d_impact < 0.01 * capital: + nature = "Neutre / range" + else: + nature = "Directionnelle haussière" if delta_dollars > 0 else "Directionnelle baissière" + return {"delta_dollars": round(delta_dollars, 2), "vega_dollars": round(vega_dollars, 2), "nature": nature} + + +def _capital_at_risk(pos: Dict[str, Any]) -> float: + """Worst point on the position's real at-expiry payoff curve (±25% spot range, + services.portfolio_pricing.compute_payoff — same curve the payoff-diagram chart shows) + as a proxy for capital genuinely at risk, since the user-entered capital_invested field + isn't guaranteed to equal true max loss for a spread. Falls back to capital_invested if + the curve never dips negative within that range (e.g. a well-covered structure) or + can't be computed.""" + try: + from services.portfolio_pricing import compute_payoff + curve = compute_payoff(pos).get("at_expiry") or [] + worst = min(curve) if curve else None + if worst is not None and worst < 0: + return abs(worst) + except Exception: + pass + return max(pos.get("capital_invested") or 1000.0, 1.0) + + +def _curve_regime_lookup() -> Dict[str, str]: + from services.database import get_all_curve_regime_cache + out: Dict[str, str] = {} + for row in get_all_curve_regime_cache(): + if row.get("regime_label"): + out[row["ticker"].upper()] = row["regime_label"] + return out + + def compute_scenario_exposure() -> Dict[str, Any]: """Reprices every open position under each of the 8 canonical macro scenarios, then - aggregates two views: - - `scenarios`: per-scenario portfolio-wide estimated P&L (the "sensitivity matrix"), - sorted by |impact| so the scenarios that matter most float to the top. - - `concentration`: for each position, the scenario that would benefit it MOST, then - the capital-weighted % of the portfolio sharing that same dominant scenario — the - "our book is really one bet, repeated" ranking, using the exact same scenario - names/colors as the global macro regime badge. + builds three complementary views: + - `scenarios`: per-scenario portfolio-wide estimated P&L (the sensitivity matrix), + sorted by |impact|, covering every priced position regardless of alignment. + - `concentration`: capital-at-risk-weighted % of the book aligned with each scenario + — a position only counts toward a scenario if it genuinely profits there, and its + weight is split across every scenario it profits in (proportional to the gain), not + assigned to a single "winner". Percentages need not sum to 100 — the gap is + `pct_unaligned`, positions that don't profit in ANY of the 8 scenarios. + - `position_details`: per-position nature (real dollar delta/vega), current Curve + Regime, and which scenarios it aligns with — the transparency layer behind the + ranking, so the numbers are inspectable rather than a black box. + - `effective_n_positions`: correlation-adjusted count of genuinely independent bets + (services.portfolio_risk's pairwise-correlation helper), since e.g. two positions on + Crude and Brent are not two independent risks. """ from services.database import get_positions + from services.portfolio_risk import _compute_avg_pairwise_correlation positions = get_positions("open") if not positions: - return {"positions": 0, "total_capital": 0, "scenarios": [], "concentration": [], - "dominant_scenario": None, "unpriced": [], "warning": None} + return {"positions": 0, "total_capital": 0, "effective_n_positions": 0, + "avg_pairwise_correlation": None, "scenarios": [], "concentration": [], + "position_details": [], "dominant_scenario": None, "pct_unaligned": 0, + "unpriced": [], "warning": None} + curve_regimes = _curve_regime_lookup() priced: List[Dict[str, Any]] = [] unpriced: List[Dict[str, Any]] = [] for pos in positions: @@ -184,19 +293,26 @@ def compute_scenario_exposure() -> Dict[str, Any]: unpriced.append({"id": pos["id"], "title": pos.get("title", pos["underlying"])}) continue + ticker_key = pos["underlying"].upper() + ticker_key = _CURVE_REGIME_TICKER_ALIASES.get(ticker_key, ticker_key) priced.append({ "id": pos["id"], "title": pos.get("title", pos["underlying"]), "underlying": pos["underlying"], "asset_class": ac, - "capital_invested": max(pos.get("capital_invested") or 0, 0), + "risk_weight": _capital_at_risk(pos), "scenario_pnl": scenario_pnl, + "greeks": _position_greeks(pos), + "curve_regime": curve_regimes.get(ticker_key), }) if not priced: - return {"positions": len(positions), "total_capital": 0, "scenarios": [], "concentration": [], - "dominant_scenario": None, "unpriced": unpriced, "warning": None} + return {"positions": len(positions), "total_capital": 0, "effective_n_positions": 0, + "avg_pairwise_correlation": None, "scenarios": [], "concentration": [], + "position_details": [], "dominant_scenario": None, "pct_unaligned": 0, + "unpriced": unpriced, "warning": None} - total_capital = sum(p["capital_invested"] for p in priced) or 1.0 + total_capital = sum(p["risk_weight"] for p in priced) or 1.0 + # ── Sensitivity matrix — every priced position, every scenario, no alignment filter ── scenario_results = [] for scen in SCENARIOS: key = scen["key"] @@ -209,41 +325,86 @@ def compute_scenario_exposure() -> Dict[str, Any]: { "id": p["id"], "title": p["title"], "pnl": round(p["scenario_pnl"].get(key) or 0, 2), - "pnl_pct": round((p["scenario_pnl"].get(key) or 0) / max(p["capital_invested"], 1) * 100, 1), + "pnl_pct": round((p["scenario_pnl"].get(key) or 0) / max(p["risk_weight"], 1) * 100, 1), } for p in priced ], }) scenario_results.sort(key=lambda s: -abs(s["portfolio_pnl_pct"])) - # Concentration: which scenario is each position's single most favorable outcome? + # ── Alignment — proportional split across every genuinely profitable scenario ── weight_by_scenario: Dict[str, float] = {s["key"]: 0.0 for s in SCENARIOS} + position_details = [] + unaligned_weight = 0.0 for p in priced: - best_key = max(p["scenario_pnl"], key=lambda k: (p["scenario_pnl"].get(k) if p["scenario_pnl"].get(k) is not None else float("-inf"))) - weight_by_scenario[best_key] = weight_by_scenario.get(best_key, 0.0) + p["capital_invested"] + positive = {k: v for k, v in p["scenario_pnl"].items() if v is not None and v > 0} + aligned_scenarios = [] + if positive: + total_positive = sum(positive.values()) + for k, v in sorted(positive.items(), key=lambda kv: -kv[1]): + share = v / total_positive + weight_by_scenario[k] += p["risk_weight"] * share + meta = next(s for s in SCENARIOS if s["key"] == k) + aligned_scenarios.append({ + "key": k, "label": meta["label"], "color": meta["color"], "emoji": meta["emoji"], + "share_of_position": round(share * 100, 1), + "pnl_pct": round(v / max(p["risk_weight"], 1) * 100, 1), + }) + else: + unaligned_weight += p["risk_weight"] + + position_details.append({ + "id": p["id"], "title": p["title"], "underlying": p["underlying"], + "asset_class": p["asset_class"], "risk_weight": round(p["risk_weight"], 2), + "nature": p["greeks"]["nature"], "delta_dollars": p["greeks"]["delta_dollars"], + "vega_dollars": p["greeks"]["vega_dollars"], "curve_regime": p["curve_regime"], + "aligned_scenarios": aligned_scenarios[:2], + "is_unaligned": not positive, + }) concentration = [ - {"key": key, **{k: v for k, v in next(s for s in SCENARIOS if s["key"] == key).items() if k != "key"}, - "pct_of_portfolio": round(w / total_capital * 100, 1)} - for key, w in weight_by_scenario.items() if w > 0 + {"key": s["key"], "label": s["label"], "color": s["color"], "emoji": s["emoji"], + "pct_of_portfolio": round(weight_by_scenario[s["key"]] / total_capital * 100, 1)} + for s in SCENARIOS if weight_by_scenario[s["key"]] > 0 ] concentration.sort(key=lambda c: -c["pct_of_portfolio"]) dominant_scenario = concentration[0] if concentration else None + pct_unaligned = round(unaligned_weight / total_capital * 100, 1) + + underlyings = sorted({p["underlying"] for p in priced}) + avg_corr = _compute_avg_pairwise_correlation(underlyings) if len(underlyings) >= 2 else None + n = len(priced) + if avg_corr is not None: + effective_n = round(n / (1 + (n - 1) * max(avg_corr, 0.0)), 2) + else: + effective_n = n warning = None - if dominant_scenario and dominant_scenario["pct_of_portfolio"] >= 60 and len(priced) >= 3: + if dominant_scenario and dominant_scenario["pct_of_portfolio"] >= 40 and n >= 3: warning = ( - f"{dominant_scenario['pct_of_portfolio']:.0f}% du portefeuille gagne surtout dans le même " - f"scénario ({dominant_scenario['label']}) — vos {len(priced)} positions ne sont pas aussi " - f"diversifiées qu'il n'y paraît, c'est en grande partie un seul pari macro répété." + f"{dominant_scenario['pct_of_portfolio']:.0f}% du portefeuille (pondéré par capital à " + f"risque) gagne dans le même scénario ({dominant_scenario['label']})" + + (f", avec seulement {effective_n:.1f} paris réellement indépendants sur {n} positions " + f"(sous-jacents corrélés)" if effective_n < n * 0.7 else "") + + " — vos positions ne sont pas aussi diversifiées qu'il n'y paraît." + ) + elif pct_unaligned >= 25 and n >= 3: + warning = ( + f"{pct_unaligned:.0f}% du portefeuille ne profite d'aucun des 8 scénarios macro " + f"(pari structurel non couvert par cette grille de lecture, ex. positions courtes " + f"sur un actif refuge)." ) return { - "positions": len(priced), + "positions": n, "total_capital": round(total_capital, 2), + "effective_n_positions": effective_n, + "avg_pairwise_correlation": round(avg_corr, 3) if avg_corr is not None else None, "scenarios": scenario_results, "concentration": concentration, + "position_details": position_details, "dominant_scenario": dominant_scenario, + "pct_unaligned": pct_unaligned, "unpriced": unpriced, "warning": warning, } diff --git a/frontend/src/pages/RiskDashboard.tsx b/frontend/src/pages/RiskDashboard.tsx index 3c94736..e032c90 100644 --- a/frontend/src/pages/RiskDashboard.tsx +++ b/frontend/src/pages/RiskDashboard.tsx @@ -141,6 +141,8 @@ function ScenarioExposureCard() { const concentration: any[] = exp.concentration ?? [] const scenarios: any[] = exp.scenarios ?? [] + const positionDetails: any[] = exp.position_details ?? [] + const redundant = exp.effective_n_positions != null && exp.effective_n_positions < exp.positions * 0.75 return (
@@ -148,9 +150,29 @@ function ScenarioExposureCard() { Alignement du portefeuille sur le Régime Macro
- Repricing Black-Scholes réel (pricing Saxo-first) de chaque position sous les 8 scénarios du - régime macro — révèle quand plusieurs positions différentes sont en réalité le même pari répété. - Vue distincte des Risk Factors ci-dessous (thème/classe d'actif touché, pas le sens du pari). + Une position ne compte pour un scénario que si elle y gagne réellement (repricing Black-Scholes + réel, Saxo-first), répartie entre tous les scénarios où elle est gagnante — pas un vote unique + au scénario "le moins pire". Vue distincte des Risk Factors ci-dessous (thème touché, pas le sens du pari). +
+ + {/* Headline stats */} +
+
+
{exp.positions}
+
positions
+
+
+
+ {exp.effective_n_positions ?? '—'} +
+
paris indépendants (N eff.)
+
+
+
= 25 ? 'text-amber-400' : 'text-slate-200')}> + {exp.pct_unaligned}% +
+
non couvert par les 8 scénarios
+
{exp.warning && ( @@ -164,25 +186,29 @@ function ScenarioExposureCard() { {/* Concentration ranking */}
- % du book dont c'est le scénario le plus favorable + % du capital à risque aligné avec chaque scénario
-
- {concentration.map((c: any) => ( -
-
- - {c.emoji}{c.label} - - - {c.pct_of_portfolio}% - + {concentration.length === 0 ? ( +
Aucune position ne gagne dans un des 8 scénarios.
+ ) : ( +
+ {concentration.map((c: any) => ( +
+
+ + {c.emoji}{c.label} + + + {c.pct_of_portfolio}% + +
+
+
+
-
-
-
-
- ))} -
+ ))} +
+ )}
{/* Sensitivity matrix */} @@ -206,6 +232,39 @@ function ScenarioExposureCard() {
+ {/* Per-position transparency layer */} + {positionDetails.length > 0 && ( +
+
Détail par position
+
+ {positionDetails.map((p: any) => ( +
+ {p.title} + {p.nature} + + Δ {p.delta_dollars != null ? p.delta_dollars.toFixed(0) : '—'} · v {p.vega_dollars != null ? p.vega_dollars.toFixed(0) : '—'} + + {p.curve_regime && ( + {p.curve_regime} + )} + + {p.is_unaligned ? ( + non aligné (perd dans les 8 scénarios) + ) : ( + p.aligned_scenarios.map((a: any) => ( + + {a.emoji} {a.label} {a.share_of_position}% + + )) + )} + +
+ ))} +
+
+ )} + {exp.unpriced?.length > 0 && (
{exp.unpriced.length} position(s) non pricée(s) (pas de legs/données) : {exp.unpriced.map((u: any) => u.title).join(', ')}