250 lines
12 KiB
Python
250 lines
12 KiB
Python
"""
|
|
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
|
|
(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
|
|
show a scenario name or direction the rest of the app doesn't already agree with.
|
|
|
|
Distinct from two other pre-existing, coarser tools:
|
|
- services.data_fetcher.score_macro_scenarios(): scores which of the 8 scenarios the
|
|
CURRENT macro gauges look like (top-down, market-wide) — this module instead scores
|
|
which of the 8 scenarios OUR OWN POSITIONS are betting on (bottom-up, portfolio-wide).
|
|
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).
|
|
|
|
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.
|
|
"""
|
|
from typing import Any, Dict, List, Optional, Tuple
|
|
|
|
from services.data_fetcher import SCENARIO_META, SCENARIO_ASSET_BIAS
|
|
|
|
SCENARIOS: List[Dict[str, str]] = [
|
|
{"key": key, "label": meta["label"], "color": meta["color"], "emoji": meta["emoji"]}
|
|
for key, meta in SCENARIO_META.items()
|
|
]
|
|
|
|
# Qualitative bias label (as used in SCENARIO_ASSET_BIAS) -> (spot_shock_pct, vol_shock_abs
|
|
# added to the leg's resolved sigma). The only numeric calibration this module adds on top
|
|
# of the existing qualitative macro-regime table.
|
|
_BIAS_TO_SHOCK: Dict[str, Tuple[float, float]] = {
|
|
"bullish+": (0.08, -0.03),
|
|
"bullish": (0.04, -0.015),
|
|
"neutral": (0.0, 0.0),
|
|
"bearish": (-0.04, 0.015),
|
|
"bearish+": (-0.08, 0.03),
|
|
}
|
|
|
|
# SCENARIO_ASSET_BIAS has no "rates" axis (bond ETFs like IEF/TLT) — extended here with the
|
|
# same 8 scenario keys, standard macro logic: bonds rally when yields fall (recession,
|
|
# desinflation, flight-to-quality in a liquidity crisis), sell off when inflation runs hot
|
|
# (reflation, stagflation, inflation shock).
|
|
_RATES_BIAS: Dict[str, str] = {
|
|
"goldilocks": "neutral", "desinflation": "bullish+", "soft_landing": "bullish",
|
|
"reflation": "bearish", "stagflation": "bearish+", "inflation_shock": "bearish+",
|
|
"recession": "bullish+", "crise_liquidite": "bullish",
|
|
}
|
|
|
|
# SCENARIO_ASSET_BIAS's forex axis is only ever "neutral" or "defensive" (a flight-to-USD
|
|
# flag, not a direction) — translated here into a dollar-strength shock. "defensive"
|
|
# scenarios push capital into USD (dollar up); "neutral" scenarios leave it flat. The actual
|
|
# sign applied to a given pair (EURUSD falls / USDJPY rises on dollar strength) is resolved
|
|
# per-position by _fx_dollar_sign().
|
|
_FOREX_DEFENSIVE_SHOCK: Tuple[float, float] = (0.04, 0.02)
|
|
|
|
_DIRECT_ASSET_CLASSES = {"energy", "metals", "indices", "equities", "agriculture"}
|
|
|
|
|
|
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
|
|
from SCENARIO_ASSET_BIAS wherever that axis exists, with a documented extension for
|
|
rates (not covered by the shared table) and forex (covered only as a non-directional
|
|
flight-to-USD flag there)."""
|
|
bias_row = SCENARIO_ASSET_BIAS[scenario_key]
|
|
if asset_class in _DIRECT_ASSET_CLASSES:
|
|
return _BIAS_TO_SHOCK.get(bias_row.get(asset_class, "neutral"), (0.0, 0.0))
|
|
if asset_class == "rates":
|
|
return _BIAS_TO_SHOCK.get(_RATES_BIAS.get(scenario_key, "neutral"), (0.0, 0.0))
|
|
if asset_class == "forex":
|
|
return _FOREX_DEFENSIVE_SHOCK if bias_row.get("forex") == "defensive" else (0.0, 0.0)
|
|
# Unknown/uncategorized asset_class — fall back to the indices axis (equity-like default).
|
|
return _BIAS_TO_SHOCK.get(bias_row.get("indices", "neutral"), (0.0, 0.0))
|
|
|
|
|
|
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 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.
|
|
"""
|
|
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()
|
|
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 = _dimension_shock(ac, key)
|
|
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"], "color": scen["color"], "emoji": scen["emoji"],
|
|
"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, **{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
|
|
]
|
|
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,
|
|
}
|