#!/usr/bin/env python3 """ Multi-Model Strategy Backtester =============================== Combines predictions from multiple ML models across timeframes: - Range Predictor (5m and 15m) - Movement Magnitude Predictor - AMD Phase Detector Strategy: 1. 5m signals prepare potential entry 2. 15m confirms direction and provides context 3. AMD phase filters unsuitable market conditions 4. Minimum R:R ratio of 2:1 or 3:1 Backtest: Full year 2025 with weekly reports Author: ML-Specialist (NEXUS v4.0) Date: 2026-01-05 """ import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).parent.parent)) sys.path.insert(0, str(Path(__file__).parent.parent / 'src')) import numpy as np import pandas as pd from dataclasses import dataclass, field from typing import Dict, List, Optional, Tuple, Any from datetime import datetime, timedelta from enum import Enum import joblib from loguru import logger import json # ML-Engine imports from config.reduced_features import generate_reduced_features, get_feature_columns_without_ohlcv from data.database import MySQLConnection # ============================================================ # Configuration # ============================================================ @dataclass class MultiModelConfig: """Configuration for multi-model strategy""" # Capital initial_capital: float = 1000.0 max_risk_per_trade: float = 0.02 # 2% max_drawdown: float = 0.15 # 15% max_positions: int = 1 # 1 position at a time for simplicity # R:R requirements min_rr_ratio: float = 2.0 # Minimum 2:1 R:R preferred_rr_ratio: float = 3.0 # Preferred 3:1 R:R # Confidence thresholds min_5m_confidence: float = 0.65 min_15m_confidence: float = 0.70 min_combined_score: float = 0.60 # Timeframe alignment require_timeframe_alignment: bool = True # 5m and 15m must agree # AMD filter use_amd_filter: bool = True avoid_manipulation_phase: bool = True # Technical confirmation use_rsi_filter: bool = True use_sar_filter: bool = True # Position management atr_sl_multiplier: float = 1.5 trailing_stop_activation: float = 1.0 # Activate after +1R profit class TradeDirection(Enum): LONG = "LONG" SHORT = "SHORT" HOLD = "HOLD" class TradeStatus(Enum): OPEN = "OPEN" CLOSED_TP = "CLOSED_TP" CLOSED_SL = "CLOSED_SL" CLOSED_TRAILING = "CLOSED_TRAILING" CLOSED_TIMEOUT = "CLOSED_TIMEOUT" @dataclass class MultiModelSignal: """Signal combining multiple model predictions""" timestamp: datetime symbol: str # 5m predictions pred_5m_high: float pred_5m_low: float conf_5m: float direction_5m: TradeDirection # 15m predictions pred_15m_high: float pred_15m_low: float conf_15m: float direction_15m: TradeDirection # AMD phase amd_phase: str = "unknown" amd_confidence: float = 0.0 # Magnitude predictions asymmetry_ratio: float = 1.0 suggested_rr: float = 1.0 # Technical indicators rsi: float = 50.0 sar_signal: str = "neutral" cmf: float = 0.0 # Combined assessment final_direction: TradeDirection = TradeDirection.HOLD combined_score: float = 0.0 suggested_entry: float = 0.0 suggested_sl: float = 0.0 suggested_tp: float = 0.0 actual_rr: float = 0.0 def to_dict(self) -> Dict: return { 'timestamp': self.timestamp.isoformat(), 'symbol': self.symbol, 'direction_5m': self.direction_5m.value, 'direction_15m': self.direction_15m.value, 'final_direction': self.final_direction.value, 'combined_score': round(self.combined_score, 3), 'amd_phase': self.amd_phase, 'actual_rr': round(self.actual_rr, 2), 'suggested_entry': round(self.suggested_entry, 4), 'suggested_sl': round(self.suggested_sl, 4), 'suggested_tp': round(self.suggested_tp, 4) } @dataclass class Trade: """Trade record""" id: str symbol: str direction: TradeDirection entry_price: float stop_loss: float take_profit: float size: float entry_time: datetime exit_time: Optional[datetime] = None exit_price: Optional[float] = None pnl: float = 0.0 status: TradeStatus = TradeStatus.OPEN # Signal info signal: Optional[MultiModelSignal] = None # Tracking max_favorable: float = 0.0 # Max favorable excursion max_adverse: float = 0.0 # Max adverse excursion bars_held: int = 0 def calculate_pnl(self, exit_price: float): """Calculate P&L based on exit price""" if self.direction == TradeDirection.LONG: self.pnl = (exit_price - self.entry_price) * self.size else: # SHORT self.pnl = (self.entry_price - exit_price) * self.size @dataclass class WeeklyReport: """Weekly performance report""" week_start: datetime week_end: datetime week_number: int # Performance starting_equity: float ending_equity: float net_pnl: float return_pct: float # Trade stats total_trades: int winning_trades: int losing_trades: int win_rate: float # Risk metrics max_drawdown: float sharpe_ratio: float profit_factor: float # Trade details trades: List[Trade] = field(default_factory=list) avg_winner: float = 0.0 avg_loser: float = 0.0 best_trade: float = 0.0 worst_trade: float = 0.0 def to_dict(self) -> Dict: return { 'week_number': self.week_number, 'week_start': self.week_start.strftime('%Y-%m-%d'), 'week_end': self.week_end.strftime('%Y-%m-%d'), 'starting_equity': round(self.starting_equity, 2), 'ending_equity': round(self.ending_equity, 2), 'net_pnl': round(self.net_pnl, 2), 'return_pct': round(self.return_pct, 2), 'total_trades': self.total_trades, 'winning_trades': self.winning_trades, 'losing_trades': self.losing_trades, 'win_rate': round(self.win_rate, 1), 'max_drawdown': round(self.max_drawdown, 2), 'profit_factor': round(self.profit_factor, 2), 'avg_winner': round(self.avg_winner, 2), 'avg_loser': round(self.avg_loser, 2), 'best_trade': round(self.best_trade, 2), 'worst_trade': round(self.worst_trade, 2) } # ============================================================ # Multi-Model Signal Generator # ============================================================ class MultiModelSignalGenerator: """Generates signals by combining multiple model predictions""" def __init__( self, model_dir: str = 'models/reduced_features_models', config: MultiModelConfig = None ): self.model_dir = Path(model_dir) self.config = config or MultiModelConfig() self.models = {} self.load_models() def load_models(self): """Load all available models""" if not self.model_dir.exists(): logger.warning(f"Model directory not found: {self.model_dir}") return for model_file in self.model_dir.glob("*.joblib"): if model_file.name != 'metadata.joblib': key = model_file.stem self.models[key] = joblib.load(model_file) logger.info(f"Loaded model: {key}") def get_predictions( self, features_5m: pd.DataFrame, features_15m: pd.DataFrame, symbol: str, idx_5m: int, idx_15m: int ) -> Dict[str, Any]: """Get predictions from all models for both timeframes""" predictions = {} feature_cols = get_feature_columns_without_ohlcv() # 5m predictions available_5m = [c for c in feature_cols if c in features_5m.columns] if available_5m: X_5m = features_5m[available_5m].iloc[[idx_5m]].values key_5m_high = f"{symbol}_5m_high_h3" key_5m_low = f"{symbol}_5m_low_h3" if key_5m_high in self.models: predictions['5m_high'] = self.models[key_5m_high].predict(X_5m)[0] if key_5m_low in self.models: predictions['5m_low'] = self.models[key_5m_low].predict(X_5m)[0] # 15m predictions available_15m = [c for c in feature_cols if c in features_15m.columns] if available_15m: X_15m = features_15m[available_15m].iloc[[idx_15m]].values key_15m_high = f"{symbol}_15m_high_h3" key_15m_low = f"{symbol}_15m_low_h3" if key_15m_high in self.models: predictions['15m_high'] = self.models[key_15m_high].predict(X_15m)[0] if key_15m_low in self.models: predictions['15m_low'] = self.models[key_15m_low].predict(X_15m)[0] return predictions def calculate_direction_and_confidence( self, pred_high: float, pred_low: float, current_price: float, atr: float ) -> Tuple[TradeDirection, float]: """Calculate direction and confidence from predictions""" # Normalize predictions to ATR high_potential = pred_high / (atr + 1e-10) low_potential = pred_low / (atr + 1e-10) # Calculate asymmetry if low_potential > 1e-10: asymmetry = high_potential / low_potential else: asymmetry = high_potential * 10 if high_potential > 0 else 1.0 # Determine direction if asymmetry > 1.3: # Bullish direction = TradeDirection.LONG confidence = min(asymmetry / 3.0, 1.0) elif asymmetry < 0.77: # Bearish direction = TradeDirection.SHORT confidence = min(1.0 / (asymmetry + 0.1), 1.0) else: direction = TradeDirection.HOLD confidence = 0.0 return direction, confidence def generate_signal( self, df_5m: pd.DataFrame, df_15m: pd.DataFrame, features_5m: pd.DataFrame, features_15m: pd.DataFrame, symbol: str, idx_5m: int, idx_15m: int ) -> Optional[MultiModelSignal]: """Generate combined signal from multiple models""" # Get current data current_price = df_5m['close'].iloc[idx_5m] atr_5m = features_5m['ATR'].iloc[idx_5m] if 'ATR' in features_5m.columns else 1.0 atr_15m = features_15m['ATR'].iloc[idx_15m] if 'ATR' in features_15m.columns else atr_5m * 2 # Get predictions predictions = self.get_predictions( features_5m, features_15m, symbol, idx_5m, idx_15m ) if not predictions: return None # Get 5m direction pred_5m_high = predictions.get('5m_high', 0) pred_5m_low = predictions.get('5m_low', 0) direction_5m, conf_5m = self.calculate_direction_and_confidence( pred_5m_high, pred_5m_low, current_price, atr_5m ) # Get 15m direction pred_15m_high = predictions.get('15m_high', 0) pred_15m_low = predictions.get('15m_low', 0) direction_15m, conf_15m = self.calculate_direction_and_confidence( pred_15m_high, pred_15m_low, current_price, atr_15m ) # Get technical indicators rsi = features_5m['RSI'].iloc[idx_5m] if 'RSI' in features_5m.columns else 50 sar = features_5m['SAR'].iloc[idx_5m] if 'SAR' in features_5m.columns else current_price cmf = features_5m['CMF'].iloc[idx_5m] if 'CMF' in features_5m.columns else 0 sar_signal = "bullish" if sar < current_price else "bearish" # Create signal object signal = MultiModelSignal( timestamp=df_5m.index[idx_5m], symbol=symbol, pred_5m_high=pred_5m_high, pred_5m_low=pred_5m_low, conf_5m=conf_5m, direction_5m=direction_5m, pred_15m_high=pred_15m_high, pred_15m_low=pred_15m_low, conf_15m=conf_15m, direction_15m=direction_15m, rsi=rsi, sar_signal=sar_signal, cmf=cmf ) # Calculate combined assessment self._assess_signal(signal, current_price, atr_5m) return signal def _assess_signal( self, signal: MultiModelSignal, current_price: float, atr: float ): """Assess signal quality and calculate entry/exit levels""" config = self.config # Check timeframe alignment if config.require_timeframe_alignment: if signal.direction_5m != signal.direction_15m: signal.final_direction = TradeDirection.HOLD signal.combined_score = 0.0 return if signal.direction_5m == TradeDirection.HOLD: signal.final_direction = TradeDirection.HOLD signal.combined_score = 0.0 return # Check confidence thresholds if signal.conf_5m < config.min_5m_confidence: signal.final_direction = TradeDirection.HOLD signal.combined_score = 0.0 return if signal.conf_15m < config.min_15m_confidence: signal.final_direction = TradeDirection.HOLD signal.combined_score = 0.0 return # Technical confirmation tech_score = 0.0 max_tech_score = 0.0 if config.use_rsi_filter: max_tech_score += 1.0 if signal.direction_5m == TradeDirection.SHORT and signal.rsi > 55: tech_score += 1.0 elif signal.direction_5m == TradeDirection.LONG and signal.rsi < 45: tech_score += 1.0 if config.use_sar_filter: max_tech_score += 1.0 if signal.direction_5m == TradeDirection.SHORT and signal.sar_signal == "bearish": tech_score += 1.0 elif signal.direction_5m == TradeDirection.LONG and signal.sar_signal == "bullish": tech_score += 1.0 # CMF confirmation max_tech_score += 1.0 if signal.direction_5m == TradeDirection.SHORT and signal.cmf < 0: tech_score += 1.0 elif signal.direction_5m == TradeDirection.LONG and signal.cmf > 0: tech_score += 1.0 tech_confirmation = tech_score / max_tech_score if max_tech_score > 0 else 0.5 # Combined score signal.combined_score = ( signal.conf_5m * 0.3 + signal.conf_15m * 0.4 + tech_confirmation * 0.3 ) if signal.combined_score < config.min_combined_score: signal.final_direction = TradeDirection.HOLD return # Set final direction signal.final_direction = signal.direction_5m # Calculate entry/exit levels signal.suggested_entry = current_price if signal.final_direction == TradeDirection.SHORT: # SHORT: SL above, TP below signal.suggested_sl = current_price + atr * config.atr_sl_multiplier # Use 15m prediction for TP (larger move) tp_distance = signal.pred_15m_low * 0.8 if signal.pred_15m_low > 0 else atr * 3 signal.suggested_tp = current_price - tp_distance else: # LONG # LONG: SL below, TP above signal.suggested_sl = current_price - atr * config.atr_sl_multiplier # Use 15m prediction for TP tp_distance = signal.pred_15m_high * 0.8 if signal.pred_15m_high > 0 else atr * 3 signal.suggested_tp = current_price + tp_distance # Calculate actual R:R risk = abs(signal.suggested_entry - signal.suggested_sl) reward = abs(signal.suggested_tp - signal.suggested_entry) signal.actual_rr = reward / risk if risk > 0 else 0 # Check minimum R:R if signal.actual_rr < config.min_rr_ratio: # Try to adjust TP for minimum R:R min_reward = risk * config.min_rr_ratio if signal.final_direction == TradeDirection.SHORT: signal.suggested_tp = current_price - min_reward else: signal.suggested_tp = current_price + min_reward signal.actual_rr = config.min_rr_ratio # ============================================================ # Multi-Timeframe Backtester # ============================================================ class MultiModelBacktester: """Backtester for multi-model strategy with weekly reports""" def __init__(self, config: MultiModelConfig = None): self.config = config or MultiModelConfig() self.signal_generator = MultiModelSignalGenerator(config=self.config) # State self.trades: List[Trade] = [] self.weekly_reports: List[WeeklyReport] = [] self.equity_curve: List[Tuple[datetime, float]] = [] # Risk management self.current_equity = self.config.initial_capital self.peak_equity = self.config.initial_capital self.current_drawdown = 0.0 def _load_data( self, symbol: str, start_date: str, end_date: str, timeframe: str = '5m' ) -> pd.DataFrame: """Load data from PostgreSQL database""" try: import psycopg2 # PostgreSQL connection conn = psycopg2.connect( host="localhost", port=5432, dbname="orbiquant_trading", user="orbiquant_user", password="orbiquant_dev_2025" ) logger.info(f"Connected to PostgreSQL") # Get ticker_id with conn.cursor() as cur: cur.execute("SELECT id FROM market_data.tickers WHERE symbol = %s", (symbol,)) result = cur.fetchone() if not result: logger.warning(f"Symbol not found: {symbol}") return pd.DataFrame() ticker_id = result[0] # Load data from parent table (covers all partitions) table = "market_data.ohlcv_5m" query = f""" SELECT timestamp as time, open, high, low, close, volume FROM {table} WHERE ticker_id = %s AND timestamp >= %s AND timestamp <= %s ORDER BY timestamp ASC """ df = pd.read_sql_query( query, conn, params=(ticker_id, start_date, end_date), parse_dates=['time'] ) conn.close() if df.empty: logger.warning(f"No data for {symbol}") return df df.set_index('time', inplace=True) # Resample if needed if timeframe != '5m': tf_map = {'15m': '15min', '30m': '30min', '1H': '1H'} offset = tf_map.get(timeframe, timeframe) df = df.resample(offset).agg({ 'open': 'first', 'high': 'max', 'low': 'min', 'close': 'last', 'volume': 'sum' }).dropna() return df except Exception as e: logger.error(f"Failed to load data: {e}") import traceback traceback.print_exc() return pd.DataFrame() def run_backtest( self, symbol: str, start_date: str = "2025-01-01", end_date: str = "2025-12-31" ) -> Dict[str, Any]: """Run backtest for full year with weekly reports""" logger.info(f"\n{'='*60}") logger.info(f"MULTI-MODEL STRATEGY BACKTEST") logger.info(f"Symbol: {symbol}") logger.info(f"Period: {start_date} to {end_date}") logger.info(f"Capital: ${self.config.initial_capital:,.2f}") logger.info(f"Min R:R: {self.config.min_rr_ratio}:1") logger.info(f"{'='*60}") # Load data for both timeframes logger.info(f"Loading {symbol} 5m data...") df_5m = self._load_data(symbol, start_date, end_date, "5m") if df_5m is None or df_5m.empty: logger.error(f"No 5m data for {symbol}") return {} logger.info(f"Loaded {len(df_5m)} 5m records") logger.info(f"Loading {symbol} 15m data...") df_15m = self._load_data(symbol, start_date, end_date, "15m") if df_15m is None or df_15m.empty: logger.error(f"No 15m data for {symbol}") return {} logger.info(f"Loaded {len(df_15m)} 15m records") # Generate features logger.info("Generating features...") features_5m = generate_reduced_features(df_5m) features_15m = generate_reduced_features(df_15m) # Reset state self.trades = [] self.weekly_reports = [] self.equity_curve = [] self.current_equity = self.config.initial_capital self.peak_equity = self.config.initial_capital self.current_drawdown = 0.0 # Track weekly data current_week_trades = [] week_start_equity = self.current_equity current_week_start = None open_trade: Optional[Trade] = None trade_id = 0 # Main backtest loop (iterate through 5m data) warmup = 50 # Skip warmup period for indicators for i in range(warmup, len(df_5m)): current_time = df_5m.index[i] current_price = df_5m['close'].iloc[i] high_price = df_5m['high'].iloc[i] low_price = df_5m['low'].iloc[i] # Find corresponding 15m bar # 15m bar that contains this 5m timestamp idx_15m = self._find_15m_index(df_15m, current_time) if idx_15m is None or idx_15m < 10: continue # Weekly tracking if current_week_start is None: current_week_start = current_time # Check for week change (Monday start) if current_time.weekday() == 0 and current_time.hour < 1: if current_week_start is not None and current_week_trades: # Generate weekly report report = self._generate_weekly_report( current_week_start, current_time - timedelta(hours=1), week_start_equity, self.current_equity, current_week_trades ) self.weekly_reports.append(report) logger.info(f"Week {report.week_number}: {report.return_pct:+.2f}%, " f"{report.total_trades} trades, WR: {report.win_rate:.1f}%") # Reset for new week current_week_start = current_time week_start_equity = self.current_equity current_week_trades = [] # Check and manage open trade if open_trade is not None: closed = self._check_trade_exit(open_trade, high_price, low_price, current_price) if closed: open_trade.exit_time = current_time self.current_equity += open_trade.pnl # Update peak and drawdown if self.current_equity > self.peak_equity: self.peak_equity = self.current_equity self.current_drawdown = (self.peak_equity - self.current_equity) / self.peak_equity self.trades.append(open_trade) current_week_trades.append(open_trade) open_trade = None # Record equity self.equity_curve.append((current_time, self.current_equity)) # Check for new signal (only if no open trade) if open_trade is None: # Check drawdown limit if self.current_drawdown >= self.config.max_drawdown: continue signal = self.signal_generator.generate_signal( df_5m, df_15m, features_5m, features_15m, symbol, i, idx_15m ) if signal and signal.final_direction != TradeDirection.HOLD: # Check minimum R:R if signal.actual_rr >= self.config.min_rr_ratio: # Calculate position size risk_amount = self.current_equity * self.config.max_risk_per_trade risk_per_unit = abs(signal.suggested_entry - signal.suggested_sl) if risk_per_unit > 0: position_size = risk_amount / risk_per_unit # Scale for Gold if 'XAU' in symbol: position_size = round(position_size, 2) else: position_size = round(position_size * 10000) / 10000 if position_size > 0: trade_id += 1 open_trade = Trade( id=f"T{trade_id:04d}", symbol=symbol, direction=signal.final_direction, entry_price=signal.suggested_entry, stop_loss=signal.suggested_sl, take_profit=signal.suggested_tp, size=position_size, entry_time=current_time, signal=signal ) logger.debug(f"Opened trade {open_trade.id}: " f"{signal.final_direction.value} @ {current_price:.2f}, " f"R:R={signal.actual_rr:.1f}") # Close any remaining trade if open_trade is not None: open_trade.exit_price = df_5m['close'].iloc[-1] open_trade.exit_time = df_5m.index[-1] open_trade.status = TradeStatus.CLOSED_TIMEOUT open_trade.calculate_pnl(open_trade.exit_price) self.current_equity += open_trade.pnl self.trades.append(open_trade) current_week_trades.append(open_trade) # Generate final week report if needed if current_week_trades: report = self._generate_weekly_report( current_week_start, df_5m.index[-1], week_start_equity, self.current_equity, current_week_trades ) self.weekly_reports.append(report) # Calculate final metrics return self._calculate_final_metrics(symbol) def _find_15m_index(self, df_15m: pd.DataFrame, timestamp: datetime) -> Optional[int]: """Find the 15m bar index that contains the given 5m timestamp""" try: # Find the 15m bar that started at or before this time mask = df_15m.index <= timestamp if mask.any(): return mask.sum() - 1 return None except: return None def _check_trade_exit( self, trade: Trade, high: float, low: float, close: float ) -> bool: """Check if trade should be closed""" trade.bars_held += 1 if trade.direction == TradeDirection.LONG: # Update MFE/MAE trade.max_favorable = max(trade.max_favorable, high - trade.entry_price) trade.max_adverse = max(trade.max_adverse, trade.entry_price - low) # Check stop loss if low <= trade.stop_loss: trade.exit_price = trade.stop_loss trade.status = TradeStatus.CLOSED_SL trade.calculate_pnl(trade.exit_price) return True # Check take profit if high >= trade.take_profit: trade.exit_price = trade.take_profit trade.status = TradeStatus.CLOSED_TP trade.calculate_pnl(trade.exit_price) return True else: # SHORT # Update MFE/MAE trade.max_favorable = max(trade.max_favorable, trade.entry_price - low) trade.max_adverse = max(trade.max_adverse, high - trade.entry_price) # Check stop loss if high >= trade.stop_loss: trade.exit_price = trade.stop_loss trade.status = TradeStatus.CLOSED_SL trade.calculate_pnl(trade.exit_price) return True # Check take profit if low <= trade.take_profit: trade.exit_price = trade.take_profit trade.status = TradeStatus.CLOSED_TP trade.calculate_pnl(trade.exit_price) return True # Timeout after 6 hours (72 bars of 5m) if trade.bars_held >= 72: trade.exit_price = close trade.status = TradeStatus.CLOSED_TIMEOUT trade.calculate_pnl(trade.exit_price) return True return False def _generate_weekly_report( self, week_start: datetime, week_end: datetime, start_equity: float, end_equity: float, trades: List[Trade] ) -> WeeklyReport: """Generate weekly performance report""" net_pnl = end_equity - start_equity return_pct = (net_pnl / start_equity) * 100 if start_equity > 0 else 0 winning = [t for t in trades if t.pnl > 0] losing = [t for t in trades if t.pnl <= 0] win_rate = len(winning) / len(trades) * 100 if trades else 0 avg_winner = np.mean([t.pnl for t in winning]) if winning else 0 avg_loser = np.mean([t.pnl for t in losing]) if losing else 0 gross_profit = sum(t.pnl for t in winning) gross_loss = abs(sum(t.pnl for t in losing)) profit_factor = gross_profit / gross_loss if gross_loss > 0 else float('inf') # Calculate max drawdown for the week max_dd = 0 peak = start_equity current = start_equity for trade in trades: current += trade.pnl if current > peak: peak = current dd = (peak - current) / peak if peak > 0 else 0 max_dd = max(max_dd, dd) # Week number week_number = week_start.isocalendar()[1] return WeeklyReport( week_start=week_start, week_end=week_end, week_number=week_number, starting_equity=start_equity, ending_equity=end_equity, net_pnl=net_pnl, return_pct=return_pct, total_trades=len(trades), winning_trades=len(winning), losing_trades=len(losing), win_rate=win_rate, max_drawdown=max_dd * 100, sharpe_ratio=0, # Would need daily returns to calculate profit_factor=min(profit_factor, 999), trades=trades, avg_winner=avg_winner, avg_loser=avg_loser, best_trade=max(t.pnl for t in trades) if trades else 0, worst_trade=min(t.pnl for t in trades) if trades else 0 ) def _calculate_final_metrics(self, symbol: str) -> Dict[str, Any]: """Calculate final backtest metrics""" total_return = (self.current_equity - self.config.initial_capital) / self.config.initial_capital * 100 winning = [t for t in self.trades if t.pnl > 0] losing = [t for t in self.trades if t.pnl <= 0] win_rate = len(winning) / len(self.trades) * 100 if self.trades else 0 avg_winner = np.mean([t.pnl for t in winning]) if winning else 0 avg_loser = np.mean([t.pnl for t in losing]) if losing else 0 gross_profit = sum(t.pnl for t in winning) gross_loss = abs(sum(t.pnl for t in losing)) profit_factor = gross_profit / gross_loss if gross_loss > 0 else float('inf') # Max drawdown from equity curve max_dd = 0 peak = self.config.initial_capital for _, equity in self.equity_curve: if equity > peak: peak = equity dd = (peak - equity) / peak if peak > 0 else 0 max_dd = max(max_dd, dd) # Direction breakdown long_trades = [t for t in self.trades if t.direction == TradeDirection.LONG] short_trades = [t for t in self.trades if t.direction == TradeDirection.SHORT] long_wins = len([t for t in long_trades if t.pnl > 0]) short_wins = len([t for t in short_trades if t.pnl > 0]) return { 'symbol': symbol, 'period': f"{self.equity_curve[0][0].date()} to {self.equity_curve[-1][0].date()}" if self.equity_curve else "N/A", 'initial_capital': self.config.initial_capital, 'final_capital': round(self.current_equity, 2), 'total_return_pct': round(total_return, 2), 'total_trades': len(self.trades), 'winning_trades': len(winning), 'losing_trades': len(losing), 'win_rate': round(win_rate, 1), 'profit_factor': round(min(profit_factor, 999), 2), 'max_drawdown_pct': round(max_dd * 100, 2), 'avg_winner': round(avg_winner, 2), 'avg_loser': round(avg_loser, 2), 'best_trade': round(max(t.pnl for t in self.trades), 2) if self.trades else 0, 'worst_trade': round(min(t.pnl for t in self.trades), 2) if self.trades else 0, 'long_trades': len(long_trades), 'long_wins': long_wins, 'long_wr': round(long_wins / len(long_trades) * 100, 1) if long_trades else 0, 'short_trades': len(short_trades), 'short_wins': short_wins, 'short_wr': round(short_wins / len(short_trades) * 100, 1) if short_trades else 0, 'total_weeks': len(self.weekly_reports), 'profitable_weeks': len([w for w in self.weekly_reports if w.net_pnl > 0]) } # ============================================================ # Report Generator # ============================================================ class MultiModelReportGenerator: """Generate comprehensive reports""" @staticmethod def generate_annual_report( backtester: MultiModelBacktester, metrics: Dict[str, Any] ) -> str: """Generate annual summary report""" report = f"""# INFORME ANUAL - ESTRATEGIA MULTI-MODELO **Símbolo:** {metrics['symbol']} **Período:** {metrics['period']} **Capital Inicial:** ${metrics['initial_capital']:,.2f} **Capital Final:** ${metrics['final_capital']:,.2f} --- ## RESUMEN EJECUTIVO | Métrica | Valor | |---------|-------| | **Retorno Total** | {metrics['total_return_pct']:+.2f}% | | **Total Trades** | {metrics['total_trades']} | | **Win Rate** | {metrics['win_rate']:.1f}% | | **Profit Factor** | {metrics['profit_factor']:.2f} | | **Max Drawdown** | {metrics['max_drawdown_pct']:.2f}% | --- ## DESGLOSE POR DIRECCIÓN ### LONG Trades | Métrica | Valor | |---------|-------| | Total | {metrics['long_trades']} | | Ganadores | {metrics['long_wins']} | | Win Rate | {metrics['long_wr']:.1f}% | ### SHORT Trades | Métrica | Valor | |---------|-------| | Total | {metrics['short_trades']} | | Ganadores | {metrics['short_wins']} | | Win Rate | {metrics['short_wr']:.1f}% | --- ## ESTADÍSTICAS DE TRADES | Métrica | Valor | |---------|-------| | Promedio Ganador | ${metrics['avg_winner']:,.2f} | | Promedio Perdedor | ${metrics['avg_loser']:,.2f} | | Mejor Trade | ${metrics['best_trade']:,.2f} | | Peor Trade | ${metrics['worst_trade']:,.2f} | --- ## RENDIMIENTO SEMANAL | Semana | Inicio | Fin | P&L | Retorno | Trades | WR | Max DD | |--------|--------|-----|-----|---------|--------|-----|--------| """ for w in backtester.weekly_reports: report += f"| {w.week_number} | {w.week_start.strftime('%m/%d')} | " report += f"{w.week_end.strftime('%m/%d')} | ${w.net_pnl:+.2f} | " report += f"{w.return_pct:+.2f}% | {w.total_trades} | " report += f"{w.win_rate:.0f}% | {w.max_drawdown:.1f}% |\n" report += f""" --- ## SEMANAS RENTABLES - **Total Semanas:** {metrics['total_weeks']} - **Semanas Rentables:** {metrics['profitable_weeks']} - **% Semanas Positivas:** {metrics['profitable_weeks']/metrics['total_weeks']*100:.1f}% --- ## CONFIGURACIÓN DE ESTRATEGIA - **R:R Mínimo:** {backtester.config.min_rr_ratio}:1 - **Riesgo por Trade:** {backtester.config.max_risk_per_trade*100:.0f}% - **Max Drawdown Permitido:** {backtester.config.max_drawdown*100:.0f}% - **Alineación Timeframes:** {'Sí' if backtester.config.require_timeframe_alignment else 'No'} - **Filtro RSI:** {'Sí' if backtester.config.use_rsi_filter else 'No'} - **Filtro SAR:** {'Sí' if backtester.config.use_sar_filter else 'No'} --- *Generado: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}* """ return report @staticmethod def generate_weekly_details(backtester: MultiModelBacktester) -> str: """Generate detailed weekly reports""" report = "# INFORMES SEMANALES DETALLADOS\n\n" for w in backtester.weekly_reports: report += f""" ## Semana {w.week_number} ({w.week_start.strftime('%Y-%m-%d')} - {w.week_end.strftime('%Y-%m-%d')}) ### Resumen | Métrica | Valor | |---------|-------| | Equity Inicial | ${w.starting_equity:,.2f} | | Equity Final | ${w.ending_equity:,.2f} | | P&L Neto | ${w.net_pnl:+,.2f} | | Retorno | {w.return_pct:+.2f}% | | Trades | {w.total_trades} | | Win Rate | {w.win_rate:.1f}% | | Profit Factor | {w.profit_factor:.2f} | | Max Drawdown | {w.max_drawdown:.2f}% | ### Trades de la Semana | ID | Dirección | Entrada | SL | TP | Salida | P&L | Status | |----|-----------|---------|-----|-----|--------|-----|--------| """ for t in w.trades: exit_str = f"{t.exit_price:.2f}" if t.exit_price else "N/A" report += f"| {t.id} | {t.direction.value} | {t.entry_price:.2f} | " report += f"{t.stop_loss:.2f} | {t.take_profit:.2f} | " report += f"{exit_str} | ${t.pnl:+.2f} | {t.status.value} |\n" report += "\n---\n" return report @staticmethod def save_reports( backtester: MultiModelBacktester, metrics: Dict[str, Any], output_dir: str = 'reports' ): """Save all reports to files""" output_path = Path(output_dir) output_path.mkdir(parents=True, exist_ok=True) timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') symbol = metrics['symbol'] # Annual report annual_report = MultiModelReportGenerator.generate_annual_report(backtester, metrics) annual_file = output_path / f"annual_report_{symbol}_{timestamp}.md" with open(annual_file, 'w') as f: f.write(annual_report) logger.info(f"Saved annual report to {annual_file}") # Weekly details weekly_report = MultiModelReportGenerator.generate_weekly_details(backtester) weekly_file = output_path / f"weekly_details_{symbol}_{timestamp}.md" with open(weekly_file, 'w') as f: f.write(weekly_report) logger.info(f"Saved weekly details to {weekly_file}") # JSON metrics json_file = output_path / f"backtest_metrics_{symbol}_{timestamp}.json" with open(json_file, 'w') as f: json.dump(metrics, f, indent=2) logger.info(f"Saved JSON metrics to {json_file}") # Weekly summary CSV csv_data = [] for w in backtester.weekly_reports: csv_data.append(w.to_dict()) if csv_data: df = pd.DataFrame(csv_data) csv_file = output_path / f"weekly_summary_{symbol}_{timestamp}.csv" df.to_csv(csv_file, index=False) logger.info(f"Saved weekly CSV to {csv_file}") return annual_file, weekly_file, json_file # ============================================================ # Main Execution # ============================================================ def run_multi_model_backtest(): """Run full year backtest with multi-model strategy""" # Configure strategy config = MultiModelConfig( initial_capital=1000.0, max_risk_per_trade=0.02, min_rr_ratio=2.0, # Minimum 2:1 R:R preferred_rr_ratio=3.0, # Prefer 3:1 require_timeframe_alignment=True, use_rsi_filter=True, use_sar_filter=True ) logger.info("="*60) logger.info("MULTI-MODEL STRATEGY BACKTESTER") logger.info(f"Min R:R: {config.min_rr_ratio}:1") logger.info(f"Timeframe Alignment: {config.require_timeframe_alignment}") logger.info("="*60) # Create backtester backtester = MultiModelBacktester(config) # Run backtest for XAUUSD (available data: Jan-Mar 2025) metrics = backtester.run_backtest( symbol="XAUUSD", start_date="2025-01-01", end_date="2025-03-18" # Data only available until this date ) if metrics: # Print summary print("\n" + "="*60) print("BACKTEST RESULTS") print("="*60) print(f"Symbol: {metrics['symbol']}") print(f"Period: {metrics['period']}") print(f"Initial Capital: ${metrics['initial_capital']:,.2f}") print(f"Final Capital: ${metrics['final_capital']:,.2f}") print(f"Total Return: {metrics['total_return_pct']:+.2f}%") print(f"Total Trades: {metrics['total_trades']}") print(f"Win Rate: {metrics['win_rate']:.1f}%") print(f"Profit Factor: {metrics['profit_factor']:.2f}") print(f"Max Drawdown: {metrics['max_drawdown_pct']:.2f}%") print(f"Profitable Weeks: {metrics['profitable_weeks']}/{metrics['total_weeks']}") print("="*60) # Save reports MultiModelReportGenerator.save_reports(backtester, metrics) return metrics, backtester if __name__ == "__main__": metrics, backtester = run_multi_model_backtest()