PythagorasGoal/scripts/analysis/validate_xsec_worker.py

"""Valida il CrossSectionalWorker: replay bar-per-bar == backtest XS01.xsec_sim?

Come validate_worker_pairs: alimenta il worker con finestre trailing crescenti del
pannello 8-asset e confronta capitale finale e n.trade col backtest di riferimento
scripts.strategies.XS01_cross_sectional.xsec_sim. Se combaciano, la semantica live e' fedele.
"""
from __future__ import annotations

import shutil
import sys
import tempfile
from pathlib import Path

import pandas as pd

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

from src.live.xsec_worker import CrossSectionalWorker
from scripts.strategies.XS01_cross_sectional import aligned_panel, xsec_sim, UNIVERSE, LB, HOLD


def _replay(M, dfs, params):
    """Alimenta il worker bar-per-bar su finestre trailing; ritorna il worker."""
    n = len(M)
    tmp = Path(tempfile.mkdtemp(prefix="xsec_val_"))
    try:
        w = CrossSectionalWorker(UNIVERSE, tf="1h", params=params,
                                 fee_rt=0.0005, data_dir=tmp)
        w._save = lambda: None; w._log = lambda *a, **k: None; w._notify = lambda *a, **k: None
        window = LB + 6
        for k in range(LB + 1, n + 1):           # prima finestra = lb+1 barre -> ingresso al bar lb
            lo = max(0, k - window)
            w.tick({a: dfs[a].iloc[lo:k] for a in UNIVERSE})
        return w
    finally:
        shutil.rmtree(tmp, ignore_errors=True)


def main():
    print("=" * 88)
    print("  VALIDAZIONE CrossSectionalWorker — replay live vs backtest xsec_sim (fee 0.10% RT/book)")
    print("=" * 88)
    M = aligned_panel(UNIVERSE)
    dfs = {a: pd.DataFrame({"timestamp": M.index.values, "close": M[a].values}) for a in UNIVERSE}

    # [1] K=1: parita' storica col backtest canonico
    w = _replay(M, dfs, {"lb": LB, "hold": HOLD})
    bt = xsec_sim(UNIVERSE)
    bt_cap = 1000.0 * (1 + bt["ret"] / 100)
    cap_ok = abs(w.capital - bt_cap) / bt_cap < 0.02 if bt_cap else False
    trd_ok = abs(w.total_trades - bt["trades"]) <= max(2, bt["trades"] * 0.02)
    ww = w.total_wins / w.total_trades * 100 if w.total_trades else 0
    print(f"\n  [K=1] {'':<6}{'cap':>14}{'trades':>8}{'win%':>7}")
    print(f"  WORKER{w.capital:>14.0f}{w.total_trades:>8d}{ww:>7.1f}")
    print(f"  BCKTST{bt_cap:>14.0f}{bt['trades']:>8d}{bt['win']:>7.1f}")
    ok1 = cap_ok and trd_ok
    print(f"  ESITO: {'OK (replay == backtest)' if ok1 else 'DIFF -> INDAGARE'}")

    # [2] K=3 (tranching live): parita' con l'unione delle fasi 0,4,8 (gate
    # xs01_tranche_gate) — capitale comune, PnL/K per trade in ordine di exit.
    # POS/LEV importati dal modulo canonico, NON hardcoded: una costante stantia
    # qui farebbe passare la validazione contro un sistema diverso (code-review)
    from scripts.analysis.xs01_tranche_research import xsec_trades
    from scripts.strategies.XS01_cross_sectional import POS, LEV
    K = 3
    w3 = _replay(M, dfs, {"lb": LB, "hold": HOLD, "tranches": K})
    step = HOLD // K
    allt = sorted([t for j in range(K) for t in xsec_trades(phase=j * step, M=M)],
                  key=lambda t: t[1])
    cap = 1000.0
    for _, _, net in allt:
        cap = max(cap + cap * POS * LEV * net / K, 10.0)
    cap_ok3 = abs(w3.capital - cap) / cap < 0.02
    trd_ok3 = abs(w3.total_trades - len(allt)) <= max(2, len(allt) * 0.02)
    print(f"\n  [K=3] {'':<6}{'cap':>14}{'trades':>8}")
    print(f"  WORKER{w3.capital:>14.0f}{w3.total_trades:>8d}")
    print(f"  BCKTST{cap:>14.0f}{len(allt):>8d}")
    ok3 = cap_ok3 and trd_ok3
    print(f"  ESITO: {'OK (replay == unione fasi)' if ok3 else 'DIFF -> INDAGARE'}")
    print(f"\n  ESITO COMPLESSIVO: {'OK' if (ok1 and ok3) else 'DIFF -> INDAGARE'}")
    print("  (diff minime attese da bar finale aperta / troncamento)")


if __name__ == "__main__":
    main()