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>
178 lines
8.4 KiB
Python
178 lines
8.4 KiB
Python
"""Ricerca sistematica edge nella FORMA (analog forecasting / kNN) — netto fee, OOS.
|
|
|
|
Obiettivo: trovare una config di analog forecasting ROBUSTA, cioe' positiva
|
|
FULL+OOS, che regge fee 0.20% RT e ha quasi tutti gli anni positivi, su >=2 asset.
|
|
Si combatte la "morte per fee" della baseline (BTC1h W24H12K50 agree0.60:
|
|
FULL +112%/OOS +48% Sharpe 1.38 ma a 0.2% RT -> FULL -72 / OOS -18, win 49.5%,
|
|
esposizione 73.9%, 4531 trade) con SELETTIVITA':
|
|
- agree alto (0.70..0.90) -> entra solo con analoghi molto concordi
|
|
- conf_atr > 0 -> richiede |rendimento medio analoghi| >= conf_atr*ATR
|
|
- trend_max/ema_long -> salta forme in trend estremo
|
|
- tp_atr/sl_atr -> exit intrabar invece che solo a tempo
|
|
|
|
Tutto causale: la forma usa solo close<=i, la libreria analoghi termina < i-H.
|
|
Per performance, il forecast kNN grezzo per barra si calcola UNA volta per
|
|
(W,H,K,rebuild) con analog_signals(); i filtri (agree/conf/trend/tp/sl) sono
|
|
applicati a valle con entries_from_signals() (cheap, risultato identico ad
|
|
analog_entries — verificato). Engine netto-fee + OOS da explore_lab.
|
|
|
|
Uso:
|
|
uv run python scripts/analysis/shape_analog_research.py
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[2]
|
|
sys.path.insert(0, str(PROJECT_ROOT))
|
|
|
|
from scripts.analysis.shape_lab import ( # noqa: E402
|
|
analog_signals, entries_from_signals, check_no_lookahead,
|
|
)
|
|
from scripts.analysis.explore_lab import get_df, evaluate, robust # noqa: E402
|
|
|
|
ROBUSTE: list[tuple] = []
|
|
MIN_TRADES = 100 # un edge "robusto" su <100 trade e' rumore campionario, non edge
|
|
|
|
|
|
def _hdr(s: str) -> None:
|
|
print("\n" + "=" * 100, flush=True)
|
|
print(" " + s, flush=True)
|
|
print("=" * 100, flush=True)
|
|
|
|
|
|
def _eval(df, sig, asset, tf, tag, **filt):
|
|
ents = entries_from_signals(df, sig, **filt)
|
|
res = evaluate(f"[{asset} {tf}] {tag}", ents, df)
|
|
# robusto E con campione sufficiente (un edge su <100 trade non e' affidabile)
|
|
if robust(res) and res["full"]["trades"] >= MIN_TRADES:
|
|
print(f" ^^^ ROBUSTA ({asset} {tf}): {tag} filt={filt}", flush=True)
|
|
ROBUSTE.append((asset, tf, tag, dict(filt), res))
|
|
elif robust(res):
|
|
print(f" (robust ma trade={res['full']['trades']}<{MIN_TRADES}: campione "
|
|
f"insufficiente, ignorato)", flush=True)
|
|
return res
|
|
|
|
|
|
def run():
|
|
# --- 0) sanity no-lookahead ---------------------------------------------
|
|
_hdr("0) SANITY no-lookahead (forma causale)")
|
|
df_btc = get_df("BTC", "1h")
|
|
check_no_lookahead(df_btc, W=24, H=12)
|
|
|
|
# sig base W24H12K50 (riusato per selettivita' agree/conf/tp/sl/trend)
|
|
sig0 = analog_signals(df_btc, W=24, H=12, K=50, rebuild=250)
|
|
|
|
# --- 1) selettivita' via agree ------------------------------------------
|
|
_hdr("1) BTC 1h — selettivita' agree (W24 H12 K50, time-exit)")
|
|
for ag in (0.60, 0.70, 0.80, 0.90):
|
|
_eval(df_btc, sig0, "BTC", "1h", f"agree{ag}", agree=ag)
|
|
|
|
# --- 2) conf_atr (forza segnale) ----------------------------------------
|
|
_hdr("2) BTC 1h — conf_atr (W24 H12 K50 agree0.70)")
|
|
for ca in (0.0, 0.25, 0.5, 1.0, 1.5):
|
|
_eval(df_btc, sig0, "BTC", "1h", f"ag0.70 conf{ca}", agree=0.70, conf_atr=ca)
|
|
|
|
# --- 3) tp/sl intrabar ---------------------------------------------------
|
|
_hdr("3) BTC 1h — exit intrabar tp/sl (W24 H12 K50 agree0.70 conf0.5)")
|
|
for tp, sl in [(1.0, 1.0), (1.5, 1.0), (2.0, 1.5), (1.5, 2.0), (3.0, 2.0)]:
|
|
_eval(df_btc, sig0, "BTC", "1h", f"tp{tp}sl{sl}",
|
|
agree=0.70, conf_atr=0.5, tp_atr=tp, sl_atr=sl)
|
|
|
|
# --- 4) filtro trend -----------------------------------------------------
|
|
_hdr("4) BTC 1h — filtro trend_max (W24 H12 K50 agree0.70 conf0.5)")
|
|
for tm in (None, 2.0, 3.0, 4.0):
|
|
_eval(df_btc, sig0, "BTC", "1h", f"trend_max{tm}",
|
|
agree=0.70, conf_atr=0.5, trend_max=tm, ema_long=200)
|
|
|
|
# --- 5) griglia W/H/K (agree0.80, time-exit) plateau ---------------------
|
|
# Griglia focalizzata: con agree0.80 e H>=24 i trade -> ~0 (vedi sez.1), e W>=24
|
|
# porta OOS negativo; il segnale vive su W piccolo, H breve. Testo il plateau
|
|
# attorno a quella regione + una banda di controllo (W24/48) per confermare il bordo.
|
|
_hdr("5) BTC 1h — griglia W/H/K (agree0.80, time-exit) — plateau check")
|
|
for W in (12, 24, 48):
|
|
for H in (6, 12, 24):
|
|
for K in (30, 50, 80):
|
|
sig = analog_signals(df_btc, W=W, H=H, K=K, rebuild=250)
|
|
_eval(df_btc, sig, "BTC", "1h", f"W{W}H{H}K{K}", agree=0.80)
|
|
|
|
# --- 6) rebuild sensitivity ---------------------------------------------
|
|
_hdr("6) BTC 1h — rebuild 250 vs 500 (W24 H12 K80 agree0.80)")
|
|
for rb in (250, 500):
|
|
sig = analog_signals(df_btc, W=24, H=12, K=80, rebuild=rb)
|
|
_eval(df_btc, sig, "BTC", "1h", f"rebuild{rb}", agree=0.80)
|
|
|
|
# --- 7) cross-asset 1h: candidati selettivi -----------------------------
|
|
_hdr("7) cross-asset 1h — candidati selettivi (>=2 robusti richiesto)")
|
|
# (build_kw: per analog_signals) (filt: per entries_from_signals)
|
|
# Su BTC 1h le uniche regioni con OOS positivo che regge fee0.2% sono W piccolo,
|
|
# H breve, K basso (W12H12K30: FULL+88/OOS+36, fee0.2% +69/+32, 243 trade, 8/9 anni;
|
|
# W12H6K30: +35/+11, fee0.2% +20/+7). conf0.25 con W24H12 e' il miglior in-sample
|
|
# ma OOS@fee~0. Verifico questi candidati cross-asset (>=2 robusti richiesto).
|
|
candidates = [
|
|
("C1 W12H12K30 ag.80", dict(W=12, H=12, K=30), dict(agree=0.80)),
|
|
("C2 W12H6K30 ag.80", dict(W=12, H=6, K=30), dict(agree=0.80)),
|
|
("C3 W12H12K30 ag.70", dict(W=12, H=12, K=30), dict(agree=0.70)),
|
|
("C4 W24H12K50 ag.70 conf.25", dict(W=24, H=12, K=50), dict(agree=0.70, conf_atr=0.25)),
|
|
("C5 W12H12K30 ag.80 trend3", dict(W=12, H=12, K=30), dict(agree=0.80, trend_max=3.0, ema_long=200)),
|
|
("C6 W12H6K50 ag.70", dict(W=12, H=6, K=50), dict(agree=0.70)),
|
|
]
|
|
per_cand: dict[str, int] = {}
|
|
for asset in ("BTC", "ETH", "ADA", "LTC", "SOL", "XRP"):
|
|
try:
|
|
df = get_df(asset, "1h")
|
|
except Exception as ex:
|
|
print(f" [{asset} 1h] SKIP load: {ex}", flush=True)
|
|
continue
|
|
# cache analog_signals per ogni build_kw distinto su questo asset
|
|
sig_cache: dict[tuple, dict] = {}
|
|
for tag, bkw, filt in candidates:
|
|
key = tuple(sorted(bkw.items()))
|
|
if key not in sig_cache:
|
|
sig_cache[key] = analog_signals(df, rebuild=250, **bkw)
|
|
res = _eval(df, sig_cache[key], asset, "1h", tag, **filt)
|
|
if robust(res):
|
|
per_cand[tag] = per_cand.get(tag, 0) + 1
|
|
|
|
# --- 8) verifica 15m dei candidati robusti su >=2 asset 1h --------------
|
|
_hdr("8) verifica 15m dei candidati robusti su >=2 asset 1h")
|
|
good = [t for t, c in per_cand.items() if c >= 2]
|
|
if not good:
|
|
print(" Nessun candidato robusto su >=2 asset 1h -> niente verifica 15m.", flush=True)
|
|
else:
|
|
for tag in good:
|
|
_, bkw, filt = next(c for c in candidates if c[0] == tag)
|
|
for asset in ("BTC", "ETH"):
|
|
try:
|
|
df = get_df(asset, "15m")
|
|
except Exception as ex:
|
|
print(f" [{asset} 15m] SKIP load: {ex}", flush=True)
|
|
continue
|
|
sig = analog_signals(df, rebuild=250, **bkw)
|
|
_eval(df, sig, asset, "15m", f"{tag} (15m)", **filt)
|
|
|
|
# --- VERDETTO ------------------------------------------------------------
|
|
_hdr("VERDETTO")
|
|
if ROBUSTE:
|
|
agg: dict[str, list] = {}
|
|
for asset, tf, tag, filt, res in ROBUSTE:
|
|
agg.setdefault(tag, []).append(f"{asset}/{tf}")
|
|
print(f" {len(ROBUSTE)} sleeve robusti (FULL+OOS+ fee0.2% + anniPos):", flush=True)
|
|
edge = False
|
|
for tag, asl in agg.items():
|
|
n_assets = len({a.split('/')[0] for a in asl})
|
|
mark = " *** EDGE (>=2 asset)" if n_assets >= 2 else " (1 asset: non sufficiente)"
|
|
if n_assets >= 2:
|
|
edge = True
|
|
print(f" - {tag}: {asl}{mark}", flush=True)
|
|
if not edge:
|
|
print("\n CONCLUSIONE: nessuna config robusta su >=2 asset -> RUMORE.", flush=True)
|
|
else:
|
|
print(" NESSUNA config robusta. Famiglia analog/forma = RUMORE sotto fee reali.", flush=True)
|
|
return ROBUSTE
|
|
|
|
|
|
if __name__ == "__main__":
|
|
run()
|