111 lines
3.4 KiB
Python
111 lines
3.4 KiB
Python
"""Strategia 2: DTW pattern matching.
|
|
Idea: per ogni finestra di N candele, cerca le K finestre più simili nel passato
|
|
via DTW sui prezzi normalizzati. Se la maggioranza delle match passate è salita
|
|
dopo, vai long. Se è scesa, vai short.
|
|
"""
|
|
from __future__ import annotations
|
|
import sys
|
|
sys.path.insert(0, ".")
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
from src.data.downloader import load_data
|
|
from src.fractal.similarity import dtw_distance
|
|
from src.fractal.patterns import normalize_pattern_window
|
|
from src.backtest.engine import run_backtest
|
|
|
|
print("=" * 60)
|
|
print(" STRATEGIA 2: DTW PATTERN MATCHING — BTC 1H")
|
|
print("=" * 60)
|
|
|
|
df = load_data("BTC", "1h")
|
|
close = df["close"].values
|
|
|
|
WINDOW = 12
|
|
LOOKAHEAD = 6
|
|
K_NEIGHBORS = 20
|
|
LOOKBACK = 2000
|
|
THRESHOLD = 0.65
|
|
|
|
split_idx = int(len(df) * 0.7)
|
|
|
|
def normalize_window(arr: np.ndarray) -> np.ndarray:
|
|
mn, mx = arr.min(), arr.max()
|
|
if mx - mn == 0:
|
|
return np.zeros_like(arr)
|
|
return (arr - mn) / (mx - mn)
|
|
|
|
def compute_returns(close_arr: np.ndarray, idx: int, ahead: int) -> float:
|
|
if idx + ahead >= len(close_arr):
|
|
return 0.0
|
|
return (close_arr[idx + ahead] - close_arr[idx]) / close_arr[idx]
|
|
|
|
print(f"\nParametri: window={WINDOW}, lookahead={LOOKAHEAD}, K={K_NEIGHBORS}")
|
|
print(f"Lookback: {LOOKBACK} candele, threshold: {THRESHOLD}")
|
|
print(f"Train: 0→{split_idx}, Test: {split_idx}→{len(df)}")
|
|
|
|
signals = pd.Series(0, index=df.index)
|
|
accuracies = []
|
|
|
|
step = 6
|
|
test_range = range(split_idx, len(df) - LOOKAHEAD, step)
|
|
total_steps = len(list(test_range))
|
|
print(f"\nValutazione: {total_steps} punti (step={step})...")
|
|
|
|
for count, i in enumerate(test_range):
|
|
if count % 500 == 0:
|
|
print(f" Progresso: {count}/{total_steps} ({count/total_steps*100:.0f}%)")
|
|
|
|
current = normalize_window(close[i - WINDOW : i])
|
|
|
|
search_start = max(WINDOW, i - LOOKBACK)
|
|
search_end = i - LOOKAHEAD
|
|
|
|
if search_end - search_start < K_NEIGHBORS:
|
|
continue
|
|
|
|
distances = []
|
|
for j in range(search_start, search_end):
|
|
candidate = normalize_window(close[j - WINDOW : j])
|
|
if len(candidate) != len(current):
|
|
continue
|
|
d = dtw_distance(current, candidate)
|
|
future_ret = compute_returns(close, j, LOOKAHEAD)
|
|
distances.append((d, future_ret))
|
|
|
|
if len(distances) < K_NEIGHBORS:
|
|
continue
|
|
|
|
distances.sort(key=lambda x: x[0])
|
|
top_k = distances[:K_NEIGHBORS]
|
|
up_count = sum(1 for _, ret in top_k if ret > 0)
|
|
up_ratio = up_count / K_NEIGHBORS
|
|
|
|
if up_ratio >= THRESHOLD:
|
|
signals.iloc[i] = 1
|
|
elif up_ratio <= (1 - THRESHOLD):
|
|
signals.iloc[i] = -1
|
|
|
|
actual_ret = compute_returns(close, i, LOOKAHEAD)
|
|
predicted_up = up_ratio >= THRESHOLD
|
|
predicted_down = up_ratio <= (1 - THRESHOLD)
|
|
if predicted_up:
|
|
accuracies.append(1 if actual_ret > 0 else 0)
|
|
elif predicted_down:
|
|
accuracies.append(1 if actual_ret < 0 else 0)
|
|
|
|
print(f"\nSegnali generati: {(signals != 0).sum()}")
|
|
print(f" Long: {(signals == 1).sum()}, Short: {(signals == -1).sum()}")
|
|
|
|
if accuracies:
|
|
print(f"Accuratezza direzione: {np.mean(accuracies)*100:.1f}% su {len(accuracies)} segnali")
|
|
|
|
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("\nRISULTATI TEST:")
|
|
for k, v in result.summary().items():
|
|
print(f" {k}: {v}")
|