Adriano/PythagorasGoal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

research(shape): 5 famiglie di pattern-forma su harness onesto

Browse Source 
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>
This commit is contained in:
AdrianoDev
2026-05-29 12:09:28 +02:00
parent b9e5176a5b
commit 4ac87ab385
6 changed files with 1965 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/analysis/shape_pivot_research.py
+333
View File
@@ -0,0 +1,333 @@
"""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()