"""Strategia 4: Regime-aware fractal ML. Combina: 1. Hurst exponent per regime detection (trend vs mean-revert vs random) 2. Feature engineering da indicatori frattali 3. RandomForest per predizione direzione 4. Trade filtering aggressivo (solo alta confidenza) """ from __future__ import annotations import sys sys.path.insert(0, ".") import numpy as np import pandas as pd from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier from sklearn.metrics import accuracy_score, classification_report 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 from src.backtest.engine import run_backtest print("=" * 60) print(" STRATEGIA 4: REGIME-AWARE FRACTAL ML — BTC 1H") print("=" * 60) df = load_data("BTC", "1h") close = df["close"].values n = len(close) LOOKBACK = 48 LOOKAHEAD = 6 MIN_CONFIDENCE = 0.60 print(f"\nDati: {n} candele") print(f"Lookback: {LOOKBACK}, Lookahead: {LOOKAHEAD}") # --- Feature engineering --- print("\nCalcolo features...") features_list = [] labels = [] indices = [] returns = np.diff(np.log(np.where(close == 0, 1e-10, close))) candle_types = encode_candles(df) body_ratios = extract_body_ratios(df) shadow_ratios = extract_shadow_ratios(df) for i in range(LOOKBACK, n - LOOKAHEAD, 3): if i % 5000 == 0: print(f" Feature extraction: {i}/{n}") window = close[i - LOOKBACK : i] ret_window = returns[i - LOOKBACK : i - 1] if len(ret_window) < 10: continue h = hurst_exponent(ret_window, max_lag=min(len(ret_window) // 4, 20)) fd = fractal_dimension_higuchi(ret_window, k_max=min(8, len(ret_window) // 4)) larger_window = close[max(0, i - LOOKBACK * 6) : i] ss = self_similarity_score(larger_window, min(LOOKBACK, len(larger_window))) vr = volatility_ratio(window, fast=12, slow=LOOKBACK) # Candle pattern features ct = candle_types[i - 6 : i] br = body_ratios[i - 6 : i] sr = shadow_ratios[i - 6 : i] recent_returns = ret_window[-12:] momentum_short = np.sum(recent_returns[-3:]) momentum_mid = np.sum(recent_returns[-6:]) momentum_long = np.sum(recent_returns) vol_short = np.std(recent_returns[-6:]) if len(recent_returns) >= 6 else 0 vol_long = np.std(ret_window) if len(ret_window) > 0 else 0 volume_window = df["volume"].values[i - 12 : i] vol_avg = np.mean(volume_window) if len(volume_window) > 0 else 0 vol_last = df["volume"].values[i - 1] if i > 0 else 0 vol_ratio = vol_last / vol_avg if vol_avg > 0 else 1.0 up_count_6 = np.sum(ct[-6:] == 1) / 6 down_count_6 = np.sum(ct[-6:] == -1) / 6 features = [ h, # Hurst exponent fd, # Fractal dimension ss, # Self-similarity vr, # Volatility ratio momentum_short, # 3-candle momentum momentum_mid, # 6-candle momentum momentum_long, # Full window momentum vol_short, # Short-term volatility vol_long, # Long-term volatility vol_ratio, # Volume spike ratio up_count_6, # Bullish ratio (last 6) down_count_6, # Bearish ratio (last 6) np.mean(br[-6:]), # Avg body ratio np.mean(sr[-6:]), # Avg shadow ratio np.mean(br[-3:]), # Avg body ratio (last 3) np.std(br[-6:]), # Body ratio std close[i - 1] / np.mean(window), # Price vs MA ] # Label: 1 if price goes up in next LOOKAHEAD candles, 0 otherwise future_ret = (close[i + LOOKAHEAD - 1] - close[i - 1]) / close[i - 1] label = 1 if future_ret > 0.002 else (0 if future_ret > -0.002 else -1) features_list.append(features) labels.append(label) indices.append(i) X = np.array(features_list) y = np.array(labels) idx_arr = np.array(indices) print(f"\nDataset: {len(X)} samples") print(f"Label distribution: up={np.sum(y==1)}, flat={np.sum(y==0)}, down={np.sum(y==-1)}") # Train/test split cronologico split_point = int(len(X) * 0.7) X_train, X_test = X[:split_point], X[split_point:] y_train, y_test = y[:split_point], y[split_point:] idx_train, idx_test = idx_arr[:split_point], idx_arr[split_point:] # Handle NaN/Inf X_train = np.nan_to_num(X_train, nan=0.0, posinf=1e6, neginf=-1e6) X_test = np.nan_to_num(X_test, nan=0.0, posinf=1e6, neginf=-1e6) # --- Modelli --- print("\n--- TRAINING ---") models = { "RandomForest": RandomForestClassifier( n_estimators=200, max_depth=8, min_samples_leaf=20, class_weight="balanced", random_state=42, n_jobs=-1, ), "GradientBoosting": GradientBoostingClassifier( n_estimators=200, max_depth=5, min_samples_leaf=20, learning_rate=0.05, random_state=42, ), } for name, model in models.items(): print(f"\n{'='*40}") print(f" {name}") print(f"{'='*40}") model.fit(X_train, y_train) # Feature importance if hasattr(model, "feature_importances_"): feat_names = [ "hurst", "fractal_dim", "self_sim", "vol_ratio", "mom_3", "mom_6", "mom_full", "vol_short", "vol_long", "vol_spike", "up_ratio", "down_ratio", "body_avg", "shadow_avg", "body_3", "body_std", "price_vs_ma" ] imp = model.feature_importances_ sorted_idx = np.argsort(imp)[::-1] print("\nFeature importance (top 10):") for j in sorted_idx[:10]: print(f" {feat_names[j]:15s}: {imp[j]:.4f}") # Prediction con probabilità y_pred = model.predict(X_test) proba = model.predict_proba(X_test) print(f"\nAccuracy: {accuracy_score(y_test, y_pred)*100:.1f}%") print(classification_report(y_test, y_pred, target_names=["down", "flat", "up"], zero_division=0)) # Genera segnali filtrati per confidenza signals = pd.Series(0, index=df.index) accuracies_filtered = [] classes = model.classes_ up_class_idx = list(classes).index(1) if 1 in classes else -1 down_class_idx = list(classes).index(-1) if -1 in classes else -1 for k, i in enumerate(idx_test): p = proba[k] if up_class_idx >= 0 and p[up_class_idx] >= MIN_CONFIDENCE: signals.iloc[i] = 1 actual = (close[i + LOOKAHEAD - 1] - close[i - 1]) / close[i - 1] accuracies_filtered.append(1 if actual > 0 else 0) elif down_class_idx >= 0 and p[down_class_idx] >= MIN_CONFIDENCE: signals.iloc[i] = -1 actual = (close[i + LOOKAHEAD - 1] - close[i - 1]) / close[i - 1] accuracies_filtered.append(1 if actual < 0 else 0) n_signals = (signals != 0).sum() print(f"\nSegnali filtrati (conf>={MIN_CONFIDENCE}): {n_signals}") if accuracies_filtered: print(f"Accuratezza filtrata: {np.mean(accuracies_filtered)*100:.1f}%") # Backtest split_idx = int(len(df) * 0.7) test_df = df.iloc[split_idx:].reset_index(drop=True) test_signals = signals.iloc[split_idx:].reset_index(drop=True) result = run_backtest(test_df, test_signals, initial_capital=1000, fee_pct=0.001, max_hold_candles=LOOKAHEAD) print(f"\nBACKTEST:") for kk, v in result.summary().items(): print(f" {kk}: {v}") # Prova con soglie diverse print(f"\n Varianti soglia:") for threshold in [0.55, 0.60, 0.65, 0.70, 0.75, 0.80]: sigs = pd.Series(0, index=df.index) accs = [] for k, i in enumerate(idx_test): p = proba[k] if up_class_idx >= 0 and p[up_class_idx] >= threshold: sigs.iloc[i] = 1 actual = (close[i + LOOKAHEAD - 1] - close[i - 1]) / close[i - 1] accs.append(1 if actual > 0 else 0) elif down_class_idx >= 0 and p[down_class_idx] >= threshold: sigs.iloc[i] = -1 actual = (close[i + LOOKAHEAD - 1] - close[i - 1]) / close[i - 1] accs.append(1 if actual < 0 else 0) t_sigs = sigs.iloc[split_idx:].reset_index(drop=True) res = run_backtest(test_df, t_sigs, initial_capital=1000, fee_pct=0.001, max_hold_candles=LOOKAHEAD) acc = np.mean(accs) * 100 if accs else 0 print(f" thr={threshold:.2f}: signals={len(accs):4d} acc={acc:.1f}% ret={res.total_return*100:+.1f}% wr={res.win_rate*100:.0f}% sharpe={res.sharpe_ratio:.2f}")