PythagorasGoal/Old/scripts/analysis/oos_validation.py

"""Validazione out-of-sample fee-aware di tutte le strategie live.

Per ognuna delle 6 config in strategies.yml:
  - split temporale held-out (train = primi (1-test_frac), test = ultimo test_frac)
  - ML01 (SignalEngine): allena sul train, predice sul test (come il worker live)
  - rule-based: i segnali sono causali, si valutano quelli nella finestra test
  - simulazione fedele al worker live: una posizione per volta (non-overlap),
    uscita a `hold` barre o stop a -2%, fee round-trip e leva inclusi

Stampa, per ogni config: numero trade nel test, win% lordo e netto, return netto,
costo commissioni, e confronto lordo-vs-netto per isolare l'impatto delle fee.
Usa i parquet locali (data/raw), nessuna chiamata di rete.
"""
from __future__ import annotations

import sys
from pathlib import Path

import numpy as np
import pandas as pd
import yaml

PROJECT_ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(PROJECT_ROOT))

from src.data.downloader import load_data
from src.live.strategy_loader import load_strategy
from src.live.signal_engine import SignalEngine, keltner_ratio, build_features

TEST_FRAC = 0.30
STOP_PCT = -0.02


def simulate(entries: list[tuple[int, int]], close: np.ndarray, hold: int,
             fee_rt: float, lev: float, pos: float,
             initial: float = 1000.0, entry_offset: int = 0) -> dict:
    """FSM fedele al worker live: non-overlap, hold N barre o stop -2%.

    entry_offset: 0 = ingresso a close[i] (worker live); 1 = close[i-1]
    (convenzione del backtest storico, che conosce la direzione di barra i).
    """
    n = len(close)
    capital = peak = initial
    max_dd = 0.0
    fees_eur = gross_eur = 0.0
    wins_gross = wins_net = n_trades = 0
    last_exit = -1

    for i, d in entries:
        e = i - entry_offset
        if e <= last_exit or e < 0 or e + 1 >= n:
            continue
        entry = close[e]
        exit_price = close[min(e + hold, n - 1)]
        for k in range(1, hold + 1):
            j = e + k
            if j >= n:
                exit_price = close[n - 1]
                break
            if k < hold and (close[j] - entry) / entry * d <= STOP_PCT:
                exit_price = close[j]
                break
            if k == hold:
                exit_price = close[j]

        actual = (exit_price - entry) / entry * d   # movimento prezzo * direzione (no leva)
        gross = actual * lev
        fee = fee_rt * lev
        net = gross - fee

        cap_before = capital
        capital = max(cap_before + cap_before * pos * net, 10.0)
        gross_eur += cap_before * pos * gross
        fees_eur += cap_before * pos * fee
        peak = max(peak, capital)
        max_dd = max(max_dd, (peak - capital) / peak)

        n_trades += 1
        wins_gross += actual > 0
        wins_net += net > 0
        last_exit = e + hold

    return {
        "trades": n_trades,
        "win_gross": wins_gross / n_trades * 100 if n_trades else 0.0,
        "win_net": wins_net / n_trades * 100 if n_trades else 0.0,
        "net_return_pct": (capital / initial - 1) * 100,
        "net_eur": capital - initial,
        "gross_eur": gross_eur,
        "fees_eur": fees_eur,
        "final_capital": capital,
        "max_dd": max_dd * 100,
    }


def rule_entries(name: str, df: pd.DataFrame, params: dict, split: int) -> list[tuple[int, int]]:
    strat = load_strategy(name)
    ts = pd.to_datetime(df["timestamp"], unit="ms", utc=True)
    sigs = strat.generate_signals(df, ts, **params)
    return [(s.idx, s.direction) for s in sigs if s.idx >= split]


def ml_entries(df: pd.DataFrame, params: dict, split: int, hold: int) -> tuple[list[tuple[int, int]], dict]:
    bb_w = params.get("bb_window", 14)
    sq_thr = params.get("sq_threshold", 0.8)
    ml_thr = params.get("ml_threshold", 0.70)

    eng = SignalEngine(bb_w=bb_w, sq_thr=sq_thr, ml_thr=ml_thr)
    train_res = eng.train(df.iloc[:split].reset_index(drop=True), lookahead=hold)
    if not eng.trained:
        return [], train_res

    close = df["close"].values
    high = df["high"].values
    low = df["low"].values
    volume = df["volume"].values
    n = len(df)
    kcr = keltner_ratio(close, high, low, bb_w)
    up_idx = list(eng.model.classes_).index(1)

    entries: list[tuple[int, int]] = []
    in_sq = False
    sq_start = 0
    for i in range(bb_w + 1, n):
        if np.isnan(kcr[i]):
            continue
        is_sq = kcr[i] < sq_thr
        if is_sq and not in_sq:
            in_sq, sq_start = True, i
        elif not is_sq and in_sq:
            in_sq = False
            dur = i - sq_start
            if dur < eng.min_squeeze_bars or i < split or i + hold >= n:
                continue
            avg_vol = float(np.mean(volume[sq_start:i]))
            feats = build_features(df, i, dur, avg_vol, kcr[i])
            if feats is None:
                continue
            p_up = eng.model.predict_proba(eng.scaler.transform(feats.reshape(1, -1)))[0][up_idx]
            if p_up >= ml_thr:
                entries.append((i, 1))
            elif p_up <= (1 - ml_thr):
                entries.append((i, -1))
    return entries, train_res


