"""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)