PythagorasGoal/Old/scripts/analysis/explore_lab.py

"""Harness ONESTO condiviso per esplorare nuove famiglie di strategie.

Regole NON negoziabili (per non ripetere l'errore squeeze look-ahead):
  - direzione e prezzo decisi con dati FINO a close[i] incluso, mai con la barra i
    usata per scegliere la direzione e poi entrare a i-1;
  - ingresso ESEGUIBILE a close[i];
  - exit: take-profit / stop-loss intrabar (high/low) e/o time-limit max_bars;
    tp/sl possono essere None -> exit solo a tempo (utile per stagionalita');
  - una posizione per volta (non-overlap), capitale composto;
  - NETTO dopo fee round-trip (default 0.10% RT reale Deribit) e leva;
  - validazione OOS (held-out, ultimo 30%) + sweep fee 0.00-0.20% RT.

Le strategie ad alta frequenza muoiono di fee: ogni entry costa fee_rt*lev sul
notional. Tienine conto: meno operazioni e edge > costi.

Asset disponibili: ADA BNB BTC DOGE ETH LTC SOL XRP (1h, 15m; BTC/ETH anche 5m).
"""
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

FEE_RT = 0.001       # Deribit perp realistico: taker ~0.05%/lato = 0.10% RT
LEV = 3.0
POS = 0.15
OOS_FRAC = 0.30
ASSETS = ["ADA", "BNB", "BTC", "DOGE", "ETH", "LTC", "SOL", "XRP"]
BARS_PER_YEAR = {"5m": 105120, "15m": 35040, "1h": 8760, "4h": 2190, "1d": 365}


