#!/usr/bin/env python3 """ Ottimizzazione Avanzata - Trova le migliori configurazioni ========================================================== Test massivo di parametri e strategie. """ import pandas as pd import numpy as np from datetime import datetime from pathlib import Path import json from itertools import product from concurrent.futures import ProcessPoolExecutor, as_completed import warnings warnings.filterwarnings('ignore') # Fee Kraken realistiche KRAKEN_MAKER_FEE = 0.0016 SPREAD_PCT = 0.0003 SLIPPAGE_BASE = 0.0001 def load_data() -> pd.DataFrame: DATA_DIR = Path('/var/www/html/pippo.cuttalo.com/data') dfs = [] for f in sorted(DATA_DIR.glob('prices_BTC_EUR_*.csv')): df = pd.read_csv(f) 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 backtest_config(prices_data: tuple, config: dict) -> dict: """Backtest singola configurazione.""" close, high, low, high_ref = prices_data capital = 5000.0 start_capital = capital position = 0 entry_price = 0 position_amount = 0 fee_entry = 0 trades = [] equity = [capital] dip_pct = config['dip'] tp_pct = config['tp'] sl_pct = config['sl'] pos_size = config.get('pos_size', 1.0) start_idx = config.get('warmup', 1440) for i in range(start_idx, len(close)): current_price = close[i] candle_high = high[i] candle_low = low[i] ref_high = high_ref[i] if ref_high <= 0: continue dip_from_high = (current_price - ref_high) / ref_high if position == 1: # Check stop loss stop_price = entry_price * (1 - sl_pct) if candle_low <= stop_price: exec_price = stop_price * (1 - SPREAD_PCT - SLIPPAGE_BASE) gross_pnl = position_amount * (exec_price - entry_price) fee = position_amount * exec_price * KRAKEN_MAKER_FEE net_pnl = gross_pnl - fee - fee_entry capital += position_amount * entry_price + net_pnl trades.append({'pnl': net_pnl, 'type': 'sl'}) position = 0 equity.append(capital) continue # Check take profit target_price = entry_price * (1 + tp_pct) if candle_high >= target_price: exec_price = target_price * (1 - SPREAD_PCT) gross_pnl = position_amount * (exec_price - entry_price) fee = position_amount * exec_price * KRAKEN_MAKER_FEE net_pnl = gross_pnl - fee - fee_entry capital += position_amount * entry_price + net_pnl trades.append({'pnl': net_pnl, 'type': 'tp'}) position = 0 equity.append(capital) continue elif position == 0: if dip_from_high <= -dip_pct: trade_value = capital * pos_size if trade_value > 10: exec_price = current_price * (1 + SPREAD_PCT + SLIPPAGE_BASE) fee_entry = trade_value * KRAKEN_MAKER_FEE position_amount = (trade_value - fee_entry) / exec_price entry_price = exec_price capital -= trade_value position = 1 # Chiudi posizione finale if position == 1: final_price = close[-1] gross_pnl = position_amount * (final_price - entry_price) fee = position_amount * final_price * KRAKEN_MAKER_FEE net_pnl = gross_pnl - fee - fee_entry capital += position_amount * entry_price + net_pnl trades.append({'pnl': net_pnl, 'type': 'end'}) if not trades or len(trades) < 5: return None # Calcola metriche total_pnl = sum(t['pnl'] for t in trades) wins = [t for t in trades if t['pnl'] > 0] losses = [t for t in trades if t['pnl'] <= 0] win_rate = len(wins) / len(trades) # Max drawdown eq = np.array(equity) peak = np.maximum.accumulate(eq) dd = (eq - peak) / peak max_dd = abs(min(dd)) if len(dd) > 0 else 0 # Profit factor total_wins = sum(t['pnl'] for t in wins) if wins else 0 total_losses = abs(sum(t['pnl'] for t in losses)) if losses else 1 profit_factor = total_wins / total_losses if total_losses > 0 else 0 return { **config, 'return_pct': (capital - start_capital) / start_capital * 100, 'final_capital': capital, 'num_trades': len(trades), 'win_rate': win_rate, 'max_dd': max_dd, 'profit_factor': profit_factor, 'tp_count': sum(1 for t in trades if t['type'] == 'tp'), 'sl_count': sum(1 for t in trades if t['type'] == 'sl'), 'avg_trade': total_pnl / len(trades), } def main(): print("="*70) print("OTTIMIZZAZIONE AVANZATA - RICERCA CONFIGURAZIONI VINCENTI") print("="*70) prices = load_data() print(f"\nDati: {len(prices):,} candele") close = prices['close'].values high = prices['high'].values low = prices['low'].values # Buy & Hold start_idx = 1440 start_price = close[start_idx] end_price = close[-1] buy_hold = (end_price - start_price) / start_price * 100 print(f"Buy & Hold: {buy_hold:+.2f}%") # Calcola vari riferimenti per il "high" print("\nCalcolo riferimenti temporali...") # 24h high (standard) high_24h = pd.Series(high).rolling(window=1440, min_periods=1).max().values # 12h high high_12h = pd.Series(high).rolling(window=720, min_periods=1).max().values # 48h high high_48h = pd.Series(high).rolling(window=2880, min_periods=1).max().values # 7d high high_7d = pd.Series(high).rolling(window=10080, min_periods=1).max().values # Configurazioni da testare configs = [] # STRATEGIA 1: Dip Buyer Standard (24h high) print("\n1. Testing Dip Buyer 24h...") for dip in [0.03, 0.04, 0.05, 0.06, 0.07, 0.08, 0.10, 0.12, 0.15]: for tp in [0.05, 0.07, 0.10, 0.12, 0.15, 0.20, 0.25]: for sl in [0.03, 0.05, 0.07, 0.10, 0.12]: if tp > sl: # TP deve essere > SL configs.append({ 'strategy': 'dip_24h', 'dip': dip, 'tp': tp, 'sl': sl, 'pos_size': 1.0, 'warmup': 1440, 'high_ref': '24h' }) # STRATEGIA 2: Dip Buyer 12h (più reattivo) print("2. Testing Dip Buyer 12h...") for dip in [0.03, 0.04, 0.05, 0.06, 0.08]: for tp in [0.05, 0.07, 0.10, 0.12, 0.15]: for sl in [0.03, 0.05, 0.07, 0.10]: if tp > sl: configs.append({ 'strategy': 'dip_12h', 'dip': dip, 'tp': tp, 'sl': sl, 'pos_size': 1.0, 'warmup': 720, 'high_ref': '12h' }) # STRATEGIA 3: Dip Buyer 48h (meno trade, più grandi) print("3. Testing Dip Buyer 48h...") for dip in [0.08, 0.10, 0.12, 0.15, 0.18, 0.20]: for tp in [0.10, 0.15, 0.20, 0.25, 0.30]: for sl in [0.07, 0.10, 0.12, 0.15]: if tp > sl: configs.append({ 'strategy': 'dip_48h', 'dip': dip, 'tp': tp, 'sl': sl, 'pos_size': 1.0, 'warmup': 2880, 'high_ref': '48h' }) # STRATEGIA 4: Dip Buyer 7d (swing trading) print("4. Testing Dip Buyer 7d...") for dip in [0.10, 0.12, 0.15, 0.18, 0.20, 0.25]: for tp in [0.15, 0.20, 0.25, 0.30, 0.35]: for sl in [0.10, 0.12, 0.15, 0.18]: if tp > sl: configs.append({ 'strategy': 'dip_7d', 'dip': dip, 'tp': tp, 'sl': sl, 'pos_size': 1.0, 'warmup': 10080, 'high_ref': '7d' }) print(f"\nTotale configurazioni da testare: {len(configs):,}") # Prepara dati per ogni strategia data_map = { '24h': (close, high, low, high_24h), '12h': (close, high, low, high_12h), '48h': (close, high, low, high_48h), '7d': (close, high, low, high_7d), } # Esegui backtest results = [] total = len(configs) print("\nEsecuzione backtest...") for i, config in enumerate(configs): if (i + 1) % 500 == 0: print(f" Progresso: {i+1}/{total}") data = data_map[config['high_ref']] result = backtest_config(data, config) if result: results.append(result) print(f"\nRisultati validi: {len(results)}") # Ordina per return results.sort(key=lambda x: x['return_pct'], reverse=True) # Filtra per metriche minime good_results = [r for r in results if r['win_rate'] >= 0.45 and r['max_dd'] <= 0.35 and r['profit_factor'] >= 1.2 and r['num_trades'] >= 10] print(f"Configurazioni con buone metriche: {len(good_results)}") # Top 30 assolute print(f"\n{'='*80}") print("TOP 30 CONFIGURAZIONI (per return)") print("="*80) print(f"\n{'Strategy':<10} {'Dip':>5} {'TP':>5} {'SL':>5} {'Return':>10} " f"{'Trades':>7} {'Win%':>7} {'MaxDD':>7} {'PF':>6} {'Alpha':>8}") print("-"*80) for r in results[:30]: alpha = r['return_pct'] - buy_hold print(f"{r['strategy']:<10} {r['dip']*100:>4.0f}% {r['tp']*100:>4.0f}% {r['sl']*100:>4.0f}% " f"{r['return_pct']:>+9.2f}% {r['num_trades']:>7} {r['win_rate']*100:>6.1f}% " f"{r['max_dd']*100:>6.1f}% {r['profit_factor']:>5.2f} {alpha:>+7.2f}%") print("-"*80) print(f"{'Buy & Hold':<10} {'':<17} {buy_hold:>+9.2f}%") # Configurazioni che battono B&H con buone metriche beating_bh = [r for r in good_results if r['return_pct'] > buy_hold] print(f"\n{'='*80}") print(f"CONFIGURAZIONI CHE BATTONO BUY & HOLD ({len(beating_bh)})") print("="*80) if beating_bh: print(f"\n{'Strategy':<10} {'Dip':>5} {'TP':>5} {'SL':>5} {'Return':>10} " f"{'Trades':>7} {'Win%':>7} {'MaxDD':>7} {'Alpha':>8}") print("-"*80) for r in beating_bh[:20]: alpha = r['return_pct'] - buy_hold print(f"{r['strategy']:<10} {r['dip']*100:>4.0f}% {r['tp']*100:>4.0f}% {r['sl']*100:>4.0f}% " f"{r['return_pct']:>+9.2f}% {r['num_trades']:>7} {r['win_rate']*100:>6.1f}% " f"{r['max_dd']*100:>6.1f}% {alpha:>+7.2f}%") # Migliori per ogni strategia print(f"\n{'='*80}") print("MIGLIORE PER OGNI STRATEGIA") print("="*80) for strategy in ['dip_24h', 'dip_12h', 'dip_48h', 'dip_7d']: strat_results = [r for r in results if r['strategy'] == strategy] if strat_results: best = max(strat_results, key=lambda x: x['return_pct']) alpha = best['return_pct'] - buy_hold print(f"\n{strategy}:") print(f" Config: Dip {best['dip']*100:.0f}% | TP {best['tp']*100:.0f}% | SL {best['sl']*100:.0f}%") print(f" Return: {best['return_pct']:+.2f}% (Alpha: {alpha:+.2f}%)") print(f" Trades: {best['num_trades']} | Win: {best['win_rate']*100:.1f}% | MaxDD: {best['max_dd']*100:.1f}%") # Seleziona i 6 migliori trader diversificati print(f"\n{'='*80}") print("SELEZIONE 6 TRADER OTTIMALI (diversificati)") print("="*80) selected = [] # 1. Migliore assoluto 24h best_24h = [r for r in results if r['strategy'] == 'dip_24h'] if best_24h: selected.append(('24h_best', best_24h[0])) # 2. Secondo migliore 24h (diverso) for r in best_24h[1:]: if r['dip'] != selected[0][1]['dip'] or r['tp'] != selected[0][1]['tp']: selected.append(('24h_alt', r)) break # 3. Migliore 12h best_12h = [r for r in results if r['strategy'] == 'dip_12h'] if best_12h: selected.append(('12h_best', best_12h[0])) # 4. Migliore 48h best_48h = [r for r in results if r['strategy'] == 'dip_48h'] if best_48h: selected.append(('48h_best', best_48h[0])) # 5. Migliore 7d best_7d = [r for r in results if r['strategy'] == 'dip_7d'] if best_7d: selected.append(('7d_best', best_7d[0])) # 6. Migliore con win rate > 65% high_wr = [r for r in good_results if r['win_rate'] >= 0.65] if high_wr: high_wr.sort(key=lambda x: x['return_pct'], reverse=True) selected.append(('high_wr', high_wr[0])) print(f"\n{'#':<3} {'Name':<12} {'Strategy':<10} {'Dip':>5} {'TP':>5} {'SL':>5} " f"{'Return':>10} {'Trades':>7} {'Win%':>7}") print("-"*80) for i, (name, r) in enumerate(selected, 1): print(f"{i:<3} {name:<12} {r['strategy']:<10} {r['dip']*100:>4.0f}% {r['tp']*100:>4.0f}% " f"{r['sl']*100:>4.0f}% {r['return_pct']:>+9.2f}% {r['num_trades']:>7} {r['win_rate']*100:>6.1f}%") # Salva risultati output = { 'timestamp': datetime.now().isoformat(), 'buy_hold': buy_hold, 'total_configs_tested': len(configs), 'valid_results': len(results), 'beating_buy_hold': len(beating_bh), 'top_30': [{k: v for k, v in r.items()} for r in results[:30]], 'selected_traders': [{'name': n, **{k: v for k, v in r.items()}} for n, r in selected], } output_path = Path('/var/www/html/pippo.cuttalo.com/optimization_results/advanced_optimization.json') with open(output_path, 'w') as f: json.dump(output, f, indent=2) print(f"\nRisultati salvati in: {output_path}") return selected if __name__ == '__main__': main()