Adriano
0e47956f7a
- src/strategies/base.py: Strategy ABC con Signal, BacktestResult, YearlyStats - src/strategies/indicators.py: keltner_ratio, detect_squeezes, ema, atr, rv, corr - scripts/strategies/: SQ01-SQ04 (squeeze puro/filtri), ML01 (squeeze+GBM) - scripts/waste/: W01-W22 script scartati + REF originali - scripts/analysis/: compare, best_yearly, final_report, paper_status - CLAUDE.md aggiornato con nuova struttura e tabella strategie Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
135 lines
4.6 KiB
Python
135 lines
4.6 KiB
Python
"""Strategia 3: Fourier decomposition e proiezione.
|
|
Ispirata al paper Pythagoras Trading Prediction.
|
|
Idea: scomponi il prezzo in componenti sinusoidali via FFT,
|
|
ricostruisci con le N componenti più forti, proietta nel futuro.
|
|
"""
|
|
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.backtest.engine import run_backtest
|
|
|
|
print("=" * 60)
|
|
print(" STRATEGIA 3: FOURIER PROJECTION — BTC 1H")
|
|
print("=" * 60)
|
|
|
|
df = load_data("BTC", "1h")
|
|
close = df["close"].values
|
|
n_total = len(close)
|
|
|
|
WINDOW = 588 # dal paper: 588 candele per l'indicatore H-C
|
|
N_COMPONENTS = 25 # dal paper: 25 linee verticali
|
|
LOOKAHEAD = 6
|
|
STEP = 6
|
|
|
|
split_idx = int(n_total * 0.7)
|
|
|
|
def fourier_project(series: np.ndarray, n_components: int, ahead: int) -> np.ndarray:
|
|
"""Ricostruisci serie con top-N componenti Fourier e proietta avanti."""
|
|
n = len(series)
|
|
detrended = series - np.linspace(series[0], series[-1], n)
|
|
fft_vals = np.fft.fft(detrended)
|
|
freqs = np.fft.fftfreq(n)
|
|
|
|
magnitudes = np.abs(fft_vals)
|
|
magnitudes[0] = 0
|
|
top_indices = np.argsort(magnitudes)[-n_components * 2:]
|
|
|
|
fft_filtered = np.zeros_like(fft_vals)
|
|
fft_filtered[top_indices] = fft_vals[top_indices]
|
|
|
|
t_extended = np.arange(n + ahead)
|
|
reconstruction = np.zeros(n + ahead)
|
|
for idx in top_indices:
|
|
amp = np.abs(fft_vals[idx]) / n
|
|
phase = np.angle(fft_vals[idx])
|
|
freq = freqs[idx]
|
|
reconstruction += amp * np.cos(2 * np.pi * freq * t_extended / 1 + phase)
|
|
|
|
trend_slope = (series[-1] - series[0]) / n
|
|
trend_extended = series[0] + trend_slope * t_extended
|
|
reconstruction += trend_extended
|
|
|
|
return reconstruction
|
|
|
|
|
|
print(f"\nParametri: window={WINDOW}, components={N_COMPONENTS}, lookahead={LOOKAHEAD}")
|
|
print(f"Train: 0→{split_idx}, Test: {split_idx}→{n_total}")
|
|
|
|
signals = pd.Series(0, index=df.index)
|
|
accuracies = []
|
|
|
|
test_range = range(max(split_idx, WINDOW), n_total - LOOKAHEAD, STEP)
|
|
total_steps = len(list(test_range))
|
|
print(f"Valutazione: {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}%)")
|
|
|
|
window_data = close[i - WINDOW : i]
|
|
projected = fourier_project(window_data, N_COMPONENTS, LOOKAHEAD)
|
|
|
|
current_price = close[i - 1]
|
|
projected_price = projected[-1]
|
|
change_pct = (projected_price - current_price) / current_price
|
|
|
|
if change_pct > 0.005:
|
|
signals.iloc[i] = 1
|
|
elif change_pct < -0.005:
|
|
signals.iloc[i] = -1
|
|
|
|
actual_ret = (close[i + LOOKAHEAD - 1] - current_price) / current_price
|
|
if signals.iloc[i] == 1:
|
|
accuracies.append(1 if actual_ret > 0 else 0)
|
|
elif signals.iloc[i] == -1:
|
|
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}")
|
|
|
|
# Varianti con parametri diversi
|
|
print("\n\n--- VARIANTI PARAMETRI ---")
|
|
for n_comp in [5, 10, 15, 25, 50]:
|
|
for window in [144, 288, 588]:
|
|
sigs = pd.Series(0, index=df.index)
|
|
accs = []
|
|
test_r = range(max(split_idx, window), n_total - LOOKAHEAD, STEP)
|
|
for i in test_r:
|
|
w = close[i - window : i]
|
|
proj = fourier_project(w, n_comp, LOOKAHEAD)
|
|
cp = close[i - 1]
|
|
pp = proj[-1]
|
|
ch = (pp - cp) / cp
|
|
if ch > 0.005:
|
|
sigs.iloc[i] = 1
|
|
elif ch < -0.005:
|
|
sigs.iloc[i] = -1
|
|
ar = (close[i + LOOKAHEAD - 1] - cp) / cp
|
|
if sigs.iloc[i] == 1:
|
|
accs.append(1 if ar > 0 else 0)
|
|
elif sigs.iloc[i] == -1:
|
|
accs.append(1 if ar < 0 else 0)
|
|
|
|
if not accs:
|
|
continue
|
|
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
|
|
print(f" W={window:3d} N={n_comp:2d} → acc={acc:.1f}% trades={res.total_trades} ret={res.total_return*100:+.1f}% sharpe={res.sharpe_ratio:.2f}")
|