# --------------------------- dati ---------------------------
def get_df(asset: str, tf: str) -> pd.DataFrame:
    """OHLCV con colonna dt (UTC). tf nativo (5m,15m,1h) o resample da 1h (4h,1d).
    timestamp resta ms-epoch reale anche dopo il resample (no placeholder)."""
    if tf in ("5m", "15m", "1h"):
        df = load_data(asset, tf).reset_index(drop=True)
    else:
        base = load_data(asset, "1h").copy()
        base["dt"] = pd.to_datetime(base["timestamp"], unit="ms", utc=True)
        base = base.set_index("dt")
        rule = {"4h": "4h", "1d": "1D"}[tf]
        agg = base.resample(rule).agg(
            {"open": "first", "high": "max", "low": "min", "close": "last", "volume": "sum"}
        ).dropna()
        epoch = pd.Timestamp("1970-01-01", tz="UTC")          # ms-epoch portabile (qualsiasi risoluzione)
        agg["timestamp"] = ((agg.index - epoch) // pd.Timedelta(milliseconds=1)).astype("int64")
        df = agg.reset_index(drop=True)
    df["dt"] = pd.to_datetime(df["timestamp"], unit="ms", utc=True)
    return df


def _dt(df: pd.DataFrame) -> pd.DatetimeIndex:
    return pd.to_datetime(df["timestamp"], unit="ms", utc=True)


# --------------------------- indicatori ---------------------------
def atr(df: pd.DataFrame, n: int = 14) -> np.ndarray:
    h, l, c = df["high"].values, df["low"].values, df["close"].values
    pc = np.roll(c, 1); pc[0] = c[0]
    tr = np.maximum(h - l, np.maximum(np.abs(h - pc), np.abs(l - pc)))
    return pd.Series(tr).rolling(n).mean().values


def ema(x: np.ndarray, n: int) -> np.ndarray:
    return pd.Series(x).ewm(span=n, adjust=False).mean().values


def rsi(close: np.ndarray, n: int = 14) -> np.ndarray:
    d = np.diff(close, prepend=close[0])
    up = pd.Series(np.where(d > 0, d, 0.0)).ewm(alpha=1/n, adjust=False).mean()
    dn = pd.Series(np.where(d < 0, -d, 0.0)).ewm(alpha=1/n, adjust=False).mean()
    rs = up / dn.replace(0, np.nan)
    return (100 - 100 / (1 + rs)).values


# --------------------------- engine ---------------------------
def simulate(entries: list[dict], df: pd.DataFrame, fee_rt: float = FEE_RT,
             lev: float = LEV, pos: float = POS, split: int = -1) -> dict:
    """entries: dict con i(idx), d(+1/-1), max_bars; tp/sl opzionali (None=solo tempo).
    split: se >0, conta solo entries con i>=split (finestra OOS)."""
    h, l, c = df["high"].values, df["low"].values, df["close"].values
    n = len(c)
    ts = _dt(df)
    cap = peak = 1000.0
    max_dd = 0.0
    fee = fee_rt * lev
    trades = wins = 0
    last_exit = -1
    bars_in = 0
    yearly: dict[int, float] = {}
    rets: list[float] = []

    for e in entries:
        i, d = e["i"], e["d"]
        if i <= last_exit or i + 1 >= n or i < split:
            continue
        entry = c[i]
        tp, sl, mb = e.get("tp"), e.get("sl"), e["max_bars"]
        exit_p = c[min(i + mb, n - 1)]
        j = min(i + mb, n - 1)
        for k in range(1, mb + 1):
            j = i + k
            if j >= n:
                exit_p = c[n - 1]; break
            hit_sl = sl is not None and ((d == 1 and l[j] <= sl) or (d == -1 and h[j] >= sl))
            hit_tp = tp is not None and ((d == 1 and h[j] >= tp) or (d == -1 and l[j] <= tp))
            if hit_sl:
                exit_p = sl; break
            if hit_tp:
                exit_p = tp; break
            if k == mb:
                exit_p = c[j]
        ret = (exit_p - entry) / entry * d * lev - fee
        cb = cap
        cap = max(cb + cb * pos * ret, 10.0)
        peak = max(peak, cap); max_dd = max(max_dd, (peak - cap) / peak)
        trades += 1; wins += ret > 0; bars_in += (j - i)
        last_exit = j
        rets.append(ret * pos)
        yearly[ts.iloc[i].year] = yearly.get(ts.iloc[i].year, 0.0) + ret * 100

    sharpe = float(np.mean(rets) / np.std(rets) * np.sqrt(len(rets))) if len(rets) > 1 and np.std(rets) > 0 else 0.0
    return {
        "trades": trades,
        "win": wins / trades * 100 if trades else 0.0,
        "ret": (cap / 1000 - 1) * 100,
        "dd": max_dd * 100,
        "sharpe": sharpe,
        "yearly": yearly,
        "exposure": bars_in / n * 100 if n else 0.0,
    }


def evaluate(name: str, entries: list[dict], df: pd.DataFrame,
             fees=(0.0, 0.0005, 0.001, 0.002)) -> dict:
    """Valuta una lista di entries: FULL, OOS e sweep fee. Stampa una riga sintetica."""
    split = int(len(df) * (1 - OOS_FRAC))
    full = simulate(entries, df)
    oos = simulate(entries, df, split=split)
    sweep = {f: simulate(entries, df, fee_rt=f)["ret"] for f in fees}
    sweep_oos = {f: simulate(entries, df, fee_rt=f, split=split)["ret"] for f in fees}
    yrs = full["yearly"]; pos_yrs = sum(1 for v in yrs.values() if v > 0)
    print(f"  {name:<24s} trd={full['trades']:>5d} win={full['win']:>4.1f}% "
          f"FULL={full['ret']:>+7.0f}% OOS={oos['ret']:>+7.0f}% DD={full['dd']:>4.0f}% "
          f"oDD={oos['dd']:>4.0f}% Shrp={full['sharpe']:>4.2f} exp={full['exposure']:>4.1f}% "
          f"anniPos={pos_yrs}/{len(yrs)} | fee0.2%: FULL={sweep[0.002]:>+6.0f} OOS={sweep_oos[0.002]:>+6.0f}")
    return {"full": full, "oos": oos, "sweep": sweep, "sweep_oos": sweep_oos, "pos_yrs": pos_yrs, "n_yrs": len(yrs)}


def robust(res: dict) -> bool:
    """Verdetto onesto: positivo FULL e OOS, regge a fee 0.20% RT, quasi tutti gli anni positivi."""
    return (res["full"]["ret"] > 0 and res["oos"]["ret"] > 0
            and res["sweep"][0.002] > 0 and res["sweep_oos"][0.002] > 0
            and res["pos_yrs"] >= max(res["n_yrs"] - 1, 1))


if __name__ == "__main__":
    # smoke test: una stagionalita' banale (hour-of-day) su BTC 1h
    df = get_df("BTC", "1h"); ts = _dt(df)
    ents = [{"i": i, "d": 1, "max_bars": 6, "tp": None, "sl": None}
            for i in range(len(df) - 7) if ts.iloc[i].hour == 0]
    print("smoke test — BTC long ad ogni 00:00 UTC, hold 6h:")
    evaluate("seasonality_h0", ents, df)