PythagorasGoal/tests/test_skyhook.py

"""Test della strategia SKH01 (Skyhook) — dual-timeframe regime+breakout su BTC/ETH.

Coprono: fedelta' al brief (ancore demo BuzVola/BuzVolume), allineamento dual-TF, assenza di
look-ahead (causalita'), e robustezza onesta del config V1 su entrambi gli asset.
"""
import sys
from pathlib import Path

import numpy as np
import pandas as pd

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

from src.data.downloader import load_data
from src.strategies.skyhook import (
    HTF_MIN, LTF_MIN, SKH01_V2_DD, SkyhookParams, build_frames, chande01, skyhook_entries)

# config V1 (vincente del lever-scout/grid; vedi diario 2026-06-23-skyhook)
V1 = dict(ptn_n=55, sl_atr=2.5, tp_atr=6.0, vola_lo=35, vola_hi=95, vol_lo=0.0)


# ---------------------------------------------------------------------------
# Fedelta' al brief: indicatori tipo-Chande, normalizzati 0-100.
# ---------------------------------------------------------------------------
def test_chande01_anchors():
    """Semantica del brief: volatilita'/volume STEADY -> 50 (neutro); in RAMPA -> 100; in CALO -> 0."""
    n = 100
    assert abs(chande01(np.full(n, 7.0), 13)[-1] - 50.0) < 1e-9      # costante -> neutro
    assert abs(chande01(np.arange(n, dtype=float), 13)[-1] - 100.0) < 1e-9   # rampa su -> 100
    assert abs(chande01(np.arange(n, 0, -1, dtype=float), 13)[-1] - 0.0) < 1e-9  # rampa giu' -> 0


def test_demo_buzvola_buzvolume():
    """Ancore della demo: ATR costante (vol steady) -> BuzVola 50; volume in rampa -> BuzVolume 100."""
    n = 100
    buz_vola = chande01(np.full(n, 2.0), 13)            # ATR steady
    buz_volume = chande01(np.linspace(1000, 5000, n), 13)  # volume in rampa
    assert abs(buz_vola[-1] - 50.0) < 1e-9
    assert abs(buz_volume[-1] - 100.0) < 1e-9
    # oscillatori sempre in [0,100]
    assert chande01(np.random.default_rng(0).normal(size=500).cumsum() + 100, 13)[20:].min() >= -1e-9
    assert chande01(np.random.default_rng(1).normal(size=500).cumsum() + 100, 13)[20:].max() <= 100 + 1e-9


# ---------------------------------------------------------------------------
# Allineamento dual-timeframe: 690 = 3 x 230, confini HTF subset dei confini LTF.
# ---------------------------------------------------------------------------
def test_dual_tf_alignment():
    assert HTF_MIN == 3 * LTF_MIN
    ltf, htf = build_frames(load_data("BTC", "5m"))
    # ogni timestamp (open) HTF e' anche un open LTF (stessa griglia epoch)
    ltf_opens = set(ltf["timestamp"].astype("int64").tolist())
    htf_opens = htf["timestamp"].astype("int64").tolist()
    inside = sum(t in ltf_opens for t in htf_opens)
    assert inside / len(htf_opens) > 0.99, "i confini HTF devono essere un sottoinsieme dei confini LTF"


# ---------------------------------------------------------------------------
# Causalita': gli ingressi su un prefisso devono coincidere con la run completa.
# ---------------------------------------------------------------------------
def test_no_lookahead_entries():
    p = SkyhookParams(**V1)
    ltf, htf = build_frames(load_data("BTC", "5m"))
    full = skyhook_entries(ltf, htf, p)
    n = len(ltf)
    cut = int(n * 0.85)
    cut_ts = int(ltf["timestamp"].iloc[cut - 1])
    htf_cut = htf[htf["timestamp"] <= cut_ts].reset_index(drop=True)
    sub = skyhook_entries(ltf.iloc[:cut].reset_index(drop=True), htf_cut, p)
    for i in range(cut - 200, cut):
        a, b = full[i], sub[i]
        assert (a is None) == (b is None)
        if a is not None:
            assert a["dir"] == b["dir"]
            assert abs(a["sl"] - b["sl"]) < 1e-6 and abs(a["tp"] - b["tp"]) < 1e-6


