refactor: riorganizzazione script — Strategy ABC, folder strategies/waste/analysis
- 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>
This commit is contained in:
@@ -0,0 +1,243 @@
|
||||
"""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
|
||||
Reference in New Issue
Block a user