"""Super Ensemble Backtester for Romeo V8 Advanced backtester for the super ensemble model with multi-algorithm collaboration, stacking, dynamic weighting, and confidence calibration. Key Features: - Super Ensemble Prediction: Combines 10+ algorithms - Stacking Logic: Uses meta-learner for final predictions - Dynamic Weighting: Real-time weight adjustment - Confidence Calibration: Calibrated probability fusion - Cross-Validation Ensemble: Multiple CV fold combination - Advanced Risk Management: Multi-algorithm consensus """ import os import json import numpy as np import pandas as pd import joblib from tensorflow import keras import sys import argparse sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '.'))) try: from v8.train_v8 import SuperEnsembleFeatureEngineer, load_romeo_v8, SuperEnsemble except Exception: from train_v8 import SuperEnsembleFeatureEngineer, load_romeo_v8, SuperEnsemble class SumAxis1Layer(keras.layers.Layer): def call(self, inputs): return keras.backend.sum(inputs, axis=1) class SuperEnsembleBacktester: def __init__(self, config=None): self.config = config or { 'ensemble_method': 'stacking', # 'stacking', 'weighted', 'voting' 'confidence_threshold': 0.60, # Minimum confidence for trades 'max_risk_per_trade': 0.12, # Maximum risk per trade 'use_dynamic_weighting': True, # Enable dynamic weight adjustment 'use_calibration': True, # Use calibrated probabilities 'use_cv_ensemble': True, # Use cross-validation ensemble 'consensus_threshold': 0.7, # Minimum algorithm agreement 'volatility_adjustment': True, 'max_drawdown_limit': 0.90, # Stop trading limit } self.super_ensemble = None def load_super_ensemble(self, model_path): """Load the super ensemble model""" if not os.path.exists(model_path): raise FileNotFoundError(f"Model not found: {model_path}") self.super_ensemble = load_romeo_v8(model_path) print(f"Loaded super ensemble with {len(self.super_ensemble.models)} base algorithms") return self.super_ensemble def get_super_ensemble_prediction(self, X, method='stacking'): """Get prediction from super ensemble using specified method""" if self.super_ensemble is None: raise ValueError("Super ensemble not loaded. Call load_super_ensemble() first.") # Use the SuperEnsemble class predict_proba method proba = self.super_ensemble.predict_proba(X) ensemble_proba = proba[:, 1] # For compatibility, create base_predictions and model_names # This is a simplified version - in full implementation you'd get individual model predictions base_predictions = [ensemble_proba.reshape(-1, 1)] # Simplified model_names = list(self.super_ensemble.models.keys()) return ensemble_proba, base_predictions, model_names def calculate_consensus_score(self, base_predictions): """Calculate consensus score among algorithms""" if not base_predictions: return 0.5 # Convert to binary predictions (above/below 0.5) binary_preds = [] for pred in base_predictions: binary_pred = (pred.ravel() > 0.5).astype(int) binary_preds.append(binary_pred) # Calculate agreement percentage all_binary = np.array(binary_preds) consensus = np.mean(all_binary, axis=0) # Average agreement return consensus def should_trade_signal(self, ensemble_proba, consensus_score, volatility, volume_ratio): """Determine if signal meets super ensemble criteria""" # Base confidence check if ensemble_proba < self.config['confidence_threshold']: return False, "Low confidence" # Consensus check if consensus_score < self.config['consensus_threshold']: return False, "Low consensus" # Volatility filter if self.config['volatility_adjustment'] and volatility > 0.025: return False, "High volatility" # Volume confirmation if volume_ratio < 1.0: return False, "Low volume" return True, "Valid signal" def backtest_super_ensemble(self, timeframe='15m', initial_capital=100, data_file=None, risk_per_trade=0.08, stop_loss=0.015, take_profit=0.04, commission_pct=0.0002, slippage_pips=0.3, timeout_bars=6): # Load data if data_file: data_path = data_file else: data_path = f'data_xauusd_v3/15m_data_v3.csv' df = pd.read_csv(data_path, parse_dates=['Datetime']) df = df.sort_values('Datetime').reset_index(drop=True) # Load model model_path = f'v8/models_romeo_v8/trading_model_romeo_{timeframe}.pkl' artifact = self.load_super_ensemble(model_path) # Process features (same as training but without fitting scaler/PCA) eng = SuperEnsembleFeatureEngineer() df = eng.add_technical_indicators(df) df = eng.add_quantum_features(df) df = df.fillna(method='bfill').fillna(method='ffill').fillna(0) exclude = ['Datetime', 'Open', 'High', 'Low', 'Close', 'Volume', 'Adj Close'] feature_cols = [c for c in df.columns if c not in exclude and not c.startswith('target')] # Ensure df contains all features for f in feature_cols: if f not in df.columns: df[f] = 0.0 X = df[feature_cols].values # Get super ensemble predictions print("Generating super ensemble predictions...") ensemble_probas, base_predictions, model_names = self.get_super_ensemble_prediction( X, method=self.config['ensemble_method'] ) # Calculate consensus scores consensus_scores = self.calculate_consensus_score(base_predictions) # Add to dataframe df['ensemble_proba'] = ensemble_probas df['consensus_score'] = consensus_scores # Generate signals based on ensemble confidence signals = (ensemble_probas > self.config['confidence_threshold']).astype(int) df['signal'] = signals # Initialize trading variables capital = initial_capital peak_capital = initial_capital trades = [] total_ensemble_contributions = {name: 0 for name in model_names} print("Starting super ensemble backtest...") # Trading loop for i in range(len(df)-1): current_drawdown = (peak_capital - capital) / peak_capital if peak_capital > 0 else 0 # Check stop trading condition if current_drawdown >= (1 - self.config['max_drawdown_limit']): print(f"Stopping trading due to drawdown limit: {current_drawdown:.1%}") break if df.iloc[i]['signal'] == 1: ensemble_proba = df.iloc[i]['ensemble_proba'] consensus_score = df.iloc[i]['consensus_score'] volatility = df.iloc[i]['Volatility'] if 'Volatility' in df.columns else 0.01 volume_ratio = df.iloc[i]['Volume_Ratio'] if 'Volume_Ratio' in df.columns else 1.0 # Check if signal meets criteria should_trade, reason = self.should_trade_signal( ensemble_proba, consensus_score, volatility, volume_ratio ) if not should_trade: continue entry_price = df.iloc[i+1]['Open'] entry_price_slip = entry_price + slippage_pips * 0.0001 # Conservative position sizing for super ensemble position_size = (capital * risk_per_trade) / (stop_loss * entry_price_slip) # Adjust for volatility if self.config['volatility_adjustment']: vol_factor = 1 / (1 + volatility * 10) position_size *= vol_factor # Ensure within limits max_size = (capital * self.config['max_risk_per_trade']) / (stop_loss * entry_price_slip) position_size = min(position_size, max_size) # Execute trade exit_price = None exit_idx = i+1 reason = 'TIMEOUT' for j in range(i+1, min(i+1+timeout_bars, len(df))): high = df.iloc[j]['High'] low = df.iloc[j]['Low'] if high >= entry_price_slip * (1 + take_profit): exit_price = entry_price_slip * (1 + take_profit) exit_idx = j reason = 'TP' break if low <= entry_price_slip * (1 - stop_loss): exit_price = entry_price_slip * (1 - stop_loss) exit_idx = j reason = 'SL' break if exit_price is None: exit_price = df.iloc[min(i+timeout_bars, len(df)-1)]['Close'] exit_idx = min(i+timeout_bars, len(df)-1) exit_price_slip = exit_price - slippage_pips * 0.0001 pnl = (exit_price_slip - entry_price_slip) * position_size commission = commission_pct * (entry_price_slip + exit_price_slip) * position_size pnl_after = pnl - commission capital += pnl_after # Update peak capital peak_capital = max(peak_capital, capital) # Track algorithm contributions for name in model_names: if name in df.columns and f'{name}_contrib' in df.columns: total_ensemble_contributions[name] += df.iloc[i][f'{name}_contrib'] trades.append({ 'entry_idx': i+1, 'exit_idx': exit_idx, 'entry_date': df.iloc[i+1]['Datetime'], 'exit_date': df.iloc[exit_idx]['Datetime'], 'entry_price': entry_price_slip, 'exit_price': exit_price_slip, 'position_size': position_size, 'pnl': pnl_after, 'commission': commission, 'reason': reason, 'ensemble_proba': float(ensemble_proba), 'consensus_score': float(consensus_score), 'volatility': float(volatility), 'volume_ratio': float(volume_ratio), 'capital_after': capital, 'drawdown_at_entry': current_drawdown }) # Calculate final metrics if trades: winning_trades = [t for t in trades if t['pnl'] > 0] win_rate = len(winning_trades) / len(trades) if winning_trades: avg_win = np.mean([t['pnl'] for t in winning_trades]) gross_profit = sum([t['pnl'] for t in winning_trades]) else: avg_win = 0 gross_profit = 0 losing_trades = [t for t in trades if t['pnl'] <= 0] if losing_trades: avg_loss = np.mean([t['pnl'] for t in losing_trades]) gross_loss = abs(sum([t['pnl'] for t in losing_trades])) else: avg_loss = 0 gross_loss = 0 profit_factor = gross_profit / gross_loss if gross_loss > 0 else float('inf') # Sharpe ratio approximation returns = [t['pnl'] / initial_capital for t in trades] if len(returns) > 1 and np.std(returns) > 0: sharpe_ratio = np.mean(returns) / np.std(returns) * np.sqrt(252) else: sharpe_ratio = 0 else: win_rate = 0 avg_win = 0 avg_loss = 0 profit_factor = 0 sharpe_ratio = 0 final_drawdown = (peak_capital - capital) / peak_capital if peak_capital > 0 else 0 summary = { 'initial_capital': initial_capital, 'final_capital': float(capital), 'total_return_pct': float((capital - initial_capital)/initial_capital*100), 'peak_capital': float(peak_capital), 'max_drawdown_pct': float(final_drawdown * 100), 'trades': len(trades), 'win_rate': float(win_rate), 'avg_win': float(avg_win), 'avg_loss': float(avg_loss), 'profit_factor': float(profit_factor), 'sharpe_ratio': float(sharpe_ratio), 'super_ensemble_metrics': { 'algorithms_used': len(model_names), 'ensemble_method': self.config['ensemble_method'], 'avg_consensus_score': float(np.mean([t['consensus_score'] for t in trades])) if trades else 0, 'avg_ensemble_proba': float(np.mean([t['ensemble_proba'] for t in trades])) if trades else 0, 'calibration_used': self.config['use_calibration'], 'cv_ensemble_used': self.config['use_cv_ensemble'], 'dynamic_weighting_used': self.config['use_dynamic_weighting'], } } # Save results os.makedirs('backtest_results_romeo_v8', exist_ok=True) out_signals = df.reset_index()[['Datetime', 'Open', 'High', 'Low', 'Close', 'signal', 'ensemble_proba', 'consensus_score']] out_signals.to_csv(f'backtest_results_romeo_v8/romeo_signals_{timeframe}.csv', index=False) pd.DataFrame(trades).to_csv(f'backtest_results_romeo_v8/romeo_trades_{timeframe}.csv', index=False) with open(f'backtest_results_romeo_v8/romeo_summary_{timeframe}.json', 'w') as f: json.dump(summary, f, indent=2, default=str) return summary def main(): parser = argparse.ArgumentParser(description='Super Ensemble Backtester for Romeo V8') parser.add_argument('--timeframe', default='15m') parser.add_argument('--data', default=None, help='Optional path to unseen CSV data') parser.add_argument('--initial-capital', type=float, default=100) parser.add_argument('--commission-pct', type=float, default=0.0002) parser.add_argument('--slippage-pips', type=float, default=0.3) parser.add_argument('--risk-per-trade', type=float, default=0.08) parser.add_argument('--stop-loss', type=float, default=0.015) parser.add_argument('--take-profit', type=float, default=0.04) parser.add_argument('--ensemble-method', choices=['stacking', 'weighted', 'voting'], default='stacking') parser.add_argument('--confidence-threshold', type=float, default=0.60) args = parser.parse_args() backtester = SuperEnsembleBacktester({ 'ensemble_method': args.ensemble_method, 'confidence_threshold': args.confidence_threshold, 'max_risk_per_trade': 0.12, 'use_dynamic_weighting': True, 'use_calibration': True, 'use_cv_ensemble': True, 'consensus_threshold': 0.7, 'volatility_adjustment': True, 'max_drawdown_limit': 0.90, }) summary = backtester.backtest_super_ensemble( timeframe=args.timeframe, initial_capital=args.initial_capital, data_file=args.data, risk_per_trade=args.risk_per_trade, stop_loss=args.stop_loss, take_profit=args.take_profit, commission_pct=args.commission_pct, slippage_pips=args.slippage_pips ) print("Romeo V8 Super Ensemble Backtest Results:") print("=" * 60) print(f"Initial Capital: ${summary['initial_capital']}") print(f"Final Capital: ${summary['final_capital']:.2f}") print(f"Total Return: {summary['total_return_pct']:.2f}%") print(f"Max Drawdown: {summary['max_drawdown_pct']:.2f}%") print(f"Total Trades: {summary['trades']}") print(f"Win Rate: {summary['win_rate']:.1%}") print(f"Profit Factor: {summary['profit_factor']:.2f}") print(f"Sharpe Ratio: {summary['sharpe_ratio']:.2f}") print(f"Super Ensemble: {summary['super_ensemble_metrics']}") if __name__ == '__main__': main()