PythagorasGoal/Old/scripts/analysis/honest_rotation.py

"""Strategia #3 candidata: ROTAZIONE cross-sectional momentum (multi-crypto).
Una sola strategia che usa l'INTERO paniere: ad ogni ribilanciamento alloca il
capitale agli asset con momentum migliore (long-only). Cattura la dispersione tra
crypto (gli alt forti corrono molto piu' di BTC nei bull) senza shortare nulla.

Onesto: i pesi a close[i] usano solo rendimenti passati; il rendimento del bar
i->i+1 e' realizzato con quei pesi. Fee sul turnover. Allineamento per timestamp.
"""
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.honest_lab import get_df, available_assets, FEE_RT  # noqa: E402

LEV = 3.0
GROSS = 0.45        # esposizione lorda = LEV*POS del singolo (0.15*3) per confronto equo


def build_panel(assets: list[str], tf: str) -> pd.DataFrame:
    """Matrice close allineata per timestamp (inner join)."""
    closes = {}
    for a in assets:
        df = get_df(a, tf)
        s = pd.Series(df["close"].values,
                      index=pd.to_datetime(df["timestamp"], unit="ms", utc=True))
        closes[a] = s[~s.index.duplicated()]
    panel = pd.DataFrame(closes).dropna()
    return panel


def simulate_rotation(panel: pd.DataFrame, lookback=30, top_k=2,
                      fee_rt=FEE_RT, gross=GROSS, abs_filter=True,
                      oos_frac=0.0) -> dict:
    """Ad ogni barra: ranking per rendimento passato `lookback`; pesi uguali sui
    top_k con momentum>0 (se abs_filter); altrimenti cash. gross = esposizione tot.
    oos_frac>0: parte a investire solo dall'ultimo frac del campione."""
    P = panel.values
    T, N = P.shape
    rets = np.zeros_like(P)
    rets[1:] = P[1:] / P[:-1] - 1
    years = panel.index.year.values
    start = max(lookback + 1, int(T * (1 - oos_frac)) if oos_frac else lookback + 1)
    cap = peak = 1000.0
    max_dd = 0.0
    w = np.zeros(N)
    yearly: dict[int, float] = {}
    turn_total = 0.0
    for i in range(start, T - 1):
        mom = P[i] / P[i - lookback] - 1
        order = np.argsort(mom)[::-1]
        new_w = np.zeros(N)
        chosen = [j for j in order if (mom[j] > 0 or not abs_filter)][:top_k]
        if chosen:
            for j in chosen:
                new_w[j] = gross / len(chosen)
        # fee sul turnover (one-way = fee_rt/2 su ogni variazione di peso)
        turnover = np.abs(new_w - w).sum()
        cap -= cap * turnover * (fee_rt / 2)
        turn_total += turnover
        w = new_w
        port_ret = float(np.dot(w, rets[i + 1]))      # rendimento bar i->i+1
        cap = max(cap * (1 + port_ret), 10.0)
        peak = max(peak, cap); max_dd = max(max_dd, (peak - cap) / peak)
        yearly[years[i]] = yearly.get(years[i], 0.0) + port_ret * 100
    return {
        "ret": (cap / 1000 - 1) * 100,
        "dd": max_dd * 100,
        "turnover": turn_total,
        "yearly": yearly,
        "pos_years": sum(1 for v in yearly.values() if v > 0),
        "n_years": len(yearly),
    }


if __name__ == "__main__":
    assets = available_assets()
    print(f"ROTATION cross-sectional momentum — fee {FEE_RT*100:.2f}% RT, gross {GROSS} | OOS 30%")
    print(f"  Paniere: {assets}")
    for tf in ["1d", "4h"]:
        panel = build_panel(assets, tf)
        print(f"\n  === {tf} === panel {panel.shape[0]} barre, {panel.index[0].date()} -> {panel.index[-1].date()}")
        print(f"  {'config':<22s}{'FULL%':>9s}{'OOS%':>9s}{'DD%':>6s}{'Turn':>7s}{'AnniP':>8s}")
        for lb in [20, 30, 60, 90]:
            for k in [1, 2, 3]:
                full = simulate_rotation(panel, lookback=lb, top_k=k)
                oos = simulate_rotation(panel, lookback=lb, top_k=k, oos_frac=0.30)
                anni = f"{full['pos_years']}/{full['n_years']}"
                print(f"  lb{lb:<3d} top{k:<14d}{full['ret']:>+9.0f}{oos['ret']:>+9.0f}"
                      f"{full['dd']:>6.0f}{full['turnover']:>7.0f}{anni:>8s}")