#!/usr/bin/env python3 """ Realistic Tick-Level Backtest ============================= Simula esecuzione ordini realistici con: - Fee Kraken maker 0.16% - Spread reale ~0.03% - Slippage basato su volume - Limit order simulation (non market) - 24h high tracking preciso """ import pandas as pd import numpy as np from datetime import datetime, timedelta from pathlib import Path import json from dataclasses import dataclass, field from typing import List, Dict, Optional, Tuple import warnings warnings.filterwarnings('ignore') # ============================================ # CONFIGURAZIONE REALISTICA KRAKEN # ============================================ KRAKEN_MAKER_FEE = 0.0016 # 0.16% maker fee KRAKEN_TAKER_FEE = 0.0026 # 0.26% taker fee (non usata) SPREAD_PCT = 0.0003 # 0.03% bid-ask spread SLIPPAGE_BASE = 0.0001 # 0.01% slippage base SLIPPAGE_VOLUME_FACTOR = 0.5 # Slippage aumenta con size # ============================================ # CONFIGURAZIONI TRADER # ============================================ TRADER_CONFIGS = { 'Conservative': { 'dip_threshold': 0.10, # Buy after 10% dip from 24h high 'profit_target': 0.05, # 5% take profit 'stop_loss': 0.03, # 3% stop loss 'position_size': 0.50, # 50% of capital per trade 'capital': 1667.0, }, 'Balanced': { 'dip_threshold': 0.07, # Buy after 7% dip (best from optimization) 'profit_target': 0.05, 'stop_loss': 0.03, 'position_size': 0.50, 'capital': 1667.0, }, 'Aggressive': { 'dip_threshold': 0.05, # Buy after 5% dip 'profit_target': 0.07, # 7% take profit 'stop_loss': 0.05, # 5% stop loss 'position_size': 0.50, 'capital': 1666.0, }, } @dataclass class Trade: """Singolo trade con tutti i dettagli.""" entry_time: datetime entry_price: float amount: float fee_entry: float exit_time: Optional[datetime] = None exit_price: Optional[float] = None fee_exit: float = 0 exit_reason: Optional[str] = None gross_pnl: float = 0 net_pnl: float = 0 pnl_pct: float = 0 duration_minutes: int = 0 dip_at_entry: float = 0 # Dip % when entered @dataclass class TraderState: """Stato corrente del trader.""" name: str config: Dict capital: float initial_capital: float position: int = 0 # 0=flat, 1=long position_amount: float = 0 entry_price: float = 0 entry_time: Optional[datetime] = None trades: List[Trade] = field(default_factory=list) equity_curve: List[Tuple[datetime, float]] = field(default_factory=list) def load_data() -> pd.DataFrame: """Carica tutti i dati disponibili.""" DATA_DIR = Path('/var/www/html/pippo.cuttalo.com/data') dfs = [] for f in sorted(DATA_DIR.glob('prices_BTC_EUR_*.csv')): print(f"Loading {f.name}...") df = pd.read_csv(f) # Parse timestamp if df['timestamp'].dtype == 'int64' or str(df['timestamp'].iloc[0]).isdigit(): ts = pd.to_numeric(df['timestamp']) df['timestamp'] = pd.to_datetime(ts, unit='s' if ts.iloc[0] < 1e12 else 'ms') else: df['timestamp'] = pd.to_datetime(df['timestamp']) dfs.append(df) prices = pd.concat(dfs, ignore_index=True) prices = prices.sort_values('timestamp').drop_duplicates(subset='timestamp').reset_index(drop=True) return prices def calculate_24h_high(prices: pd.DataFrame, lookback_minutes: int = 1440) -> pd.Series: """Calcola il massimo rolling delle ultime 24 ore.""" return prices['high'].rolling(window=lookback_minutes, min_periods=1).max() def simulate_limit_order_execution( order_price: float, candle_low: float, candle_high: float, is_buy: bool, volume: float = 1.0 ) -> Tuple[bool, float]: """ Simula esecuzione limit order. Returns: (filled, execution_price) """ if is_buy: # Buy limit: eseguito se il prezzo scende al nostro livello # Aggiungiamo spread (compriamo all'ask) ask_price = order_price * (1 + SPREAD_PCT) if candle_low <= ask_price: # Ordine eseguito - calcola slippage slippage = SLIPPAGE_BASE * (1 + SLIPPAGE_VOLUME_FACTOR * np.random.random()) exec_price = ask_price * (1 + slippage) return True, exec_price else: # Sell limit: eseguito se il prezzo sale al nostro livello # Vendiamo al bid bid_price = order_price * (1 - SPREAD_PCT) if candle_high >= bid_price: slippage = SLIPPAGE_BASE * (1 + SLIPPAGE_VOLUME_FACTOR * np.random.random()) exec_price = bid_price * (1 - slippage) return True, exec_price return False, 0 def check_stop_loss( entry_price: float, candle_low: float, stop_loss_pct: float ) -> Tuple[bool, float]: """ Verifica se lo stop loss è stato triggerato. Simula esecuzione market order (peggiore caso). """ stop_price = entry_price * (1 - stop_loss_pct) if candle_low <= stop_price: # Stop triggered - esecuzione a prezzo peggiore del range # Simula market sell con slippage exec_price = stop_price * (1 - SPREAD_PCT - SLIPPAGE_BASE * 2) return True, exec_price return False, 0 def check_take_profit( entry_price: float, candle_high: float, take_profit_pct: float ) -> Tuple[bool, float]: """ Verifica se il take profit è stato raggiunto. Limit order - esecuzione migliore. """ target_price = entry_price * (1 + take_profit_pct) if candle_high >= target_price: # TP raggiunto - esecuzione al prezzo target meno spread exec_price = target_price * (1 - SPREAD_PCT) return True, exec_price return False, 0 def run_backtest(prices: pd.DataFrame, config_name: str, config: Dict) -> TraderState: """Esegue backtest per un singolo trader.""" trader = TraderState( name=config_name, config=config, capital=config['capital'], initial_capital=config['capital'], ) # Calcola 24h high per ogni candela prices['high_24h'] = calculate_24h_high(prices) # Start dopo 24h di warmup start_idx = 1440 print(f"\n{'='*60}") print(f"BACKTEST: {config_name}") print(f"{'='*60}") print(f"Capitale iniziale: €{config['capital']:,.2f}") print(f"Dip threshold: {config['dip_threshold']*100:.0f}%") print(f"Take profit: {config['profit_target']*100:.0f}%") print(f"Stop loss: {config['stop_loss']*100:.0f}%") print(f"Position size: {config['position_size']*100:.0f}%") trades_count = 0 for i in range(start_idx, len(prices)): row = prices.iloc[i] timestamp = row['timestamp'] current_price = row['close'] candle_high = row['high'] candle_low = row['low'] high_24h = row['high_24h'] # Calcola dip corrente dip_from_high = (current_price - high_24h) / high_24h if high_24h > 0 else 0 # ======================================== # SE IN POSIZIONE: CHECK EXIT # ======================================== if trader.position == 1: entry_price = trader.entry_price # Check STOP LOSS first (priorità) sl_hit, sl_price = check_stop_loss( entry_price, candle_low, config['stop_loss'] ) if sl_hit: # Esegui stop loss amount = trader.position_amount gross_value = amount * sl_price fee = gross_value * KRAKEN_MAKER_FEE gross_pnl = amount * (sl_price - entry_price) net_pnl = gross_pnl - fee - trader.trades[-1].fee_entry pnl_pct = net_pnl / (amount * entry_price) # Aggiorna trade trade = trader.trades[-1] trade.exit_time = timestamp trade.exit_price = sl_price trade.fee_exit = fee trade.exit_reason = 'stop_loss' trade.gross_pnl = gross_pnl trade.net_pnl = net_pnl trade.pnl_pct = pnl_pct trade.duration_minutes = int((timestamp - trade.entry_time).total_seconds() / 60) # Aggiorna capitale trader.capital += (amount * entry_price) + net_pnl trader.position = 0 trader.position_amount = 0 continue # Check TAKE PROFIT tp_hit, tp_price = check_take_profit( entry_price, candle_high, config['profit_target'] ) if tp_hit: amount = trader.position_amount gross_value = amount * tp_price fee = gross_value * KRAKEN_MAKER_FEE gross_pnl = amount * (tp_price - entry_price) net_pnl = gross_pnl - fee - trader.trades[-1].fee_entry pnl_pct = net_pnl / (amount * entry_price) trade = trader.trades[-1] trade.exit_time = timestamp trade.exit_price = tp_price trade.fee_exit = fee trade.exit_reason = 'take_profit' trade.gross_pnl = gross_pnl trade.net_pnl = net_pnl trade.pnl_pct = pnl_pct trade.duration_minutes = int((timestamp - trade.entry_time).total_seconds() / 60) trader.capital += (amount * entry_price) + net_pnl trader.position = 0 trader.position_amount = 0 continue # ======================================== # SE FLAT: CHECK ENTRY # ======================================== elif trader.position == 0: # Condizione: prezzo sceso di X% dal max 24h if dip_from_high <= -config['dip_threshold']: # Simula limit buy order trade_value = trader.capital * config['position_size'] filled, exec_price = simulate_limit_order_execution( current_price, candle_low, candle_high, is_buy=True ) if filled and trade_value > 10: # Minimo €10 fee = trade_value * KRAKEN_MAKER_FEE amount = (trade_value - fee) / exec_price trade = Trade( entry_time=timestamp, entry_price=exec_price, amount=amount, fee_entry=fee, dip_at_entry=dip_from_high, ) trader.trades.append(trade) trader.capital -= trade_value trader.position = 1 trader.position_amount = amount trader.entry_price = exec_price trader.entry_time = timestamp trades_count += 1 # Salva equity ogni ora if i % 60 == 0: if trader.position == 1: unrealized = trader.position_amount * (current_price - trader.entry_price) equity = trader.capital + trader.position_amount * trader.entry_price + unrealized else: equity = trader.capital trader.equity_curve.append((timestamp, equity)) # Chiudi posizione aperta alla fine if trader.position == 1: final_price = prices.iloc[-1]['close'] amount = trader.position_amount entry_price = trader.entry_price fee = amount * final_price * KRAKEN_MAKER_FEE gross_pnl = amount * (final_price - entry_price) net_pnl = gross_pnl - fee - trader.trades[-1].fee_entry pnl_pct = net_pnl / (amount * entry_price) trade = trader.trades[-1] trade.exit_time = prices.iloc[-1]['timestamp'] trade.exit_price = final_price trade.fee_exit = fee trade.exit_reason = 'end_of_data' trade.gross_pnl = gross_pnl trade.net_pnl = net_pnl trade.pnl_pct = pnl_pct trader.capital += (amount * entry_price) + net_pnl trader.position = 0 return trader def print_results(trader: TraderState, buy_hold_return: float): """Stampa risultati dettagliati.""" closed_trades = [t for t in trader.trades if t.exit_time is not None] if not closed_trades: print(f"\n{trader.name}: Nessun trade eseguito") return # Calcoli total_pnl = sum(t.net_pnl for t in closed_trades) total_fees = sum(t.fee_entry + t.fee_exit for t in closed_trades) total_gross = sum(t.gross_pnl for t in closed_trades) wins = [t for t in closed_trades if t.net_pnl > 0] losses = [t for t in closed_trades if t.net_pnl <= 0] win_rate = len(wins) / len(closed_trades) if closed_trades else 0 avg_win = np.mean([t.net_pnl for t in wins]) if wins else 0 avg_loss = np.mean([t.net_pnl for t in losses]) if losses else 0 profit_factor = abs(sum(t.net_pnl for t in wins) / sum(t.net_pnl for t in losses)) if losses and sum(t.net_pnl for t in losses) != 0 else float('inf') avg_duration = np.mean([t.duration_minutes for t in closed_trades]) # Max drawdown equity = [e[1] for e in trader.equity_curve] if equity: peak = np.maximum.accumulate(equity) drawdown = (np.array(equity) - peak) / peak max_dd = abs(min(drawdown)) if len(drawdown) > 0 else 0 else: max_dd = 0 return_pct = (trader.capital - trader.initial_capital) / trader.initial_capital * 100 # TP vs SL breakdown tp_trades = [t for t in closed_trades if t.exit_reason == 'take_profit'] sl_trades = [t for t in closed_trades if t.exit_reason == 'stop_loss'] print(f"\n{'='*60}") print(f"RISULTATI: {trader.name}") print(f"{'='*60}") print(f"\n--- CAPITALE ---") print(f"Iniziale: €{trader.initial_capital:>12,.2f}") print(f"Finale: €{trader.capital:>12,.2f}") print(f"P&L Netto: €{total_pnl:>12,.2f} ({return_pct:+.2f}%)") print(f"P&L Lordo: €{total_gross:>12,.2f}") print(f"Fee Totali: €{total_fees:>12,.2f}") print(f"\n--- TRADE ---") print(f"Totali: {len(closed_trades):>12}") print(f"Vincenti: {len(wins):>12} ({win_rate*100:.1f}%)") print(f"Perdenti: {len(losses):>12}") print(f"Take Profit: {len(tp_trades):>12}") print(f"Stop Loss: {len(sl_trades):>12}") print(f"\n--- PERFORMANCE ---") print(f"Avg Win: €{avg_win:>12,.2f}") print(f"Avg Loss: €{avg_loss:>12,.2f}") print(f"Profit Factor: {profit_factor:>12.2f}") print(f"Max Drawdown: {max_dd*100:>11.2f}%") print(f"Avg Duration: {avg_duration:>10.0f} min") print(f"\n--- VS BUY & HOLD ---") print(f"Buy & Hold: {buy_hold_return:>11.2f}%") print(f"Strategia: {return_pct:>11.2f}%") print(f"Alpha: {return_pct - buy_hold_return:>+11.2f}%") # Sample trades print(f"\n--- SAMPLE TRADES (ultimi 5) ---") for t in closed_trades[-5:]: emoji = "+" if t.net_pnl > 0 else "" print(f" {t.entry_time.strftime('%Y-%m-%d %H:%M')} | " f"Dip {t.dip_at_entry*100:+.1f}% | " f"€{t.entry_price:,.0f} -> €{t.exit_price:,.0f} | " f"{t.exit_reason:12} | " f"{emoji}€{t.net_pnl:.2f}") return { 'name': trader.name, 'initial_capital': trader.initial_capital, 'final_capital': trader.capital, 'return_pct': return_pct, 'total_pnl': total_pnl, 'total_fees': total_fees, 'num_trades': len(closed_trades), 'win_rate': win_rate, 'profit_factor': profit_factor, 'max_drawdown': max_dd, 'avg_duration_min': avg_duration, 'tp_trades': len(tp_trades), 'sl_trades': len(sl_trades), 'buy_hold_return': buy_hold_return, 'alpha': return_pct - buy_hold_return, } def main(): print("="*60) print("BACKTEST REALISTICO - DIP BUYER STRATEGY") print("="*60) print(f"Fee Kraken Maker: {KRAKEN_MAKER_FEE*100:.2f}%") print(f"Spread: {SPREAD_PCT*100:.3f}%") print(f"Slippage base: {SLIPPAGE_BASE*100:.3f}%") # Load data prices = load_data() print(f"\nDati: {len(prices):,} candele") print(f"Periodo: {prices['timestamp'].min()} -> {prices['timestamp'].max()}") # Buy & Hold benchmark start_price = prices.iloc[1440]['close'] # Dopo warmup end_price = prices.iloc[-1]['close'] buy_hold_return = (end_price - start_price) / start_price * 100 print(f"\nBuy & Hold: {start_price:,.2f} -> {end_price:,.2f} = {buy_hold_return:+.2f}%") # Run backtests results = [] for name, config in TRADER_CONFIGS.items(): trader = run_backtest(prices.copy(), name, config) result = print_results(trader, buy_hold_return) if result: results.append(result) # Summary print("\n" + "="*60) print("RIEPILOGO FINALE") print("="*60) print(f"\n{'Trader':<15} {'Return':>10} {'Trades':>8} {'Win%':>8} {'MaxDD':>8} {'Alpha':>10}") print("-"*60) for r in results: print(f"{r['name']:<15} {r['return_pct']:>+9.2f}% {r['num_trades']:>8} " f"{r['win_rate']*100:>7.1f}% {r['max_drawdown']*100:>7.1f}% {r['alpha']:>+9.2f}%") print("-"*60) print(f"{'Buy & Hold':<15} {buy_hold_return:>+9.2f}%") # Capitale totale total_initial = sum(r['initial_capital'] for r in results) total_final = sum(r['final_capital'] for r in results) total_return = (total_final - total_initial) / total_initial * 100 total_fees = sum(r['total_fees'] for r in results) print(f"\n--- TOTALE PORTFOLIO (3 trader) ---") print(f"Capitale iniziale: €{total_initial:,.2f}") print(f"Capitale finale: €{total_final:,.2f}") print(f"Return totale: {total_return:+.2f}%") print(f"Fee totali pagate: €{total_fees:,.2f}") print(f"Alpha vs B&H: {total_return - buy_hold_return:+.2f}%") # Save results output = { 'timestamp': datetime.now().isoformat(), 'period': f"{prices['timestamp'].min()} to {prices['timestamp'].max()}", 'total_candles': len(prices), 'fee_maker': KRAKEN_MAKER_FEE, 'spread': SPREAD_PCT, 'buy_hold_return_pct': buy_hold_return, 'traders': results, 'portfolio': { 'initial_capital': total_initial, 'final_capital': total_final, 'return_pct': total_return, 'total_fees': total_fees, 'alpha': total_return - buy_hold_return, } } output_path = Path('/var/www/html/pippo.cuttalo.com/optimization_results/realistic_backtest.json') with open(output_path, 'w') as f: json.dump(output, f, indent=2, default=str) print(f"\nRisultati salvati in: {output_path}") if __name__ == '__main__': main()