"""Strategia 5: Enhanced fractal features + binary classification + position management. Miglioramenti rispetto a #4: - Binary classification (up vs down, ignora flat) - Feature engineering esteso: multi-window fractal indicators - Migliore filtraggio segnali - Position sizing basato su confidenza - Trailing stop """ from __future__ import annotations import sys sys.path.insert(0, ".") import numpy as np import pandas as pd from sklearn.ensemble import GradientBoostingClassifier from sklearn.metrics import accuracy_score from src.data.downloader import load_data from src.fractal.indicators import ( hurst_exponent, fractal_dimension_higuchi, self_similarity_score, volatility_ratio, ) from src.fractal.patterns import encode_candles, extract_body_ratios, extract_shadow_ratios print("=" * 60) print(" STRATEGIA 5: ENHANCED FRACTAL — BTC + ETH 1H") print("=" * 60) LOOKAHEADS = [3, 6, 12] MIN_RETURN = 0.003 # 0.3% threshold for "up" label for asset in ["BTC", "ETH"]: for LOOKAHEAD in LOOKAHEADS: print(f"\n{'#'*60}") print(f" {asset} 1H — LOOKAHEAD={LOOKAHEAD}") print(f"{'#'*60}") df = load_data(asset, "1h") close = df["close"].values volume = df["volume"].values n = len(close) log_close = np.log(np.where(close == 0, 1e-10, close)) returns = np.diff(log_close) candle_types = encode_candles(df) body_ratios = extract_body_ratios(df) shadow_ratios = extract_shadow_ratios(df) WINDOWS = [24, 48, 96, 192] features_list = [] labels = [] indices = [] max_window = max(WINDOWS) + 50 for i in range(max_window, n - LOOKAHEAD, 2): feats = [] for w in WINDOWS: ret_w = returns[i - w : i - 1] close_w = close[i - w : i] h = hurst_exponent(ret_w, max_lag=min(len(ret_w) // 4, 20)) fd = fractal_dimension_higuchi(ret_w, k_max=min(6, len(ret_w) // 4)) vr = volatility_ratio(close_w, fast=min(12, w // 4), slow=w) mom = np.sum(ret_w) vol = np.std(ret_w) skew = float(pd.Series(ret_w).skew()) if len(ret_w) > 2 else 0 kurt = float(pd.Series(ret_w).kurtosis()) if len(ret_w) > 3 else 0 ma = np.mean(close_w) price_vs_ma = close[i - 1] / ma if ma > 0 else 1 # Autocorrelation lag-1 if len(ret_w) > 1 and np.std(ret_w) > 0: ac1 = np.corrcoef(ret_w[:-1], ret_w[1:])[0, 1] if not np.isfinite(ac1): ac1 = 0 else: ac1 = 0 feats.extend([h, fd, vr, mom, vol, skew, kurt, price_vs_ma, ac1]) # Self-similarity multi-scale large_window = close[max(0, i - 192 * 4) : i] ss = self_similarity_score(large_window, 48) feats.append(ss) # Candle pattern features (last 12 candles) ct = candle_types[i - 12 : i] br = body_ratios[i - 12 : i] sr = shadow_ratios[i - 12 : i] feats.extend([ np.mean(ct[-3:]), np.mean(ct[-6:]), np.mean(ct[-12:]), np.std(br[-6:]), np.mean(br[-3:]), np.mean(sr[-6:]), np.max(br[-6:]), np.min(br[-6:]), ]) # Volume features vol_w = volume[i - 24 : i] if np.mean(vol_w) > 0: feats.append(volume[i - 1] / np.mean(vol_w)) feats.append(np.std(vol_w) / np.mean(vol_w)) else: feats.extend([1.0, 0.0]) # Range/ATR proxy h_arr = df["high"].values[i - 14 : i] l_arr = df["low"].values[i - 14 : i] c_arr = close[i - 14 : i] tr = np.maximum(h_arr - l_arr, np.maximum(np.abs(h_arr - np.roll(c_arr, 1)), np.abs(l_arr - np.roll(c_arr, 1)))) atr = np.mean(tr[1:]) feats.append(atr / close[i - 1] if close[i - 1] > 0 else 0) # Label future_ret = (close[i + LOOKAHEAD - 1] - close[i - 1]) / close[i - 1] if abs(future_ret) < MIN_RETURN: continue # skip flat zones label = 1 if future_ret > 0 else 0 features_list.append(feats) labels.append(label) indices.append(i) X = np.array(features_list) y = np.array(labels) idx_arr = np.array(indices) X = np.nan_to_num(X, nan=0.0, posinf=1e6, neginf=-1e6) # Split 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}%") # Train model = GradientBoostingClassifier( n_estimators=300, max_depth=5, min_samples_leaf=30, learning_rate=0.03, subsample=0.8, random_state=42, ) model.fit(X_train, y_train) y_pred = model.predict(X_test) proba = model.predict_proba(X_test) base_acc = accuracy_score(y_test, y_pred) print(f"Base accuracy: {base_acc*100:.1f}%") # Threshold sweep print(f"\n Threshold sweep:") for thr in [0.50, 0.55, 0.60, 0.65, 0.70, 0.75, 0.80]: up_idx = model.classes_.tolist().index(1) sigs = [] accs = [] for k in range(len(X_test)): p_up = proba[k][up_idx] i = idx_test[k] actual = (close[i + LOOKAHEAD - 1] - close[i - 1]) / close[i - 1] if p_up >= thr: sigs.append(("long", i)) accs.append(1 if actual > 0 else 0) elif p_up <= (1 - thr): sigs.append(("short", i)) accs.append(1 if actual < 0 else 0) if not accs: print(f" thr={thr:.2f}: no signals") continue acc = np.mean(accs) * 100 # Simple PnL estimate pnl = 0 capital = 1000 for direction, i in sigs: entry = close[i - 1] exit_ = close[i + LOOKAHEAD - 1] if direction == "long": ret = (exit_ - entry) / entry else: ret = (entry - exit_) / entry ret -= 0.002 # fees round-trip pnl += capital * ret * 0.5 # 50% per trade capital += capital * ret * 0.5 total_ret = (capital - 1000) / 1000 * 100 trades_per_year = len(sigs) / ((n - max_window) / (24 * 365)) print(f" thr={thr:.2f}: signals={len(sigs):5d} acc={acc:.1f}% ret={total_ret:+.1f}% trades/yr={trades_per_year:.0f}")