Adriano Dal Pastro
9879b46688
L'analisi out-of-sample fee-aware ha dimostrato che l'intera famiglia squeeze-breakout (SQ01-04, MT01, ML01, AD01, CM01, PD01) non ha edge: le accuratezze storiche 76-82% erano un artefatto di look-ahead (ingresso a close[i-1] con direzione decisa da close[i]). Sotto ingresso onesto a close[i] e fee reali tutte perdono, anche a fee zero. - nuova MR01_bollinger_fade (mean-reversion): edge netto validato OOS, robusto su griglia parametri e fino a 0.20% fee RT. BTC 1h n50 k2.5: +201% OOS, DD 15% - 9 strategie squeeze spostate in scripts/waste/ - strategy_loader + strategies.yml: solo MR01 (BTC/ETH 1h) - signal_engine.train: validazione OOS (accuratezza test + signal precision) - scripts/analysis/strategy_research.py: harness di ricerca fee-aware NOTA: lo StrategyWorker va aggiornato per usare gli exit TP/SL passati in metadata prima di tradare MR01 dal vivo (ora esce solo a hold_bars/stop fisso). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
285 lines
9.7 KiB
Python
285 lines
9.7 KiB
Python
"""Motore segnali: squeeze detection + ML confirmation su dati live."""
|
|
from __future__ import annotations
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
from sklearn.ensemble import GradientBoostingClassifier
|
|
from sklearn.preprocessing import StandardScaler
|
|
|
|
|
|
def keltner_ratio(close: np.ndarray, high: np.ndarray, low: np.ndarray, window: int = 14) -> np.ndarray:
|
|
n = len(close)
|
|
result = np.full(n, np.nan)
|
|
for i in range(window, n):
|
|
wc = close[i - window : i]
|
|
wh = high[i - window : i]
|
|
wl = low[i - window : i]
|
|
ma = np.mean(wc)
|
|
bb_std = np.std(wc)
|
|
tr = np.maximum(wh - wl, np.maximum(np.abs(wh - np.roll(wc, 1)), np.abs(wl - np.roll(wc, 1))))
|
|
atr = np.mean(tr[1:])
|
|
kc_r = (ma + 1.5 * atr) - (ma - 1.5 * atr)
|
|
bb_r = (ma + 2 * bb_std) - (ma - 2 * bb_std)
|
|
if kc_r > 0:
|
|
result[i] = bb_r / kc_r
|
|
return result
|
|
|
|
|
|
def build_features(df: pd.DataFrame, i: int, squeeze_duration: int, squeeze_avg_vol: float, kcr_val: float) -> np.ndarray | None:
|
|
if i < 100 or i >= len(df):
|
|
return None
|
|
|
|
o = df["open"].values
|
|
h = df["high"].values
|
|
l = df["low"].values
|
|
c = df["close"].values
|
|
v = df["volume"].values
|
|
|
|
feats = []
|
|
for w in [12, 24, 48]:
|
|
if i < w:
|
|
feats.extend([0] * 12)
|
|
continue
|
|
|
|
win_c = c[i - w : i]
|
|
win_o = o[i - w : i]
|
|
win_h = h[i - w : i]
|
|
win_l = l[i - w : i]
|
|
win_v = v[i - w : i]
|
|
|
|
mn, mx = win_l.min(), max(win_h.max(), win_c.max())
|
|
rng = mx - mn if mx - mn > 0 else 1e-10
|
|
total = win_h - win_l
|
|
total = np.where(total == 0, 1e-10, total)
|
|
body = np.abs(win_c - win_o) / total
|
|
direction = np.sign(win_c - win_o)
|
|
log_c = np.log(np.where(win_c == 0, 1e-10, win_c))
|
|
rets = np.diff(log_c)
|
|
v_mean = np.mean(win_v)
|
|
|
|
feats.extend([
|
|
np.mean(rets) if len(rets) > 0 else 0,
|
|
np.std(rets) if len(rets) > 0 else 0,
|
|
np.sum(rets) if len(rets) > 0 else 0,
|
|
float(pd.Series(rets).skew()) if len(rets) > 2 else 0,
|
|
float(pd.Series(rets).kurtosis()) if len(rets) > 3 else 0,
|
|
np.mean(body),
|
|
np.std(body),
|
|
np.mean(direction),
|
|
np.mean(direction[-min(3, w):]),
|
|
(win_c[-1] - mn) / rng,
|
|
win_v[-1] / v_mean if v_mean > 0 else 1,
|
|
np.corrcoef(rets[:-1], rets[1:])[0, 1] if len(rets) > 1 and np.std(rets) > 0 else 0,
|
|
])
|
|
|
|
feats.extend([
|
|
squeeze_duration,
|
|
squeeze_duration / (24 * 4),
|
|
kcr_val,
|
|
v[i - 1] / squeeze_avg_vol if squeeze_avg_vol > 0 else 1,
|
|
np.mean(v[max(0, i - 3) : i]) / squeeze_avg_vol if squeeze_avg_vol > 0 else 1,
|
|
])
|
|
|
|
h48 = np.max(h[max(0, i - 48) : i])
|
|
l48 = np.min(l[max(0, i - 48) : i])
|
|
r48 = h48 - l48
|
|
feats.append((c[i - 1] - l48) / r48 if r48 > 0 else 0.5)
|
|
|
|
tr = np.maximum(h[i - 14 : i] - l[i - 14 : i],
|
|
np.maximum(np.abs(h[i - 14 : i] - np.roll(c[i - 14 : i], 1)),
|
|
np.abs(l[i - 14 : i] - np.roll(c[i - 14 : i], 1))))
|
|
atr = np.mean(tr[1:])
|
|
feats.append(atr / c[i - 1] if c[i - 1] > 0 else 0)
|
|
|
|
first_ret = (c[i - 1] - c[i - 2]) / c[i - 2] if i >= 2 and c[i - 2] > 0 else 0
|
|
feats.append(first_ret)
|
|
|
|
return np.nan_to_num(np.array(feats), nan=0, posinf=1e6, neginf=-1e6)
|
|
|
|
|
|
class SignalEngine:
|
|
"""Rileva squeeze e genera segnali ML in real-time."""
|
|
|
|
def __init__(self, bb_w: int = 14, sq_thr: float = 0.8, ml_thr: float = 0.70, min_squeeze_bars: int = 5):
|
|
self.bb_w = bb_w
|
|
self.sq_thr = sq_thr
|
|
self.ml_thr = ml_thr
|
|
self.min_squeeze_bars = min_squeeze_bars
|
|
|
|
self.model: GradientBoostingClassifier | None = None
|
|
self.scaler: StandardScaler | None = None
|
|
self.in_squeeze = False
|
|
self.squeeze_start_idx = 0
|
|
self.trained = False
|
|
|
|
def _new_model(self) -> GradientBoostingClassifier:
|
|
return GradientBoostingClassifier(
|
|
n_estimators=150, max_depth=4, min_samples_leaf=10,
|
|
learning_rate=0.05, subsample=0.8, random_state=42,
|
|
)
|
|
|
|
def _validate_oos(self, X: np.ndarray, y: np.ndarray, test_frac: float = 0.2) -> dict:
|
|
"""Split temporale (no shuffle) per stimare la performance out-of-sample.
|
|
|
|
Allena su training iniziale e valuta sull'ultimo `test_frac` dei campioni.
|
|
Oltre all'accuratezza OOS, riporta la precisione sui soli segnali con
|
|
confidenza >= ml_thr — cioè i trade che la strategia aprirebbe davvero.
|
|
"""
|
|
n_test = int(len(X) * test_frac)
|
|
n_train = len(X) - n_test
|
|
if n_train < 30 or n_test < 5:
|
|
return {"oos_warning": "test set troppo piccolo", "oos_test_samples": n_test}
|
|
|
|
scaler = StandardScaler()
|
|
X_tr = scaler.fit_transform(X[:n_train])
|
|
X_te = scaler.transform(X[n_train:])
|
|
y_tr, y_te = y[:n_train], y[n_train:]
|
|
|
|
model = self._new_model()
|
|
model.fit(X_tr, y_tr)
|
|
|
|
up_idx = list(model.classes_).index(1)
|
|
p_up = model.predict_proba(X_te)[:, up_idx]
|
|
test_acc = float(np.mean((p_up >= 0.5).astype(int) == y_te) * 100)
|
|
oos_train_acc = float(np.mean(model.predict(X_tr) == y_tr) * 100)
|
|
|
|
long_sig = p_up >= self.ml_thr
|
|
short_sig = p_up <= (1 - self.ml_thr)
|
|
n_sig = int((long_sig | short_sig).sum())
|
|
if n_sig > 0:
|
|
correct = int(((long_sig & (y_te == 1)) | (short_sig & (y_te == 0))).sum())
|
|
sig_prec = round(correct / n_sig * 100, 1)
|
|
else:
|
|
sig_prec = None
|
|
|
|
return {
|
|
"oos_train_accuracy": round(oos_train_acc, 1),
|
|
"oos_test_accuracy": round(test_acc, 1),
|
|
"oos_test_samples": n_test,
|
|
"oos_signals": n_sig,
|
|
"oos_signal_precision": sig_prec,
|
|
}
|
|
|
|
def train(self, df: pd.DataFrame, lookahead: int = 3) -> dict:
|
|
"""Addestra il modello su dati storici."""
|
|
close = df["close"].values
|
|
high = df["high"].values
|
|
low = df["low"].values
|
|
volume = df["volume"].values
|
|
n = len(df)
|
|
|
|
kcr = keltner_ratio(close, high, low, self.bb_w)
|
|
|
|
X_all, y_all = [], []
|
|
in_sq = False
|
|
sq_start = 0
|
|
|
|
for i in range(self.bb_w + 1, n - lookahead):
|
|
if np.isnan(kcr[i]):
|
|
continue
|
|
is_sq = kcr[i] < self.sq_thr
|
|
if is_sq and not in_sq:
|
|
in_sq = True
|
|
sq_start = i
|
|
elif not is_sq and in_sq:
|
|
in_sq = False
|
|
duration = i - sq_start
|
|
if duration < self.min_squeeze_bars:
|
|
continue
|
|
|
|
avg_vol = np.mean(volume[sq_start:i])
|
|
feats = build_features(df, i, duration, avg_vol, kcr[i])
|
|
if feats is None:
|
|
continue
|
|
|
|
actual = (close[i + lookahead - 1] - close[i - 1]) / close[i - 1]
|
|
X_all.append(feats)
|
|
y_all.append(1 if actual > 0 else 0)
|
|
|
|
if len(X_all) < 30:
|
|
return {"error": "not enough training samples", "samples": len(X_all)}
|
|
|
|
X = np.array(X_all)
|
|
y = np.array(y_all)
|
|
|
|
oos = self._validate_oos(X, y)
|
|
|
|
self.scaler = StandardScaler()
|
|
X_s = self.scaler.fit_transform(X)
|
|
|
|
self.model = self._new_model()
|
|
self.model.fit(X_s, y)
|
|
self.trained = True
|
|
|
|
preds = self.model.predict(X_s)
|
|
train_acc = float(np.mean(preds == y) * 100)
|
|
|
|
return {
|
|
"samples": len(X),
|
|
"up_ratio": round(float(np.mean(y) * 100), 1),
|
|
"train_accuracy": round(train_acc, 1),
|
|
**oos,
|
|
}
|
|
|
|
def check_signal(self, df: pd.DataFrame) -> dict | None:
|
|
"""Controlla se c'è un segnale sulle ultime candele.
|
|
Ritorna dict con direzione e probabilità, oppure None.
|
|
"""
|
|
if not self.trained:
|
|
return None
|
|
|
|
close = df["close"].values
|
|
high = df["high"].values
|
|
low = df["low"].values
|
|
volume = df["volume"].values
|
|
n = len(df)
|
|
|
|
kcr = keltner_ratio(close, high, low, self.bb_w)
|
|
|
|
if n < self.bb_w + 10:
|
|
return None
|
|
|
|
last_kcr = kcr[-1]
|
|
prev_kcr = kcr[-2] if n > 1 else np.nan
|
|
|
|
if np.isnan(last_kcr) or np.isnan(prev_kcr):
|
|
return None
|
|
|
|
was_squeeze = prev_kcr < self.sq_thr
|
|
is_released = last_kcr >= self.sq_thr
|
|
|
|
if not (was_squeeze and is_released):
|
|
self.in_squeeze = prev_kcr < self.sq_thr
|
|
if self.in_squeeze and not hasattr(self, '_sq_start_tracking'):
|
|
self._sq_start_tracking = n - 1
|
|
if not self.in_squeeze:
|
|
self._sq_start_tracking = None
|
|
return None
|
|
|
|
sq_start = getattr(self, '_sq_start_tracking', n - 10)
|
|
if sq_start is None:
|
|
sq_start = n - 10
|
|
duration = (n - 1) - sq_start
|
|
if duration < self.min_squeeze_bars:
|
|
self._sq_start_tracking = None
|
|
return None
|
|
|
|
avg_vol = np.mean(volume[max(0, sq_start) : n - 1])
|
|
feats = build_features(df, n - 1, duration, avg_vol, last_kcr)
|
|
self._sq_start_tracking = None
|
|
|
|
if feats is None:
|
|
return None
|
|
|
|
feats_s = self.scaler.transform(feats.reshape(1, -1))
|
|
proba = self.model.predict_proba(feats_s)[0]
|
|
up_idx = list(self.model.classes_).index(1)
|
|
p_up = proba[up_idx]
|
|
|
|
if p_up >= self.ml_thr:
|
|
return {"direction": "buy", "probability": p_up, "squeeze_duration": duration}
|
|
elif p_up <= (1 - self.ml_thr):
|
|
return {"direction": "sell", "probability": 1 - p_up, "squeeze_duration": duration}
|
|
|
|
return None
|