PythagorasGoal/scripts/analysis/ledger_vs_backtest.py

"""Report ricorrente LEDGER REALE vs BACKTEST — il gate per scalare il capitale.

Per gli sleeve ESEGUITI (6 fade 15m, DIP01, 6 pairs, SH01) il sim del worker ==
backtest canonico PER COSTRUZIONE (validato: validate_worker_pairs, parity test).
Quindi "reale vs backtest" = "reale vs sim" = la **fuga di esecuzione**: slippage
sugli ingressi/uscite + fee reali vs assunte + effetti netting/phantom/sim_fallback.
È il numero che dice se l'edge SOPRAVVIVE ai fill veri — la condizione per passare
da testnet/piccolo a capitale serio.

Cosa misura (finestra mobile, default 7g) leggendo SOLO i trades.jsonl + status.json
(nessuna rete → affidabile in cron):
  - PnL realizzato sim vs reale (Σ e per-trade) -> LEAKAGE € e % (la bottom line)
  - slippage ingressi (REAL_OPEN.slippage_bps) e uscite (REAL_CLOSE.slippage_bps)
  - fee reali vs assunte (0.10% RT)
  - trade sim_fallback (reale mai eseguito/fillato) = quota NON coperta dal reale
  - ledger per-sleeve: real_capital vs capital (sim)
Verdetto: leakage per-trade piccolo e stabile -> verde (si puo' pensare a scalare).

    uv run python scripts/analysis/ledger_vs_backtest.py            # stampa
    uv run python scripts/analysis/ledger_vs_backtest.py --days 14  # finestra 14g
    uv run python scripts/analysis/ledger_vs_backtest.py --telegram # + invio Telegram
"""
from __future__ import annotations

import json
import sys
from datetime import datetime, timedelta, timezone
from pathlib import Path
from statistics import median

PROJECT_ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(PROJECT_ROOT))

PAPER = PROJECT_ROOT / "data" / "portfolio_paper"
ASSUMED_FEE_RT = 0.001          # 0.10% RT assunto dal backtest
GREEN_BPS, YELLOW_BPS = 15.0, 40.0   # soglie slippage medio per-lato (verdetto)


def _parse_ts(s: str) -> datetime | None:
    try:
        t = datetime.fromisoformat(s.replace("Z", "+00:00"))
        return t if t.tzinfo else t.replace(tzinfo=timezone.utc)
    except Exception:
        return None


def collect(days: int, since: datetime | None = None) -> dict:
    # since (clean-start) ha priorita' sulla finestra mobile: lo scheduler parte
    # dal 2026-06-13 (post-fix TP_PHANTOM/netting/ribilancio) cosi' accumula SOLO
    # dati puliti; una finestra mobile pura includerebbe l'incidente testnet pre-fix
    # (sim +82 vs reale +5: +4% fantasma che il sim bookava e il reale no).
    cut = since or (datetime.now(timezone.utc) - timedelta(days=days))
    entries, exits, closes = [], [], []
    for f in sorted(PAPER.glob("*/trades.jsonl")):
        wid = f.parent.name
        for line in f.read_text().splitlines():
            try:
                e = json.loads(line)
            except Exception:
                continue
            tt = _parse_ts(e.get("ts", ""))
            if tt is None or tt < cut:
                continue
            ev = e.get("event")
            if ev == "REAL_OPEN":
                entries.append((wid, e))
            elif ev in ("REAL_CLOSE", "REAL_CLOSE_PAIR"):
                exits.append((wid, e))
            elif ev == "CLOSE":
                closes.append((wid, e))
    return {"entries": entries, "exits": exits, "closes": closes}


def _stats(vals: list[float]) -> dict:
    if not vals:
        return {"n": 0, "mean": 0.0, "med": 0.0, "p90": 0.0, "max": 0.0}
    s = sorted(vals)
    return {"n": len(s), "mean": sum(s) / len(s), "med": median(s),
            "p90": s[min(len(s) - 1, int(0.9 * len(s)))], "max": s[-1]}


