"""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}%")