PythagorasGoal/Old/scripts/strategies/XS01_cross_sectional.py

"""XS01 — Cross-Sectional Reversion (market-neutral su 8 cripto). FAMIGLIA NUOVA.

Distinta dai pairs PR01 (pairwise) e dai fade (single-asset): ogni HOLD ore classifica
gli 8 asset per rendimento su LB ore e va LONG i perdenti relativi / SHORT i vincenti
(peso ∝ -(ret - media_cross-section)), market-neutral gross 1. Cattura il FATTORE
reversione cross-sezionale. Scorrelato (~0) da pairs e fade -> diversificatore.

Engine ONESTO (no look-ahead, verificato): pesi a barra i da close[<=i]; ingresso a
close[i], uscita a close[i+HOLD]; roll NON sovrapposto (riallinea ogni HOLD barre).
Fee = 0.10% RT/book (turnover gross 1 -> 2*fee_rt). PnL su capitale composto (pos, lev).

Validazione (sessione 2026-06-09, lb48 hold12, fee 0.10% RT, OOS ultimo 30%):
  FULL Sharpe ~3.3 / OOS ~3.4, plateau lb 12-72 x hold 6-24 (OOS 2-3.9), 4/5 anni+.
  Decorrelato (-0.006 da PR01 ETH/BTC). Cost-sensitive: muore ~0.35% RT/book.
  Gate PORT06: +XS01 -> OOS Sharpe 9.66->10.07, FULL DD 3.68->3.46 (OOS DD +0.06pp a mezza size).
"""
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.data.downloader import load_data

UNIVERSE = ["BTC", "ETH", "LTC", "ADA", "SOL", "BNB", "XRP", "DOGE"]
FEE_RT, LEV, POS, OOS_FRAC = 0.0005, 3.0, 0.15, 0.30
LB, HOLD = 48, 12


def aligned_panel(assets=UNIVERSE, tf="1h"):
    dfs = {a: load_data(a, tf)[["timestamp", "close"]].rename(columns={"close": a}).set_index("timestamp")
           for a in assets}
    M = pd.concat(dfs.values(), axis=1, join="inner").sort_index()
    return M[assets]


def xsec_sim(assets=UNIVERSE, tf="1h", lb=LB, hold=HOLD, fee_rt=FEE_RT, lev=LEV,
             pos=POS, split_frac=0.0):
    M = aligned_panel(assets, tf)
    C = M[assets].values
    ts = pd.to_datetime(M.index, unit="ms", utc=True)
    n = len(C); logC = np.log(C)
    split = int(n * split_frac)
    cap = peak = 1000.0; dd = 0.0
    trades = wins = 0; rets = []; yearly = {}; yearly_n = {}
    eq_ts, eq_v = [], []
    last = -1; i = max(lb, split)
    fee = 2 * fee_rt                       # gross 1 -> turnover 2 (entra+esce)
    while i < n - hold:
        if i <= last:
            i += 1; continue
        dm = (logC[i] - logC[i - lb]); dm = dm - dm.mean()
        w = -dm; gw = np.sum(np.abs(w))
        if gw < 1e-9:
            i += 1; continue
        w = w / gw                         # market-neutral, gross 1
        book = float(np.sum(w * (logC[i + hold] - logC[i])))
        net = book - fee
        cap = max(cap + cap * pos * lev * net, 10.0)
        peak = max(peak, cap); dd = max(dd, (peak - cap) / peak)
        trades += 1; wins += net > 0; rets.append(net * pos); last = i + hold
        eq_ts.append(ts[i + hold]); eq_v.append(cap)
        yearly[ts[i].year] = yearly.get(ts[i].year, 0.0) + net * 100
        yearly_n[ts[i].year] = yearly_n.get(ts[i].year, 0) + 1
        i += 1
    yrs_span = (ts[-1] - ts[max(split, 0)]).days / 365.25 or 1
    sharpe = 0.0
    if len(rets) > 1 and np.std(rets) > 0:
        sharpe = float(np.mean(rets) / np.std(rets) * np.sqrt(trades / yrs_span))
    ret_tot = (cap / 1000 - 1) * 100
    cagr = ((cap / 1000) ** (1 / yrs_span) - 1) * 100 if cap > 0 else -100
    return dict(trades=trades, win=wins / trades * 100 if trades else 0, ret=ret_tot,
                cagr=cagr, dd=dd * 100, sharpe=sharpe, yearly=yearly, yearly_n=yearly_n,
                eq_ts=eq_ts, eq_v=eq_v)


def check_no_lookahead():
    M = aligned_panel(); logC = np.log(M.values); i = 1000
    a = (logC[i] - logC[i - LB])
    Cp = logC.copy(); Cp[i + 1:] += 0.5
    b = (Cp[i] - Cp[i - LB])
    print(f"  no-look-ahead: segnale invariato col futuro perturbato -> "
          f"{'OK' if np.allclose(a, b) else 'VIOLAZIONE'}")


def run():
    print("=" * 84)
    print("  XS01 — Cross-Sectional Reversion (8 asset, market-neutral) | netto fee 0.10% RT/book")
    print("=" * 84)
    check_no_lookahead()
    f = xsec_sim()
    o = xsec_sim(split_frac=1 - OOS_FRAC)
    yrs = f["yearly"]; pos_y = sum(1 for v in yrs.values() if v > 0)
    print(f"  trade {f['trades']} | win {f['win']:.1f}% | CAGR {f['cagr']:.0f}% | DD {f['dd']:.0f}% | "
          f"Sharpe FULL {f['sharpe']:.2f} / OOS {o['sharpe']:.2f} | anni+ {pos_y}/{len(yrs)}")
    print("  per anno:", "  ".join(f"{y}:{v:+.0f}%" for y, v in sorted(yrs.items())))


if __name__ == "__main__":
    run()