def squeeze_releases(df: pd.DataFrame, bb_w: int, sq_thr: float, min_dur: int,
                     split: int) -> list[int]:
    """Indici delle barre di rilascio squeeze nella finestra test (idx >= split)."""
    close = df["close"].values
    high = df["high"].values
    low = df["low"].values
    kcr = keltner_ratio(close, high, low, bb_w)
    rels: list[int] = []
    in_sq = False
    sq_start = 0
    for i in range(bb_w + 1, len(df)):
        if np.isnan(kcr[i]):
            continue
        is_sq = kcr[i] < sq_thr
        if is_sq and not in_sq:
            in_sq, sq_start = True, i
        elif not is_sq and in_sq:
            in_sq = False
            if i - sq_start >= min_dur and i >= split:
                rels.append(i)
    return rels


def honest_entries(df: pd.DataFrame, rels: list[int], rule: str, mom: int = 4) -> list[tuple[int, int]]:
    """Direzione da regole honest (solo dati <= i-1) o baseline breakout.

    breakout: sign(close[i]-close[i-1])  -> conoscibile solo a close[i] (= live attuale)
    premom:   sign(close[i-1]-close[i-1-mom]) -> trend pre-release, 100% honest
    fade:     -sign(close[i]-close[i-1]) -> mean-reversion del breakout
    """
    close = df["close"].values
    out: list[tuple[int, int]] = []
    for i in rels:
        if i - 1 - mom < 0:
            continue
        if rule == "premom":
            d = np.sign(close[i - 1] - close[i - 1 - mom])
        elif rule == "fade":
            d = -np.sign(close[i] - close[i - 1])
        else:  # breakout
            d = np.sign(close[i] - close[i - 1])
        if d != 0:
            out.append((i, int(d)))
    return out


def main():
    cfg = yaml.safe_load((PROJECT_ROOT / "strategies.yml").read_text())
    defaults = cfg.get("defaults", {})
    hold = defaults.get("hold_bars", 3)
    lev = defaults.get("leverage", 3)
    fee_rt = 0.002

    fee_grid = [0.0, 0.0005, 0.001, 0.0015, 0.002]

    # ---- (b) SENSIBILITA' ALLE FEE (config live, ingresso close[i]) ----
    print("=" * 104)
    print(f"  (b) SENSIBILITA' ALLE FEE — config live, ingresso close[i] | OOS {int(TEST_FRAC*100)}% | hold={hold} leva={lev}x")
    print("=" * 104)
    print(f"  {'Strategia':<26s}{'Asset':>5s}{'Trd':>5s}{'Lordo€':>9s}"
          + "".join(f"{f'{f*100:.2f}%':>10s}" for f in fee_grid))
    print("  " + "-" * 100)

    for entry in cfg.get("strategies", []):
        if not entry.get("enabled", True):
            continue
        name, asset, tf = entry["name"], entry["asset"], entry["tf"]
        pos = entry.get("position_size", defaults.get("position_size", 0.15))
        params = dict(entry.get("params", {}))
        params["asset"], params["tf"] = asset, tf

        df = load_data(asset, tf).reset_index(drop=True)
        split = int(len(df) * (1 - TEST_FRAC))
        close = df["close"].values
        entries = (ml_entries(df, params, split, hold)[0] if name.startswith("ML01")
                   else rule_entries(name, df, params, split))

        gross = simulate(entries, close, hold, 0.0, lev, pos)["net_eur"]
        rets = [simulate(entries, close, hold, f, lev, pos)["net_return_pct"] for f in fee_grid]
        print(f"  {name:<26s}{asset:>5s}{len(entries):>5d}{gross:>+9.0f}"
              + "".join(f"{r:>+10.1f}" for r in rets))

    print("  " + "-" * 100)
    print("  Colonne = Ret% netto al variare della fee RT. 0.00% isola l'edge puro (senza costi).")
    print("  Deribit perp reale: taker ~0.10% RT, maker ~0%. Il modello live usa 0.20% RT.")

    # ---- (a) HONEST-ENTRY squeeze: direzione decisa <= i-1, ingresso close[i] ----
    print("\n" + "=" * 104)
    print(f"  (a) HONEST-ENTRY squeeze (bb14 sq0.8 dur>=5) — ingresso close[i], fee={fee_rt*100:.1f}% RT")
    print("=" * 104)
    print(f"  {'Asset':>5s}{'Regola direzione':>20s}{'Trd':>6s}{'Win%g':>8s}{'Win%n':>8s}{'Netto€':>9s}{'Ret%':>9s}{'DD%':>7s}")
    print("  " + "-" * 100)

    rules = [("breakout (=live)", "breakout"), ("pre-trend mom4", "premom"),
             ("pre-trend mom8", "premom8"), ("fade breakout", "fade")]
    for asset in ["BTC", "ETH"]:
        df = load_data(asset, "15m").reset_index(drop=True)
        split = int(len(df) * (1 - TEST_FRAC))
        close = df["close"].values
        rels = squeeze_releases(df, 14, 0.8, 5, split)
        for label, rule in rules:
            mom = 8 if rule == "premom8" else 4
            ents = honest_entries(df, rels, "premom" if rule == "premom8" else rule, mom=mom)
            r = simulate(ents, close, hold, fee_rt, lev, 0.15)
            print(f"  {asset:>5s}{label:>20s}{r['trades']:>6d}{r['win_gross']:>8.1f}"
                  f"{r['win_net']:>8.1f}{r['net_eur']:>+9.0f}{r['net_return_pct']:>+9.1f}{r['max_dd']:>7.1f}")
        print("  " + "-" * 100)
    print("  pre-trend = direzione dal trend PRIMA del rilascio (solo dati <= i-1): 100% honest.")
    print("  Se nessuna regola honest batte ~breakeven, non esiste edge direzionale tradeable.")


if __name__ == "__main__":
    main()