432 lines
18 KiB
Python
432 lines
18 KiB
Python
"""Simulation portfolio risk analysis — mirrors IBKR risk module for logged trades."""
|
|
from __future__ import annotations
|
|
from typing import Any, Dict, List, Optional
|
|
from services.database import get_conn
|
|
|
|
_BEARISH_KEYWORDS = {"bear", "put", "short", "sell", "vente", "baissier"}
|
|
|
|
# Ticker → asset_class mapping for common instruments (fallback when pattern not in DB)
|
|
_TICKER_AC: Dict[str, str] = {
|
|
# Energy
|
|
"CL=F": "energy", "BZ=F": "energy", "NG=F": "energy", "RB=F": "energy",
|
|
"HO=F": "energy", "USO": "energy", "XLE": "energy", "XOP": "energy", "OIL": "energy",
|
|
# Metals
|
|
"GC=F": "metals", "SI=F": "metals", "HG=F": "metals", "PL=F": "metals",
|
|
"PA=F": "metals", "GLD": "metals", "SLV": "metals", "GDX": "metals", "GDXJ": "metals",
|
|
# Agriculture
|
|
"ZW=F": "agriculture", "ZC=F": "agriculture", "ZS=F": "agriculture",
|
|
"CT=F": "agriculture", "KC=F": "agriculture", "SB=F": "agriculture", "CC=F": "agriculture",
|
|
"WEAT": "agriculture", "CORN": "agriculture", "SOYB": "agriculture",
|
|
# Equity indices
|
|
"^GSPC": "indices", "^DJI": "indices", "^NDX": "indices", "^RUT": "indices",
|
|
"^VIX": "indices", "^FTSE": "indices", "^GDAXI": "indices", "^FCHI": "indices",
|
|
"^N225": "indices", "^HSI": "indices", "^NSEI": "indices", "^BSESN": "indices",
|
|
"^STOXX50E": "indices", "^IBEX": "indices",
|
|
"SPY": "indices", "QQQ": "indices", "IWM": "indices", "DIA": "indices",
|
|
"VXX": "indices", "UVXY": "indices", "SVXY": "indices",
|
|
# Forex
|
|
"EURUSD=X": "forex", "GBPUSD=X": "forex", "USDJPY=X": "forex",
|
|
"AUDUSD=X": "forex", "USDCAD=X": "forex", "USDCHF=X": "forex",
|
|
"NZDUSD=X": "forex", "EURGBP=X": "forex", "EURJPY=X": "forex",
|
|
"GBPJPY=X": "forex", "USDCNH=X": "forex",
|
|
"FXE": "forex", "UUP": "forex", "FXB": "forex", "FXY": "forex",
|
|
# Rates
|
|
"ZB=F": "rates", "ZN=F": "rates", "ZF=F": "rates", "ZT=F": "rates",
|
|
"TLT": "rates", "IEF": "rates", "SHY": "rates", "HYG": "rates",
|
|
"LQD": "rates", "EMB": "rates",
|
|
}
|
|
|
|
|
|
def _infer_asset_class(ticker: str) -> str:
|
|
"""Derive asset_class from ticker when not stored in DB."""
|
|
t = (ticker or "").upper().strip()
|
|
if t in _TICKER_AC:
|
|
return _TICKER_AC[t]
|
|
if ":" in t: # NSE:RELIANCE, BSE:TCS, etc.
|
|
return "equities"
|
|
if t.endswith("=X") and len(t) >= 7: # forex pairs like EURUSD=X
|
|
return "forex"
|
|
if t.endswith("=F"): # generic futures
|
|
return "energy" # most unknown futures are commodities
|
|
if t.startswith("^"): # index
|
|
return "indices"
|
|
if t.isalpha() and len(t) <= 5: # short alpha = equity
|
|
return "equities"
|
|
return "unknown"
|
|
|
|
|
|
def _direction(strategy: str) -> str:
|
|
s = (strategy or "").lower()
|
|
return "bearish" if any(kw in s for kw in _BEARISH_KEYWORDS) else "bullish"
|
|
|
|
|
|
def get_open_simulation_trades() -> List[Dict[str, Any]]:
|
|
"""Open trades enriched with asset_class — via stored column, JOIN fallback, then ticker inference."""
|
|
conn = get_conn()
|
|
rows = conn.execute("""
|
|
SELECT tep.*,
|
|
COALESCE(tep.asset_class, cp.asset_class) AS asset_class
|
|
FROM trade_entry_prices tep
|
|
LEFT JOIN custom_patterns cp ON cp.id = tep.pattern_id
|
|
WHERE (tep.status IS NULL OR tep.status = 'open')
|
|
ORDER BY tep.entry_date DESC
|
|
""").fetchall()
|
|
conn.close()
|
|
result = []
|
|
for r in rows:
|
|
d = dict(r)
|
|
if not d.get("asset_class"):
|
|
d["asset_class"] = _infer_asset_class(d.get("underlying", ""))
|
|
result.append(d)
|
|
return result
|
|
|
|
|
|
def analyze_simulation_portfolio() -> Dict[str, Any]:
|
|
"""Full risk breakdown of the open simulation portfolio."""
|
|
trades = get_open_simulation_trades()
|
|
if not trades:
|
|
return {
|
|
"open_count": 0, "conflicts": [], "concentration": {},
|
|
"by_underlying": {}, "alerts": [], "direction_exposure": {},
|
|
}
|
|
|
|
total = len(trades)
|
|
|
|
# Group by asset_class
|
|
by_class: Dict[str, List] = {}
|
|
for t in trades:
|
|
ac = (t.get("asset_class") or "unknown").lower()
|
|
by_class.setdefault(ac, []).append(t)
|
|
|
|
concentration = {
|
|
ac: {
|
|
"count": len(items),
|
|
"pct": round(len(items) / total * 100, 1),
|
|
"tickers": sorted({t["underlying"] for t in items}),
|
|
"bullish": sum(1 for t in items if _direction(t.get("strategy", "")) == "bullish"),
|
|
"bearish": sum(1 for t in items if _direction(t.get("strategy", "")) == "bearish"),
|
|
}
|
|
for ac, items in by_class.items()
|
|
}
|
|
|
|
# Group by underlying
|
|
by_underlying: Dict[str, List] = {}
|
|
for t in trades:
|
|
u = (t.get("underlying") or "").upper()
|
|
if u:
|
|
by_underlying.setdefault(u, []).append(t)
|
|
|
|
# Detect directional conflicts
|
|
conflicts = []
|
|
for underlying, group in by_underlying.items():
|
|
dirs = [_direction(t.get("strategy", "")) for t in group]
|
|
if "bullish" in dirs and "bearish" in dirs:
|
|
conflicts.append({
|
|
"underlying": underlying,
|
|
"trades": [
|
|
{
|
|
"id": t["id"],
|
|
"strategy": t.get("strategy", ""),
|
|
"direction": _direction(t.get("strategy", "")),
|
|
"entry_date": t.get("entry_date", ""),
|
|
"pattern_name": t.get("pattern_name", ""),
|
|
"score_at_entry": t.get("score_at_entry", 0),
|
|
"asset_class": (t.get("asset_class") or "").lower(),
|
|
}
|
|
for t in group
|
|
],
|
|
})
|
|
|
|
# Direction exposure summary per class
|
|
direction_exposure = {
|
|
ac: {
|
|
"bullish": data["bullish"],
|
|
"bearish": data["bearish"],
|
|
"net": data["bullish"] - data["bearish"],
|
|
"bias": "bullish" if data["bullish"] > data["bearish"]
|
|
else "bearish" if data["bearish"] > data["bullish"]
|
|
else "neutral",
|
|
}
|
|
for ac, data in concentration.items()
|
|
}
|
|
|
|
# Build alerts — sorted by severity
|
|
alerts: List[Dict] = []
|
|
for c in conflicts:
|
|
alerts.append({
|
|
"type": "conflict", "level": "danger",
|
|
"underlying": c["underlying"],
|
|
"message": f"Positions contradictoires sur {c['underlying']} — {len(c['trades'])} trades opposés",
|
|
})
|
|
for ac, data in concentration.items():
|
|
if data["pct"] >= 35:
|
|
alerts.append({
|
|
"type": "concentration", "level": "warning",
|
|
"asset_class": ac, "pct": data["pct"],
|
|
"message": f"{ac} = {data['pct']}% du portefeuille simulé ({data['count']} trades)",
|
|
})
|
|
for underlying, group in by_underlying.items():
|
|
if len(group) >= 3:
|
|
alerts.append({
|
|
"type": "overweight", "level": "warning",
|
|
"underlying": underlying, "count": len(group),
|
|
"message": f"{len(group)} trades sur {underlying} — sur-exposition",
|
|
})
|
|
alerts.sort(key=lambda a: {"danger": 0, "warning": 1}.get(a["level"], 2))
|
|
|
|
return {
|
|
"open_count": total,
|
|
"conflicts": conflicts,
|
|
"concentration": concentration,
|
|
"by_underlying": {u: len(g) for u, g in by_underlying.items()},
|
|
"direction_exposure": direction_exposure,
|
|
"alerts": alerts,
|
|
}
|
|
|
|
|
|
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_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 (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
|
|
|
|
open_count = len(trades)
|
|
if not trades:
|
|
return {"axes": [], "open_count": 0}
|
|
|
|
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] = {}
|
|
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)
|
|
|
|
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 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()
|
|
total = len(open_trades)
|
|
|
|
new_dir = _direction(strategy)
|
|
underlying_upper = underlying.upper()
|
|
warnings: List[Dict] = []
|
|
|
|
same_underlying = [t for t in open_trades if (t.get("underlying") or "").upper() == underlying_upper]
|
|
for t in same_underlying:
|
|
existing_dir = _direction(t.get("strategy", ""))
|
|
if existing_dir != new_dir:
|
|
warnings.append({
|
|
"type": "conflict", "level": "danger",
|
|
"trade_id": t["id"], "existing_direction": existing_dir,
|
|
"message": f"Conflit : trade #{t['id']} est {existing_dir} sur {underlying} ({t.get('strategy', '')})",
|
|
})
|
|
else:
|
|
warnings.append({
|
|
"type": "accumulation", "level": "info",
|
|
"trade_id": t["id"],
|
|
"message": f"Accumulation {new_dir} sur {underlying} (trade #{t['id']} déjà en position)",
|
|
})
|
|
|
|
# Forecast concentration after adding this trade
|
|
total_after = total + 1
|
|
ac_key = (asset_class or "").lower()
|
|
current_ac = sum(
|
|
1 for t in open_trades
|
|
if (t.get("asset_class") or "").lower() == ac_key
|
|
)
|
|
future_pct = round((current_ac + 1) / total_after * 100, 1) if total_after > 0 else 100
|
|
if future_pct >= 35 and ac_key:
|
|
warnings.append({
|
|
"type": "concentration", "level": "warning",
|
|
"asset_class": ac_key, "future_pct": future_pct,
|
|
"message": f"Ce trade porterait {ac_key} à {future_pct}% du portefeuille simulé",
|
|
})
|
|
|
|
return {
|
|
"warnings": warnings,
|
|
"ok": not any(w["level"] == "danger" for w in warnings),
|
|
"underlying": underlying,
|
|
"strategy": strategy,
|
|
"direction": new_dir,
|
|
"total_open_after": total_after,
|
|
}
|
|
|
|
|
|
def build_monitor_context(risk: Dict[str, Any]) -> str:
|
|
"""Build a compact text summary of portfolio risk for GPT-4o."""
|
|
lines = [
|
|
f"PORTEFEUILLE SIMULÉ — {risk['open_count']} positions ouvertes",
|
|
"",
|
|
"RÉPARTITION PAR CLASSE D'ACTIF:",
|
|
]
|
|
for ac, data in sorted(risk["concentration"].items(), key=lambda x: -x[1]["pct"]):
|
|
exp = risk["direction_exposure"].get(ac, {})
|
|
lines.append(
|
|
f" {ac}: {data['pct']}% ({data['count']} trades) — "
|
|
f"bull={exp.get('bullish', 0)} bear={exp.get('bearish', 0)} — "
|
|
f"tickers: {', '.join(data['tickers'][:5])}"
|
|
)
|
|
|
|
if risk["conflicts"]:
|
|
lines += ["", "CONFLITS DIRECTIONNELS (DANGER):"]
|
|
for c in risk["conflicts"]:
|
|
trades_desc = "; ".join(
|
|
f"#{t['id']} {t['direction']} ({t['strategy']}) depuis {t['entry_date']}"
|
|
for t in c["trades"]
|
|
)
|
|
lines.append(f" {c['underlying']}: {trades_desc}")
|
|
|
|
if risk["alerts"]:
|
|
lines += ["", "ALERTES:"]
|
|
for a in risk["alerts"]:
|
|
lines.append(f" [{a['level'].upper()}] {a['message']}")
|
|
|
|
return "\n".join(lines)
|