Adriano
4ac87ab385
Harness shape_lab (analog kNN causale, no look-ahead verificato) + 5 ricerche parallele. 4/5 famiglie = RUMORE (confermano dominanza mean-reversion): - analog kNN forma grezza: solo BTC-overfit, non robusto >=2 asset - encoding candele UP/DOWN/DOJI + body/shadow: hit-rate ~50%, muore a fee - DTW + template geometrici: DTW peggiora euclidea; template overfit - PIP/pivot/zig-zag: 0/48 config robuste 1/5 = EDGE REALE: ML walk-forward (LogisticRegression) sulle feature di forma. BTC logit W24H12 th0.58: FULL +219% / OOS +42% / Sharpe 2.72 / 8-9 anni+ / regge fee 0.20% RT (+60/+26). Causalita' verificata. Da validare a fondo. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
334 lines
14 KiB
Python
334 lines
14 KiB
Python
"""Famiglia SHAPE-PIVOT: geometria a punti di svolta (PIP / pivot) -> bias futuro.
|
|
|
|
Idea (causale, no look-ahead):
|
|
- a ogni barra i comprimo la finestra di L barre terminante a close[i] nei suoi
|
|
P punti percettivamente importanti (PIP, Perceptually Important Points: i punti
|
|
di massima deviazione dalla retta congiungente — Fu et al.);
|
|
- la sequenza di P punti e' una POLILINEA = forma geometrica grezza;
|
|
- la classifico con feature interpretabili e CAUSALI:
|
|
* trend dei pivot interni: higher-highs/higher-lows (HH/HL) vs lower-* (LH/LL);
|
|
* convergenza/divergenza delle pendenze (triangoli/cunei);
|
|
* distanza % di close[i] dall'ultimo pivot alto/basso (vicino a R / a S);
|
|
* pendenza dell'ultimo segmento (slancio recente);
|
|
- per ogni CLASSE geometrica stimo l'esito medio a H barre usando SOLO occorrenze
|
|
passate il cui esito era gia' realizzato prima di i (statistica causale rolling);
|
|
- entro a close[i] nella direzione del bias di classe se l'edge passato e' netto;
|
|
exit a H barre o TP/SL in ATR.
|
|
|
|
VINCOLI (CLAUDE.md "metodologia obbligatoria" + "lezione squeeze look-ahead"):
|
|
- PIP/pivot calcolati SOLO su close[i-L+1 .. i]; nessun pivot "confermato dal futuro".
|
|
- ogni statistica per-classe usa solo campioni con esito (entry+H) <= i-1.
|
|
- ingresso eseguibile a close[i]; netto fee (0.10% RT base, sweep a 0.20%); leva 3x,
|
|
pos 0.15; validazione OOS (ultimo 30%) + robustezza griglia + >=2 asset.
|
|
- check di causalita' esplicito (perturbo il futuro: la forma a i non cambia).
|
|
|
|
Riusa l'engine netto-fee + OOS di explore_lab (simulate/evaluate/robust).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
|
|
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, _dt, OOS_FRAC,
|
|
)
|
|
|
|
|
|
# =========================================================================
|
|
# PIP — Perceptually Important Points (causale, solo su close[a..b])
|
|
# =========================================================================
|
|
def pip_indices(seg: np.ndarray, p: int) -> list[int]:
|
|
"""Estrae p indici PIP dalla serie `seg` (inclusi i 2 estremi).
|
|
|
|
Algoritmo Fu et al.: parti dai 2 estremi; aggiungi iterativamente il punto a
|
|
massima distanza VERTICALE dalla retta che unisce i due PIP adiacenti, finche'
|
|
non hai p punti. Tutto sul segmento dato -> nessun look-ahead se seg=close[..i].
|
|
"""
|
|
n = len(seg)
|
|
if p >= n:
|
|
return list(range(n))
|
|
pts = [0, n - 1]
|
|
while len(pts) < p:
|
|
best_d, best_k = -1.0, -1
|
|
for s in range(len(pts) - 1):
|
|
l, r = pts[s], pts[s + 1]
|
|
if r - l < 2:
|
|
continue
|
|
x1, y1 = l, seg[l]
|
|
x2, y2 = r, seg[r]
|
|
dx = x2 - x1
|
|
# distanza verticale dalla retta (interpolazione lineare in x)
|
|
for k in range(l + 1, r):
|
|
if dx == 0:
|
|
dist = abs(seg[k] - y1)
|
|
else:
|
|
yline = y1 + (y2 - y1) * (k - x1) / dx
|
|
dist = abs(seg[k] - yline)
|
|
if dist > best_d:
|
|
best_d, best_k = dist, k
|
|
if best_k < 0:
|
|
break
|
|
# inserisci mantenendo l'ordine
|
|
for s in range(len(pts) - 1):
|
|
if pts[s] < best_k < pts[s + 1]:
|
|
pts.insert(s + 1, best_k)
|
|
break
|
|
return pts
|
|
|
|
|
|
# =========================================================================
|
|
# Classe geometrica della polilinea PIP (feature causali interpretabili)
|
|
# =========================================================================
|
|
def shape_class(seg: np.ndarray, p: int) -> tuple | None:
|
|
"""Ritorna una tupla-classe discreta della forma PIP di `seg`, o None se degenere.
|
|
|
|
Feature (tutte da seg=close[..i], causali):
|
|
- dir_seq: per ogni pivot interno, segno della variazione vs precedente
|
|
(sequenza su/giu) -> cattura HH/HL vs LH/LL e zig-zag;
|
|
- conv: convergenza pendenze inizio vs fine (triangolo/cuneo): segno di
|
|
(|slope_last| - |slope_first|) discretizzato;
|
|
- loc: posizione di close[i] nel range della finestra (vicino a max=resistenza,
|
|
vicino a min=supporto), in 3 bucket.
|
|
La classe e' invariante a livello/scala (z-norm implicito su forma).
|
|
"""
|
|
idx = pip_indices(seg, p)
|
|
if len(idx) < 3:
|
|
return None
|
|
y = seg[idx]
|
|
rng = y.max() - y.min()
|
|
if rng <= 0:
|
|
return None
|
|
yn = (y - y.min()) / rng # forma normalizzata 0..1
|
|
# sequenza direzioni dei segmenti (su=1 / giu=0)
|
|
diffs = np.diff(yn)
|
|
dir_seq = tuple(int(x > 0) for x in diffs)
|
|
# convergenza: pendenza primo vs ultimo segmento
|
|
s_first = abs(diffs[0])
|
|
s_last = abs(diffs[-1])
|
|
if s_last > s_first * 1.3:
|
|
conv = 1 # divergente (slancio finale)
|
|
elif s_last < s_first * 0.77:
|
|
conv = -1 # convergente (compressione, triangolo/cuneo)
|
|
else:
|
|
conv = 0
|
|
# posizione di close[i] (=ultimo punto) nel range: 0..1 in 3 bucket
|
|
last = yn[-1]
|
|
loc = 0 if last < 0.33 else (2 if last > 0.67 else 1)
|
|
return (dir_seq, conv, loc)
|
|
|
|
|
|
# =========================================================================
|
|
# Strategia: bias per-classe stimato CAUSALMENTE (rolling, esito realizzato)
|
|
# =========================================================================
|
|
def pivot_entries(df, L=48, P=5, H=12, min_lib=1000, min_samples=20,
|
|
edge=0.0, tp_atr=None, sl_atr=None,
|
|
trend_max=None, ema_long=200, mode="bias") -> list[dict]:
|
|
"""Entries dalla geometria PIP con bias di classe causale.
|
|
|
|
L: lunghezza finestra-forma. P: n. punti PIP. H: orizzonte (=max_bars).
|
|
min_lib: barre minime prima di operare. min_samples: campioni minimi per fidarsi
|
|
della statistica di una classe. edge: |rendimento medio classe| minimo
|
|
(frazione, es. 0.002 = 0.2%) per entrare. mode:
|
|
- "bias": entra nel verso del rendimento medio passato della classe (momentum
|
|
della forma: la classe X storicamente -> su/giu);
|
|
- "fade": entra nel verso OPPOSTO (test mean-reversion della forma).
|
|
Statistica per-classe accumulata SOLO con esiti realizzati < i (causale stretta).
|
|
"""
|
|
close = df["close"].values
|
|
high = df["high"].values
|
|
low = df["low"].values
|
|
n = len(close)
|
|
a = atr(df, 14)
|
|
el = ema(close, ema_long) if trend_max is not None else None
|
|
|
|
# stato rolling per classe: somma rendimenti e conteggio (solo esiti < i)
|
|
cls_sum: dict[tuple, float] = {}
|
|
cls_cnt: dict[tuple, int] = {}
|
|
# coda di campioni la cui forma e' stata calcolata ma esito non ancora maturo
|
|
# pending[t] = (classe, indice_entry t) -> matura quando t+H <= i-1
|
|
pending: list[tuple] = [] # (mature_at, cls, t)
|
|
pend_ptr = 0
|
|
|
|
entries: list[dict] = []
|
|
|
|
for i in range(min_lib, n - 1):
|
|
# 1) integra nello storico tutti i campioni il cui esito e' realizzato (< i)
|
|
# un campione formato a t matura quando t+H <= i-1 => mature_at = t+H+1 <= i
|
|
while pend_ptr < len(pending) and pending[pend_ptr][0] <= i:
|
|
_, cls_p, t = pending[pend_ptr]
|
|
ret_real = (close[t + H] - close[t]) / close[t]
|
|
cls_sum[cls_p] = cls_sum.get(cls_p, 0.0) + ret_real
|
|
cls_cnt[cls_p] = cls_cnt.get(cls_p, 0) + 1
|
|
pend_ptr += 1
|
|
|
|
# 2) forma corrente (solo close fino a i)
|
|
seg = close[i - L + 1: i + 1]
|
|
cls = shape_class(seg, P)
|
|
if cls is None:
|
|
continue
|
|
# registra il campione corrente come pending (esito da realizzare in futuro)
|
|
pending.append((i + H + 1, cls, i))
|
|
|
|
# 3) decisione con statistica PASSATA della classe
|
|
cnt = cls_cnt.get(cls, 0)
|
|
if cnt < min_samples:
|
|
continue
|
|
mean_ret = cls_sum[cls] / cnt
|
|
if abs(mean_ret) < edge:
|
|
continue
|
|
d = 1 if mean_ret > 0 else -1
|
|
if mode == "fade":
|
|
d = -d
|
|
# filtro trend opzionale
|
|
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
|
|
|
|
|
|
# =========================================================================
|
|
# Filone (c): distanza da supporto/resistenza locale (ultimo pivot alto/basso)
|
|
# =========================================================================
|
|
def sr_entries(df, L=48, P=7, H=12, near=0.5, mode="fade",
|
|
tp_atr=None, sl_atr=None, trend_max=None, ema_long=200) -> list[dict]:
|
|
"""Filone (c): close[i] vicino all'ultimo pivot alto (R) o basso (S) della forma.
|
|
|
|
Usa i PIP per individuare l'ultimo massimo/minimo locale (resistenza/supporto) e
|
|
misura la distanza % di close[i]. Se close e' entro `near`*ATR da R -> bias short
|
|
(mode='fade': rimbalzo da R) o long (mode='break': rottura). Simmetrico per S.
|
|
Tutto causale: PIP su close[..i], decisione a close[i].
|
|
"""
|
|
close = df["close"].values
|
|
n = len(close)
|
|
a = atr(df, 14)
|
|
el = ema(close, ema_long) if trend_max is not None else None
|
|
entries: list[dict] = []
|
|
|
|
for i in range(L, n - 1):
|
|
seg = close[i - L + 1: i + 1]
|
|
idx = pip_indices(seg, P)
|
|
if len(idx) < 3 or a[i] <= 0:
|
|
continue
|
|
y = seg[idx]
|
|
# pivot interni (escludi i 2 estremi e l'ultimo punto = close[i])
|
|
inner = y[1:-1]
|
|
if len(inner) == 0:
|
|
continue
|
|
res = inner.max() # resistenza locale
|
|
sup = inner.min() # supporto locale
|
|
cur = close[i]
|
|
dist_r = (res - cur) / a[i]
|
|
dist_s = (cur - sup) / a[i]
|
|
d = None
|
|
if 0 <= dist_r <= near: # appena sotto R
|
|
d = -1 if mode == "fade" else 1
|
|
elif 0 <= dist_s <= near: # appena sopra S
|
|
d = 1 if mode == "fade" else -1
|
|
if d is None:
|
|
continue
|
|
if trend_max is not None and abs(cur - el[i]) / a[i] > trend_max:
|
|
continue
|
|
e = {"i": i, "d": d, "max_bars": H}
|
|
if tp_atr is not None:
|
|
e["tp"] = cur + d * tp_atr * a[i]
|
|
if sl_atr is not None:
|
|
e["sl"] = cur - d * sl_atr * a[i]
|
|
entries.append(e)
|
|
return entries
|
|
|
|
|
|
# =========================================================================
|
|
# Check causalita' esplicito
|
|
# =========================================================================
|
|
def check_no_lookahead(df, L=48, P=5) -> bool:
|
|
"""La classe-forma a i non deve cambiare se perturbo il FUTURO (>i)."""
|
|
close = df["close"].values.copy()
|
|
i = len(close) // 2
|
|
seg0 = close[i - L + 1: i + 1].copy()
|
|
c0 = shape_class(seg0, P)
|
|
close2 = close.copy()
|
|
close2[i + 1:] *= 1.7 # stravolge il futuro
|
|
seg1 = close2[i - L + 1: i + 1]
|
|
c1 = shape_class(seg1, P)
|
|
ok = (c0 == c1)
|
|
print(f" no-lookahead classe-forma a i={i}: {'OK' if ok else 'VIOLATO'} "
|
|
f"(c0={c0} c1={c1})")
|
|
# check su PIP indices
|
|
p0 = pip_indices(seg0, P)
|
|
p1 = pip_indices(seg1, P)
|
|
ok2 = (p0 == p1)
|
|
print(f" no-lookahead indici PIP: {'OK' if ok2 else 'VIOLATO'}")
|
|
return ok and ok2
|
|
|
|
|
|
# =========================================================================
|
|
# run() riproducibile
|
|
# =========================================================================
|
|
def run():
|
|
print("=" * 100)
|
|
print(" SHAPE-PIVOT RESEARCH — geometria PIP/pivot -> bias futuro | netto fee, OOS")
|
|
print("=" * 100)
|
|
|
|
df_btc = get_df("BTC", "1h")
|
|
print("\n[CAUSALITA']")
|
|
check_no_lookahead(df_btc, L=48, P=5)
|
|
|
|
assets = ["BTC", "ETH", "SOL", "ADA"]
|
|
dfs = {a: get_df(a, "1h") for a in assets}
|
|
|
|
# ---- A) bias di classe PIP (momentum della forma) ----
|
|
print("\n[A] BIAS di classe PIP (entra nel verso del rendimento medio passato della classe)")
|
|
print(" sweep L/P/H, edge=0.002, min_samples=25, time-exit a H")
|
|
A_grid = [(48, 5, 12), (48, 5, 24), (72, 6, 24), (36, 5, 12), (96, 7, 24), (48, 7, 12)]
|
|
for L, P, H in A_grid:
|
|
print(f" -- L{L} P{P} H{H} --")
|
|
for a in assets:
|
|
ents = pivot_entries(dfs[a], L=L, P=P, H=H, edge=0.002, min_samples=25, mode="bias")
|
|
evaluate(f"{a} bias L{L}P{P}H{H}", ents, dfs[a])
|
|
|
|
# ---- B) fade di classe PIP (mean-reversion della forma) ----
|
|
print("\n[B] FADE di classe PIP (entra opposto al bias storico -> test mean-reversion)")
|
|
for L, P, H in A_grid:
|
|
print(f" -- L{L} P{P} H{H} --")
|
|
for a in assets:
|
|
ents = pivot_entries(dfs[a], L=L, P=P, H=H, edge=0.002, min_samples=25, mode="fade")
|
|
evaluate(f"{a} fade L{L}P{P}H{H}", ents, dfs[a])
|
|
|
|
# ---- C) supporto/resistenza locale dai pivot ----
|
|
print("\n[C] S/R locale dai PIP — FADE (rimbalzo da R/S) vs BREAK (rottura)")
|
|
for mode in ("fade", "break"):
|
|
for near in (0.5, 1.0):
|
|
print(f" -- mode={mode} near={near} ATR, TP/SL 1.5/1.5 ATR, H=12 --")
|
|
for a in assets:
|
|
ents = sr_entries(dfs[a], L=48, P=7, H=12, near=near, mode=mode,
|
|
tp_atr=1.5, sl_atr=1.5)
|
|
evaluate(f"{a} SR-{mode} near{near}", ents, dfs[a])
|
|
|
|
# ---- D) miglior candidato con TP/SL ATR + filtro trend (se A o B mostra segnali) ----
|
|
print("\n[D] FADE di classe con TP/SL ATR (2.0/1.5) + filtro trend 3.0, L48 P5 H24")
|
|
for a in assets:
|
|
ents = pivot_entries(dfs[a], L=48, P=5, H=24, edge=0.002, min_samples=25,
|
|
mode="fade", tp_atr=2.0, sl_atr=1.5, trend_max=3.0)
|
|
res = evaluate(f"{a} fadeTPSL L48P5H24", ents, dfs[a])
|
|
if robust(res):
|
|
print(f" ^^^ {a} ROBUSTO")
|
|
|
|
print("\n" + "=" * 100)
|
|
print(" Verdetto: cerca righe con FULL>0 E OOS>0 E fee0.2% OOS>0 su >=2 asset.")
|
|
print("=" * 100)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
run()
|