""" Portfolio scenario-exposure — answers "if macro scenario X happens, how does my ACTUAL book of open positions react, and how many of my positions are really the same bet wearing different tickers?" Distinct from two pre-existing, coarser tools: - services.data_fetcher.score_macro_scenarios(): the GLOBAL 8-scenario macro regime, qualitative asset-class bias only (bullish/bearish/neutral), used for the top-level regime badge — not calibrated to numeric spot shocks and has no gold-bearish or forex-directional case, so it can't tell two option positions apart. - 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. This module instead reprices each position's REAL legs (same Saxo-first pricing as services.portfolio_pricing, used by mark-to-market and the payoff chart) under a small set of named spot/vol shocks, so positions on different tickers that both profit from the same shock get flagged as the SAME risk bet — e.g. a short S&P call spread, a long gold put spread and a short crude call spread can all really be "one Risk-Off bet, three times." """ from typing import Any, Dict, List, Optional, Tuple SCENARIOS: List[Dict[str, str]] = [ {"key": "risk_off", "label": "Risk-Off / Ralentissement"}, {"key": "risk_on", "label": "Reprise économique / Risk-On"}, {"key": "inflation_persistante", "label": "Inflation persistante"}, {"key": "dollar_fort", "label": "Dollar fort"}, {"key": "commodities_baisse", "label": "Baisse des matières premières"}, ] # asset_class -> scenario_key -> (spot_shock_pct, vol_shock_abs added to the leg's resolved sigma) _DIMENSION_SHOCKS: Dict[str, Dict[str, Tuple[float, float]]] = { "indices": { "risk_off": (-0.08, 0.06), "risk_on": (0.07, -0.03), "inflation_persistante": (-0.05, 0.04), "dollar_fort": (-0.02, 0.01), "commodities_baisse": (0.01, -0.01), }, "equities": { "risk_off": (-0.08, 0.06), "risk_on": (0.07, -0.03), "inflation_persistante": (-0.05, 0.04), "dollar_fort": (-0.02, 0.01), "commodities_baisse": (0.01, -0.01), }, "energy": { "risk_off": (-0.10, 0.08), "risk_on": (0.08, -0.04), "inflation_persistante": (0.10, 0.05), "dollar_fort": (-0.05, 0.02), "commodities_baisse": (-0.12, 0.03), }, "metals": { "risk_off": (0.05, 0.03), "risk_on": (-0.04, -0.02), "inflation_persistante": (0.08, 0.04), "dollar_fort": (-0.06, 0.02), "commodities_baisse": (-0.08, 0.02), }, "agriculture": { "risk_off": (-0.03, 0.03), "risk_on": (0.03, -0.02), "inflation_persistante": (0.09, 0.04), "dollar_fort": (-0.04, 0.01), "commodities_baisse": (-0.10, 0.03), }, "forex": { # Expressed as "USD strength" moves — sign is flipped per-pair by _fx_dollar_sign() # depending on whether USD is the base or quote currency. "risk_off": (0.03, 0.03), "risk_on": (-0.03, -0.02), "inflation_persistante": (-0.02, 0.03), "dollar_fort": (0.05, 0.01), "commodities_baisse": (0.01, -0.01), }, "rates": { "risk_off": (0.04, 0.02), "risk_on": (-0.03, -0.01), "inflation_persistante": (-0.06, 0.03), "dollar_fort": (0.01, 0.01), "commodities_baisse": (0.01, -0.01), }, } def _fx_dollar_sign(ticker: str) -> int: """+1 if USD is the base currency (pair rises when USD strengthens, e.g. USDJPY), -1 if USD is the quote currency (pair falls when USD strengthens, e.g. EURUSD), 0 for a non-USD cross where the "dollar strength" dimension doesn't clearly apply.""" t = (ticker or "").upper().replace("=X", "").replace("/", "") if t.startswith("USD"): return 1 if t.endswith("USD"): return -1 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.""" from datetime import date, datetime from services.portfolio_pricing import resolve_saxo_chain, price_leg from services.options_pricer import black_scholes 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: exp = datetime.strptime(expiry_date[:10], "%Y-%m-%d").date() days_remaining = max(0, (exp - date.today()).days) except ValueError: days_remaining = 0 else: 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 if chain is None: q = get_quote(underlying) 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 S_shocked = S * (1 + spot_shock_pct) T_remaining = days_remaining / 365 pnl = -pos.get("ib_fees_entry", 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 priced_now = price_leg(K, opt_type, days_remaining, r, chain, surface, expiry_date, fallback_spot, fallback_sigma) entry_premium = leg.get("premium_paid") if entry_premium is None: entry_premium = priced_now["price"] sigma = max(0.01, priced_now["sigma"] + vol_shock_abs) if T_remaining > 0: shocked_price = black_scholes(S_shocked, K, T_remaining, r, sigma, opt_type)["price"] else: shocked_price = max(0.0, S_shocked - K) if opt_type == "call" else max(0.0, K - S_shocked) pnl += sign * qty * 100 * (shocked_price - entry_premium) return pnl def compute_scenario_exposure() -> Dict[str, Any]: """Reprices every open position under each named scenario, then aggregates two views: - `scenarios`: per-scenario portfolio-wide estimated P&L (the "sensitivity matrix"). - `concentration`: for each position, the scenario that would benefit it MOST, then the capital-weighted % of the portfolio sharing that same dominant scenario (the "X% of your book is really one bet" bars) — the whole point being to surface when several differently-named positions are actually the same directional wager. """ from services.database import get_positions positions = get_positions("open") if not positions: return {"positions": 0, "total_capital": 0, "scenarios": [], "concentration": [], "dominant_scenario": None, "unpriced": [], "warning": None} priced: List[Dict[str, Any]] = [] unpriced: List[Dict[str, Any]] = [] for pos in positions: ac = (pos.get("asset_class") or "indices").lower() dims = _DIMENSION_SHOCKS.get(ac, _DIMENSION_SHOCKS["indices"]) fx_sign = _fx_dollar_sign(pos["underlying"]) if ac == "forex" else 1 scenario_pnl: Dict[str, Optional[float]] = {} for scen in SCENARIOS: key = scen["key"] spot_shock, vol_shock = dims.get(key, (0.0, 0.0)) if ac == "forex": spot_shock = spot_shock * fx_sign scenario_pnl[key] = _reprice_position(pos, spot_shock, vol_shock) if all(v is None for v in scenario_pnl.values()): unpriced.append({"id": pos["id"], "title": pos.get("title", pos["underlying"])}) continue 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), "scenario_pnl": scenario_pnl, }) if not priced: return {"positions": len(positions), "total_capital": 0, "scenarios": [], "concentration": [], "dominant_scenario": None, "unpriced": unpriced, "warning": None} total_capital = sum(p["capital_invested"] for p in priced) or 1.0 scenario_results = [] for scen in SCENARIOS: key = scen["key"] total_pnl = sum(p["scenario_pnl"].get(key) or 0 for p in priced) scenario_results.append({ "key": key, "label": scen["label"], "portfolio_pnl": round(total_pnl, 2), "portfolio_pnl_pct": round(total_pnl / total_capital * 100, 2), "positions": [ { "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), } 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? weight_by_scenario: Dict[str, float] = {s["key"]: 0.0 for s in SCENARIOS} 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"] concentration = [ {"key": key, "label": next(s["label"] for s in SCENARIOS if s["key"] == key), "pct_of_portfolio": round(w / total_capital * 100, 1)} for key, w in weight_by_scenario.items() if w > 0 ] concentration.sort(key=lambda c: -c["pct_of_portfolio"]) dominant_scenario = concentration[0] if concentration else None warning = None if dominant_scenario and dominant_scenario["pct_of_portfolio"] >= 60 and len(priced) >= 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é." ) return { "positions": len(priced), "total_capital": round(total_capital, 2), "scenarios": scenario_results, "concentration": concentration, "dominant_scenario": dominant_scenario, "unpriced": unpriced, "warning": warning, }