"""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()