PythagorasGoal/scripts/analysis/honest_lab.py

"""honest_lab — laboratorio di ricerca strategie ONESTO e fee-aware.

Principi (per non ripetere l'errore look-ahead della famiglia squeeze):
  1. Ogni segnale a barra i usa SOLO dati fino a close[i]. Ingresso a close[i]
     (eseguibile dal vivo: il worker vede la candela chiusa ed entra). Opzione
     di robustezza: ingresso a open[i+1] (ancora piu' conservativo).
  2. Uscita TP/SL valutata intrabar su high/low, conservativa: SL prima del TP
     nello stesso bar. Time-limit max_bars. Una posizione per volta (non-overlap).
  3. Tutto NETTO dopo fee round-trip realistiche (0.10% Deribit) * leva.
  4. Validazione: FULL + OOS (held-out ultimo 30%) + per-anno + sweep fee
     + griglia parametri + su PIU' asset. Niente di tutto cio' -> scartata.

Engine condiviso riusabile da tutte le strategie candidate.
"""
from __future__ import annotations

import sys
from dataclasses import dataclass
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  # noqa: E402

FEE_RT = 0.001          # Deribit perp realistico: taker ~0.05%/lato = 0.10% RT
LEV = 3.0
POS = 0.15
OOS_FRAC = 0.30
DATA_DIR = PROJECT_ROOT / "data" / "raw"

# ----------------------------------------------------------------------------
# dati
# ----------------------------------------------------------------------------
_CACHE: dict[tuple[str, str], pd.DataFrame] = {}


def available_assets() -> list[str]:
    out = []
    for p in sorted(DATA_DIR.glob("*_1h.parquet")):
        name = p.stem.replace("_1h", "").upper()
        if name not in ("BTC_DVOL", "ETH_DVOL"):
            out.append(name)
    return out


def get_df(asset: str, tf: str) -> pd.DataFrame:
    """tf nativo (15m,1h) o resample da 1h (2h,4h,6h,12h,1d)."""
    key = (asset, tf)
    if key in _CACHE:
        return _CACHE[key]
    if tf in ("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 = {"2h": "2h", "4h": "4h", "6h": "6h", "12h": "12h", "1d": "1D"}[tf]
        agg = base.resample(rule).agg(
            {"open": "first", "high": "max", "low": "min", "close": "last", "volume": "sum"}
        ).dropna()
        # l'indice puo' essere datetime64[ms] o [ns]: forza ms in modo robusto
        agg["timestamp"] = agg.index.values.astype("datetime64[ms]").astype("int64")
        df = agg.reset_index(drop=True)
    df = df[["timestamp", "open", "high", "low", "close", "volume"]].copy()
    _CACHE[key] = df
    return df


# ----------------------------------------------------------------------------
# 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 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


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


# ----------------------------------------------------------------------------
# engine
# ----------------------------------------------------------------------------
@dataclass
class SimResult:
    trades: int
    win: float
    ret: float          # ritorno % netto composto su 1000
    dd: float
    exposure: float
    yearly: dict[int, float]

    @property
    def pos_years(self) -> int:
        return sum(1 for v in self.yearly.values() if v > 0)

    @property
    def n_years(self) -> int:
        return len(self.yearly)


def simulate(entries: list[dict], df: pd.DataFrame, fee_rt: float = FEE_RT,
             lev: float = LEV, pos: float = POS, entry_on_open: bool = False) -> SimResult:
    """entries: dict {i, d(+1/-1), tp, sl, max_bars}.

    entry_on_open=True -> ingresso a open[i+1] invece di close[i] (robustezza).
    """
    o, h, l, c = (df["open"].values, df["high"].values,
                  df["low"].values, df["close"].values)
    n = len(c)
    cap = peak = 1000.0
    max_dd = 0.0
    fee = fee_rt * lev
    trades = wins = 0
    last_exit = -1
    bars_in = 0
    ts = pd.to_datetime(df["timestamp"], unit="ms", utc=True)
    yearly: dict[int, float] = {}

    for e in entries:
        i, d = e["i"], e["d"]
        ei = i + 1 if entry_on_open else i        # barra di ingresso
        if ei <= last_exit or ei + 1 >= n:
            continue
        entry = o[ei] if entry_on_open else c[i]
        tp, sl, mb = e["tp"], e["sl"], e["max_bars"]
        exit_p = c[min(ei + mb, n - 1)]
        j = min(ei + mb, n - 1)
        for k in range(1, mb + 1):
            j = ei + k
            if j >= n:
                j = n - 1; exit_p = c[j]; break
            hit_sl = (d == 1 and l[j] <= sl) or (d == -1 and h[j] >= sl)
            hit_tp = (d == 1 and h[j] >= tp) or (d == -1 and l[j] <= tp)
            if hit_sl:                      # conservativo: SL prima del TP nello stesso bar
                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
        cap = max(cap + cap * pos * ret, 10.0)
        peak = max(peak, cap); max_dd = max(max_dd, (peak - cap) / peak)
        trades += 1; wins += ret > 0; bars_in += (j - ei)
        last_exit = j
        yr = ts.iloc[i].year
        yearly[yr] = yearly.get(yr, 0.0) + ret * 100
    return SimResult(
        trades=trades,
        win=wins / trades * 100 if trades else 0.0,
        ret=(cap / 1000 - 1) * 100,
        dd=max_dd * 100,
        exposure=bars_in / n * 100,
        yearly=yearly,
    )


def oos_split(entries: list[dict], df: pd.DataFrame, frac: float = OOS_FRAC):
    split = int(len(df) * (1 - frac))
    ins = [e for e in entries if e["i"] < split]
    oos = [e for e in entries if e["i"] >= split]
    return ins, oos


# ----------------------------------------------------------------------------
# criterio di accettazione
# ----------------------------------------------------------------------------
def verdict(full: SimResult, oos: SimResult) -> bool:
    """Strategia attendibile su un singolo asset/tf."""
    if full.trades < 30:
        return False
    if full.ret <= 0 or oos.ret <= 0:
        return False
    if full.pos_years < max(full.n_years - 1, 1):
        return False
    if full.dd > 45:
        return False
    return True