576 lines
21 KiB
Python
576 lines
21 KiB
Python
"""VaR service — Black-Scholes delta approach with numpy/scipy (no numba dependency)."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import numpy as np
|
||
import pandas as pd
|
||
from scipy.stats import norm
|
||
from datetime import datetime, timedelta
|
||
from typing import List, Dict, Optional
|
||
|
||
from .database import get_conn
|
||
|
||
|
||
# ─── Option strategy → (type, directional multiplier) ───────────────────────
|
||
|
||
def _parse_strategy(strategy: str) -> tuple[str, float]:
|
||
"""Return (option_type, direction_sign) from strategy name."""
|
||
s = strategy.lower()
|
||
if "straddle" in s or "strangle" in s:
|
||
return "straddle", (1.0 if "long" in s else -1.0)
|
||
if "iron condor" in s or "butterfly" in s or "neutral" in s:
|
||
return "neutral", 0.0
|
||
if "bull" in s:
|
||
return "call", 0.5
|
||
if "bear" in s:
|
||
return "put", -0.5
|
||
if "call" in s:
|
||
return "call", (1.0 if "long" in s else -1.0)
|
||
if "put" in s:
|
||
return "put", (1.0 if "long" in s else -1.0)
|
||
return "call", 0.5 # default
|
||
|
||
|
||
def _bs_delta(S: float, K: float, T_days: float, sigma: float, opt_type: str, direction: float) -> float:
|
||
"""Black-Scholes delta, direction-adjusted."""
|
||
r = 0.05
|
||
T = max(T_days, 1) / 252.0
|
||
d1 = (np.log(S / K) + (r + 0.5 * sigma**2) * T) / (sigma * np.sqrt(T))
|
||
if opt_type == "call":
|
||
raw = float(norm.cdf(d1))
|
||
elif opt_type == "put":
|
||
raw = float(norm.cdf(d1) - 1.0)
|
||
elif opt_type == "straddle":
|
||
# Long straddle: net delta ≈ 0 ATM; represent as small residual
|
||
raw = float(norm.cdf(d1) + (norm.cdf(d1) - 1.0)) # ≈ 0
|
||
else:
|
||
raw = 0.0
|
||
return raw * direction
|
||
|
||
|
||
# ─── Market data ─────────────────────────────────────────────────────────────
|
||
|
||
def _fetch_returns(tickers: List[str], lookback: int) -> pd.DataFrame:
|
||
"""Download historical daily returns via yfinance. Returns {} on failure."""
|
||
from .database import _normalize_ticker
|
||
valid = [t for t in tickers if ":" not in t]
|
||
if not valid:
|
||
return pd.DataFrame()
|
||
# Map raw → yfinance ticker; keep reverse map for column rename
|
||
yf_map = {t: _normalize_ticker(t) for t in valid}
|
||
yf_tickers = list(yf_map.values())
|
||
reverse = {v: k for k, v in yf_map.items()}
|
||
try:
|
||
import yfinance as yf
|
||
end = datetime.now()
|
||
start = end - timedelta(days=lookback + 60)
|
||
raw = yf.download(yf_tickers, start=start, end=end, progress=False, auto_adjust=True)
|
||
if raw.empty:
|
||
return pd.DataFrame()
|
||
close = raw["Close"] if len(yf_tickers) > 1 else raw[["Close"]].rename(columns={"Close": yf_tickers[0]})
|
||
# Rename yfinance tickers back to original
|
||
close = close.rename(columns=reverse)
|
||
return close.pct_change().dropna().tail(lookback)
|
||
except Exception:
|
||
return pd.DataFrame()
|
||
|
||
|
||
def _synthetic_returns(tickers: List[str], lookback: int, seed: int = 42) -> pd.DataFrame:
|
||
"""Fallback: simulate realistic returns when market data unavailable."""
|
||
rng = np.random.default_rng(seed)
|
||
idx = pd.date_range(end=datetime.now(), periods=lookback, freq="B")
|
||
data = {t: rng.normal(0.0002, 0.018, lookback) for t in tickers}
|
||
return pd.DataFrame(data, index=idx)
|
||
|
||
|
||
# ─── Core VaR computation ────────────────────────────────────────────────────
|
||
|
||
def compute_var(
|
||
confidence: float = 0.95,
|
||
horizon_days: int = 1,
|
||
lookback_days: int = 252,
|
||
default_iv: float = 0.20,
|
||
) -> Dict:
|
||
conn = get_conn()
|
||
rows = conn.execute(
|
||
"SELECT underlying, strategy, entry_price, capital_invested, "
|
||
"strike_guidance, expiry_days_at_entry, pattern_name "
|
||
"FROM trade_entry_prices WHERE status = 'open'"
|
||
).fetchall()
|
||
|
||
if not rows:
|
||
return {"error": "Aucune position ouverte"}
|
||
|
||
positions = [dict(r) for r in rows]
|
||
|
||
# Filter positions usable for delta calc
|
||
valid = [
|
||
p for p in positions
|
||
if p.get("underlying") and ":" not in (p["underlying"] or "")
|
||
and p.get("entry_price") and p["entry_price"] > 0
|
||
]
|
||
|
||
if not valid:
|
||
return {"error": "Aucune position avec données de marché disponibles"}
|
||
|
||
tickers = list({p["underlying"] for p in valid})
|
||
|
||
# Fetch or synthesize returns
|
||
returns_df = _fetch_returns(tickers, lookback_days)
|
||
data_source = "live"
|
||
if returns_df.empty:
|
||
returns_df = _synthetic_returns(tickers, lookback_days)
|
||
data_source = "simulated"
|
||
|
||
# Align to available history
|
||
n = len(returns_df)
|
||
|
||
# Build delta-weighted portfolio PnL series
|
||
weighted_pnl = pd.Series(0.0, index=returns_df.index)
|
||
total_notional = 0.0
|
||
pos_details = []
|
||
|
||
for p in valid:
|
||
ticker = p["underlying"]
|
||
if ticker not in returns_df.columns:
|
||
continue
|
||
|
||
S = float(p["entry_price"])
|
||
T = float(p.get("expiry_days_at_entry") or 60)
|
||
capital = float(p.get("capital_invested") or S)
|
||
strategy = p.get("strategy") or "Long Call"
|
||
opt_type, direction = _parse_strategy(strategy)
|
||
|
||
# Strike: ATM unless guidance specifies otherwise
|
||
K = S
|
||
|
||
delta = _bs_delta(S, K, T, default_iv, opt_type, direction)
|
||
|
||
weighted_pnl += returns_df[ticker] * delta * capital
|
||
total_notional += capital
|
||
|
||
pos_details.append({
|
||
"ticker": ticker,
|
||
"pattern": p.get("pattern_name") or "",
|
||
"strategy": strategy,
|
||
"delta": round(delta, 4),
|
||
"notional": round(capital, 2),
|
||
})
|
||
|
||
if total_notional == 0:
|
||
return {"error": "Notionnel total nul"}
|
||
|
||
portfolio_pnl = (weighted_pnl / total_notional).dropna()
|
||
pnl = portfolio_pnl.values.astype(float)
|
||
alpha = 1.0 - confidence
|
||
|
||
# ── Historical VaR ──
|
||
hist_var_1d = float(np.percentile(pnl, alpha * 100))
|
||
hist_var_nd = hist_var_1d * np.sqrt(horizon_days)
|
||
tail = pnl[pnl <= hist_var_1d]
|
||
hist_cvar = float(np.mean(tail)) if len(tail) > 0 else hist_var_1d
|
||
|
||
# ── Parametric VaR (Gaussian) ──
|
||
mu = float(np.mean(pnl))
|
||
sigma = float(np.std(pnl))
|
||
z = float(norm.ppf(alpha))
|
||
param_var_1d = mu + z * sigma
|
||
param_var_nd = param_var_1d * np.sqrt(horizon_days)
|
||
# ES closed-form: μ − σ·φ(z)/α
|
||
param_cvar = mu - sigma * norm.pdf(-z) / alpha
|
||
|
||
# ── Monte Carlo (stressed: vol × 1.5) ──
|
||
rng = np.random.default_rng(42)
|
||
stressed_sigma = sigma * 1.5
|
||
mc_draws = rng.normal(mu, stressed_sigma, 10_000)
|
||
mc_var_1d = float(np.percentile(mc_draws, alpha * 100))
|
||
mc_var_nd = mc_var_1d * np.sqrt(horizon_days)
|
||
mc_tail = mc_draws[mc_draws <= mc_var_1d]
|
||
mc_cvar = float(np.mean(mc_tail)) if len(mc_tail) > 0 else mc_var_1d
|
||
|
||
# ── Rolling 30-day Historical VaR ──
|
||
rolling_var = []
|
||
for i in range(30, n):
|
||
w = pnl[i - 30:i]
|
||
rolling_var.append({
|
||
"date": portfolio_pnl.index[i].strftime("%Y-%m-%d"),
|
||
"var_95": round(float(np.percentile(w, 5)) * 100, 4),
|
||
})
|
||
rolling_var = rolling_var[-90:] # last 90 data points max
|
||
|
||
# ── Returns histogram ──
|
||
counts, edges = np.histogram(pnl * 100, bins=30)
|
||
histogram = [
|
||
{"x": round(float((edges[i] + edges[i + 1]) / 2), 4), "count": int(counts[i])}
|
||
for i in range(len(counts))
|
||
]
|
||
|
||
# ── Backtest (Kupiec test) ──
|
||
n_breaches = int(np.sum(pnl < hist_var_1d))
|
||
breach_rate = round(n_breaches / n * 100, 2) if n > 0 else 0.0
|
||
|
||
# ── VaR in EUR (based on total notional) ──
|
||
def pct_to_eur(pct_val: float) -> float:
|
||
return round(pct_val / 100 * total_notional, 2)
|
||
|
||
return {
|
||
"var": {
|
||
"historical": {
|
||
"var_1d_pct": round(hist_var_1d * 100, 3),
|
||
"var_nd_pct": round(hist_var_nd * 100, 3),
|
||
"cvar_pct": round(hist_cvar * 100, 3),
|
||
"var_1d_eur": pct_to_eur(hist_var_1d * 100),
|
||
"var_nd_eur": pct_to_eur(hist_var_nd * 100),
|
||
"cvar_eur": pct_to_eur(hist_cvar * 100),
|
||
},
|
||
"parametric": {
|
||
"var_1d_pct": round(param_var_1d * 100, 3),
|
||
"var_nd_pct": round(param_var_nd * 100, 3),
|
||
"cvar_pct": round(param_cvar * 100, 3),
|
||
"var_1d_eur": pct_to_eur(param_var_1d * 100),
|
||
"var_nd_eur": pct_to_eur(param_var_nd * 100),
|
||
"cvar_eur": pct_to_eur(param_cvar * 100),
|
||
},
|
||
"monte_carlo": {
|
||
"var_1d_pct": round(mc_var_1d * 100, 3),
|
||
"var_nd_pct": round(mc_var_nd * 100, 3),
|
||
"cvar_pct": round(mc_cvar * 100, 3),
|
||
"var_1d_eur": pct_to_eur(mc_var_1d * 100),
|
||
"var_nd_eur": pct_to_eur(mc_var_nd * 100),
|
||
"cvar_eur": pct_to_eur(mc_cvar * 100),
|
||
"stressed": True,
|
||
},
|
||
},
|
||
"portfolio": {
|
||
"total_notional_eur": round(total_notional, 2),
|
||
"n_positions": len(pos_details),
|
||
"horizon_days": horizon_days,
|
||
"confidence_pct": round(confidence * 100, 1),
|
||
"lookback_days": n,
|
||
"data_source": data_source,
|
||
},
|
||
"positions": pos_details,
|
||
"rolling_var": rolling_var,
|
||
"histogram": histogram,
|
||
"backtest": {
|
||
"n_observations": n,
|
||
"n_breaches": n_breaches,
|
||
"breach_rate_pct": breach_rate,
|
||
"expected_breach_rate_pct": round(alpha * 100, 1),
|
||
"kupiec_ok": breach_rate <= alpha * 100 * 2,
|
||
},
|
||
}
|
||
|
||
|
||
def save_var_snapshot(result: Dict, confidence: float, horizon_days: int,
|
||
lookback_days: int, default_iv: float,
|
||
macro_regime: Optional[str] = None,
|
||
ticker_prices: Optional[str] = None) -> int:
|
||
"""Persist a VaR result dict to var_snapshots. Returns new row id."""
|
||
if "error" in result:
|
||
raise ValueError(result["error"])
|
||
|
||
v = result["var"]
|
||
p = result["portfolio"]
|
||
bt = result.get("backtest", {})
|
||
computed_at = datetime.utcnow().isoformat(timespec="seconds")
|
||
|
||
conn = get_conn()
|
||
cur = conn.execute(
|
||
"""INSERT INTO var_snapshots
|
||
(computed_at, confidence, horizon_days, lookback_days, default_iv,
|
||
hist_var_1d_pct, hist_cvar_pct, hist_var_1d_eur,
|
||
param_var_1d_pct, param_cvar_pct,
|
||
mc_var_1d_pct, mc_cvar_pct,
|
||
n_positions, total_notional_eur, data_source,
|
||
breach_rate_pct, kupiec_ok,
|
||
macro_regime, ticker_prices, full_result)
|
||
VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
|
||
(
|
||
computed_at, confidence, horizon_days, lookback_days, default_iv,
|
||
v["historical"]["var_1d_pct"], v["historical"]["cvar_pct"], v["historical"]["var_1d_eur"],
|
||
v["parametric"]["var_1d_pct"], v["parametric"]["cvar_pct"],
|
||
v["monte_carlo"]["var_1d_pct"], v["monte_carlo"]["cvar_pct"],
|
||
p["n_positions"], p["total_notional_eur"], p["data_source"],
|
||
bt.get("breach_rate_pct"), 1 if bt.get("kupiec_ok") else 0,
|
||
macro_regime, ticker_prices,
|
||
json.dumps(result, ensure_ascii=False),
|
||
)
|
||
)
|
||
conn.commit()
|
||
row_id = cur.lastrowid
|
||
conn.close()
|
||
return row_id
|
||
|
||
|
||
def get_var_snapshots(limit: int = 20) -> List[Dict]:
|
||
"""Return most recent VaR snapshots (summary, no full_result)."""
|
||
conn = get_conn()
|
||
rows = conn.execute(
|
||
"""SELECT id, computed_at, confidence, horizon_days, lookback_days,
|
||
hist_var_1d_pct, hist_cvar_pct, hist_var_1d_eur,
|
||
param_var_1d_pct, mc_var_1d_pct,
|
||
n_positions, total_notional_eur, data_source,
|
||
breach_rate_pct, kupiec_ok
|
||
FROM var_snapshots ORDER BY computed_at DESC LIMIT ?""",
|
||
(limit,)
|
||
).fetchall()
|
||
conn.close()
|
||
return [dict(r) for r in rows]
|
||
|
||
|
||
def get_var_snapshot(snapshot_id: int) -> Optional[Dict]:
|
||
"""Return a single snapshot with full_result parsed."""
|
||
conn = get_conn()
|
||
row = conn.execute(
|
||
"SELECT * FROM var_snapshots WHERE id=?", (snapshot_id,)
|
||
).fetchone()
|
||
conn.close()
|
||
if not row:
|
||
return None
|
||
d = dict(row)
|
||
if d.get("full_result"):
|
||
try:
|
||
d["full_result"] = json.loads(d["full_result"])
|
||
except Exception:
|
||
pass
|
||
return d
|
||
|
||
|
||
def get_latest_var_snapshot() -> Optional[Dict]:
|
||
"""Return the most recent snapshot with full result."""
|
||
conn = get_conn()
|
||
row = conn.execute(
|
||
"SELECT * FROM var_snapshots ORDER BY computed_at DESC LIMIT 1"
|
||
).fetchone()
|
||
conn.close()
|
||
if not row:
|
||
return None
|
||
d = dict(row)
|
||
if d.get("full_result"):
|
||
try:
|
||
d["full_result"] = json.loads(d["full_result"])
|
||
except Exception:
|
||
pass
|
||
return d
|
||
|
||
|
||
# ─── PnL snapshot ────────────────────────────────────────────────────────────
|
||
|
||
def save_pnl_snapshot() -> int:
|
||
"""Compute live PnL and persist to pnl_snapshots. Returns new row id."""
|
||
from .database import _fetch_live_prices
|
||
|
||
conn = get_conn()
|
||
rows = conn.execute(
|
||
"SELECT * FROM trade_entry_prices WHERE status = 'open'"
|
||
).fetchall()
|
||
trades = [dict(r) for r in rows]
|
||
|
||
closed_count = conn.execute(
|
||
"SELECT COUNT(*) FROM trade_entry_prices WHERE status = 'closed'"
|
||
).fetchone()[0]
|
||
|
||
# Fetch live prices
|
||
tickers = list({t["underlying"] for t in trades if t.get("underlying") and ":" not in (t["underlying"] or "")})
|
||
prices: Dict = {}
|
||
if tickers:
|
||
try:
|
||
prices = _fetch_live_prices(tickers, timeout=15) or {}
|
||
except Exception:
|
||
pass
|
||
|
||
BEARISH = {"long put", "bear put spread", "short call", "put"}
|
||
|
||
def is_bearish(strategy: str) -> bool:
|
||
return any(k in strategy.lower() for k in BEARISH)
|
||
|
||
enriched = []
|
||
total_capital = 0.0
|
||
total_pnl_eur = 0.0
|
||
|
||
for t in trades:
|
||
entry = t.get("entry_price") or 0.0
|
||
capital = t.get("capital_invested") or entry
|
||
current = prices.get(t.get("underlying") or "")
|
||
pnl_pct = None
|
||
if entry and current and entry > 0:
|
||
raw = (current - entry) / entry * 100
|
||
pnl_pct = round(-raw if is_bearish(t.get("strategy") or "") else raw, 2)
|
||
pnl_eur = round(capital * (pnl_pct / 100), 2) if pnl_pct is not None and capital else None
|
||
total_capital += capital
|
||
if pnl_eur is not None:
|
||
total_pnl_eur += pnl_eur
|
||
enriched.append({
|
||
"id": t["id"],
|
||
"ticker": t.get("underlying"),
|
||
"strategy": t.get("strategy"),
|
||
"entry_price": entry,
|
||
"current_price": current,
|
||
"pnl_pct": pnl_pct,
|
||
"pnl_eur": pnl_eur,
|
||
"capital_invested": capital,
|
||
})
|
||
|
||
total_pnl_pct = round(total_pnl_eur / total_capital * 100, 3) if total_capital > 0 else 0.0
|
||
|
||
# Macro regime snapshot
|
||
macro_row = conn.execute(
|
||
"SELECT dominant, scores_json FROM macro_regime_history ORDER BY timestamp DESC LIMIT 1"
|
||
).fetchone()
|
||
macro_context = json.dumps(dict(macro_row)) if macro_row else None
|
||
|
||
snapped_at = datetime.utcnow().isoformat(timespec="seconds")
|
||
cur = conn.execute(
|
||
"""INSERT INTO pnl_snapshots
|
||
(snapped_at, n_open, n_closed, total_capital_eur, total_pnl_pct, total_pnl_eur,
|
||
ticker_prices, macro_regime, trades_snapshot)
|
||
VALUES (?,?,?,?,?,?,?,?,?)""",
|
||
(
|
||
snapped_at, len(trades), closed_count,
|
||
round(total_capital, 2), total_pnl_pct, round(total_pnl_eur, 2),
|
||
json.dumps(prices, ensure_ascii=False),
|
||
macro_context,
|
||
json.dumps(enriched, ensure_ascii=False),
|
||
)
|
||
)
|
||
conn.commit()
|
||
row_id = cur.lastrowid
|
||
conn.close()
|
||
return row_id
|
||
|
||
|
||
def get_pnl_snapshots(limit: int = 48) -> List[Dict]:
|
||
conn = get_conn()
|
||
rows = conn.execute(
|
||
"""SELECT id, snapped_at, n_open, n_closed,
|
||
total_capital_eur, total_pnl_pct, total_pnl_eur
|
||
FROM pnl_snapshots ORDER BY snapped_at DESC LIMIT ?""",
|
||
(limit,)
|
||
).fetchall()
|
||
conn.close()
|
||
return [dict(r) for r in rows]
|
||
|
||
|
||
def get_latest_pnl_snapshot() -> Optional[Dict]:
|
||
conn = get_conn()
|
||
row = conn.execute(
|
||
"SELECT * FROM pnl_snapshots ORDER BY snapped_at DESC LIMIT 1"
|
||
).fetchone()
|
||
conn.close()
|
||
if not row:
|
||
return None
|
||
d = dict(row)
|
||
for key in ("ticker_prices", "macro_regime", "trades_snapshot"):
|
||
if d.get(key):
|
||
try:
|
||
d[key] = json.loads(d[key])
|
||
except Exception:
|
||
pass
|
||
return d
|
||
|
||
|
||
def get_pnl_snapshot(snapshot_id: int) -> Optional[Dict]:
|
||
conn = get_conn()
|
||
row = conn.execute("SELECT * FROM pnl_snapshots WHERE id=?", (snapshot_id,)).fetchone()
|
||
conn.close()
|
||
if not row:
|
||
return None
|
||
d = dict(row)
|
||
for key in ("ticker_prices", "macro_regime", "trades_snapshot"):
|
||
if d.get(key):
|
||
try:
|
||
d[key] = json.loads(d[key])
|
||
except Exception:
|
||
pass
|
||
return d
|
||
|
||
|
||
def diff_pnl_snapshots(id_a: int, id_b: int) -> Dict:
|
||
"""Compute position-level diff between two PnL snapshots (a=earlier, b=later)."""
|
||
snap_a = get_pnl_snapshot(id_a)
|
||
snap_b = get_pnl_snapshot(id_b)
|
||
if not snap_a:
|
||
raise ValueError(f"Snapshot {id_a} introuvable")
|
||
if not snap_b:
|
||
raise ValueError(f"Snapshot {id_b} introuvable")
|
||
|
||
trades_a: Dict[int, Dict] = {t["id"]: t for t in (snap_a.get("trades_snapshot") or [])}
|
||
trades_b: Dict[int, Dict] = {t["id"]: t for t in (snap_b.get("trades_snapshot") or [])}
|
||
|
||
ids_a = set(trades_a)
|
||
ids_b = set(trades_b)
|
||
|
||
new_positions = [] # in B not A
|
||
closed_positions = [] # in A not B
|
||
changed_positions = [] # in both — with delta
|
||
|
||
for tid in ids_b - ids_a:
|
||
t = trades_b[tid]
|
||
new_positions.append({**t, "change_type": "new"})
|
||
|
||
for tid in ids_a - ids_b:
|
||
t = trades_a[tid]
|
||
closed_positions.append({**t, "change_type": "closed"})
|
||
|
||
for tid in ids_a & ids_b:
|
||
ta, tb = trades_a[tid], trades_b[tid]
|
||
pnl_a = ta.get("pnl_pct") or 0.0
|
||
pnl_b = tb.get("pnl_pct") or 0.0
|
||
delta_pct = round(pnl_b - pnl_a, 3)
|
||
pnl_eur_a = ta.get("pnl_eur") or 0.0
|
||
pnl_eur_b = tb.get("pnl_eur") or 0.0
|
||
delta_eur = round(pnl_eur_b - pnl_eur_a, 2)
|
||
changed_positions.append({
|
||
**tb,
|
||
"pnl_pct_a": pnl_a,
|
||
"pnl_pct_b": pnl_b,
|
||
"delta_pct": delta_pct,
|
||
"pnl_eur_a": pnl_eur_a,
|
||
"pnl_eur_b": pnl_eur_b,
|
||
"delta_eur": delta_eur,
|
||
"change_type": "changed",
|
||
})
|
||
|
||
# Sort changed by abs delta descending
|
||
changed_positions.sort(key=lambda x: abs(x["delta_pct"]), reverse=True)
|
||
|
||
# Portfolio-level delta
|
||
cap_a = snap_a.get("total_capital_eur") or 0.0
|
||
cap_b = snap_b.get("total_capital_eur") or 0.0
|
||
pnl_pct_a = snap_a.get("total_pnl_pct") or 0.0
|
||
pnl_pct_b = snap_b.get("total_pnl_pct") or 0.0
|
||
pnl_eur_a = snap_a.get("total_pnl_eur") or 0.0
|
||
pnl_eur_b = snap_b.get("total_pnl_eur") or 0.0
|
||
|
||
# Macro regime diff
|
||
regime_a = (snap_a.get("macro_regime") or {}).get("dominant") if isinstance(snap_a.get("macro_regime"), dict) else None
|
||
regime_b = (snap_b.get("macro_regime") or {}).get("dominant") if isinstance(snap_b.get("macro_regime"), dict) else None
|
||
|
||
return {
|
||
"snapshot_a": {
|
||
"id": snap_a["id"], "snapped_at": snap_a["snapped_at"],
|
||
"n_open": snap_a.get("n_open"), "total_capital_eur": cap_a,
|
||
"total_pnl_pct": pnl_pct_a, "total_pnl_eur": pnl_eur_a,
|
||
"macro_regime": regime_a,
|
||
},
|
||
"snapshot_b": {
|
||
"id": snap_b["id"], "snapped_at": snap_b["snapped_at"],
|
||
"n_open": snap_b.get("n_open"), "total_capital_eur": cap_b,
|
||
"total_pnl_pct": pnl_pct_b, "total_pnl_eur": pnl_eur_b,
|
||
"macro_regime": regime_b,
|
||
},
|
||
"portfolio_delta": {
|
||
"capital_delta_eur": round(cap_b - cap_a, 2),
|
||
"pnl_pct_delta": round(pnl_pct_b - pnl_pct_a, 3),
|
||
"pnl_eur_delta": round(pnl_eur_b - pnl_eur_a, 2),
|
||
"positions_opened": len(new_positions),
|
||
"positions_closed": len(closed_positions),
|
||
"positions_unchanged": sum(1 for p in changed_positions if p["delta_pct"] == 0),
|
||
},
|
||
"new_positions": new_positions,
|
||
"closed_positions": closed_positions,
|
||
"changed_positions": changed_positions,
|
||
}
|