PythagorasGoal/Old/scripts/analysis/validate_honest_workers.py

"""Validazione dei worker live multi-asset (TR01/ROT02/TSM01): il replay bar-by-bar del
worker riproduce la funzione di backtest di riferimento?

Replay onesto: si alimenta il worker con finestre crescenti dei dati storici (stesso
universo e stessa config della reference) e si confronta il rendimento finale con la
funzione di riferimento. Non si pretende parità al centesimo (differenze attese da
bar-timing e dalla convenzione capitale-singolo vs media-di-equity), ma il tracking
deve essere stretto e dello stesso segno/ordine di grandezza.

Riferimenti:
  TR01  -> honest_improve2._tr_basket_daily
  ROT02 -> honest_improve2._rot_daily_equity
  TSM01 -> tsmom_research.tsmom_sim

Run: uv run python scripts/analysis/validate_honest_workers.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 scripts.analysis.explore_lab import get_df
from scripts.analysis.honest_lab import available_assets
from src.live.basket_trend_worker import BasketTrendWorker
from src.live.rotation_worker import RotationWorker
from src.live.tsmom_worker import TsmomWorker


def _aligned_panel(assets, tf):
    """{asset: df get_df} -> DataFrame allineato sui timestamp comuni (timestamp + close per asset)."""
    frames = {}
    for a in assets:
        try:
            d = get_df(a, tf)[["timestamp", "close"]].rename(columns={"close": a})
            frames[a] = d
        except Exception:
            pass
    panel = None
    for a, f in frames.items():
        panel = f if panel is None else panel.merge(f, on="timestamp", how="inner")
    return panel.sort_values("timestamp").reset_index(drop=True), list(frames)


def _asset_df(panel, a):
    """df OHLCV minimale (close = open = ...) per un asset dal panel allineato."""
    c = panel[a].values
    return pd.DataFrame({"timestamp": panel["timestamp"].values,
                         "open": c, "high": c, "low": c, "close": c, "volume": 1.0})


def replay(worker, panel, cols, start):
    """Replay bar-by-bar: a ogni step feed delle finestre crescenti. Ritorna ret% finale."""
    n = len(panel)
    for i in range(start, n):
        sub = panel.iloc[: i + 1]
        data = {a: _asset_df(sub, a) for a in cols}
        worker.tick(data)
    return (worker.capital / worker.initial_capital - 1) * 100


def main():
    import tempfile, shutil
    tmp = Path(tempfile.mkdtemp())
    print("=" * 92)
    print("  VALIDAZIONE worker live multi-asset (replay vs backtest di riferimento)")
    print("=" * 92)
    try:
        # ---- ROT02 ----
        from scripts.analysis.honest_improve2 import _rot_daily_equity
        idx = pd.date_range("2021-01-01", "2026-05-26", freq="1D", tz="UTC")
        ref_rot = (_rot_daily_equity(idx).iloc[-1] - 1) * 100
        uni = available_assets()
        panel, cols = _aligned_panel(uni, "1d")
        wr = RotationWorker(universe=cols, top_k=3, gross=0.45, tf="1d",
                            capital=1000.0, data_dir=tmp)
        rot = replay(wr, panel, cols, start=101)
        print(f"  ROT02  worker={rot:+.0f}%   reference={ref_rot:+.0f}%   "
              f"univ={len(cols)} barre={len(panel)}")

        # ---- TSM01 ----
        from scripts.analysis.tsmom_research import tsmom_sim
        ref_tsm = tsmom_sim()["ret"]
        wt = TsmomWorker(universe=cols, horizons=(63, 126, 252), thr=1.0, gross=0.30,
                         tf="1d", capital=1000.0, data_dir=tmp)
        tsm = replay(wt, panel, cols, start=253)
        print(f"  TSM01  worker={tsm:+.0f}%   reference={ref_tsm:+.0f}%")

        # ---- TR01 ----
        from scripts.analysis.honest_improve2 import _tr_basket_daily
        tr_assets = ["BNB", "BTC", "DOGE", "SOL", "XRP"]
        ref_tr = (_tr_basket_daily(tr_assets, idx).iloc[-1] - 1) * 100
        panel4, cols4 = _aligned_panel(tr_assets, "4h")
        wb = BasketTrendWorker(universe=cols4, tf="4h", capital=1000.0, data_dir=tmp)
        tr = replay(wb, panel4, cols4, start=101)
        print(f"  TR01   worker={tr:+.0f}%   reference={ref_tr:+.0f}%   "
              f"univ={len(cols4)} barre={len(panel4)}")

        print("\n  NB: il worker tiene UN capitale unico (compounding del paniere), la reference")
        print("  media equity normalizzate per-asset -> differenza di convenzione attesa, non un bug.")
        print("  Validazione = stesso segno e ordine di grandezza, tracking ragionevole.")
    finally:
        shutil.rmtree(tmp, ignore_errors=True)


if __name__ == "__main__":
    main()