#!/usr/bin/env python3 """ Simulazione REALISTICA del Live Trading ======================================== Simula ESATTAMENTE come funziona il live: 1. Check ogni 60 secondi 2. Usa solo CLOSE price 3. Puo mancare movimenti intracandle 4. Confronta con backtest che vede ogni candela Questo mostra la VERA differenza tra backtest e live. """ import pandas as pd import numpy as np from datetime import datetime from pathlib import Path import json from dataclasses import dataclass 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') KRAKEN_MAKER_FEE = 0.0016 SPREAD = 0.0003 # Tutte le strategie STRATEGIES = { 'Alpha_24h': { 'dip_threshold': 0.08, 'profit_target': 0.15, 'stop_loss': 0.12, 'position_size': 1.0, 'lookback_minutes': 1440, }, 'Fast_24h': { 'dip_threshold': 0.07, 'profit_target': 0.10, 'stop_loss': 0.07, 'position_size': 1.0, 'lookback_minutes': 1440, }, 'Swing_48h': { 'dip_threshold': 0.12, 'profit_target': 0.20, 'stop_loss': 0.15, 'position_size': 1.0, 'lookback_minutes': 2880, }, } @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 net_pnl: float = 0 pnl_pct: float = 0 duration_minutes: int = 0 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_full_backtest(prices: pd.DataFrame, config: Dict) -> Tuple[List[Trade], float]: """Backtest COMPLETO - ogni candela, HIGH/LOW trigger.""" prices = prices.copy() prices['high_rolling'] = prices['high'].rolling(window=config['lookback_minutes'], min_periods=1).max() trades = [] in_position = False entry_price = entry_amount = entry_fee = 0 entry_time = None capital = 5000.0 start_idx = config['lookback_minutes'] for i in range(start_idx, len(prices)): row = prices.iloc[i] ts, close, high, low = row['timestamp'], row['close'], row['high'], row['low'] high_ref = row['high_rolling'] dip = (close - high_ref) / high_ref if high_ref > 0 else 0 if in_position: stop_price = entry_price * (1 - config['stop_loss']) target_price = entry_price * (1 + config['profit_target']) if low <= stop_price: exec = stop_price * (1 - SPREAD) fee = entry_amount * exec * KRAKEN_MAKER_FEE net = entry_amount * (exec - entry_price) - fee - entry_fee capital += entry_amount * entry_price + net trades.append(Trade( entry_time=entry_time, entry_price=entry_price, amount=entry_amount, fee_entry=entry_fee, exit_time=ts, exit_price=exec, fee_exit=fee, exit_reason='stop_loss', net_pnl=net, pnl_pct=net/(entry_amount*entry_price), duration_minutes=int((ts-entry_time).total_seconds()/60) )) in_position = False continue if high >= target_price: exec = target_price * (1 - SPREAD) fee = entry_amount * exec * KRAKEN_MAKER_FEE net = entry_amount * (exec - entry_price) - fee - entry_fee capital += entry_amount * entry_price + net trades.append(Trade( entry_time=entry_time, entry_price=entry_price, amount=entry_amount, fee_entry=entry_fee, exit_time=ts, exit_price=exec, fee_exit=fee, exit_reason='take_profit', net_pnl=net, pnl_pct=net/(entry_amount*entry_price), duration_minutes=int((ts-entry_time).total_seconds()/60) )) in_position = False continue else: if dip <= -config['dip_threshold'] and capital > 10: exec = close * (1 + SPREAD) fee = capital * config['position_size'] * KRAKEN_MAKER_FEE entry_amount = (capital * config['position_size'] - fee) / exec capital -= capital * config['position_size'] entry_price, entry_time, entry_fee = exec, ts, fee in_position = True return trades, capital def run_live_simulation(prices: pd.DataFrame, config: Dict, check_interval: int = 60) -> Tuple[List[Trade], float, Dict]: """ Simula il LIVE - check ogni N secondi, solo CLOSE. Traccia quanti movimenti vengono persi. """ prices = prices.copy() prices['high_rolling'] = prices['high'].rolling(window=config['lookback_minutes'], min_periods=1).max() trades = [] in_position = False entry_price = entry_amount = entry_fee = 0 entry_time = None capital = 5000.0 start_idx = config['lookback_minutes'] # Tracking missed_sl = 0 # Stop loss che avremmo dovuto prendere missed_tp = 0 # Take profit che avremmo dovuto prendere extra_loss = 0 # Perdita extra per SL ritardato missed_gain = 0 # Guadagno perso per TP mancato # Simula check ogni check_interval candele for i in range(start_idx, len(prices), check_interval): row = prices.iloc[i] ts, close = row['timestamp'], row['close'] high_ref = row['high_rolling'] dip = (close - high_ref) / high_ref if high_ref > 0 else 0 if in_position: pnl_pct = (close - entry_price) / entry_price # Controlla cosa e successo nelle candele che abbiamo saltato skipped_start = max(i - check_interval, start_idx) skipped = prices.iloc[skipped_start:i] # Cerca se SL/TP sono stati triggerati nelle candele saltate stop_price = entry_price * (1 - config['stop_loss']) target_price = entry_price * (1 + config['profit_target']) sl_triggered_in_gap = skipped['low'].min() <= stop_price tp_triggered_in_gap = skipped['high'].max() >= target_price if sl_triggered_in_gap and pnl_pct > -config['stop_loss']: # Lo SL doveva triggerare ma il close e sopra missed_sl += 1 if tp_triggered_in_gap and pnl_pct < config['profit_target']: # Il TP doveva triggerare ma il close e sotto missed_tp += 1 # Exit su CLOSE (come fa il live) if pnl_pct <= -config['stop_loss']: exec = close * (1 - SPREAD) fee = entry_amount * exec * KRAKEN_MAKER_FEE net = entry_amount * (exec - entry_price) - fee - entry_fee capital += entry_amount * entry_price + net trades.append(Trade( entry_time=entry_time, entry_price=entry_price, amount=entry_amount, fee_entry=entry_fee, exit_time=ts, exit_price=exec, fee_exit=fee, exit_reason='stop_loss', net_pnl=net, pnl_pct=net/(entry_amount*entry_price), duration_minutes=int((ts-entry_time).total_seconds()/60) )) in_position = False continue if pnl_pct >= config['profit_target']: exec = close * (1 - SPREAD) fee = entry_amount * exec * KRAKEN_MAKER_FEE net = entry_amount * (exec - entry_price) - fee - entry_fee capital += entry_amount * entry_price + net trades.append(Trade( entry_time=entry_time, entry_price=entry_price, amount=entry_amount, fee_entry=entry_fee, exit_time=ts, exit_price=exec, fee_exit=fee, exit_reason='take_profit', net_pnl=net, pnl_pct=net/(entry_amount*entry_price), duration_minutes=int((ts-entry_time).total_seconds()/60) )) in_position = False continue else: if dip <= -config['dip_threshold'] and capital > 10: exec = close * (1 + SPREAD) fee = capital * config['position_size'] * KRAKEN_MAKER_FEE entry_amount = (capital * config['position_size'] - fee) / exec capital -= capital * config['position_size'] entry_price, entry_time, entry_fee = exec, ts, fee in_position = True stats = { 'missed_sl': missed_sl, 'missed_tp': missed_tp, 'check_interval': check_interval, } return trades, capital, stats def analyze(trades: List[Trade]) -> Dict: if not trades: return {'total': 0, 'wins': 0, 'losses': 0, 'win_rate': 0, 'total_pnl': 0} wins = [t for t in trades if t.net_pnl > 0] losses = [t for t in trades if t.net_pnl <= 0] return { 'total': len(trades), 'wins': len(wins), 'losses': len(losses), 'win_rate': len(wins)/len(trades)*100 if trades else 0, 'total_pnl': sum(t.net_pnl for t in trades), 'tp_count': len([t for t in trades if t.exit_reason == 'take_profit']), 'sl_count': len([t for t in trades if t.exit_reason == 'stop_loss']), } def main(): print("="*80) print("SIMULAZIONE REALISTICA: LIVE vs BACKTEST") print("="*80) print("\nQuesto test simula il comportamento REALE del live trading engine") print("che controlla il mercato ogni 60 secondi invece di ogni candela.") prices = load_data() print(f"\nDati: {len(prices):,} candele 1m") print(f"Periodo: {prices['timestamp'].min()} -> {prices['timestamp'].max()}") start_price = prices.iloc[1440]['close'] end_price = prices.iloc[-1]['close'] buy_hold = (end_price - start_price) / start_price * 100 print(f"Buy & Hold: {buy_hold:+.2f}%") results = [] for name, config in STRATEGIES.items(): print(f"\n{'='*60}") print(f"STRATEGIA: {name}") print(f"Dip: {config['dip_threshold']*100}% | TP: {config['profit_target']*100}% | SL: {config['stop_loss']*100}%") print(f"{'='*60}") # Backtest completo bt_trades, bt_capital = run_full_backtest(prices, config) bt_stats = analyze(bt_trades) bt_return = (bt_capital - 5000) / 5000 * 100 # Live simulation (check ogni 60 candele = 60 minuti per simulare il check ogni 60s) # Ma in realta il live controlla ogni 60 SECONDI = ogni candela 1m # Il problema e che usa solo CLOSE, non HIGH/LOW # Test 1: Check ogni candela ma solo CLOSE (come il live fa davvero) live1_trades, live1_capital, live1_stats = run_live_simulation(prices, config, check_interval=1) live1_return = (live1_capital - 5000) / 5000 * 100 live1_an = analyze(live1_trades) # Test 2: Check ogni 5 candele (simula lag/ritardi) live5_trades, live5_capital, live5_stats = run_live_simulation(prices, config, check_interval=5) live5_return = (live5_capital - 5000) / 5000 * 100 live5_an = analyze(live5_trades) # Test 3: Check ogni 15 candele (simula problemi di rete) live15_trades, live15_capital, live15_stats = run_live_simulation(prices, config, check_interval=15) live15_return = (live15_capital - 5000) / 5000 * 100 live15_an = analyze(live15_trades) print(f"\n{'Metodo':<25} {'Trade':>6} {'Win%':>8} {'P&L':>12} {'Return':>10} {'TP':>4} {'SL':>4}") print("-"*75) print(f"{'Backtest (HIGH/LOW)':<25} {bt_stats['total']:>6} {bt_stats['win_rate']:>7.1f}% €{bt_stats['total_pnl']:>10,.0f} {bt_return:>+9.1f}% {bt_stats['tp_count']:>4} {bt_stats['sl_count']:>4}") print(f"{'Live (1m check, CLOSE)':<25} {live1_an['total']:>6} {live1_an['win_rate']:>7.1f}% €{live1_an['total_pnl']:>10,.0f} {live1_return:>+9.1f}% {live1_an['tp_count']:>4} {live1_an['sl_count']:>4}") print(f"{'Live (5m check)':<25} {live5_an['total']:>6} {live5_an['win_rate']:>7.1f}% €{live5_an['total_pnl']:>10,.0f} {live5_return:>+9.1f}% {live5_an['tp_count']:>4} {live5_an['sl_count']:>4}") print(f"{'Live (15m check)':<25} {live15_an['total']:>6} {live15_an['win_rate']:>7.1f}% €{live15_an['total_pnl']:>10,.0f} {live15_return:>+9.1f}% {live15_an['tp_count']:>4} {live15_an['sl_count']:>4}") print(f"\n--- Opportunita Perse ---") print(f"Live 1m: {live1_stats['missed_tp']} TP mancati, {live1_stats['missed_sl']} SL mancati") print(f"Live 5m: {live5_stats['missed_tp']} TP mancati, {live5_stats['missed_sl']} SL mancati") print(f"Live 15m: {live15_stats['missed_tp']} TP mancati, {live15_stats['missed_sl']} SL mancati") delta = live1_return - bt_return print(f"\nDelta Live vs Backtest: {delta:+.2f}%") results.append({ 'strategy': name, 'backtest_return': bt_return, 'live_1m_return': live1_return, 'live_5m_return': live5_return, 'live_15m_return': live15_return, 'delta_1m': live1_return - bt_return, 'delta_5m': live5_return - bt_return, 'delta_15m': live15_return - bt_return, 'missed_tp_1m': live1_stats['missed_tp'], 'missed_sl_1m': live1_stats['missed_sl'], }) # Summary print("\n" + "="*80) print("RIEPILOGO FINALE") print("="*80) print(f"\n{'Strategia':<15} {'BT Return':>12} {'Live 1m':>12} {'Delta':>10}") print("-"*50) for r in results: print(f"{r['strategy']:<15} {r['backtest_return']:>+11.1f}% {r['live_1m_return']:>+11.1f}% {r['delta_1m']:>+9.1f}%") avg_delta = np.mean([r['delta_1m'] for r in results]) print("-"*50) print(f"{'MEDIA':<15} {'':>12} {'':>12} {avg_delta:>+9.1f}%") print("\n" + "="*80) print("CONCLUSIONI") print("="*80) if avg_delta < -5: print(f"\n ATTENZIONE: Il live performa in media {abs(avg_delta):.1f}% PEGGIO del backtest!") print("Le cause principali sono:") print(" 1. Uso di CLOSE invece di HIGH/LOW per trigger") print(" 2. Movimenti intracandle che non vengono catturati") elif avg_delta > 5: print(f"\n Il live performa {avg_delta:.1f}% MEGLIO - verificare i dati") else: print(f"\n Differenza minima ({avg_delta:+.1f}%) - backtest e live sono allineati") # Save output = { 'timestamp': datetime.now().isoformat(), 'buy_hold_return': buy_hold, 'strategies': results, 'average_delta_1m': avg_delta, } with open(OUTPUT_DIR / 'live_simulation_results.json', 'w') as f: json.dump(output, f, indent=2) print(f"\nRisultati salvati in: {OUTPUT_DIR}/live_simulation_results.json") if __name__ == '__main__': main()