Adriano/PythagorasGoal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
719509fb2f27e9ded09b1157d1570497a3dc5315
PythagorasGoal/scripts/analysis/shape_candle_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

329 lines
14 KiB
Python
Raw Blame History

"""Edge nella FORMA discreta delle candele -> distribuzione condizionale dell'esito.
Famiglia: encoding DISCRETO della morfologia di una finestra di L candele
(sequenza UP/DOWN/DOJI, opzionalmente arricchita con bucket di body-ratio e
shadow-ratio) -> codice intero. Per ogni codice si stima la distribuzione del
rendimento a H barre usando SOLO le occorrenze PASSATE il cui esito era gia'
realizzato prima della barra di decisione i (expanding window causale). Se un
codice mostra bias direzionale forte e statisticamente solido (n campioni >=
soglia, win-rate o |media| oltre soglia) si ENTRA a close[i] nella direzione del
bias; exit a H barre o TP/SL ATR.
VINCOLI ANTI-LOOK-AHEAD (l'errore squeeze e' nato qui):
- il codice a i usa SOLO open/high/low/close fino alla barra i inclusa;
- la statistica condizionale a i conta SOLO occorrenze del codice terminate in
e con e+H <= i-1 -> il loro esito H e' interamente noto PRIMA di i;
- direzione decisa dal CODICE (forma fino a close[i]) + STATISTICHE PASSATE,
ingresso eseguibile a close[i].
- check_no_lookahead() perturba il futuro: ne' il codice a i ne' le stat usate
devono cambiare.
Riusa l'engine netto-fee + OOS di explore_lab (simulate/evaluate/robust).
Implementazione causale O(N) per codice via accumulatori incrementali (niente
ricalcolo dell'intera storia ad ogni barra).
Asset: ADA BNB BTC DOGE ETH LTC SOL XRP (1h, 15m). Default 1h.
"""
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, OOS_FRAC,
)
# --------------------------- encoding discreto della forma ---------------------------
def candle_codes(df, L: int, body_buckets: int = 1, shadow_buckets: int = 1) -> np.ndarray:
"""Codice intero della forma per la finestra di L candele terminante in i.
Componenti per ogni candela:
- direzione UP/DOWN/DOJI (sempre): 3 stati.
- bucket del body-ratio |c-o|/(h-l) (se body_buckets>1): quantizzazione fissa
in body_buckets livelli (corpo piccolo/medio/grande...).
- bucket dello shadow-ratio (h-max(o,c)-(min(o,c)-l))/(h-l) in [-1,1]
(se shadow_buckets>1): ombra sup vs inf.
Quantizzazione a SOGLIE FISSE (non quantili): non dipende dal futuro ne' dal
dataset globale -> causale per costruzione. codes[i] dipende solo da
barre [i-L+1 .. i]. Per i < L-1 -> -1 (non valido).
"""
o = df["open"].values; c = df["close"].values
h = df["high"].values; l = df["low"].values
n = len(c)
rng = np.where((h - l) == 0, 1e-12, h - l)
body = np.abs(c - o) / rng # [0,1]
direction = np.where(body < 0.1, 0, # DOJI
np.where(c > o, 1, 2)) # UP=1, DOWN=2 (3 stati: 0,1,2)
# shadow asymmetry in [-1,1]: >0 ombra sup dominante, <0 ombra inf
up_sh = (h - np.maximum(o, c)) / rng
lo_sh = (np.minimum(o, c) - l) / rng
shadow = up_sh - lo_sh
# bucket body (soglie fisse su frazioni del range): 0..body_buckets-1
if body_buckets > 1:
edges_b = np.linspace(0.0, 1.0, body_buckets + 1)[1:-1]
bbk = np.digitize(body, edges_b) # 0..body_buckets-1
else:
bbk = np.zeros(n, dtype=int)
if shadow_buckets > 1:
edges_s = np.linspace(-1.0, 1.0, shadow_buckets + 1)[1:-1]
sbk = np.digitize(shadow, edges_s)
else:
sbk = np.zeros(n, dtype=int)
# simbolo per candela: dir * (body_buckets*shadow_buckets) + bbk*shadow_buckets + sbk
nbb, nsb = body_buckets, shadow_buckets
per_dir = nbb * nsb
sym = direction * per_dir + bbk * nsb + sbk # 0 .. 3*per_dir-1
base = 3 * per_dir
codes = np.full(n, -1, dtype=np.int64)
# codice della finestra L: base-L polinomiale sui simboli [i-L+1 .. i]
acc = np.zeros(n, dtype=np.int64)
for k in range(L):
# contributo della candela a posizione (i-L+1+k): peso base**(L-1-k)
shifted = np.full(n, 0, dtype=np.int64)
shifted[L - 1 - k:] = sym[: n - (L - 1 - k)] if (L - 1 - k) > 0 else sym
acc += shifted * (base ** (L - 1 - k))
codes[L - 1:] = acc[L - 1:]
return codes
def fwd_return(close: np.ndarray, H: int) -> np.ndarray:
out = np.full(len(close), np.nan)
out[: len(close) - H] = (close[H:] - close[:-H]) / close[:-H]
return out
# --------------------------- stima condizionale causale ---------------------------
def shape_entries(df, L=3, H=12, body_buckets=1, shadow_buckets=1,
min_n=30, edge=0.55, min_lib=500,
tp_atr=None, sl_atr=None, fade=False) -> list[dict]:
"""Entries dal bias condizionale del codice di forma (causale, no look-ahead).
L: lunghezza finestra-forma. H: orizzonte = max_bars.
body_buckets/shadow_buckets: granularita' dell'encoding (1 = solo direzione).
min_n: occorrenze passate minime del codice (con esito noto) per fidarsi.
edge: win-rate minimo (frazione di esiti concordi col segno della media) per
entrare; |edge-0.5| e' il margine direzionale.
min_lib: barre minime di storia prima di iniziare a operare.
tp_atr/sl_atr: TP/SL in multipli di ATR (None = solo time-limit H).
Causalita': a barra di decisione i si aggiorna lo stato del codice della
finestra terminata in e = i-1-H (il cui esito fr[e] e' ora noto). Le statistiche
usate per decidere a i derivano quindi solo da occorrenze con e+H <= i-1.
"""
close = df["close"].values
n = len(close)
a = atr(df, 14)
codes = candle_codes(df, L, body_buckets, shadow_buckets)
fr = fwd_return(close, H)
# accumulatori per codice: somma rendimenti, n positivi, n totali
from collections import defaultdict
cnt = defaultdict(int)
pos = defaultdict(int)
ssum = defaultdict(float)
entries: list[dict] = []
for i in range(min_lib, n - 1):
# aggiorna lo stato col codice la cui finestra termina in e = i-1-H
e = i - 1 - H
if e >= L - 1:
ce = codes[e]
re = fr[e]
if ce >= 0 and not np.isnan(re):
cnt[ce] += 1
pos[ce] += 1 if re > 0 else 0
ssum[ce] += re
ci = codes[i]
if ci < 0:
continue
ntot = cnt.get(ci, 0)
if ntot < min_n:
continue
mean = ssum[ci] / ntot
wr_up = pos[ci] / ntot # frazione esiti positivi nel passato
d = 1 if mean > 0 else -1
# win-rate nella direzione scelta
wr = wr_up if d == 1 else (1.0 - wr_up)
if wr < edge:
continue
if fade:
d = -d # FADE: entra contro il bias storico
ent = {"i": i, "d": d, "max_bars": H}
if tp_atr is not None and a[i] > 0:
ent["tp"] = close[i] + d * tp_atr * a[i]
if sl_atr is not None and a[i] > 0:
ent["sl"] = close[i] - d * sl_atr * a[i]
entries.append(ent)
return entries
# --------------------------- verifica no look-ahead ---------------------------
def check_no_lookahead(df, L=3, H=12, body_buckets=2, shadow_buckets=2) -> bool:
"""Perturbare il FUTURO (>i) non cambia (a) il codice a i, (b) le stat usate a i.
(a) codes[i] dipende solo da barre <= i.
(b) le entries fino a i (incluse) non cambiano se stravolgo le barre > i+1.
"""
n = len(df)
i = n // 2
codes0 = candle_codes(df, L, body_buckets, shadow_buckets)
df2 = df.copy()
fut = slice(i + 1, n)
for col in ("open", "high", "low", "close"):
df2.loc[df2.index[i + 1:], col] = df2[col].values[i + 1:] * 1.5
codes1 = candle_codes(df2, L, body_buckets, shadow_buckets)
ok_code = bool(codes0[i] == codes1[i] and np.array_equal(codes0[: i + 1], codes1[: i + 1]))
# entries: confronta quelle con indice <= i-1-H (decise con stat tutte note prima del futuro)
e0 = shape_entries(df, L=L, H=H, body_buckets=body_buckets, shadow_buckets=shadow_buckets,
min_n=5, edge=0.50, min_lib=200)
e1 = shape_entries(df2, L=L, H=H, body_buckets=body_buckets, shadow_buckets=shadow_buckets,
min_n=5, edge=0.50, min_lib=200)
cutoff = i - 1 - H
s0 = {(x["i"], x["d"]) for x in e0 if x["i"] <= cutoff}
s1 = {(x["i"], x["d"]) for x in e1 if x["i"] <= cutoff}
ok_ent = (s0 == s1)
print(f" no-lookahead codice a i={i}: {'OK' if ok_code else 'VIOLATO'}; "
f"entries<=i-1-H invarianti: {'OK' if ok_ent else 'VIOLATO'} "
f"({len(s0)} vs {len(s1)})")
return ok_code and ok_ent
# --------------------------- run riproducibile ---------------------------
def predictive_power(df, L=3, H=12, body_buckets=1, shadow_buckets=1, min_n=30, min_lib=500):
"""Diagnostica ONESTA: la direzione predetta dal bias storico (causale) anticipa
il segno del rendimento realizzato? Misura hit-rate aggregato della predizione
(segno media passata del codice) vs realizzato, su tutte le barre operabili.
Niente fee: pura capacita' predittiva del codice di forma."""
close = df["close"].values
n = len(close)
codes = candle_codes(df, L, body_buckets, shadow_buckets)
fr = fwd_return(close, H)
from collections import defaultdict
cnt = defaultdict(int); pos = defaultdict(int); ssum = defaultdict(float)
hits = tot = 0
pred_ret = 0.0
for i in range(min_lib, n - 1):
e = i - 1 - H
if e >= L - 1:
ce = codes[e]; re = fr[e]
if ce >= 0 and not np.isnan(re):
cnt[ce] += 1; pos[ce] += 1 if re > 0 else 0; ssum[ce] += re
ci = codes[i]
if ci < 0 or cnt.get(ci, 0) < min_n or np.isnan(fr[i]):
continue
d = 1 if ssum[ci] / cnt[ci] > 0 else -1
actual = fr[i]
hits += (np.sign(actual) == d); tot += 1
pred_ret += d * actual # PnL teorico senza fee
hr = hits / tot * 100 if tot else 0.0
print(f" predittivita' L{L}H{H} b{body_buckets}s{shadow_buckets}: "
f"hit={hr:.2f}% su {tot} (50%=rumore) | PnL_grezzo_noFee={pred_ret*100:+.0f}%")
return hr
def run():
print("=" * 100)
print(" SHAPE_CANDLE_RESEARCH — encoding discreto forma -> bias condizionale | netto fee, OOS")
print("=" * 100)
df_btc = get_df("BTC", "1h")
print("\n[causalita']")
check_no_lookahead(df_btc, L=3, H=12, body_buckets=2, shadow_buckets=2)
# ----- sweep base BTC/ETH 1h: solo direzione (body=shadow=1) -----
print("\n[BTC/ETH 1h] solo direzione UP/DOWN/DOJI (body=1,shadow=1), time-exit a H:")
for asset in ("BTC", "ETH"):
df = get_df(asset, "1h")
print(f" -- {asset} 1h --")
for L in (2, 3, 4):
for H in (6, 12, 24):
ents = shape_entries(df, L=L, H=H, min_n=30, edge=0.55)
evaluate(f"dir L{L}H{H}", ents, df)
# ----- encoding arricchito body+shadow -----
print("\n[BTC/ETH 1h] encoding arricchito (body=2,shadow=2), time-exit a H:")
for asset in ("BTC", "ETH"):
df = get_df(asset, "1h")
print(f" -- {asset} 1h --")
for L in (2, 3):
for H in (6, 12, 24):
ents = shape_entries(df, L=L, H=H, body_buckets=2, shadow_buckets=2,
min_n=30, edge=0.55)
evaluate(f"rich L{L}H{H} b2s2", ents, df)
# ----- selettivita': soglie edge piu' alte, meno trade -----
print("\n[BTC/ETH 1h] selettivo (edge>=0.58, min_n>=50), dir-only:")
for asset in ("BTC", "ETH"):
df = get_df(asset, "1h")
print(f" -- {asset} 1h --")
for L in (3, 4, 5):
for H in (12, 24):
ents = shape_entries(df, L=L, H=H, min_n=50, edge=0.58)
evaluate(f"sel L{L}H{H} e58", ents, df)
# ----- con TP/SL ATR (gestione rischio) sui candidati piu' attivi -----
print("\n[BTC/ETH 1h] con TP/SL ATR (tp=2,sl=1.5), dir-only L3:")
for asset in ("BTC", "ETH"):
df = get_df(asset, "1h")
for H in (12, 24):
ents = shape_entries(df, L=3, H=H, min_n=30, edge=0.55, tp_atr=2.0, sl_atr=1.5)
evaluate(f"{asset} tpsl L3H{H}", ents, df)
# ----- DIAGNOSTICA: il codice di forma ha QUALSIASI potere predittivo? -----
print("\n[DIAGNOSTICA] hit-rate predizione (segno bias storico vs realizzato), senza fee:")
for asset in ("BTC", "ETH"):
df = get_df(asset, "1h")
print(f" -- {asset} 1h --")
for L in (2, 3, 4):
for H in (6, 12, 24):
predictive_power(df, L=L, H=H, min_n=30)
for L in (2, 3):
predictive_power(df, L=L, H=12, body_buckets=2, shadow_buckets=2, min_n=30)
# ----- IPOTESI FADE: il bias e' anti-predittivo (mean-reversion)? -----
print("\n[FADE del bias] entra CONTRO il bias storico (dir-only):")
for asset in ("BTC", "ETH"):
df = get_df(asset, "1h")
print(f" -- {asset} 1h --")
for L in (2, 3):
for H in (6, 12, 24):
ents = shape_entries(df, L=L, H=H, min_n=30, edge=0.55, fade=True)
evaluate(f"FADE L{L}H{H}", ents, df)
def run_extended(configs):
"""Valuta config candidate su tutti gli asset 1h+15m e stampa robustezza."""
assets = ["ADA", "BNB", "BTC", "DOGE", "ETH", "LTC", "SOL", "XRP"]
for cfg in configs:
print(f"\n[ESTESO] config {cfg}")
for tf in ("1h", "15m"):
print(f" -- timeframe {tf} --")
nrob = 0
for asset in assets:
try:
df = get_df(asset, tf)
except Exception as ex:
print(f" {asset}: skip ({ex})")
continue
ents = shape_entries(df, **cfg)
res = evaluate(f"{asset} {tf}", ents, df)
nrob += robust(res)
print(f" -> robuste {nrob}/{len(assets)} su {tf}")
if __name__ == "__main__":
run()
Reference in New Issue View Git Blame Copy Permalink