#!/usr/bin/env python3 """ Close-Only Backtest - Simula ESATTAMENTE il comportamento LIVE =============================================================== Questo backtest usa SOLO il prezzo close per trigger SL/TP, esattamente come fa il live engine. Obiettivo: Vedere quanto peggiora il risultato rispetto al backtest che usa HIGH/LOW per i trigger. """ import pandas as pd import numpy as np from datetime import datetime from pathlib import Path import json from dataclasses import dataclass, field from typing import List, Dict, Optional, Tuple import warnings warnings.filterwarnings('ignore') DATA_DIR = Path('/var/www/html/pippo.cuttalo.com/data') OUTPUT_DIR = Path('/var/www/html/pippo.cuttalo.com/optimization_results') # Fee Kraken (identiche a live) KRAKEN_MAKER_FEE = 0.0016 SPREAD = 0.0003 # Strategia Alpha_24h STRATEGY_CONFIG = { 'name': 'Alpha_24h', 'dip_threshold': 0.08, 'profit_target': 0.15, 'stop_loss': 0.12, 'position_size': 1.0, 'lookback_minutes': 1440, } @dataclass class Trade: 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 # Tracking differenze intracandle_sl_triggered: bool = False # Avrebbe triggerato con LOW intracandle_tp_triggered: bool = False # Avrebbe triggerato con HIGH def load_data() -> pd.DataFrame: dfs = [] for f in sorted(DATA_DIR.glob('prices_BTC_EUR_*.csv')): print(f"Loading {f.name}...") df = pd.read_csv(f) if df['timestamp'].dtype != 'object': if df['timestamp'].iloc[0] < 1e12: df['timestamp'] = pd.to_datetime(df['timestamp'], unit='s') else: df['timestamp'] = pd.to_datetime(df['timestamp'], unit='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 run_close_only_backtest(prices: pd.DataFrame, config: Dict) -> Tuple[List[Trade], Dict]: """ Backtest che usa SOLO close per trigger, come il live. Traccia anche quando HIGH/LOW avrebbero triggerato. """ prices = prices.copy() prices['high_24h'] = prices['high'].rolling(window=config['lookback_minutes'], min_periods=1).max() trades: List[Trade] = [] in_position = False entry_price = 0 entry_time = None entry_amount = 0 entry_fee = 0 capital = 5000.0 # Tracking differenze missed_tp_count = 0 missed_sl_count = 0 late_sl_extra_loss = 0 late_tp_missed_gain = 0 start_idx = 1440 print(f"\nClose-Only Backtest (simula LIVE)") print(f"Periodo: {prices.iloc[start_idx]['timestamp']} -> {prices.iloc[-1]['timestamp']}") print(f"Config: {config['name']} - Dip {config['dip_threshold']*100}% / TP {config['profit_target']*100}% / SL {config['stop_loss']*100}%") for i in range(start_idx, len(prices)): row = prices.iloc[i] timestamp = row['timestamp'] current_close = row['close'] # SOLO CLOSE! candle_high = row['high'] candle_low = row['low'] high_24h = row['high_24h'] dip_from_high = (current_close - high_24h) / high_24h if high_24h > 0 else 0 if in_position: # Calcola P&L usando SOLO CLOSE (come live) pnl_pct_close = (current_close - entry_price) / entry_price # Calcola anche con HIGH/LOW per tracking pnl_pct_low = (candle_low - entry_price) / entry_price pnl_pct_high = (candle_high - entry_price) / entry_price # Check se HIGH/LOW avrebbero triggerato would_sl_on_low = pnl_pct_low <= -config['stop_loss'] would_tp_on_high = pnl_pct_high >= config['profit_target'] # Ma noi triggeriamo solo su CLOSE trigger_sl = pnl_pct_close <= -config['stop_loss'] trigger_tp = pnl_pct_close >= config['profit_target'] # Tracking missed opportunities if would_sl_on_low and not trigger_sl: missed_sl_count += 1 if would_tp_on_high and not trigger_tp: missed_tp_count += 1 # Exit logic (close only) if trigger_sl: exec_price = current_close * (1 - SPREAD) fee = entry_amount * exec_price * KRAKEN_MAKER_FEE gross_pnl = entry_amount * (exec_price - entry_price) net_pnl = gross_pnl - fee - entry_fee final_pnl_pct = net_pnl / (entry_amount * entry_price) duration = int((timestamp - entry_time).total_seconds() / 60) trade = trades[-1] trade.exit_time = timestamp trade.exit_price = exec_price trade.fee_exit = fee trade.exit_reason = 'stop_loss' trade.gross_pnl = gross_pnl trade.net_pnl = net_pnl trade.pnl_pct = final_pnl_pct trade.duration_minutes = duration # Se il LOW era peggio del close, abbiamo perso di meno # Ma il backtest originale avrebbe stoppato prima! if would_sl_on_low and pnl_pct_low < pnl_pct_close: # Il close e migliore del low, quindi nel backtest originale # avremmo perso di piu (usciti al low) trade.intracandle_sl_triggered = True capital += (entry_amount * entry_price) + net_pnl in_position = False continue if trigger_tp: exec_price = current_close * (1 - SPREAD) fee = entry_amount * exec_price * KRAKEN_MAKER_FEE gross_pnl = entry_amount * (exec_price - entry_price) net_pnl = gross_pnl - fee - entry_fee final_pnl_pct = net_pnl / (entry_amount * entry_price) duration = int((timestamp - entry_time).total_seconds() / 60) trade = trades[-1] trade.exit_time = timestamp trade.exit_price = exec_price trade.fee_exit = fee trade.exit_reason = 'take_profit' trade.gross_pnl = gross_pnl trade.net_pnl = net_pnl trade.pnl_pct = final_pnl_pct trade.duration_minutes = duration capital += (entry_amount * entry_price) + net_pnl in_position = False continue else: # Entry logic - usa close (come live) if dip_from_high <= -config['dip_threshold']: trade_value = capital * config['position_size'] if trade_value > 10: exec_price = current_close * (1 + SPREAD) 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, ) trades.append(trade) capital -= trade_value in_position = True entry_price = exec_price entry_time = timestamp entry_amount = amount entry_fee = fee # Chiudi posizione aperta if in_position: final_price = prices.iloc[-1]['close'] exec_price = final_price * (1 - SPREAD) fee = entry_amount * exec_price * KRAKEN_MAKER_FEE gross_pnl = entry_amount * (exec_price - entry_price) net_pnl = gross_pnl - fee - entry_fee final_pnl_pct = net_pnl / (entry_amount * entry_price) trade = trades[-1] trade.exit_time = prices.iloc[-1]['timestamp'] trade.exit_price = exec_price trade.fee_exit = fee trade.exit_reason = 'end_of_data' trade.gross_pnl = gross_pnl trade.net_pnl = net_pnl trade.pnl_pct = final_pnl_pct capital += (entry_amount * entry_price) + net_pnl stats = { 'missed_tp_intracandle': missed_tp_count, 'missed_sl_intracandle': missed_sl_count, 'final_capital': capital, } return trades, stats def run_highlow_backtest(prices: pd.DataFrame, config: Dict) -> Tuple[List[Trade], Dict]: """ Backtest originale che usa HIGH/LOW per trigger. """ prices = prices.copy() prices['high_24h'] = prices['high'].rolling(window=config['lookback_minutes'], min_periods=1).max() trades: List[Trade] = [] in_position = False entry_price = 0 entry_time = None entry_amount = 0 entry_fee = 0 capital = 5000.0 start_idx = 1440 print(f"\nHigh/Low Backtest (originale)") print(f"Periodo: {prices.iloc[start_idx]['timestamp']} -> {prices.iloc[-1]['timestamp']}") for i in range(start_idx, len(prices)): row = prices.iloc[i] timestamp = row['timestamp'] current_close = row['close'] candle_high = row['high'] candle_low = row['low'] high_24h = row['high_24h'] dip_from_high = (current_close - high_24h) / high_24h if high_24h > 0 else 0 if in_position: stop_price = entry_price * (1 - config['stop_loss']) target_price = entry_price * (1 + config['profit_target']) # STOP LOSS su LOW if candle_low <= stop_price: exec_price = stop_price * (1 - SPREAD) fee = entry_amount * exec_price * KRAKEN_MAKER_FEE gross_pnl = entry_amount * (exec_price - entry_price) net_pnl = gross_pnl - fee - entry_fee final_pnl_pct = net_pnl / (entry_amount * entry_price) duration = int((timestamp - entry_time).total_seconds() / 60) trade = trades[-1] trade.exit_time = timestamp trade.exit_price = exec_price trade.fee_exit = fee trade.exit_reason = 'stop_loss' trade.gross_pnl = gross_pnl trade.net_pnl = net_pnl trade.pnl_pct = final_pnl_pct trade.duration_minutes = duration capital += (entry_amount * entry_price) + net_pnl in_position = False continue # TAKE PROFIT su HIGH if candle_high >= target_price: exec_price = target_price * (1 - SPREAD) fee = entry_amount * exec_price * KRAKEN_MAKER_FEE gross_pnl = entry_amount * (exec_price - entry_price) net_pnl = gross_pnl - fee - entry_fee final_pnl_pct = net_pnl / (entry_amount * entry_price) duration = int((timestamp - entry_time).total_seconds() / 60) trade = trades[-1] trade.exit_time = timestamp trade.exit_price = exec_price trade.fee_exit = fee trade.exit_reason = 'take_profit' trade.gross_pnl = gross_pnl trade.net_pnl = net_pnl trade.pnl_pct = final_pnl_pct trade.duration_minutes = duration capital += (entry_amount * entry_price) + net_pnl in_position = False continue else: if dip_from_high <= -config['dip_threshold']: trade_value = capital * config['position_size'] if trade_value > 10: exec_price = current_close * (1 + SPREAD) 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, ) trades.append(trade) capital -= trade_value in_position = True entry_price = exec_price entry_time = timestamp entry_amount = amount entry_fee = fee if in_position: final_price = prices.iloc[-1]['close'] exec_price = final_price * (1 - SPREAD) fee = entry_amount * exec_price * KRAKEN_MAKER_FEE gross_pnl = entry_amount * (exec_price - entry_price) net_pnl = gross_pnl - fee - entry_fee final_pnl_pct = net_pnl / (entry_amount * entry_price) trade = trades[-1] trade.exit_time = prices.iloc[-1]['timestamp'] trade.exit_price = exec_price trade.fee_exit = fee trade.exit_reason = 'end_of_data' trade.gross_pnl = gross_pnl trade.net_pnl = net_pnl trade.pnl_pct = final_pnl_pct capital += (entry_amount * entry_price) + net_pnl return trades, {'final_capital': capital} def analyze_trades(trades: List[Trade], name: str) -> Dict: closed = [t for t in trades if t.exit_time is not None] if not closed: return {} wins = [t for t in closed if t.net_pnl > 0] losses = [t for t in closed if t.net_pnl <= 0] tp_trades = [t for t in closed if t.exit_reason == 'take_profit'] sl_trades = [t for t in closed if t.exit_reason == 'stop_loss'] total_pnl = sum(t.net_pnl for t in closed) win_rate = len(wins) / len(closed) * 100 if closed 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') return { 'name': name, 'total_trades': len(closed), 'wins': len(wins), 'losses': len(losses), 'win_rate': win_rate, 'total_pnl': total_pnl, 'avg_win': avg_win, 'avg_loss': avg_loss, 'profit_factor': profit_factor, 'tp_count': len(tp_trades), 'sl_count': len(sl_trades), } def main(): print("="*70) print("CONFRONTO: CLOSE-ONLY (Live) vs HIGH/LOW (Backtest originale)") print("="*70) # Load data prices = load_data() print(f"\nDati: {len(prices):,} candele 1m") print(f"Periodo: {prices['timestamp'].min()} -> {prices['timestamp'].max()}") # Buy & Hold start_price = prices.iloc[1440]['close'] end_price = prices.iloc[-1]['close'] buy_hold_return = (end_price - start_price) / start_price * 100 print(f"\nBuy & Hold: €{start_price:,.0f} -> €{end_price:,.0f} = {buy_hold_return:+.2f}%") # Run both backtests close_trades, close_stats = run_close_only_backtest(prices, STRATEGY_CONFIG) highlow_trades, highlow_stats = run_highlow_backtest(prices, STRATEGY_CONFIG) # Analyze close_results = analyze_trades(close_trades, 'CLOSE-ONLY (Live)') highlow_results = analyze_trades(highlow_trades, 'HIGH/LOW (Backtest)') # Print comparison print("\n" + "="*70) print("CONFRONTO RISULTATI") print("="*70) print(f"\n{'Metrica':<25} {'HIGH/LOW':>15} {'CLOSE-ONLY':>15} {'Delta':>15}") print("-"*70) metrics = [ ('Trade totali', 'total_trades', '', 0), ('Vincenti', 'wins', '', 0), ('Perdenti', 'losses', '', 0), ('Win Rate', 'win_rate', '%', 1), ('P&L Totale', 'total_pnl', '', 2), ('Media Vincita', 'avg_win', '', 2), ('Media Perdita', 'avg_loss', '', 2), ('Profit Factor', 'profit_factor', '', 2), ('Take Profit', 'tp_count', '', 0), ('Stop Loss', 'sl_count', '', 0), ] for label, key, suffix, decimals in metrics: hl_val = highlow_results.get(key, 0) co_val = close_results.get(key, 0) delta = co_val - hl_val if decimals == 0: print(f"{label:<25} {hl_val:>15.0f}{suffix} {co_val:>15.0f}{suffix} {delta:>+15.0f}") elif decimals == 1: print(f"{label:<25} {hl_val:>14.1f}{suffix} {co_val:>14.1f}{suffix} {delta:>+14.1f}{suffix}") else: print(f"{label:<25} {hl_val:>15.2f}{suffix} {co_val:>15.2f}{suffix} {delta:>+15.2f}") # Return comparison initial_cap = 5000 hl_return = (highlow_stats['final_capital'] - initial_cap) / initial_cap * 100 co_return = (close_stats['final_capital'] - initial_cap) / initial_cap * 100 print("-"*70) print(f"{'Capitale Finale':<25} €{highlow_stats['final_capital']:>14,.2f} €{close_stats['final_capital']:>14,.2f} €{close_stats['final_capital']-highlow_stats['final_capital']:>+14,.2f}") print(f"{'Return %':<25} {hl_return:>14.2f}% {co_return:>14.2f}% {co_return-hl_return:>+14.2f}%") print(f"{'Alpha vs B&H':<25} {hl_return-buy_hold_return:>+14.2f}% {co_return-buy_hold_return:>+14.2f}%") # Additional stats print(f"\n--- OPPORTUNITA PERSE (Close-Only) ---") print(f"TP mancati (spike su HIGH): {close_stats.get('missed_tp_intracandle', 'N/A')}") print(f"SL mancati (drop su LOW): {close_stats.get('missed_sl_intracandle', 'N/A')}") # Conclusioni print("\n" + "="*70) print("CONCLUSIONI") print("="*70) delta_return = co_return - hl_return if delta_return < 0: print(f"\nIl LIVE (close-only) performa {abs(delta_return):.1f}% PEGGIO del backtest.") print(f"Questa e la discrepanza reale tra backtest e live!") else: print(f"\nIl LIVE (close-only) performa {delta_return:.1f}% MEGLIO del backtest.") print(f"Questo e inaspettato - verificare i dati.") # Save results output = { 'timestamp': datetime.now().isoformat(), 'period': f"{prices['timestamp'].min()} to {prices['timestamp'].max()}", 'buy_hold_return_pct': buy_hold_return, 'highlow_backtest': { 'final_capital': highlow_stats['final_capital'], 'return_pct': hl_return, 'alpha': hl_return - buy_hold_return, **highlow_results, }, 'close_only_backtest': { 'final_capital': close_stats['final_capital'], 'return_pct': co_return, 'alpha': co_return - buy_hold_return, 'missed_tp_intracandle': close_stats.get('missed_tp_intracandle', 0), 'missed_sl_intracandle': close_stats.get('missed_sl_intracandle', 0), **close_results, }, 'delta': { 'return_pct': delta_return, 'pnl': close_results['total_pnl'] - highlow_results['total_pnl'], 'win_rate': close_results['win_rate'] - highlow_results['win_rate'], } } output_path = OUTPUT_DIR / 'close_only_vs_highlow.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()