# ---------------------------------------------------------------------------
# Robustezza onesta del config V1: PASS su BTC E ETH, netto fee, OOS.
# ---------------------------------------------------------------------------
def test_v1_robust_both_assets():
    import skyhooklib as sk
    p = SkyhookParams(**V1)
    for a in ("BTC", "ETH"):
        r = sk.run_asset(a, p, sk.FEE_RT)
        assert r["full"]["sharpe"] >= 0.5, f"{a} FULL Sharpe basso: {r['full']['sharpe']}"
        assert r["holdout"]["sharpe"] >= 0.2, f"{a} HOLD-OUT Sharpe basso: {r['holdout']['sharpe']}"
        assert r["full"]["n_trades"] >= 20, f"{a} troppo pochi trade: {r['full']['n_trades']}"
    assert sk.causality(p, "BTC")["ok"] and sk.causality(p, "ETH")["ok"]


# ---------------------------------------------------------------------------
# Exit asimmetrici SHORT (SKH01-V2-DD): l'override cambia SOLO gli short; i default
# (None) preservano esattamente il comportamento simmetrico precedente.
# ---------------------------------------------------------------------------
def test_short_override_backward_compatible():
    """Con gli override SHORT a None, gli ingressi sono identici alla versione simmetrica."""
    ltf, htf = build_frames(load_data("BTC", "5m"))
    base = SkyhookParams(**V1)
    # stessi parametri ma con campi override esplicitamente None (= default)
    same = SkyhookParams(**V1, exit_mode_short=None, sl_pct_short=None, tp_pct_short=None)
    e0, e1 = skyhook_entries(ltf, htf, base), skyhook_entries(ltf, htf, same)
    assert e0 == e1, "i campi override a None NON devono cambiare nulla (backward-compat)"


def test_short_override_changes_only_shorts():
    """Un SL short piu' stretto (pct) modifica gli stop SHORT ma lascia intatti i LONG."""
    ltf, htf = build_frames(load_data("ETH", "5m"))
    sym = SkyhookParams(ptn_n=45, vola_lo=35, vola_hi=95, vol_lo=0.0,
                        exit_mode="pct", sl_pct=0.04, tp_pct=0.10)
    asym = SkyhookParams(ptn_n=45, vola_lo=35, vola_hi=95, vol_lo=0.0,
                         exit_mode="pct", sl_pct=0.04, tp_pct=0.10,
                         sl_pct_short=0.02, tp_pct_short=0.08)
    es, ea = skyhook_entries(ltf, htf, sym), skyhook_entries(ltf, htf, asym)
    longs_same = shorts_diff = 0
    for a, b in zip(es, ea):
        if a is None or b is None:
            assert (a is None) == (b is None)
            continue
        assert a["dir"] == b["dir"]
        if a["dir"] == 1:                                  # LONG invariati
            assert abs(a["sl"] - b["sl"]) < 1e-6 and abs(a["tp"] - b["tp"]) < 1e-6
            longs_same += 1
        else:                                              # SHORT con SL/TP diversi
            assert abs(a["sl"] - b["sl"]) > 1e-6
            shorts_diff += 1
    assert longs_same > 0 and shorts_diff > 0


def test_skyhook_pos_fn_structure():
    """pos_fn introspettiva: dict BTC/ETH, ciascuno 'flat' o dict con dir/entry/sl/tp coerenti."""
    from src.portfolio.sleeves import _skyhook_positions
    pos = _skyhook_positions()
    assert set(pos.keys()) == {"BTC", "ETH"}
    for a, p in pos.items():
        assert p == "flat" or (isinstance(p, dict) and p["dir"] in ("LONG", "SHORT")
                               and p["sl"] > 0 and p["tp"] > 0 and 0 <= p["bars_in"] <= p["max_bars"])


def test_v2dd_robust_both_assets():
    """SKH01-V2-DD: PASS netto fee su BTC&ETH, hold-out forte, e maxDD standalone <30%."""
    import skyhooklib as sk
    p = SKH01_V2_DD
    for a in ("BTC", "ETH"):
        r = sk.run_asset(a, p, sk.FEE_RT)
        assert r["full"]["sharpe"] >= 0.5, f"{a} FULL Sharpe basso: {r['full']['sharpe']}"
        assert r["holdout"]["sharpe"] >= 0.5, f"{a} HOLD-OUT Sharpe basso: {r['holdout']['sharpe']}"
        assert r["full"]["maxdd"] < 0.30, f"{a} maxDD non sotto 30%: {r['full']['maxdd']}"
        assert r["full"]["n_trades"] >= 20, f"{a} troppo pochi trade: {r['full']['n_trades']}"
    assert sk.causality(p, "BTC")["ok"] and sk.causality(p, "ETH")["ok"]