232 lines
8.4 KiB
Python
232 lines
8.4 KiB
Python
"""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}")
|