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>
244 lines
7.8 KiB
Python
244 lines
7.8 KiB
Python
"""Classe base astratta per tutte le strategie di trading."""
|
|
from __future__ import annotations
|
|
|
|
from abc import ABC, abstractmethod
|
|
from dataclasses import dataclass, field
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
|
|
from src.data.downloader import load_data
|
|
|
|
|
|
@dataclass
|
|
class Signal:
|
|
"""Segnale di trading generato da una strategia."""
|
|
idx: int
|
|
direction: int # +1 long, -1 short
|
|
entry_price: float
|
|
metadata: dict = field(default_factory=dict)
|
|
|
|
|
|
@dataclass
|
|
class YearlyStats:
|
|
year: int
|
|
trades: int
|
|
wins: int
|
|
pnl: float
|
|
|
|
@property
|
|
def accuracy(self) -> float:
|
|
return self.wins / self.trades * 100 if self.trades > 0 else 0.0
|
|
|
|
|
|
@dataclass
|
|
class BacktestResult:
|
|
"""Risultato completo di un backtest."""
|
|
strategy_name: str
|
|
asset: str
|
|
timeframe: str
|
|
params: dict
|
|
|
|
trades: int
|
|
wins: int
|
|
pnl: float
|
|
capital: float
|
|
initial_capital: float
|
|
max_dd: float
|
|
time_in_market_pct: float
|
|
avg_trade_duration_h: float
|
|
years_active: int
|
|
yearly: list[YearlyStats]
|
|
|
|
@property
|
|
def accuracy(self) -> float:
|
|
return self.wins / self.trades * 100 if self.trades > 0 else 0.0
|
|
|
|
@property
|
|
def sharpe(self) -> float:
|
|
pnls = []
|
|
for ys in self.yearly:
|
|
pnls.append(ys.pnl)
|
|
if len(pnls) < 2 or np.std(pnls) == 0:
|
|
return 0.0
|
|
return float(np.mean(pnls) / np.std(pnls) * np.sqrt(len(pnls)))
|
|
|
|
@property
|
|
def daily_pnl(self) -> float:
|
|
return self.pnl / (self.years_active * 365) if self.years_active > 0 else 0.0
|
|
|
|
@property
|
|
def worst_year(self) -> YearlyStats | None:
|
|
valid = [y for y in self.yearly if y.trades >= 10]
|
|
if not valid:
|
|
valid = self.yearly
|
|
return min(valid, key=lambda y: y.accuracy) if valid else None
|
|
|
|
def print_summary(self):
|
|
worst = self.worst_year
|
|
worst_str = f"{worst.year}({worst.accuracy:.0f}%)" if worst else "N/A"
|
|
dur = f"{self.avg_trade_duration_h:.0f}h" if self.avg_trade_duration_h >= 1 else f"{self.avg_trade_duration_h * 60:.0f}m"
|
|
print(f" {self.strategy_name:<30s} {self.asset:>3s} {self.timeframe:>3s} "
|
|
f"{self.trades:>5d}t {self.accuracy:>5.1f}% "
|
|
f"€{self.pnl:>+9.0f} DD {self.max_dd:>4.1f}% "
|
|
f"€/d {self.daily_pnl:>+6.2f} "
|
|
f"Mkt {self.time_in_market_pct:>4.1f}% {dur:>5s} "
|
|
f"worst={worst_str} {self.years_active}y")
|
|
|
|
def print_yearly(self):
|
|
print(f"\n {self.strategy_name} [{self.asset} {self.timeframe}] — per anno:")
|
|
print(f" {'Anno':>6s} {'Trades':>7s} {'Acc':>6s} {'PnL€':>9s}")
|
|
for ys in sorted(self.yearly, key=lambda y: y.year):
|
|
print(f" {ys.year:>6d} {ys.trades:>7d} {ys.accuracy:>5.1f}% €{ys.pnl:>+8.0f}")
|
|
|
|
|
|
TF_MINUTES = {"1m": 1, "5m": 5, "15m": 15, "1h": 60, "4h": 240, "1d": 1440}
|
|
|
|
|
|
class Strategy(ABC):
|
|
"""Classe base per tutte le strategie.
|
|
|
|
Sottoclassi devono implementare:
|
|
- name, description, default_assets, default_timeframes
|
|
- generate_signals(df, timestamps, **params) -> list[Signal]
|
|
"""
|
|
|
|
name: str = "unnamed"
|
|
description: str = ""
|
|
default_assets: list[str] = ["BTC", "ETH"]
|
|
default_timeframes: list[str] = ["15m", "1h"]
|
|
|
|
# Parametri di backtest
|
|
fee_rt: float = 0.002
|
|
leverage: float = 3.0
|
|
position_size: float = 0.15
|
|
initial_capital: float = 1000.0
|
|
|
|
@abstractmethod
|
|
def generate_signals(self, df: pd.DataFrame, ts: pd.DatetimeIndex,
|
|
**params) -> list[Signal]:
|
|
"""Genera segnali di trading dal dataframe OHLCV.
|
|
|
|
Args:
|
|
df: DataFrame con colonne open, high, low, close, volume, timestamp
|
|
ts: DatetimeIndex UTC dei timestamp
|
|
**params: parametri specifici della strategia
|
|
|
|
Returns:
|
|
Lista di Signal con idx, direction, entry_price
|
|
"""
|
|
...
|
|
|
|
def backtest(self, asset: str, tf: str, hold: int = 3,
|
|
**params) -> BacktestResult | None:
|
|
"""Esegue backtest su un asset/timeframe."""
|
|
df = load_data(asset, tf)
|
|
c = df["close"].values
|
|
n = len(c)
|
|
ts = pd.to_datetime(df["timestamp"], unit="ms", utc=True)
|
|
|
|
sig_params = {**params, "asset": asset, "tf": tf}
|
|
signals = self.generate_signals(df, ts, **sig_params)
|
|
if not signals:
|
|
return None
|
|
|
|
yearly: dict[int, dict] = {}
|
|
capital = float(self.initial_capital)
|
|
peak = capital
|
|
max_dd = 0.0
|
|
total_bars = 0
|
|
|
|
for sig in signals:
|
|
i = sig.idx
|
|
if i + hold >= n or i < 1:
|
|
continue
|
|
|
|
entry = sig.entry_price
|
|
exit_price = c[min(i + hold - 1, n - 1)]
|
|
actual = (exit_price - entry) / entry * sig.direction
|
|
net = actual * self.leverage - self.fee_rt * self.leverage
|
|
|
|
capital += capital * self.position_size * net
|
|
capital = max(capital, 10)
|
|
if capital > peak:
|
|
peak = capital
|
|
dd = (peak - capital) / peak
|
|
max_dd = max(max_dd, dd)
|
|
total_bars += hold
|
|
|
|
year = ts.iloc[i].year
|
|
if year not in yearly:
|
|
yearly[year] = {"w": 0, "t": 0, "pnl": 0.0}
|
|
yearly[year]["t"] += 1
|
|
if actual > 0:
|
|
yearly[year]["w"] += 1
|
|
yearly[year]["pnl"] += net * self.initial_capital
|
|
|
|
all_t = sum(d["t"] for d in yearly.values())
|
|
all_w = sum(d["w"] for d in yearly.values())
|
|
if all_t == 0:
|
|
return None
|
|
|
|
yearly_stats = [
|
|
YearlyStats(year=y, trades=d["t"], wins=d["w"], pnl=d["pnl"])
|
|
for y, d in sorted(yearly.items())
|
|
]
|
|
|
|
return BacktestResult(
|
|
strategy_name=self.name,
|
|
asset=asset,
|
|
timeframe=tf,
|
|
params=params,
|
|
trades=all_t,
|
|
wins=all_w,
|
|
pnl=sum(d["pnl"] for d in yearly.values()),
|
|
capital=capital,
|
|
initial_capital=self.initial_capital,
|
|
max_dd=max_dd * 100,
|
|
time_in_market_pct=total_bars / n * 100,
|
|
avg_trade_duration_h=hold * TF_MINUTES.get(tf, 60) / 60,
|
|
years_active=len(yearly),
|
|
yearly=yearly_stats,
|
|
)
|
|
|
|
def run_all(self, assets: list[str] | None = None,
|
|
timeframes: list[str] | None = None,
|
|
hold: int = 3, **params) -> list[BacktestResult]:
|
|
"""Esegue backtest su tutte le combinazioni asset/timeframe."""
|
|
assets = assets or self.default_assets
|
|
timeframes = timeframes or self.default_timeframes
|
|
results = []
|
|
for asset in assets:
|
|
for tf in timeframes:
|
|
r = self.backtest(asset, tf, hold=hold, **params)
|
|
if r and r.trades >= 20:
|
|
results.append(r)
|
|
results.sort(key=lambda r: r.accuracy, reverse=True)
|
|
return results
|
|
|
|
def report(self, results: list[BacktestResult] | None = None,
|
|
assets: list[str] | None = None,
|
|
timeframes: list[str] | None = None,
|
|
hold: int = 3, **params):
|
|
"""Esegue e stampa report completo."""
|
|
if results is None:
|
|
results = self.run_all(assets, timeframes, hold, **params)
|
|
|
|
print(f"\n{'=' * 120}")
|
|
print(f" {self.name} — {self.description}")
|
|
print(f" Fee: {self.fee_rt*100:.1f}% RT | Leva: {self.leverage:.0f}x | Pos: {self.position_size*100:.0f}%")
|
|
print(f"{'=' * 120}")
|
|
print(f" {'Nome':<30s} {'A/T':>7s} {'Trades':>6s} {'Acc':>6s} "
|
|
f"{'PnL€':>10s} {'DD%':>6s} {'€/day':>7s} "
|
|
f"{'Mkt%':>5s} {'Dur':>5s} {'Worst':>12s} {'Anni':>4s}")
|
|
print(f" {'─' * 110}")
|
|
|
|
for r in results:
|
|
r.print_summary()
|
|
|
|
if results:
|
|
best = results[0]
|
|
best.print_yearly()
|
|
|
|
return results
|