feat: page Historique Positions + diff entre 2 snapshots PnL

Backend:
- get_pnl_snapshot(id) : détail complet d'un snapshot avec trades parsés
- diff_pnl_snapshots(a, b) : diff positions entre deux snapshots (nouvelles /
  fermées / évolution PnL par position + delta portfolio)
- GET /api/var/pnl/snapshots/{id} : détail snapshot
- GET /api/var/pnl/diff?a=&b= : calcul du diff

Frontend PositionHistory.tsx :
- Timeline scrollable des snapshots avec sparkline PnL
- Clic snapshot → détail des positions à ce moment (prix entrée, prix actuel,
  PnL %, PnL €, régime macro)
- Boutons A/B par snapshot → sélection de deux points à comparer
- Vue diff A→B : nouvelles positions, fermées, évolution PnL par trade,
  delta portfolio (capital, PnL %, PnL €)
- Route /position-history + nav sidebar

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
OpenSquared
2026-06-20 06:36:55 +02:00
parent 67546d39af
commit 4a1dc76d26
5 changed files with 692 additions and 3 deletions

View File

@@ -3,7 +3,8 @@ from pydantic import BaseModel
from services.var_service import (
compute_var, save_var_snapshot,
get_var_snapshots, get_var_snapshot, get_latest_var_snapshot,
get_pnl_snapshots, get_latest_pnl_snapshot, save_pnl_snapshot,
get_pnl_snapshots, get_pnl_snapshot, get_latest_pnl_snapshot, save_pnl_snapshot,
diff_pnl_snapshots,
)
from services.var_scheduler import (
get_scheduler_status, restart_var_scheduler, restart_pnl_scheduler,
@@ -64,10 +65,27 @@ def pnl_latest():
@router.get("/pnl/snapshots")
def pnl_snapshots(limit: int = Query(default=48, ge=1, le=200)):
def pnl_snapshots(limit: int = Query(default=100, ge=1, le=500)):
return {"snapshots": get_pnl_snapshots(limit)}
@router.get("/pnl/snapshots/{snapshot_id}")
def pnl_snapshot_detail(snapshot_id: int):
snap = get_pnl_snapshot(snapshot_id)
if not snap:
raise HTTPException(404, "Snapshot not found")
return snap
@router.get("/pnl/diff")
def pnl_diff(a: int = Query(..., description="ID snapshot A (earlier)"),
b: int = Query(..., description="ID snapshot B (later)")):
try:
return diff_pnl_snapshots(a, b)
except ValueError as e:
raise HTTPException(404, str(e))
@router.post("/pnl/run-now")
def pnl_run_now():
try:

View File

@@ -462,3 +462,107 @@ def get_latest_pnl_snapshot() -> Optional[Dict]:
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,
}