202 lines
6.3 KiB
Python
202 lines
6.3 KiB
Python
"""Strategia 6: Structural Pattern Matching con DTW veloce.
|
|
Idea diversa: invece di ML generico, cerca nel passato le finestre OHLC
|
|
più simili alla finestra corrente usando una versione veloce (reduced DTW).
|
|
Vota sulla direzione basandosi sui K-nearest neighbors nel passato.
|
|
Usa features normalizzate (non DTW puro sul prezzo che è lento).
|
|
"""
|
|
from __future__ import annotations
|
|
import sys
|
|
sys.path.insert(0, ".")
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
from sklearn.neighbors import KNeighborsClassifier
|
|
from sklearn.preprocessing import StandardScaler
|
|
from src.data.downloader import load_data
|
|
from src.fractal.patterns import normalize_pattern_window
|
|
|
|
print("=" * 60)
|
|
print(" STRATEGIA 6: STRUCTURAL PATTERN KNN — BTC 1H")
|
|
print("=" * 60)
|
|
|
|
df = load_data("BTC", "1h")
|
|
close = df["close"].values
|
|
n = len(close)
|
|
|
|
WINDOW = 24
|
|
LOOKAHEAD = 6
|
|
MIN_RETURN = 0.003
|
|
|
|
def extract_structural_features(df: pd.DataFrame, idx: int, window: int) -> np.ndarray | None:
|
|
"""Extract normalized structural features from OHLC window."""
|
|
if idx < window:
|
|
return None
|
|
|
|
o = df["open"].values[idx - window : idx]
|
|
h = df["high"].values[idx - window : idx]
|
|
l = df["low"].values[idx - window : idx]
|
|
c = df["close"].values[idx - window : idx]
|
|
v = df["volume"].values[idx - window : idx]
|
|
|
|
# Normalize price to [0,1]
|
|
all_prices = np.concatenate([o, h, l, c])
|
|
mn, mx = all_prices.min(), all_prices.max()
|
|
if mx - mn == 0:
|
|
return None
|
|
o_n = (o - mn) / (mx - mn)
|
|
h_n = (h - mn) / (mx - mn)
|
|
l_n = (l - mn) / (mx - mn)
|
|
c_n = (c - mn) / (mx - mn)
|
|
|
|
# Body and shadow ratios (already normalized)
|
|
total = h - l
|
|
total = np.where(total == 0, 1e-10, total)
|
|
body = np.abs(c - o) / total
|
|
upper_shadow = (h - np.maximum(o, c)) / total
|
|
lower_shadow = (np.minimum(o, c) - l) / total
|
|
direction = np.sign(c - o)
|
|
|
|
# Returns
|
|
log_c = np.log(np.where(c == 0, 1e-10, c))
|
|
returns = np.diff(log_c)
|
|
|
|
# Volume profile (normalized)
|
|
v_mean = np.mean(v)
|
|
v_n = v / v_mean if v_mean > 0 else np.ones_like(v)
|
|
|
|
# Downsample to fixed-size feature vector
|
|
# Take every N-th candle if window is large
|
|
step = max(1, window // 12)
|
|
sampled_idx = np.arange(0, window, step)[:12]
|
|
|
|
features = np.concatenate([
|
|
c_n[sampled_idx], # 12: normalized close
|
|
body[sampled_idx], # 12: body ratios
|
|
direction[sampled_idx], # 12: direction
|
|
upper_shadow[sampled_idx], # 12: upper shadow
|
|
lower_shadow[sampled_idx], # 12: lower shadow
|
|
v_n[sampled_idx], # 12: volume profile
|
|
[np.mean(returns), np.std(returns), np.sum(returns)], # 3: return stats
|
|
[np.mean(body), np.std(body)], # 2: body stats
|
|
])
|
|
return features
|
|
|
|
|
|
print("Extracting features...")
|
|
features_all = []
|
|
labels_all = []
|
|
indices_all = []
|
|
|
|
for i in range(WINDOW, n - LOOKAHEAD, 1):
|
|
feats = extract_structural_features(df, i, WINDOW)
|
|
if feats is None:
|
|
continue
|
|
|
|
future_ret = (close[i + LOOKAHEAD - 1] - close[i - 1]) / close[i - 1]
|
|
if abs(future_ret) < MIN_RETURN:
|
|
continue
|
|
|
|
features_all.append(feats)
|
|
labels_all.append(1 if future_ret > 0 else 0)
|
|
indices_all.append(i)
|
|
|
|
X = np.array(features_all)
|
|
y = np.array(labels_all)
|
|
idx_arr = np.array(indices_all)
|
|
|
|
X = np.nan_to_num(X, nan=0.0, posinf=1e6, neginf=-1e6)
|
|
|
|
split = int(len(X) * 0.7)
|
|
X_train, X_test = X[:split], X[split:]
|
|
y_train, y_test = y[:split], y[split:]
|
|
idx_test = idx_arr[split:]
|
|
|
|
print(f"Samples: {len(X)} (train={split}, test={len(X)-split})")
|
|
print(f"Label balance: up={np.mean(y)*100:.1f}%")
|
|
|
|
scaler = StandardScaler()
|
|
X_train_s = scaler.fit_transform(X_train)
|
|
X_test_s = scaler.transform(X_test)
|
|
|
|
# Test diversi K
|
|
print("\n--- KNN SWEEP ---")
|
|
for K in [5, 10, 20, 50, 100, 200]:
|
|
knn = KNeighborsClassifier(n_neighbors=K, weights="distance", n_jobs=-1)
|
|
knn.fit(X_train_s, y_train)
|
|
|
|
proba = knn.predict_proba(X_test_s)
|
|
up_idx = list(knn.classes_).index(1)
|
|
|
|
for thr in [0.55, 0.60, 0.65, 0.70]:
|
|
sigs = []
|
|
accs = []
|
|
for j in range(len(X_test)):
|
|
p_up = proba[j][up_idx]
|
|
i = idx_test[j]
|
|
actual = (close[i + LOOKAHEAD - 1] - close[i - 1]) / close[i - 1]
|
|
|
|
if p_up >= thr:
|
|
sigs.append(1)
|
|
accs.append(1 if actual > 0 else 0)
|
|
elif p_up <= (1 - thr):
|
|
sigs.append(-1)
|
|
accs.append(1 if actual < 0 else 0)
|
|
|
|
if not accs:
|
|
continue
|
|
|
|
acc = np.mean(accs) * 100
|
|
# PnL
|
|
capital = 1000
|
|
for direction, j in zip(sigs, range(len(accs))):
|
|
i_idx = idx_test[[k for k in range(len(X_test)) if proba[k][up_idx] >= thr or proba[k][up_idx] <= (1 - thr)][j]]
|
|
entry = close[i_idx - 1]
|
|
exit_ = close[i_idx + LOOKAHEAD - 1]
|
|
if direction == 1:
|
|
ret = (exit_ - entry) / entry
|
|
else:
|
|
ret = (entry - exit_) / entry
|
|
ret -= 0.002
|
|
capital *= (1 + ret * 0.5)
|
|
|
|
total_ret = (capital - 1000) / 1000 * 100
|
|
print(f" K={K:3d} thr={thr:.2f}: signals={len(accs):5d} acc={acc:.1f}% ret={total_ret:+.1f}%")
|
|
|
|
# Best combo: try with Gradient Boosting on same features
|
|
print("\n\n--- GRADIENT BOOSTING SU STRUCTURAL FEATURES ---")
|
|
from sklearn.ensemble import GradientBoostingClassifier
|
|
|
|
gb = GradientBoostingClassifier(
|
|
n_estimators=300, max_depth=5, min_samples_leaf=30,
|
|
learning_rate=0.03, subsample=0.8, random_state=42,
|
|
)
|
|
gb.fit(X_train_s, y_train)
|
|
proba_gb = gb.predict_proba(X_test_s)
|
|
up_idx_gb = list(gb.classes_).index(1)
|
|
|
|
for thr in [0.50, 0.55, 0.60, 0.65, 0.70, 0.75]:
|
|
accs = []
|
|
capital = 1000
|
|
n_trades = 0
|
|
for j in range(len(X_test)):
|
|
p_up = proba_gb[j][up_idx_gb]
|
|
i = idx_test[j]
|
|
actual = (close[i + LOOKAHEAD - 1] - close[i - 1]) / close[i - 1]
|
|
|
|
if p_up >= thr:
|
|
accs.append(1 if actual > 0 else 0)
|
|
ret = actual - 0.002
|
|
capital *= (1 + ret * 0.5)
|
|
n_trades += 1
|
|
elif p_up <= (1 - thr):
|
|
accs.append(1 if actual < 0 else 0)
|
|
ret = -actual - 0.002
|
|
capital *= (1 + ret * 0.5)
|
|
n_trades += 1
|
|
|
|
if not accs:
|
|
continue
|
|
acc = np.mean(accs) * 100
|
|
total_ret = (capital - 1000) / 1000 * 100
|
|
print(f" thr={thr:.2f}: trades={n_trades:5d} acc={acc:.1f}% ret={total_ret:+.1f}%")
|