PythagorasGoal/tests/test_skh_live_book.py

"""Test per il follow-up "book sul path live di SKH01" (r0726_skh_live_book / _deluck_factor).

Il test piu' importante e' `test_lente_sleeve_uguale_a_quella_del_simulatore`: e' l'invariante su
cui poggia TUTTA la misura, ed e' esattamente la premessa che per tre sessioni si era creduta
FALSA ("lo sleeve e' vol-targeted, il simulatore no"). Se qualcuno domani aggiunge un vol-target
a `_skyhook_returns`, quella premessa diventa vera e i numeri di questo filone smettono di valere:
il test deve fallire, non passare in silenzio.
"""
from __future__ import annotations

import inspect
import sys
from pathlib import Path

import numpy as np
import pandas as pd
import pytest

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))
sys.path.insert(0, str(ROOT / "scripts" / "research"))
sys.path.insert(0, str(ROOT / "scripts" / "research" / "alt"))


# ===========================================================================
# L'INVARIANTE DI LENTE — la premessa del follow-up
# ===========================================================================
def test_sleeve_skyhook_non_e_vol_targeted():
    """`_skyhook_returns` compone per-trade a nozionale unitario: niente vol-target.
    E' il fatto che sblocca il follow-up; se cambia, va rifatta tutta la misura."""
    from src.portfolio.sleeves import _skyhook_returns

    src = inspect.getsource(_skyhook_returns)
    assert "leverage=1.0" in src
    for kw in ("target_vol", "vol_target", "realized_vol"):
        assert kw not in src, f"lo sleeve ora usa {kw}: la lente e' cambiata, rifare r0726"


def test_lente_sleeve_uguale_a_quella_del_simulatore():
    """sim_equity(mode='canonical') deve riprodurre backtest_signals ESATTAMENTE.
    E' cio' che rende confrontabili path live e sleeve senza conversioni."""
    import r0702_anchor_skh01 as r02

    _, m, ltf, ent = r02.run_asset("BTC", 0)
    eq = r02.sim_equity(ltf, ent, "canonical")
    assert float(np.max(np.abs(eq - m.equity))) == 0.0


# ===========================================================================
# Convenzioni di combinazione
# ===========================================================================
def _s(vals, start="2020-01-01"):
    return pd.Series(np.asarray(vals, float),
                     index=pd.date_range(start, periods=len(vals), freq="1D", tz="UTC"))


def test_skh_book_e_5050_su_inner_join():
    import r0726_skh_live_book as LB

    a = _s([0.01, 0.02, 0.03])
    b = _s([0.03, 0.02, 0.01])
    out = LB.skh_book({"BTC": a, "ETH": b})
    assert list(np.round(out.values, 10)) == [0.02, 0.02, 0.02]


def test_skh_book_inner_join_taglia_le_date_non_comuni():
    """Convenzione dello sleeve: inner join. Se un asset ha meno storia, il book parte dopo."""
    import r0726_skh_live_book as LB

    a = _s([0.01] * 5, start="2020-01-01")
    b = _s([0.03] * 3, start="2020-01-03")
    out = LB.skh_book({"BTC": a, "ETH": b})
    assert len(out) == 3
    assert out.index[0] == pd.Timestamp("2020-01-03", tz="UTC")


def test_book_metrics_drift_e_annualizzato_a_365():
    import r0726_skh_live_book as LB

    r = 0.001
    cols = {"X": _s([r] * 400)}
    _, _, drift = LB.book_metrics(cols, {"X": 1.0})
    assert drift == pytest.approx(r * 365.25, rel=1e-9)


def test_drift_and_cagr_coerenti():
    import r0726_deluck_factor as DF

    r = 0.0005
    cols = {"X": _s([r] * 800)}
    drift, cagr, vol = DF.drift_and_cagr(cols, {"X": 1.0})
    assert drift == pytest.approx(r * 365.25, rel=1e-9)
    assert cagr == pytest.approx((1 + r) ** 365.25 - 1, rel=1e-6)   # serie costante -> CAGR esatto
    assert vol == pytest.approx(0.0, abs=1e-12)                     # nessuna varianza


# ===========================================================================
# Il sottocampione di offset: dichiarato a priori, non selezionato
# ===========================================================================
def test_offsets_sub_e_un_sottoinsieme_uniforme_della_griglia_piena():
    import r0702_anchor_skh01 as r02
    import r0726_skh_sigcache as SC

    full = set(r02.OFFSETS)
    assert set(SC.OFFSETS_SUB) <= full, "il sottocampione contiene offset fuori griglia"
    assert SC.OFFSETS_SUB[0] == 0, "l'offset canonico deve essere incluso"
    passi = np.diff(np.array(SC.OFFSETS_SUB))
    assert len(set(passi.tolist())) == 1, f"sottocampione non uniforme: passi {passi}"


def test_cache_path_e_deterministico_e_separa_asset_e_offset():
    import r0726_skh_sigcache as SC

    p1 = SC.cache_path("BTC", 0)
    p2 = SC.cache_path("BTC", 90)
    p3 = SC.cache_path("ETH", 0)
    assert p1 != p2 and p1 != p3
    assert p1 == SC.cache_path("BTC", 0)
    assert p1.suffix == ".parquet"