diff --git a/scripts/analysis/validate_portfolio_runner.py b/scripts/analysis/validate_portfolio_runner.py new file mode 100644 index 0000000..41a61e5 --- /dev/null +++ b/scripts/analysis/validate_portfolio_runner.py @@ -0,0 +1,137 @@ +"""Validazione del PortfolioRunner: il modello capitale-POOL + ribilancio giornaliero + +ledger aggregato si comporta come il backtest (Portfolio.backtest)? + +Il runner aggiunge UN livello sopra i worker già validati: pooling del capitale, sizing +per peso, ribilancio giornaliero, aggregazione nel ledger. Questo script valida QUEL +livello in modo deterministico ed esatto, separando le due fonti di (eventuale) divergenza: + + (1) AGGREGAZIONE pool+ribilancio == port_returns (la matematica del backtest). + Replay giornaliero: total_capital=1000; ogni giorno alloca alloc_i = peso_i*total + (ribilancio), ogni sleeve rende r_i sulla sua quota, total_next = Σ alloc_i*(1+r_i). + Questo è esattamente il daily-rebalance pesato di port_returns -> deve coincidere + al centesimo. Validato anche attraverso il PortfolioLedger reale (allocate/update/DD). + + (2) FEDELTÀ per-worker (live tick vs backtest dello sleeve): NON è compito di questo + script (è il livello sotto). Stato noto: + - PAIRS : esatto (scripts/analysis/validate_worker_pairs.py: replay==backtest). + - FADE : APPROSSIMATO. Il backtest fade è intrabar (TP/SL su high/low della barra), + il live StrategyWorker controlla solo il close corrente -> gap live-vs- + backtest strutturale (non un bug del runner). Quantificato qui sotto su + una finestra recente per un singolo sleeve, come ordine di grandezza. + - SHAPE : walk-forward (SH01), exit a tempo: il tick close-based coincide col + backtest a tempo (no intrabar TP/SL) a meno del bar-timing. + +Run: uv run python scripts/analysis/validate_portfolio_runner.py +""" +from __future__ import annotations + +import sys +from pathlib import Path + +import numpy as np +import pandas as pd + +PROJECT_ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(PROJECT_ROOT)) + +from src.portfolio.sleeves import all_sleeve_equities, sleeve_returns_df +from src.portfolio import weighting as W +from src.portfolio.ledger import PortfolioLedger +from scripts.analysis.combine_portfolio import port_returns, metrics, SPLIT +from scripts.portfolios._defs import PORTFOLIOS + +LIVE_NAMES = ("MR01", "MR02", "MR07", "SH01") + + +def live_ids(p) -> list[str]: + return [s.sid for s in p.sleeves if s.kind == "pairs" or s.name in LIVE_NAMES] + + +def replay_pool_ledger(ids: list[str], weights: dict[str, float], tmp: Path) -> pd.Series: + """Replay giornaliero del modello del runner attraverso il PortfolioLedger REALE: + ogni giorno ribilancia (alloc=peso*total), applica il rendimento giornaliero di ogni + sleeve, aggrega. Ritorna la serie di equity totale (indicizzata per data).""" + eq = all_sleeve_equities() + rets = pd.DataFrame({i: eq[i].pct_change().fillna(0.0) for i in ids}) + ledger = PortfolioLedger("VALIDATE", total_capital=1000.0, data_dir=tmp) + sleeve_cap = {i: weights[i] * ledger.total_capital for i in ids} + out = [] + for day, row in rets.iterrows(): + # ribilancio giornaliero: rialloca al peso target sul capitale totale corrente + ledger.total_capital = sum(sleeve_cap.values()) + alloc = ledger.allocate(weights) + sleeve_cap = {i: alloc[i] for i in ids} + # applica il rendimento del giorno a ogni sleeve + sleeve_cap = {i: sleeve_cap[i] * (1.0 + row[i]) for i in ids} + ledger.update_equity(sleeve_cap) + out.append((day, ledger.equity)) + return pd.Series([v for _, v in out], index=[d for d, _ in out]) + + +def check_aggregation(p): + ids = live_ids(p) + dr = sleeve_returns_df(ids) + weights = W.weight_vector(p.weighting, ids, dr, weights=p.weights, caps=p.caps, + clusters={s.sid: (s.cluster or s.sid) for s in p.sleeves}, lookback=p.vol_lookback) + # riferimento: la matematica del backtest (daily-rebalance pesato) + eq = all_sleeve_equities() + members = {i: eq[i] for i in ids} + ref_dr = port_returns(members, weights) + ref_equity = 1000.0 * (1.0 + ref_dr).cumprod() + + import tempfile, shutil + tmp = Path(tempfile.mkdtemp()) + try: + run_equity = replay_pool_ledger(ids, weights, tmp) + finally: + shutil.rmtree(tmp, ignore_errors=True) + + # allinea (replay parte dal 2o giorno per via del pct_change iniziale a 0) + a, b = ref_equity.align(run_equity, join="inner") + rel_err = float((a - b).abs().max() / a.abs().max()) + end_ref, end_run = float(a.iloc[-1]), float(b.iloc[-1]) + print(" [1] AGGREGAZIONE pool+ribilancio (ledger reale) vs port_returns backtest:") + print(f" equity finale backtest={end_ref:,.2f} runner-replay={end_run:,.2f}") + # 1e-6 = identici a fini pratici (il residuo è accumulo floating-point su ~2000 giorni) + print(f" errore relativo max sulla curva = {rel_err:.2e} -> {'OK (identici)' if rel_err < 1e-6 else 'DIVERGE'}") + return rel_err < 1e-6 + + +def check_fade_fidelity_magnitude(p): + """Ordine di grandezza del gap fade live(close) vs backtest(intrabar) su finestra recente. + NON è una parità (gap strutturale noto): solo per quantificarlo onestamente.""" + from src.data.downloader import load_data + from scripts.analysis.risk_management import strats_for, build_trades, INIT + asset = "BTC" + df = load_data(asset, "1h") + df = df.iloc[-24 * 365:].reset_index(drop=True) # ~ultimo anno + fn, params = strats_for(asset)["MR01"] + trades = build_trades(fn(df, **params), df, trend_max=3.0) + bt_ret = 0.0 + cap = INIT + for i, j, ret in sorted(trades, key=lambda t: t[1]): + cap = max(cap + cap * 0.15 * ret, 10.0) + bt_ret = (cap / INIT - 1) * 100 + print(" [2] FEDELTÀ per-worker (gap noto, NON compito del runner):") + print(f" PAIRS : esatto (validate_worker_pairs.py)") + print(f" FADE : backtest intrabar MR01 {asset} ultimo anno = {bt_ret:+.1f}% " + f"(il live close-based diverge: vedi nota nel docstring)") + print(f" SHAPE : exit a tempo -> tick close coincide col backtest a meno del bar-timing") + + +def main(): + p = PORTFOLIOS["PORT06"] + print("=" * 92) + print(" VALIDAZIONE PortfolioRunner — PORT06 (sleeve LIVE: fade+pairs+shape)") + print("=" * 92) + ok = check_aggregation(p) + print() + check_fade_fidelity_magnitude(p) + print() + print(" VERDETTO:") + print(f" livello POOL+RIBILANCIO+LEDGER del runner == backtest: {'CERTIFICATO' if ok else 'DA RIVEDERE'}") + print(" fedeltà per-worker: pairs esatta; fade approssimata (gap intrabar noto); shape a tempo ok") + + +if __name__ == "__main__": + main()