PythagorasGoal/scripts/analysis/shape_lab.py

"""Harness ONESTO per pattern *di forma* -> previsione dell'andamento successivo.

Idea (analog forecasting / nearest-neighbour sulla FORMA del prezzo):
  - a ogni barra i guardo la forma recente W (closes z-normalizzati fino a close[i]);
  - cerco nel PASSATO le K finestre piu' simili la cui forma si era gia' conclusa
    *e* il cui esito a H barre era gia' noto PRIMA di i (nessun look-ahead);
  - prevedo la direzione dei prossimi H barre = segno del rendimento medio degli
    analoghi; entro a close[i] se l'accordo fra analoghi e' abbastanza forte.

Vincoli anti-look-ahead (gli stessi della famiglia squeeze fallita):
  - la forma usa SOLO closes fino a close[i];
  - la libreria di analoghi a decisione i contiene solo finestre che terminano in
    e con e+H <= i-1 -> il loro esito e' interamente realizzato *prima* della barra i;
  - ingresso eseguibile a close[i]; exit TP/SL intrabar o time-limit H.

Riusa l'engine netto-fee + OOS di explore_lab (simulate/evaluate/robust).
KDTree ricostruito ogni `rebuild` barre (causale): query O(log N), niente O(N^2).

Asset: 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
from numpy.lib.stride_tricks import sliding_window_view
from scipy.spatial import cKDTree

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

from scripts.analysis.explore_lab import (  # noqa: E402
    get_df, evaluate, robust, simulate, atr, ema, rsi, _dt, OOS_FRAC, ASSETS, FEE_RT,
)


# --------------------------- forma normalizzata ---------------------------
def znorm_windows(close: np.ndarray, W: int) -> tuple[np.ndarray, np.ndarray]:
    """Matrice delle finestre z-normalizzate per FORMA.

    Ritorna (M, ends) dove M[k] = z-norm(close[e-W+1 .. e]) e ends[k] = e.
    Z-norm per forma: (w - media)/std -> invariante a livello e scala -> confronto
    sulla sola morfologia. Le finestre piatte (std=0) hanno norm tutta a 0.
    """
    if len(close) < W:
        return np.empty((0, W)), np.empty(0, dtype=int)
    wins = sliding_window_view(close, W)            # (N-W+1, W), wins[k] = close[k..k+W-1]
    mu = wins.mean(axis=1, keepdims=True)
    sd = wins.std(axis=1, keepdims=True)
    sd = np.where(sd == 0, 1.0, sd)
    M = (wins - mu) / sd
    ends = np.arange(W - 1, len(close))             # finestra k termina in e = k+W-1
    return M, ends


def fwd_return(close: np.ndarray, H: int) -> np.ndarray:
    """Rendimento forward a H barre per ogni indice: (close[i+H]-close[i])/close[i].
    NaN dove i+H esce dai dati (non usabile come esito)."""
    out = np.full(len(close), np.nan)
    out[: len(close) - H] = (close[H:] - close[:-H]) / close[:-H]
    return out


# --------------------------- analog forecasting causale ---------------------------
def analog_entries(df, W=24, H=12, K=50, rebuild=250, min_lib=800,
                   agree=0.60, conf_atr=0.0, tp_atr=None, sl_atr=None,
                   trend_max=None, ema_long=200) -> list[dict]:
    """Entries da nearest-neighbour sulla FORMA (causale, no look-ahead).

    W: lunghezza finestra-forma. H: orizzonte previsione (= max_bars). K: n. analoghi.
    rebuild: ogni quante barre si ricostruisce il KDTree (libreria cresce nel tempo).
    min_lib: barre minime di storia prima di iniziare a operare.
    agree: frazione minima di analoghi concordi sul segno per entrare (>0.5).
    conf_atr: soglia |rendimento medio analoghi| in multipli di ATR-equivalente (0=off).
    tp_atr/sl_atr: take-profit/stop in multipli di ATR (None = solo time-limit H).
    trend_max: salta se |close-EMA(ema_long)|/ATR14 > trend_max (filtro trend, None=off).
    """
    close = df["close"].values
    n = len(close)
    a = atr(df, 14)
    M, ends = znorm_windows(close, W)               # forme z-norm e indice di fine
    end_pos = {int(e): k for k, e in enumerate(ends)}
    fr = fwd_return(close, H)                        # esito H-barre per ogni indice
    el = None
    if trend_max is not None:
        el = ema(close, ema_long)

    entries: list[dict] = []
    tree = None
    lib_idx = None                                  # indici e (fine finestra) nella libreria
    next_rebuild = 0

    for i in range(min_lib, n - 1):
        # libreria causale: finestre la cui forma E il cui esito H sono < i
        if tree is None or i >= next_rebuild:
            eligible = ends[(ends <= i - 1 - H) & (ends >= W - 1)]
            # esito noto e finito (fr non-NaN garantito da e+H <= i-1 < n)
            eligible = eligible[~np.isnan(fr[eligible])]
            if len(eligible) < max(K * 3, 200):
                next_rebuild = i + rebuild
                continue
            tree = cKDTree(M[[end_pos[int(e)] for e in eligible]])
            lib_idx = eligible
            next_rebuild = i + rebuild

        if tree is None:
            continue
        q = M[end_pos[i]]
        if not np.isfinite(q).all():
            continue
        kk = min(K, len(lib_idx))
        _, nn = tree.query(q, k=kk)
        nn = np.atleast_1d(nn)
        outs = fr[lib_idx[nn]]                       # rendimenti H-barre degli analoghi
        outs = outs[~np.isnan(outs)]
        if len(outs) < 5:
            continue
        mean_out = float(outs.mean())
        d = 1 if mean_out > 0 else -1
        frac = float(np.mean(np.sign(outs) == d))
        if frac < agree:
            continue
        if conf_atr > 0:
            if not (abs(mean_out) * close[i] >= conf_atr * a[i]):
                continue
        if trend_max is not None and a[i] > 0:
            if abs(close[i] - el[i]) / a[i] > trend_max:
                continue
        e = {"i": i, "d": d, "max_bars": H}
        if tp_atr is not None and a[i] > 0:
            e["tp"] = close[i] + d * tp_atr * a[i]
        if sl_atr is not None and a[i] > 0:
            e["sl"] = close[i] - d * sl_atr * a[i]
        entries.append(e)
    return entries


# --------------------------- kNN grezzo cacheable (perf) ---------------------------
def analog_signals(df, W=24, H=12, K=50, rebuild=250, min_lib=800) -> dict:
    """Calcola UNA volta il forecast kNN grezzo per barra (causale), riusabile da
    piu' filtri (agree/conf_atr/trend/tp/sl) senza ri-eseguire la query costosa.

    Ritorna dict con array allineati per le barre che hanno un forecast valido:
      i        : indice barra (ingresso eseguibile a close[i])
      mean_out : rendimento H-barre medio degli analoghi
      frac     : frazione di analoghi concordi col segno di mean_out (>=0.5)
      d        : segno previsto (+1/-1)
    Identico, riga per riga, alla logica di analog_entries (stessa libreria causale,
    stessa query, stessa soglia len(outs)>=5) ma SENZA i filtri di selezione.
    """
    close = df["close"].values
    n = len(close)
    M, ends = znorm_windows(close, W)
    end_pos = {int(e): k for k, e in enumerate(ends)}
    fr = fwd_return(close, H)

    out_i: list[int] = []
    out_mean: list[float] = []
    out_frac: list[float] = []
    out_d: list[int] = []
    tree = None
    lib_idx = None
    next_rebuild = 0

    for i in range(min_lib, n - 1):
        if tree is None or i >= next_rebuild:
            eligible = ends[(ends <= i - 1 - H) & (ends >= W - 1)]
            eligible = eligible[~np.isnan(fr[eligible])]
            if len(eligible) < max(K * 3, 200):
                next_rebuild = i + rebuild
                continue
            tree = cKDTree(M[[end_pos[int(e)] for e in eligible]])
            lib_idx = eligible
            next_rebuild = i + rebuild
        if tree is None:
            continue
        q = M[end_pos[i]]
        if not np.isfinite(q).all():
            continue
        kk = min(K, len(lib_idx))
        _, nn = tree.query(q, k=kk)
        nn = np.atleast_1d(nn)
        outs = fr[lib_idx[nn]]
        outs = outs[~np.isnan(outs)]
        if len(outs) < 5:
            continue
        mean_out = float(outs.mean())
        d = 1 if mean_out > 0 else -1
        frac = float(np.mean(np.sign(outs) == d))
        out_i.append(i); out_mean.append(mean_out); out_frac.append(frac); out_d.append(d)

    return {
        "i": np.asarray(out_i, dtype=int),
        "mean_out": np.asarray(out_mean, dtype=float),
        "frac": np.asarray(out_frac, dtype=float),
        "d": np.asarray(out_d, dtype=int),
        "H": H,
    }


def entries_from_signals(df, sig: dict, agree=0.60, conf_atr=0.0,
                         tp_atr=None, sl_atr=None, trend_max=None, ema_long=200) -> list[dict]:
    """Applica i filtri di selezione al forecast grezzo di analog_signals (cheap).
    Risultato identico ad analog_entries con gli stessi parametri (stesso W/H/K/rebuild
    usati per costruire sig)."""
    close = df["close"].values
    a = atr(df, 14)
    H = sig["H"]
    el = ema(close, ema_long) if trend_max is not None else None
    entries: list[dict] = []
    for k in range(len(sig["i"])):
        i = int(sig["i"][k]); d = int(sig["d"][k])
        if sig["frac"][k] < agree:
            continue
        if conf_atr > 0 and not (abs(sig["mean_out"][k]) * close[i] >= conf_atr * a[i]):
            continue
        if trend_max is not None and a[i] > 0 and abs(close[i] - el[i]) / a[i] > trend_max:
            continue
        e = {"i": i, "d": d, "max_bars": H}
        if tp_atr is not None and a[i] > 0:
            e["tp"] = close[i] + d * tp_atr * a[i]
        if sl_atr is not None and a[i] > 0:
            e["sl"] = close[i] - d * sl_atr * a[i]
        entries.append(e)
    return entries


# --------------------------- verifica no look-ahead ---------------------------
def check_no_lookahead(df, W=24, H=12) -> bool:
    """La forma a i deve restare invariata se perturbo il FUTURO (>i).
    Conferma che znorm_windows usa solo close fino a i."""
    close = df["close"].values.copy()
    M0, ends = znorm_windows(close, W)
    pos = {int(e): k for k, e in enumerate(ends)}
    i = len(close) // 2
    q0 = M0[pos[i]].copy()
    close2 = close.copy()
    close2[i + 1:] *= 1.5                            # stravolgo il futuro
    M1, _ = znorm_windows(close2, W)
    q1 = M1[pos[i]]
    ok = np.allclose(q0, q1)
    print(f"  no-lookahead forma a i={i}: {'OK' if ok else 'VIOLATO'} "
          f"(max diff {np.max(np.abs(q0 - q1)):.2e})")
    return ok


if __name__ == "__main__":
    print("=" * 92)
    print("  SHAPE_LAB — baseline analog forecasting (kNN sulla forma) | netto fee, OOS")
    print("=" * 92)
    df = get_df("BTC", "1h")
    check_no_lookahead(df)
    print("\n  BTC 1h — sweep base W/H/K (time-exit a H barre):")
    for W, H, K in [(24, 12, 50), (24, 24, 50), (48, 24, 80), (12, 6, 40), (48, 48, 100)]:
        ents = analog_entries(df, W=W, H=H, K=K, agree=0.60)
        evaluate(f"analog W{W}H{H}K{K}", ents, df)