Adriano/PythagorasGoal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
8abeb7a83f6d2d13735a1a22c85451070e270cbe
PythagorasGoal/scripts/analysis/shape_template_research.py
T
Adriano 4ac87ab385 research(shape): 5 famiglie di pattern-forma su harness onesto 
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>
2026-05-29 12:09:28 +02:00

444 lines
18 KiB
Python
Raw Blame History

"""SHAPE_TEMPLATE_RESEARCH — edge nella FORMA del prezzo: distanze alternative e template canonici.
Due filoni, sull'harness ONESTO condiviso (shape_lab + explore_lab), netto-fee e OOS:
1. ANALOG con distanza di FORMA alternativa (DTW warping-invariant, correlazione/coseno)
confrontata HEAD-TO-HEAD con l'euclidea a PARITA' di selettivita' (stessa libreria,
stesso K, stessa soglia di accordo). DTW e' O(W^2): si usa una libreria SOTTOCAMPIONATA
(uno start ogni `step` barre) + W ridotto + banda di Sakoe-Chiba.
2. TEMPLATE di forma canonici (doppio top/bottom, testa-spalle, V-reversal, salita/discesa
lineare, U). A ogni i misuro la similarita' (correlazione di Pearson sulla finestra
z-normalizzata) fra forma recente e ogni template; se supera soglia, entro a close[i]
nella DIREZIONE ATTESA del template stimata SOLO sul passato (esito medio causale delle
occorrenze gia' concluse di quel template), exit H barre o tp/sl ATR.
VINCOLI anti-look-ahead (verificati esplicitamente):
- la forma/match a i usa SOLO close fino a i (z-norm causale);
- la direzione attesa di ogni template e la libreria analog usano SOLO occorrenze il cui
esito a H barre e' gia' realizzato PRIMA di i (end + H <= i-1);
- ingresso eseguibile a close[i]; exit TP/SL intrabar o time-limit H.
Netto fee 0.10% RT baseline + sweep fino a 0.20%. Leva 3x, pos 0.15. OOS ultimo 30%.
Run riproducibile: uv run python scripts/analysis/shape_template_research.py
DTW e' costoso: usa run_in_background per gli sweep larghi (vedi --sweep).
"""
from __future__ import annotations
import sys
import time
from pathlib import Path
import numpy as np
from numpy.lib.stride_tricks import sliding_window_view
PROJECT_ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(PROJECT_ROOT))
from scripts.analysis.explore_lab import get_df, evaluate, robust, simulate, atr, ema, OOS_FRAC # noqa: E402
from scripts.analysis.shape_lab import znorm_windows, fwd_return # noqa: E402
RNG_SEED = 7
SUBC_ASSETS = ["BTC", "ETH", "SOL"]
# =========================================================================================
# DISTANZE DI FORMA
# =========================================================================================
def _euclid(q: np.ndarray, lib: np.ndarray) -> np.ndarray:
"""Distanza euclidea fra q (W,) e ogni riga di lib (M,W). Forme gia' z-normalizzate."""
return np.sqrt(((lib - q) ** 2).sum(axis=1))
def _corr_dist(q: np.ndarray, lib: np.ndarray) -> np.ndarray:
"""Distanza = 1 - correlazione di Pearson (q,lib gia' z-norm: corr = q.lib / W)."""
# forme z-norm hanno media 0 std 1 -> dot/W e' la correlazione di Pearson
corr = (lib @ q) / q.shape[0]
return 1.0 - corr
def _cosine_dist(q: np.ndarray, lib: np.ndarray) -> np.ndarray:
"""Distanza = 1 - coseno fra q e ogni riga di lib."""
qn = q / (np.linalg.norm(q) + 1e-12)
ln = lib / (np.linalg.norm(lib, axis=1, keepdims=True) + 1e-12)
return 1.0 - (ln @ qn)
def _dtw_one(a: np.ndarray, b: np.ndarray, band: int) -> float:
"""DTW 1D con banda di Sakoe-Chiba (|i-j|<=band). a,b stessa lunghezza W."""
n = len(a)
INF = 1e18
prev = np.full(n + 1, INF)
prev[0] = 0.0
for i in range(1, n + 1):
cur = np.full(n + 1, INF)
jlo = max(1, i - band)
jhi = min(n, i + band)
ai = a[i - 1]
for j in range(jlo, jhi + 1):
cost = abs(ai - b[j - 1])
m = prev[j]
if prev[j - 1] < m:
m = prev[j - 1]
if cur[j - 1] < m:
m = cur[j - 1]
cur[j] = cost + m
prev = cur
return float(prev[n])
def _dtw_dist(q: np.ndarray, lib: np.ndarray, band: int) -> np.ndarray:
"""DTW di q contro ogni riga di lib. O(M * W * band)."""
out = np.empty(lib.shape[0])
for k in range(lib.shape[0]):
out[k] = _dtw_one(q, lib[k], band)
return out
DIST_FUNCS = {"euclid": _euclid, "corr": _corr_dist, "cosine": _cosine_dist}
# =========================================================================================
# FILONE 1 — ANALOG con distanza configurabile (libreria sottocampionata, causale)
# =========================================================================================
def analog_dist_entries(df, dist="euclid", W=24, H=12, K=40, step=5, rebuild=500,
min_lib=2000, agree=0.62, dtw_band=4, dtw_prefilter=200,
decide_step=1, tp_atr=None, sl_atr=None,
trend_max=None, ema_long=200) -> list[dict]:
"""Analog kNN sulla FORMA con metrica `dist` ('euclid'|'corr'|'cosine'|'dtw').
Libreria SOTTOCAMPIONATA: si considerano solo finestre che terminano a indici
multipli di `step` (riduce N e rende DTW trattabile). Causalita': la libreria a
decisione i contiene solo finestre con end<=i-1-H (esito gia' realizzato).
Ricostruita ogni `rebuild` barre. Stessa firma per tutte le metriche -> confronto
head-to-head a parita' di selettivita' (stesso W,H,K,agree).
DTW (costoso, O(W*band) per coppia in Python): si PREFILTRA con la correlazione ai
`dtw_prefilter` candidati piu' simili, poi si fa DTW-rerank solo su quelli (approccio
standard lower-bound/rerank). `decide_step`>1 valuta una barra ogni decide_step (non
cambia la causalita', riduce solo il numero di query DTW).
"""
close = df["close"].values
n = len(close)
a = atr(df, 14)
M, ends = znorm_windows(close, W)
end_pos = {int(e): k for k, e in enumerate(ends)}
fr = fwd_return(close, H)
el = ema(close, ema_long) if trend_max is not None else None
# candidati di libreria: solo end multipli di step (sottocampionamento causale fisso)
base_ends = ends[(ends % step == 0)]
entries: list[dict] = []
lib_M = None
lib_idx = None
next_rebuild = 0
for i in range(min_lib, n - 1):
if i % decide_step != 0:
continue
if lib_M is None or i >= next_rebuild:
elig = base_ends[(base_ends <= i - 1 - H) & (base_ends >= W - 1)]
elig = elig[~np.isnan(fr[elig])]
if len(elig) < max(K * 3, 200):
next_rebuild = i + rebuild
continue
lib_M = M[[end_pos[int(e)] for e in elig]]
lib_idx = elig
next_rebuild = i + rebuild
if lib_M is None:
continue
q = M[end_pos[i]]
if not np.isfinite(q).all():
continue
if trend_max is not None and a[i] > 0 and abs(close[i] - el[i]) / a[i] > trend_max:
continue
if dist == "dtw":
# prefiltro corr (cheap, vettoriale) -> DTW-rerank solo sui top dtw_prefilter
pre = _corr_dist(q, lib_M)
npre = min(dtw_prefilter, len(lib_idx))
cand = np.argpartition(pre, npre - 1)[:npre]
dd_cand = _dtw_dist(q, lib_M[cand], dtw_band)
kk = min(K, len(cand))
sub = np.argpartition(dd_cand, kk - 1)[:kk]
nn = cand[sub]
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))
if frac < agree:
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)
continue
dd = DIST_FUNCS[dist](q, lib_M)
kk = min(K, len(lib_idx))
nn = np.argpartition(dd, kk - 1)[:kk]
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))
if frac < agree:
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 2 — TEMPLATE di forma canonici
# =========================================================================================
def make_templates(W: int) -> dict[str, np.ndarray]:
"""Template parametrici z-normalizzati di lunghezza W (forma pura, no scala/livello).
Sono solo descrittori di FORMA recente (gli ultimi W close). La direzione attesa NON
e' decisa a priori: viene stimata causalmente sul passato (vedi template_entries).
"""
t = np.linspace(0, 1, W)
s = 0.012 # ampiezza gaussiana scalata sulla finestra (W-indipendente in t in [0,1])
g = lambda c: np.exp(-((t - c) ** 2) / s)
raw = {
# estremi di reversione a DOPPIO picco (due massimi / minimi simmetrici)
"double_top": g(0.25) + g(0.75), # M: due cime
"double_bottom": -(g(0.25) + g(0.75)), # W: due fondi
# testa-spalle: spalla-testa-spalla (centro piu' alto)
"head_shoulders": g(0.2) + 1.7 * g(0.5) + g(0.8),
"inv_head_shoulders": -(g(0.2) + 1.7 * g(0.5) + g(0.8)),
# singola reversione
"v_bottom": np.abs(t - 0.5),
"inv_v_top": -np.abs(t - 0.5),
"u_bottom": (t - 0.5) ** 2,
"arch_top": -((t - 0.5) ** 2),
# trend lineari
"ramp_up": t,
"ramp_down": -t,
}
out = {}
for k, v in raw.items():
v = np.asarray(v, dtype=float)
sd = v.std()
out[k] = (v - v.mean()) / (sd if sd > 0 else 1.0)
return out
def template_entries(df, W=24, H=12, corr_min=0.85, dir_min=0.10, min_lib=2000,
rebuild=300, tp_atr=None, sl_atr=None, trend_max=None, ema_long=200,
templates=None) -> list[dict]:
"""Entries da match con template canonici, DIREZIONE stimata SOLO sul passato.
A ogni i, per ogni template calcolo la correlazione di Pearson fra la forma recente
z-norm (close[i-W+1..i]) e il template. Prendo il template a correlazione massima; se
>= corr_min lo considero "attivo". La DIREZIONE in cui entrare e' il segno del rendimento
forward MEDIO storico delle occorrenze gia' concluse (end+H<=i-1) di quel template
(stesso criterio di match), purche' |media| in barre-equivalenti superi dir_min*media_atr-ish
-> qui dir_min e' una soglia sulla |media forward| relativa (frazione). NIENTE direzione a
priori: se il passato non e' coerente (occorrenze<min o segno debole) si salta.
"""
close = df["close"].values
n = len(close)
a = atr(df, 14)
M, ends = znorm_windows(close, W)
end_pos = {int(e): k for k, e in enumerate(ends)}
fr = fwd_return(close, H)
el = ema(close, ema_long) if trend_max is not None else None
tps = templates if templates is not None else make_templates(W)
names = list(tps.keys())
T = np.stack([tps[k] for k in names]) # (NT, W), gia' z-norm
# match-history: per ogni end di libreria, quale template e con che corr
# (precalcolo causale: per ogni end, corr con ogni template)
# corr Pearson fra forme z-norm = dot/W
lib_ends = ends[ends >= W - 1]
lib_M = M[[end_pos[int(e)] for e in lib_ends]] # (L, W)
corr_mat = (lib_M @ T.T) / W # (L, NT)
best_tpl = np.argmax(corr_mat, axis=1)
best_corr = corr_mat[np.arange(len(lib_ends)), best_tpl]
lib_fr = fr[lib_ends]
lib_end_arr = lib_ends
entries: list[dict] = []
# cache direzione per template, ricostruita ogni rebuild barre
dir_cache: dict[int, int] = {}
next_rebuild = 0
for i in range(min_lib, n - 1):
q = M[end_pos[i]]
if not np.isfinite(q).all():
continue
cq = (T @ q) / W # corr con ogni template
bt = int(np.argmax(cq))
if cq[bt] < corr_min:
continue
if trend_max is not None and a[i] > 0 and abs(close[i] - el[i]) / a[i] > trend_max:
continue
# direzione attesa: media forward causale delle occorrenze concluse dello stesso template
if i >= next_rebuild:
dir_cache = {}
next_rebuild = i + rebuild
if bt not in dir_cache:
mask = (lib_end_arr <= i - 1 - H) & (best_tpl == bt) & (best_corr >= corr_min) & (~np.isnan(lib_fr))
outs = lib_fr[mask]
if len(outs) < 30:
dir_cache[bt] = 0
else:
m = float(outs.mean())
# soglia: |media| forward deve superare dir_min volte la std forward (edge vs rumore)
sd = float(outs.std()) + 1e-12
dir_cache[bt] = (1 if m > 0 else -1) if abs(m) / sd >= dir_min else 0
d = dir_cache[bt]
if d == 0:
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
# =========================================================================================
# CHECK CAUSALITA' espliciti
# =========================================================================================
def check_causality_analog(df, **kw) -> bool:
"""Le entries non devono cambiare se perturbo il FUTURO oltre l'ultima barra usata.
Tronco il df a una certa lunghezza L e verifico che le entries con i<L-H-1 siano
identiche a quelle calcolate sul df completo (la coda futura non le tocca)."""
L = int(len(df) * 0.55)
H = kw.get("H", 12)
full = analog_dist_entries(df, **kw)
trunc = analog_dist_entries(df.iloc[:L].reset_index(drop=True), **kw)
horizon = L - H - 2
f = {e["i"]: e["d"] for e in full if e["i"] < horizon}
t = {e["i"]: e["d"] for e in trunc if e["i"] < horizon}
ok = (f == t)
print(f" causalita' analog ({kw.get('dist','euclid')}): {'OK' if ok else 'VIOLATO'} "
f"({len(f)} entries confrontate <{horizon})")
return ok
def check_causality_template(df, **kw) -> bool:
L = int(len(df) * 0.55)
H = kw.get("H", 12)
full = template_entries(df, **kw)
trunc = template_entries(df.iloc[:L].reset_index(drop=True), **kw)
horizon = L - H - 2
f = {e["i"]: e["d"] for e in full if e["i"] < horizon}
t = {e["i"]: e["d"] for e in trunc if e["i"] < horizon}
ok = (f == t)
print(f" causalita' template: {'OK' if ok else 'VIOLATO'} "
f"({len(f)} entries confrontate <{horizon})")
return ok
# =========================================================================================
# RUN
# =========================================================================================
def run_head_to_head(assets=SUBC_ASSETS, W=16, H=12, K=40, step=6, agree=0.62,
decide_step=4, dtw_prefilter=120):
"""Confronto HEAD-TO-HEAD delle metriche di forma a PARITA' di selettivita'.
Tutte le metriche valutano le STESSE barre-decisione (decide_step) con lo STESSO
W/H/K/agree: l'unica variabile e' la distanza. decide_step>1 serve a rendere DTW
trattabile (pura Python ~9ms/query); applicato a tutte per equita'.
"""
print("=" * 100)
print(f" FILONE 1 — ANALOG head-to-head metriche (W{W} H{H} K{K} step{step} "
f"agree{agree} decide_step{decide_step}) | netto fee, OOS")
print("=" * 100)
results = {}
for asset in assets:
df = get_df(asset, "1h")
print(f"\n --- {asset} 1h (n={len(df)}) ---", flush=True)
for dist in ["euclid", "corr", "cosine", "dtw"]:
t0 = time.time()
ents = analog_dist_entries(df, dist=dist, W=W, H=H, K=K, step=step, agree=agree,
dtw_band=max(2, W // 5), dtw_prefilter=dtw_prefilter,
decide_step=decide_step)
dt = time.time() - t0
res = evaluate(f"{dist:<7s}", ents, df)
results[(asset, dist)] = res
print(f" ^ time={dt:>5.1f}s robust={'YES' if robust(res) else 'no '}", flush=True)
return results
def run_templates(assets=SUBC_ASSETS, W=20, H=12, corr_min=0.85, dir_min=0.10):
print("=" * 100)
print(f" FILONE 2 — TEMPLATE canonici (W{W} H{H} corr>={corr_min} dir>={dir_min}) | netto fee, OOS")
print("=" * 100)
results = {}
for asset in assets:
df = get_df(asset, "1h")
print(f"\n --- {asset} 1h (n={len(df)}) ---")
for cm in [0.80, 0.85, 0.90]:
ents = template_entries(df, W=W, H=H, corr_min=cm, dir_min=dir_min)
res = evaluate(f"corr_min={cm}", ents, df)
results[(asset, cm)] = res
print(f" ^ robust={'YES' if robust(res) else 'no '}")
return results
def run_sweep():
"""Sweep largo (lento per via di DTW). Usa run_in_background."""
print("=" * 100)
print(" SWEEP LARGO — analog griglia W/H/K/step x metriche + template griglia")
print("=" * 100)
for W in [16, 20, 28]:
for H in [8, 12, 24]:
print(f"\n##### W={W} H={H} #####")
run_head_to_head(W=W, H=H, K=40, step=6, agree=0.62)
for W in [16, 20, 28]:
for H in [8, 12, 24]:
print(f"\n##### TEMPLATE W={W} H={H} #####")
run_templates(W=W, H=H, corr_min=0.85, dir_min=0.10)
def run():
np.random.seed(RNG_SEED)
print("#" * 100)
print(" SHAPE_TEMPLATE_RESEARCH — distanze di forma alternative + template canonici")
print("#" * 100)
# 1) check causalita' espliciti
print("\n[CAUSALITA']")
dfb = get_df("BTC", "1h")
check_causality_analog(dfb, dist="euclid", W=20, H=12, K=40, step=6, min_lib=2000)
check_causality_analog(dfb, dist="dtw", W=16, H=12, K=40, step=8, min_lib=2000,
dtw_band=3, decide_step=20)
check_causality_template(dfb, W=20, H=12, corr_min=0.85)
# 2) head-to-head metriche
print()
run_head_to_head()
# 3) template
print()
run_templates()
if __name__ == "__main__":
if "--sweep" in sys.argv:
run_sweep()
elif "--templates" in sys.argv:
run_templates()
elif "--h2h" in sys.argv:
run_head_to_head()
else:
run()
Reference in New Issue View Git Blame Copy Permalink