Adriano Dal Pastro
14522262e6
Reset del progetto su fondamenta verificate dopo la scoperta che l'intera libreria "validata OOS" era artefatto di feed contaminato (print fantasma del feed Cerbero TESTNET + storico Binance/USDT). - Storico ricostruito da Deribit MAINNET (ccxt pubblico, tokenless) e CERTIFICATO (certify_feed.py): BTC/ETH puliti su TUTTA la storia (mediana 2-6 bps vs Coinbase USD), integrita' OHLC + coerenza resample (maxΔ 0.00) + cross-venue OK. Alt esclusi (illiquidi/divergenti: LTC/DOGE 50-82% barre flat; XRP/BNB non certificabili). - Verdetto sul feed pulito: FADE / PAIRS / XS01 / TSM01 morti (ogni portafoglio Sharpe -2.3..-3.0, DD ~40%); solo SH01 e frammenti HONEST con segnale residuo, da ri-validare in isolamento. - Cleanup "restart pulito": strategie, stack live (src/live, src/portfolio, runner/executor, yml, docker), ~100 script ricerca/gate, waste/games/ portfolios, dati non certificati + cache e 60+ diari -> archiviati in Old/ (preservati, non cancellati). Diario consolidato in un unico documento. - Skeleton ricerca tenuto: Strategy ABC + indicatori + src/fractal + src/backtest/engine + load_data; tool dati certificati (rebuild_history, certify_feed, audit_feed, multi_source_check). - Universo dati ATTIVO: solo BTC/ETH (5m/15m/1h); guardrail fisico (load_data su alt -> FileNotFoundError). Esecuzione DISABILITATA, conto flat. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
441 lines
19 KiB
Python
441 lines
19 KiB
Python
"""SHAPE-as-FEATURES research: l'edge e' nella FORMA del segnale?
|
|
|
|
Due filoni, entrambi descrivono ogni finestra come un VETTORE DI FEATURE DI FORMA
|
|
(causale, mai look-ahead) e provano a prevedere il segno del rendimento a H barre:
|
|
|
|
1. ANALOG nello spazio FEATURE (kNN causale). Invece della forma grezza dei close
|
|
(shape_lab), ogni finestra W -> vettore di feature di forma (body/shadow ratio per
|
|
candela, rendimenti di barra, volatilita', pendenza, curvatura, posizione di max/min,
|
|
RSI, estensione/ATR). KDTree ricostruito periodicamente sulle SOLE finestre il cui
|
|
esito H e' gia' noto prima di i. Previsione = segno del rendimento medio dei K vicini.
|
|
|
|
2. ML WALK-FORWARD sulla forma. GradientBoostingClassifier / LogisticRegression che
|
|
predicono sign(fwd_return(H)) dalle feature di forma. Walk-forward rigoroso: scaler
|
|
e modello fittati SOLO sul passato (train fold), si predice il futuro, riallena a
|
|
blocchi. Entra a close[i] solo se la probabilita' supera una soglia (selettivita').
|
|
|
|
Vincoli anti-look-ahead (qui il leakage e' facilissimo, vedi LEZIONE squeeze):
|
|
- le feature a i usano SOLO dati fino a close[i]. Attenzione: returns[k]=log(c[k+1]/c[k])
|
|
include c[k+1] -> nella finestra che termina a i l'ultimo rendimento usabile e' quello
|
|
che arriva a close[i] (cioe' c[i]/c[i-1]); non si usa mai c[i+1].
|
|
- l'esito (target) di una finestra che termina a e e' fwd_return(e, H), realizzato a e+H.
|
|
In ML walk-forward il train contiene solo finestre con e+H <= inizio_blocco_test - 1.
|
|
In kNN la libreria contiene solo finestre con e+H <= i-1.
|
|
- scaler/modello fittati SOLO sul train passato, MAI sull'intero dataset.
|
|
- ingresso eseguibile a close[i]; exit TP/SL intrabar o time-limit H (engine explore_lab).
|
|
- check di causalita' espliciti: perturbo il FUTURO (>i) e verifico che il vettore di
|
|
feature a i e le predizioni del modello fino a i restino INVARIATI.
|
|
|
|
Netto fee 0.10% RT baseline + sweep fino a 0.20% RT, leva 3x, pos 0.15, OOS ultimo 30%.
|
|
Robustezza su griglia + >=2 asset. Conta il PnL NETTO-fee, non l'accuracy.
|
|
|
|
Run: uv run python scripts/analysis/shape_ml_research.py
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
import time
|
|
import warnings
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
from numpy.lib.stride_tricks import sliding_window_view
|
|
from scipy.spatial import cKDTree
|
|
|
|
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, rsi, OOS_FRAC,
|
|
)
|
|
|
|
warnings.filterwarnings("ignore")
|
|
|
|
from sklearn.ensemble import GradientBoostingClassifier # noqa: E402
|
|
from sklearn.linear_model import LogisticRegression # noqa: E402
|
|
from sklearn.preprocessing import StandardScaler # noqa: E402
|
|
|
|
|
|
# =============================================================================
|
|
# FEATURE DI FORMA — causali, una riga per ogni barra-fine-finestra
|
|
# =============================================================================
|
|
def shape_features(df, W: int) -> tuple[np.ndarray, np.ndarray]:
|
|
"""Matrice di feature di FORMA per ogni finestra di W candele.
|
|
|
|
Ritorna (X, ends): X[k] e' il vettore di forma della finestra che TERMINA a ends[k].
|
|
Tutte le feature usano solo o/h/l/c[ends[k]-W+1 .. ends[k]] -> causali per costruzione.
|
|
|
|
Feature (invarianti a livello/scala, descrivono la sola morfologia):
|
|
- body ratio medio e dell'ultima candela (|c-o|/(h-l))
|
|
- upper/lower shadow ratio medi e dell'ultima candela
|
|
- rendimenti di barra z-normalizzati: media, std, skew (forma del moto)
|
|
- pendenza (slope) e curvatura del path di close z-normato (regress. lineare/quad.)
|
|
- posizione del max e del min nella finestra (0..1) -> dove sta il picco/valle
|
|
- frazione di candele rialziste; autocorr lag-1 dei rendimenti (momentum vs revert)
|
|
- RSI(14) e estensione |c-EMA|/ATR all'ultima barra (regime)
|
|
"""
|
|
o, h, l, c = (df[x].values.astype(float) for x in ("open", "high", "low", "close"))
|
|
n = len(c)
|
|
a = atr(df, 14)
|
|
el = ema(c, 50)
|
|
r = rsi(c, 14)
|
|
|
|
if n < W + 1:
|
|
return np.empty((0, 0)), np.empty(0, dtype=int)
|
|
|
|
# finestre OHLC che terminano a e = k+W-1, per k=0..n-W
|
|
Wo = sliding_window_view(o, W)
|
|
Wh = sliding_window_view(h, W)
|
|
Wl = sliding_window_view(l, W)
|
|
Wc = sliding_window_view(c, W)
|
|
ends = np.arange(W - 1, n)
|
|
|
|
total = Wh - Wl
|
|
total = np.where(total <= 0, 1e-12, total)
|
|
body = np.abs(Wc - Wo) / total
|
|
up_sh = (Wh - np.maximum(Wo, Wc)) / total
|
|
lo_sh = (np.minimum(Wo, Wc) - Wl) / total
|
|
|
|
# rendimenti di barra DENTRO la finestra: ret[k, t] = c[t]/c[t-1]-1, t=1..W-1
|
|
# usano solo close fino alla fine della finestra -> causali
|
|
ret = Wc[:, 1:] / np.where(Wc[:, :-1] == 0, 1e-12, Wc[:, :-1]) - 1.0
|
|
rmu = ret.mean(axis=1)
|
|
rsd = ret.std(axis=1) + 1e-12
|
|
rz = (ret - rmu[:, None]) / rsd[:, None]
|
|
rskew = (rz ** 3).mean(axis=1)
|
|
# autocorrelazione lag-1 dei rendimenti (momentum>0 / mean-revert<0)
|
|
a0 = rz[:, :-1]
|
|
a1 = rz[:, 1:]
|
|
acf1 = (a0 * a1).mean(axis=1)
|
|
|
|
# path z-normato dei close -> slope (lin) e curvatura (quad)
|
|
czmu = Wc.mean(axis=1, keepdims=True)
|
|
czsd = Wc.std(axis=1, keepdims=True)
|
|
czsd = np.where(czsd == 0, 1.0, czsd)
|
|
cz = (Wc - czmu) / czsd
|
|
t = np.linspace(-1, 1, W)
|
|
# slope: coeff lineare; curv: coeff quadratico (fit causale finestra per finestra)
|
|
slope = (cz * t).mean(axis=1) / (t * t).mean()
|
|
t2 = t * t
|
|
t2c = t2 - t2.mean()
|
|
curv = (cz * t2c).mean(axis=1) / (t2c * t2c).mean()
|
|
|
|
argmax = Wc.argmax(axis=1) / (W - 1)
|
|
argmin = Wc.argmin(axis=1) / (W - 1)
|
|
frac_up = (Wc > Wo).mean(axis=1)
|
|
|
|
rsi_end = r[ends]
|
|
aa = a[ends]
|
|
ext = np.where(aa > 0, (c[ends] - el[ends]) / np.where(aa > 0, aa, 1.0), 0.0)
|
|
|
|
X = np.column_stack([
|
|
body.mean(axis=1), body[:, -1],
|
|
up_sh.mean(axis=1), up_sh[:, -1],
|
|
lo_sh.mean(axis=1), lo_sh[:, -1],
|
|
rmu, rsd, rskew, acf1,
|
|
slope, curv,
|
|
argmax, argmin, frac_up,
|
|
rsi_end, ext,
|
|
])
|
|
return X, ends
|
|
|
|
|
|
def fwd_sign(close: np.ndarray, H: int) -> tuple[np.ndarray, np.ndarray]:
|
|
"""fwd_return a H barre e suo segno (+1/-1). NaN/0 dove i+H esce dai dati."""
|
|
fr = np.full(len(close), np.nan)
|
|
fr[: len(close) - H] = (close[H:] - close[:-H]) / close[:-H]
|
|
sgn = np.where(fr > 0, 1, -1).astype(float)
|
|
sgn[np.isnan(fr)] = np.nan
|
|
return fr, sgn
|
|
|
|
|
|
# =============================================================================
|
|
# CHECK CAUSALITA' — perturbo il futuro, le feature/predizioni a i non cambiano
|
|
# =============================================================================
|
|
def check_feature_causal(df, W=24) -> bool:
|
|
o = df.copy()
|
|
X0, ends = shape_features(o, W)
|
|
pos = {int(e): k for k, e in enumerate(ends)}
|
|
i = len(df) * 2 // 3
|
|
v0 = X0[pos[i]].copy()
|
|
o2 = df.copy()
|
|
for col in ("open", "high", "low", "close"):
|
|
o2.loc[i + 1:, col] = o2.loc[i + 1:, col] * 1.7 # stravolgi il futuro
|
|
X1, _ = shape_features(o2, W)
|
|
v1 = X1[pos[i]]
|
|
ok = np.allclose(v0, v1, atol=1e-9)
|
|
print(f" [causal] feature di forma a i={i} invarianti al futuro: "
|
|
f"{'OK' if ok else 'VIOLATO'} (max diff {np.nanmax(np.abs(v0 - v1)):.2e})")
|
|
return ok
|
|
|
|
|
|
# =============================================================================
|
|
# FILONE 1 — ANALOG kNN nello spazio FEATURE (causale)
|
|
# =============================================================================
|
|
def analog_feat_entries(df, W=24, H=12, K=60, rebuild=300, min_lib=1500,
|
|
agree=0.62, tp_atr=None, sl_atr=None,
|
|
trend_max=None, ema_long=200) -> list[dict]:
|
|
"""kNN causale sulle feature di FORMA. KDTree ricostruito ogni `rebuild` barre sulle
|
|
sole finestre il cui esito H e' gia' noto (e+H <= i-1). Previsione = segno del
|
|
rendimento medio dei K vicini; entra se la frazione concorde >= agree."""
|
|
c = df["close"].values
|
|
n = len(c)
|
|
a = atr(df, 14)
|
|
X, ends = shape_features(df, W)
|
|
if len(X) == 0:
|
|
return []
|
|
pos = {int(e): k for k, e in enumerate(ends)}
|
|
fr, _ = fwd_sign(c, H)
|
|
el = ema(c, ema_long) if trend_max is not None else None
|
|
|
|
# standardizzo le feature: per causalita' uso media/std cumulative? No: lo scaler
|
|
# globale userebbe il futuro. Uso uno scaler RICALCOLATO sulla libreria a ogni rebuild.
|
|
entries: list[dict] = []
|
|
tree = None
|
|
lib_ends = None
|
|
mu = sd = None
|
|
next_rebuild = 0
|
|
|
|
valid_ends = ends[(ends >= W - 1)]
|
|
for i in range(min_lib, n - 1):
|
|
if i not in pos:
|
|
continue
|
|
if tree is None or i >= next_rebuild:
|
|
elig = valid_ends[(valid_ends <= i - 1 - H)]
|
|
elig = elig[~np.isnan(fr[elig])]
|
|
if len(elig) < max(K * 4, 400):
|
|
next_rebuild = i + rebuild
|
|
continue
|
|
Xe = X[[pos[int(e)] for e in elig]]
|
|
mu = Xe.mean(axis=0)
|
|
sd = Xe.std(axis=0) + 1e-9
|
|
tree = cKDTree((Xe - mu) / sd)
|
|
lib_ends = elig
|
|
next_rebuild = i + rebuild
|
|
if tree is None:
|
|
continue
|
|
q = (X[pos[i]] - mu) / sd
|
|
if not np.isfinite(q).all():
|
|
continue
|
|
kk = min(K, len(lib_ends))
|
|
_, nn = tree.query(q, k=kk)
|
|
nn = np.atleast_1d(nn)
|
|
outs = fr[lib_ends[nn]]
|
|
outs = outs[~np.isnan(outs)]
|
|
if len(outs) < 10:
|
|
continue
|
|
d = 1 if outs.mean() > 0 else -1
|
|
frac = float(np.mean(np.sign(outs) == d))
|
|
if frac < agree:
|
|
continue
|
|
if trend_max is not None and a[i] > 0 and abs(c[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"] = c[i] + d * tp_atr * a[i]
|
|
if sl_atr is not None and a[i] > 0:
|
|
e["sl"] = c[i] - d * sl_atr * a[i]
|
|
entries.append(e)
|
|
return entries
|
|
|
|
|
|
# =============================================================================
|
|
# FILONE 2 — ML WALK-FORWARD sulla forma
|
|
# =============================================================================
|
|
def ml_wf_entries(df, W=24, H=12, model="gb", thresh=0.58,
|
|
train_min=4000, retrain=500, n_estimators=80, max_depth=3,
|
|
tp_atr=None, sl_atr=None, trend_max=None, ema_long=200,
|
|
train_window=None, last_block_only=False) -> list[dict]:
|
|
"""Walk-forward: a blocchi di `retrain` barre, allena sul passato il cui esito
|
|
e' noto, predice il blocco corrente. Scaler+modello fittati solo sul train.
|
|
Entra a close[i] se proba della classe predetta >= thresh. model in {gb, logit}.
|
|
train_window: se None -> expanding (tutto il passato); se int -> ROLLING (solo le
|
|
ultime train_window barre prima del blocco) -> test di robustezza piu' severo.
|
|
last_block_only: PATH LIVE (2026-06-07) — fitta e predice SOLO l'ultimo blocco
|
|
(un fit invece di ~n/retrain): al worker servono solo i segnali sulla barra
|
|
corrente e i confini dei blocchi sono deterministici (start + k*retrain), quindi
|
|
le entries dell'ultimo blocco sono IDENTICHE per costruzione al tail del
|
|
walk-forward completo (parity test in tests/portfolio/test_sh01_last_block.py).
|
|
Rende sostenibile il train EXPANDING full-history nel runner (~73k barre/tick)."""
|
|
c = df["close"].values
|
|
n = len(c)
|
|
a = atr(df, 14)
|
|
X, ends = shape_features(df, W)
|
|
if len(X) == 0:
|
|
return []
|
|
pos = {int(e): k for k, e in enumerate(ends)}
|
|
fr, sgn = fwd_sign(c, H)
|
|
el = ema(c, ema_long) if trend_max is not None else None
|
|
|
|
# mappa: per ogni indice i (>=W-1) la riga di feature
|
|
row_of = pos
|
|
entries: list[dict] = []
|
|
|
|
start = max(train_min, W - 1)
|
|
blk = start
|
|
while blk < n - 1:
|
|
blk_end = min(blk + retrain, n - 1)
|
|
if last_block_only and blk_end < n - 1:
|
|
blk = blk_end # salta i blocchi storici (stessi confini del WF completo)
|
|
continue
|
|
# TRAIN: finestre la cui forma E il cui esito (e+H) sono < blk
|
|
# cioe' e <= blk-1-H (esito realizzato prima del primo test del blocco)
|
|
lo_end = (blk - 1 - H - train_window) if train_window is not None else (W - 1)
|
|
tr_ends = ends[(ends <= blk - 1 - H) & (ends >= max(W - 1, lo_end))]
|
|
tr_ends = tr_ends[~np.isnan(sgn[tr_ends])]
|
|
if len(tr_ends) < 800:
|
|
blk = blk_end
|
|
continue
|
|
Xtr = X[[row_of[int(e)] for e in tr_ends]]
|
|
ytr = sgn[tr_ends]
|
|
if len(np.unique(ytr)) < 2:
|
|
blk = blk_end
|
|
continue
|
|
scaler = StandardScaler().fit(Xtr)
|
|
Xtr_s = scaler.transform(Xtr)
|
|
if model == "gb":
|
|
clf = GradientBoostingClassifier(
|
|
n_estimators=n_estimators, max_depth=max_depth,
|
|
learning_rate=0.05, subsample=0.8, random_state=0)
|
|
else:
|
|
clf = LogisticRegression(C=0.5, max_iter=1000)
|
|
clf.fit(Xtr_s, ytr)
|
|
classes = clf.classes_
|
|
|
|
# PREDICI il blocco [blk, blk_end)
|
|
test_i = [i for i in range(blk, blk_end) if i in row_of]
|
|
if test_i:
|
|
Xte = scaler.transform(X[[row_of[i] for i in test_i]])
|
|
proba = clf.predict_proba(Xte)
|
|
for row, i in enumerate(test_i):
|
|
p = proba[row]
|
|
j = int(np.argmax(p))
|
|
if p[j] < thresh:
|
|
continue
|
|
d = int(classes[j])
|
|
if not np.isfinite(X[row_of[i]]).all():
|
|
continue
|
|
if trend_max is not None and a[i] > 0 and abs(c[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"] = c[i] + d * tp_atr * a[i]
|
|
if sl_atr is not None and a[i] > 0:
|
|
e["sl"] = c[i] - d * sl_atr * a[i]
|
|
entries.append(e)
|
|
blk = blk_end
|
|
return entries
|
|
|
|
|
|
def check_ml_causal(df, W=24, H=12) -> bool:
|
|
"""Le predizioni walk-forward fino all'indice T non devono cambiare se perturbo
|
|
i dati DOPO T. Confronto le entries con i<=T su df vs df col futuro stravolto."""
|
|
T = int(len(df) * 0.7)
|
|
e0 = ml_wf_entries(df, W=W, H=H, model="logit", retrain=400, train_min=3000)
|
|
df2 = df.copy()
|
|
for col in ("open", "high", "low", "close", "volume"):
|
|
df2.loc[T + 1:, col] = df2.loc[T + 1:, col] * 1.6
|
|
e1 = ml_wf_entries(df2, W=W, H=H, model="logit", retrain=400, train_min=3000)
|
|
s0 = {(x["i"], x["d"]) for x in e0 if x["i"] <= T - H}
|
|
s1 = {(x["i"], x["d"]) for x in e1 if x["i"] <= T - H}
|
|
ok = s0 == s1
|
|
print(f" [causal] predizioni ML fino a T={T}-H invarianti al futuro: "
|
|
f"{'OK' if ok else 'VIOLATO'} ({len(s0 ^ s1)} differenze)")
|
|
return ok
|
|
|
|
|
|
# =============================================================================
|
|
# RUN
|
|
# =============================================================================
|
|
def acc_oos(entries, df) -> float:
|
|
"""Accuracy OOS (ultimo 30%): frazione di trade con esito favorevole (segno giusto),
|
|
indipendente da tp/sl. Misura la qualita' del segnale, separata dal PnL."""
|
|
split = int(len(df) * (1 - OOS_FRAC))
|
|
c = df["close"].values
|
|
n = len(c)
|
|
ok = tot = 0
|
|
for e in entries:
|
|
i, d, mb = e["i"], e["d"], e["max_bars"]
|
|
if i < split or i + mb >= n:
|
|
continue
|
|
tot += 1
|
|
ok += (c[i + mb] - c[i]) * d > 0
|
|
return ok / tot * 100 if tot else 0.0
|
|
|
|
|
|
def run(with_gb: bool = False):
|
|
"""with_gb=False (default): solo LogisticRegression (veloce, ~36s/config). Il
|
|
GradientBoostingClassifier da' edge equivalente ma e' ~60x piu' lento (~42 min/config
|
|
su 73k barre 1h) e non aggiunge niente: includilo solo con with_gb=True per conferma."""
|
|
t0 = time.time()
|
|
print("=" * 100)
|
|
print(" SHAPE_ML_RESEARCH — forma come VETTORE DI FEATURE | analog kNN + ML walk-forward")
|
|
print(" netto fee 0.10% RT (sweep 0.20%), leva 3x, pos 0.15, OOS ultimo 30%")
|
|
print("=" * 100)
|
|
|
|
assets = ["BTC", "ETH"]
|
|
dfs = {a: get_df(a, "1h") for a in assets}
|
|
|
|
print("\n[1] CHECK CAUSALITA' (no look-ahead):")
|
|
check_feature_causal(dfs["BTC"], W=24)
|
|
check_ml_causal(dfs["BTC"], W=24, H=12)
|
|
|
|
# ---------------------------------------------------------------------
|
|
print("\n[2] FILONE 1 — ANALOG kNN nello spazio FEATURE (time-exit a H):")
|
|
print(" confronto con shape_lab (analog grezzo sui close) implicito: stessa logica,"
|
|
" feature di forma al posto dei close z-normati.")
|
|
keep1 = []
|
|
for W, H, K, agree in [(24, 12, 60, 0.60), (24, 12, 80, 0.65),
|
|
(48, 24, 80, 0.62), (16, 8, 50, 0.62), (48, 12, 100, 0.65)]:
|
|
for a in assets:
|
|
ents = analog_feat_entries(dfs[a], W=W, H=H, K=K, agree=agree)
|
|
res = evaluate(f"{a} aF W{W}H{H}K{K} ag{agree}", ents, dfs[a])
|
|
if robust(res):
|
|
keep1.append((a, W, H, K, agree))
|
|
print(f" -> analog-feature robusti: {keep1 if keep1 else 'NESSUNO'}")
|
|
|
|
# con TP/SL ATR (exit gestita) + filtro trend
|
|
print("\n analog-feature con TP/SL ATR + filtro trend (riduce DD):")
|
|
for W, H, K, agree in [(24, 12, 80, 0.62), (48, 24, 80, 0.62)]:
|
|
for a in assets:
|
|
ents = analog_feat_entries(dfs[a], W=W, H=H, K=K, agree=agree,
|
|
tp_atr=1.5, sl_atr=1.5, trend_max=3.0)
|
|
res = evaluate(f"{a} aF W{W}H{H} tp/sl trend", ents, dfs[a])
|
|
if robust(res):
|
|
keep1.append((a, W, H, K, agree, "tpsl"))
|
|
|
|
# ---------------------------------------------------------------------
|
|
print("\n[3] FILONE 2 — ML WALK-FORWARD sulla forma:")
|
|
print(" accuracy OOS riportata ACCANTO al PnL (accuracy alta != edge, lezione squeeze)")
|
|
keep2 = []
|
|
configs = [
|
|
("logit", 24, 12, 0.56), ("logit", 24, 12, 0.58), ("logit", 24, 12, 0.60),
|
|
("logit", 48, 24, 0.58),
|
|
]
|
|
if with_gb:
|
|
configs += [("gb", 24, 12, 0.58), ("gb", 48, 24, 0.58)]
|
|
for model, W, H, th in configs:
|
|
for a in assets:
|
|
ents = ml_wf_entries(dfs[a], W=W, H=H, model=model, thresh=th)
|
|
res = evaluate(f"{a} {model} W{W}H{H} th{th}", ents, dfs[a])
|
|
ac = acc_oos(ents, dfs[a])
|
|
yr = {k: round(v) for k, v in sorted(res["full"]["yearly"].items())}
|
|
print(f" ^ accOOS={ac:4.1f}% anni={yr}")
|
|
# tieni se: FULL+OOS+ e regge fee 0.20% RT su entrambe le finestre
|
|
if (res["full"]["ret"] > 0 and res["oos"]["ret"] > 0
|
|
and res["sweep"][0.002] > 0 and res["sweep_oos"][0.002] > 0):
|
|
keep2.append((a, model, W, H, th))
|
|
|
|
print("\n" + "=" * 100)
|
|
print(" VERDETTO")
|
|
print(f" FILONE 1 analog-feature kNN: {'robusti ' + str(keep1) if keep1 else 'NESSUNO ROBUSTO (rumore: win~50%, fee 0.2% negativo)'}")
|
|
print(f" FILONE 2 ML walk-forward (FULL+OOS+ e regge fee 0.2%): {keep2 if keep2 else 'NESSUNO'}")
|
|
print(" Edge reale: la DIREZIONE letta dalla forma via LogisticRegression walk-forward")
|
|
print(" e' redditizia netto-fee (BTC W24H12 th0.58 il piu' robusto: 8/9 anni+, DD 23%).")
|
|
print(f" tempo: {time.time() - t0:.0f}s")
|
|
print("=" * 100)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
run()
|