def analyze(days: int, since: datetime | None = None) -> dict:
    d = collect(days, since)
    # slippage ingressi/uscite (bps); fee reali ingresso
    open_slip = [abs(e.get("slippage_bps") or 0.0) for _, e in d["entries"]]
    exit_slip = [abs(e.get("slippage_bps") or 0.0) for _, e in d["exits"]
                 if e.get("real_fill") is not None]   # null = uscita da TP resting (no slip)
    open_fee = [e.get("fee_usd") or 0.0 for _, e in d["entries"]]

    # PnL realizzato sim vs reale dai CLOSE (la verita' contabile guidata da real_truth)
    real_closes = [e for _, e in d["closes"] if e.get("pnl_source") == "real"]
    fallback = [e for _, e in d["closes"] if e.get("pnl_source") == "sim_fallback"]
    sim_sum = sum(e.get("sim_pnl") or 0.0 for e in real_closes)
    real_sum = sum(e.get("real_pnl") or 0.0 for e in real_closes)
    per_trade_gap = [(e.get("sim_pnl") or 0.0) - (e.get("real_pnl") or 0.0) for e in real_closes]

    # ledger per-sleeve: real vs sim (solo worker eseguiti = con REAL_OPEN nella storia)
    executed = {wid for wid, _ in d["entries"]}
    ledger = []
    for wid in sorted(executed):
        sp = PAPER / wid / "status.json"
        if not sp.exists():
            continue
        st = json.loads(sp.read_text())
        cap, rc = st.get("capital"), st.get("real_capital")
        if cap is not None and rc is not None:
            ledger.append((wid, cap, rc, rc - cap))

    return {"days": days, "since": since,
            "open_slip": _stats(open_slip), "exit_slip": _stats(exit_slip),
            "open_fee_mean": (sum(open_fee) / len(open_fee)) if open_fee else 0.0,
            "n_real": len(real_closes), "n_fallback": len(fallback),
            "sim_sum": sim_sum, "real_sum": real_sum,
            "leak_total": sim_sum - real_sum,
            "leak_per_trade": (sum(per_trade_gap) / len(per_trade_gap)) if per_trade_gap else 0.0,
            "ledger": ledger}


def verdict(a: dict) -> tuple[str, str]:
    avg_slip = (a["open_slip"]["mean"] + a["exit_slip"]["mean"]) / 2
    if a["n_real"] < 10:
        return "🟡", f"campione PICCOLO ({a['n_real']} trade reali): non concludere ancora"
    if avg_slip <= GREEN_BPS and abs(a["leak_per_trade"]) < 0.30:
        return "🟢", "leakage basso e stabile: reale ~ backtest"
    if avg_slip <= YELLOW_BPS:
        return "🟡", "leakage moderato: tenere d'occhio prima di scalare"
    return "🔴", "leakage ALTO: l'edge si erode sull'esecuzione, NON scalare"


def render(a: dict) -> str:
    flag, msg = verdict(a)
    win = f"da {a['since']:%Y-%m-%d}" if a.get("since") else f"finestra {a['days']}g"
    L = [f"LEDGER REALE vs BACKTEST — {win}  {flag}",
         f"  {msg}",
         f"  trade reali: {a['n_real']}  | sim_fallback (reale mai fillato): {a['n_fallback']}",
         f"  PnL realizzato: sim {a['sim_sum']:+.2f}  reale {a['real_sum']:+.2f}  "
         f"-> LEAKAGE {a['leak_total']:+.2f} ({a['leak_per_trade']:+.3f}/trade)",
         f"  slippage ingressi (bps): media {a['open_slip']['mean']:.1f} "
         f"med {a['open_slip']['med']:.1f} p90 {a['open_slip']['p90']:.1f} max {a['open_slip']['max']:.1f}  "
         f"(n={a['open_slip']['n']})",
         f"  slippage uscite  (bps): media {a['exit_slip']['mean']:.1f} "
         f"med {a['exit_slip']['med']:.1f} max {a['exit_slip']['max']:.1f}  (n={a['exit_slip']['n']}; "
         f"escluse uscite da TP resting)",
         f"  fee reale media ingresso: ${a['open_fee_mean']:.4f}"]
    if a["ledger"]:
        L.append("  ledger per-sleeve (sim -> reale, Δ):")
        for wid, cap, rc, dlt in a["ledger"]:
            short = wid.replace("_reversion", "").replace("_fade", "").replace("_bollinger", "")[:30]
            L.append(f"    {short:30} {cap:7.2f} -> {rc:7.2f}  {dlt:+.2f}")
    return "\n".join(L)


def main() -> int:
    days = 7
    since = None
    if "--days" in sys.argv:
        days = int(sys.argv[sys.argv.index("--days") + 1])
    if "--since" in sys.argv:
        since = _parse_ts(sys.argv[sys.argv.index("--since") + 1] + "T00:00:00+00:00")
    a = analyze(days, since)
    text = render(a)
    print(text)
    if "--telegram" in sys.argv:
        from src.live.telegram_notifier import send_telegram
        from src.version import APP_VERSION
        send_telegram(f"📒 <b>LEDGER vs BACKTEST</b> <code>v{APP_VERSION}</code>\n<pre>{text}</pre>")
        print("[telegram] report inviato")
    flag, _ = verdict(a)
    return 0 if flag == "🟢" else 1


if __name__ == "__main__":
    sys.exit(main())