#!/usr/bin/env python3 """ Portfolio Backtest - Tutti i 6 trader ottimizzati ================================================= """ import pandas as pd import numpy as np from datetime import datetime from pathlib import Path import json KRAKEN_MAKER_FEE = 0.0016 SPREAD_PCT = 0.0003 SLIPPAGE_BASE = 0.0001 TRADER_CONFIGS = { 'Alpha_24h': {'dip': 0.08, 'tp': 0.15, 'sl': 0.12, 'lookback': 1440, 'capital': 833.33}, 'Swing_48h': {'dip': 0.12, 'tp': 0.20, 'sl': 0.15, 'lookback': 2880, 'capital': 833.33}, 'Macro_7d': {'dip': 0.15, 'tp': 0.15, 'sl': 0.10, 'lookback': 10080, 'capital': 833.33}, 'Fast_24h': {'dip': 0.07, 'tp': 0.10, 'sl': 0.07, 'lookback': 1440, 'capital': 833.33}, 'Deep_48h': {'dip': 0.10, 'tp': 0.15, 'sl': 0.12, 'lookback': 2880, 'capital': 833.33}, 'Safe_48h': {'dip': 0.12, 'tp': 0.15, 'sl': 0.10, 'lookback': 2880, 'capital': 833.35}, } def load_data(): 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_trader(prices, name, config): close = prices['close'].values high = prices['high'].values low = prices['low'].values lookback = config['lookback'] high_ref = pd.Series(high).rolling(window=lookback, min_periods=1).max().values capital = config['capital'] start_capital = capital position = 0 entry_price = 0 position_amount = 0 fee_entry = 0 trades = [] equity = [capital] start_idx = lookback 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 = (current_price - ref_high) / ref_high if position == 1: stop_price = entry_price * (1 - config['sl']) 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 target_price = entry_price * (1 + config['tp']) 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 <= -config['dip']: trade_value = capital 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 # Close final position 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: return None wins = [t for t in trades if t['pnl'] > 0] eq = np.array(equity) peak = np.maximum.accumulate(eq) dd = (eq - peak) / peak max_dd = abs(min(dd)) if len(dd) > 0 else 0 return { 'name': name, 'initial': start_capital, 'final': capital, 'return_pct': (capital - start_capital) / start_capital * 100, 'trades': len(trades), 'win_rate': len(wins) / len(trades) if trades else 0, 'max_dd': max_dd, 'tp_count': sum(1 for t in trades if t['type'] == 'tp'), 'sl_count': sum(1 for t in trades if t['type'] == 'sl'), } def main(): print("="*70) print("PORTFOLIO BACKTEST - 6 TRADER OTTIMIZZATI") print("="*70) prices = load_data() print(f"\nDati: {len(prices):,} candele") print(f"Periodo: {prices['timestamp'].min().date()} -> {prices['timestamp'].max().date()}") start_idx = 10080 # Max lookback start_price = prices.iloc[start_idx]['close'] end_price = prices.iloc[-1]['close'] buy_hold = (end_price - start_price) / start_price * 100 print(f"\nBuy & Hold: €{start_price:,.0f} -> €{end_price:,.0f} = {buy_hold:+.2f}%") results = [] for name, config in TRADER_CONFIGS.items(): result = backtest_trader(prices.copy(), name, config) if result: results.append(result) # Print results print(f"\n{'='*70}") print("RISULTATI PER TRADER") print("="*70) print(f"\n{'Trader':<12} {'Config':^20} {'Return':>10} {'Trades':>7} {'Win%':>7} {'MaxDD':>7} {'Alpha':>9}") print("-"*70) for r in results: cfg = TRADER_CONFIGS[r['name']] cfg_str = f"D{cfg['dip']*100:.0f}/TP{cfg['tp']*100:.0f}/SL{cfg['sl']*100:.0f}" alpha = r['return_pct'] - buy_hold print(f"{r['name']:<12} {cfg_str:^20} {r['return_pct']:>+9.2f}% {r['trades']:>7} " f"{r['win_rate']*100:>6.1f}% {r['max_dd']*100:>6.1f}% {alpha:>+8.2f}%") # Portfolio totals total_initial = sum(r['initial'] for r in results) total_final = sum(r['final'] for r in results) total_return = (total_final - total_initial) / total_initial * 100 total_trades = sum(r['trades'] for r in results) print("-"*70) print(f"{'PORTFOLIO':<12} {'6 trader':^20} {total_return:>+9.2f}% {total_trades:>7} " f"{'':>7} {'':>7} {total_return - buy_hold:>+8.2f}%") print(f"{'Buy & Hold':<12} {'':^20} {buy_hold:>+9.2f}%") print(f"\n{'='*70}") print("RIEPILOGO CAPITALE") print("="*70) print(f"\nCapitale iniziale: €{total_initial:,.2f}") print(f"Capitale finale: €{total_final:,.2f}") print(f"Profitto totale: €{total_final - total_initial:,.2f}") print(f"Return: {total_return:+.2f}%") print(f"Alpha vs B&H: {total_return - buy_hold:+.2f}%") # Best and worst best = max(results, key=lambda x: x['return_pct']) worst = min(results, key=lambda x: x['return_pct']) print(f"\nMigliore: {best['name']} ({best['return_pct']:+.2f}%)") print(f"Peggiore: {worst['name']} ({worst['return_pct']:+.2f}%)") beating = [r for r in results if r['return_pct'] > buy_hold] print(f"\nTrader che battono B&H: {len(beating)}/{len(results)}") # Conclusion print(f"\n{'='*70}") if total_return > buy_hold: print(f">>> PORTFOLIO BATTE BUY & HOLD! Alpha: {total_return - buy_hold:+.2f}% <<<") else: print(f">>> Buy & Hold vince ({buy_hold:+.2f}% vs {total_return:+.2f}%)") print("="*70) # Save output = { 'timestamp': datetime.now().isoformat(), 'buy_hold': buy_hold, 'portfolio_return': total_return, 'portfolio_alpha': total_return - buy_hold, 'total_capital': total_final, 'traders': results, } output_path = Path('/var/www/html/pippo.cuttalo.com/optimization_results/portfolio_backtest.json') with open(output_path, 'w') as f: json.dump(output, f, indent=2, default=str) print(f"\nSalvato: {output_path}") if __name__ == '__main__': main()