#!/usr/bin/env python3 """ LLM Strategy Backtester with Risk Management ============================================= Sistema completo de backtesting que: 1. Usa predicciones de los modelos ML (high/low) 2. Genera informes para el agente LLM 3. Implementa gestión de riesgo (cuenta 1000 USD) 4. Backtestea estrategias del agente 5. Genera informe final de operaciones Author: ML-Specialist + LLM-Agent (NEXUS v4.0) Version: 1.0.0 Created: 2026-01-05 """ import sys import os from pathlib import Path from datetime import datetime, timedelta from typing import Dict, List, Optional, Tuple, Any from dataclasses import dataclass, field from enum import Enum import json import numpy as np import pandas as pd from loguru import logger import joblib # Add src to path sys.path.insert(0, str(Path(__file__).parent.parent / 'src')) from config.reduced_features import generate_reduced_features, get_feature_columns_without_ohlcv from models.volatility_attention import compute_attention_weights, VolatilityAttentionConfig # ============================================================================== # Enums and Data Classes # ============================================================================== class TradeDirection(Enum): LONG = "LONG" SHORT = "SHORT" HOLD = "HOLD" class TradeStatus(Enum): OPEN = "OPEN" CLOSED_TP = "CLOSED_TP" CLOSED_SL = "CLOSED_SL" CLOSED_SIGNAL = "CLOSED_SIGNAL" CLOSED_TIMEOUT = "CLOSED_TIMEOUT" @dataclass class RiskConfig: """Risk management configuration for 1000 USD account""" initial_capital: float = 1000.0 max_risk_per_trade: float = 0.02 # 2% per trade max_daily_loss: float = 0.05 # 5% daily max loss max_drawdown: float = 0.15 # 15% max drawdown max_positions: int = 2 # Max concurrent positions min_rr_ratio: float = 1.5 # Minimum Risk:Reward ratio # Position sizing leverage: float = 1.0 # No leverage by default commission_pct: float = 0.0002 # 0.02% commission per trade @dataclass class Trade: """Represents a single trade""" 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 status: TradeStatus = TradeStatus.OPEN pnl: float = 0.0 pnl_pct: float = 0.0 confidence: float = 0.0 predicted_high: float = 0.0 predicted_low: float = 0.0 attention_weight: float = 1.0 notes: str = "" def calculate_pnl(self, exit_price: float) -> float: """Calculate PnL for the trade""" if self.direction == TradeDirection.LONG: self.pnl = (exit_price - self.entry_price) * self.size else: self.pnl = (self.entry_price - exit_price) * self.size self.pnl_pct = self.pnl / (self.entry_price * self.size) * 100 return self.pnl @dataclass class EquityPoint: """Point in equity curve""" timestamp: datetime equity: float balance: float drawdown: float open_positions: int @dataclass class BacktestResult: """Results from backtesting""" symbol: str timeframe: str start_date: str end_date: str initial_capital: float final_capital: float total_return: float total_return_pct: float total_trades: int winning_trades: int losing_trades: int win_rate: float profit_factor: float max_drawdown: float max_drawdown_pct: float sharpe_ratio: float avg_winner: float avg_loser: float largest_winner: float largest_loser: float avg_trade_duration: float trades: List[Trade] = field(default_factory=list) equity_curve: List[EquityPoint] = field(default_factory=list) # ============================================================================== # Risk Manager # ============================================================================== class RiskManager: """Manages risk for the trading account""" def __init__(self, config: RiskConfig): self.config = config self.current_equity = config.initial_capital self.daily_pnl = 0.0 self.peak_equity = config.initial_capital self.current_drawdown = 0.0 self.open_positions: List[Trade] = [] self.daily_trades = 0 self.consecutive_losses = 0 def can_open_trade(self) -> Tuple[bool, str]: """Check if we can open a new trade""" # Check max positions if len(self.open_positions) >= self.config.max_positions: return False, "Max positions reached" # Check daily loss limit daily_loss_pct = abs(self.daily_pnl) / self.config.initial_capital if self.daily_pnl < 0 and daily_loss_pct >= self.config.max_daily_loss: return False, f"Daily loss limit reached ({daily_loss_pct:.1%})" # Check max drawdown if self.current_drawdown >= self.config.max_drawdown: return False, f"Max drawdown reached ({self.current_drawdown:.1%})" # Check consecutive losses (circuit breaker) if self.consecutive_losses >= 5: return False, "Circuit breaker: 5 consecutive losses" return True, "OK" def calculate_position_size( self, entry_price: float, stop_loss: float, symbol: str ) -> float: """Calculate position size based on risk""" risk_amount = self.current_equity * self.config.max_risk_per_trade sl_distance = abs(entry_price - stop_loss) if sl_distance == 0: return 0 # Position size in units position_size = risk_amount / sl_distance # Apply leverage position_size *= self.config.leverage # Symbol-specific adjustments if 'USD' in symbol and 'XAU' not in symbol and 'BTC' not in symbol: # Forex: lot size position_size = round(position_size * 10000) / 10000 # 0.01 lots return position_size def update_equity(self, pnl: float): """Update equity after trade close""" self.current_equity += pnl self.daily_pnl += 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 # Update consecutive losses if pnl < 0: self.consecutive_losses += 1 else: self.consecutive_losses = 0 def reset_daily(self): """Reset daily counters""" self.daily_pnl = 0.0 self.daily_trades = 0 # ============================================================================== # Signal Generator (Using ML Predictions) # ============================================================================== class MLSignalGenerator: """Generates trading signals from ML model predictions""" def __init__(self, model_dir: str = 'models/reduced_features_models'): self.model_dir = Path(model_dir) 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_prediction( self, features: pd.DataFrame, symbol: str, timeframe: str, horizon: int = 3 ) -> Dict[str, np.ndarray]: """Get predictions for a symbol/timeframe""" key_high = f"{symbol}_{timeframe}_high_h{horizon}" key_low = f"{symbol}_{timeframe}_low_h{horizon}" feature_cols = get_feature_columns_without_ohlcv() available_cols = [c for c in feature_cols if c in features.columns] X = features[available_cols].values predictions = {} if key_high in self.models: predictions['high'] = self.models[key_high].predict(X) if key_low in self.models: predictions['low'] = self.models[key_low].predict(X) return predictions def generate_signal( self, df: pd.DataFrame, predictions: Dict[str, np.ndarray], attention_weights: np.ndarray, idx: int, min_confidence: float = 0.6, use_directional_filters: bool = True ) -> Tuple[TradeDirection, float, float, float, float]: """ Generate trading signal based on predictions WITH directional filters. Based on backtest analysis showing 100% of winning trades were SHORT, we implement strict filters for LONG entries and prioritize SHORT. Returns: Tuple of (direction, entry, stop_loss, take_profit, confidence) """ if 'high' not in predictions or 'low' not in predictions: return TradeDirection.HOLD, 0, 0, 0, 0 pred_high = predictions['high'][idx] pred_low = predictions['low'][idx] attention = attention_weights[idx] if idx < len(attention_weights) else 1.0 current_price = df['close'].iloc[idx] atr = df['ATR'].iloc[idx] if 'ATR' in df.columns else abs(df['high'].iloc[idx] - df['low'].iloc[idx]) # Get technical indicators for directional filters rsi = df['RSI'].iloc[idx] if 'RSI' in df.columns else 50 sar = df['SAR'].iloc[idx] if 'SAR' in df.columns else current_price cmf = df['CMF'].iloc[idx] if 'CMF' in df.columns else 0 mfi = df['MFI'].iloc[idx] if 'MFI' in df.columns else 50 # Calculate asymmetry ratio asymmetry = pred_high / (pred_low + 1e-10) # Directional confirmation scores short_confirms = 0 long_confirms = 0 if use_directional_filters: # SHORT confirmations if rsi > 55: # RSI elevated short_confirms += 1 if sar > current_price: # SAR bearish short_confirms += 1 if cmf < 0: # Money flow negative short_confirms += 1 if mfi > 55: # MFI elevated (selling pressure) short_confirms += 1 # LONG confirmations (stricter requirements) if rsi < 35: # RSI oversold long_confirms += 1 if sar < current_price: # SAR bullish long_confirms += 1 if cmf > 0.1: # Strong positive money flow long_confirms += 1 if mfi < 35: # MFI oversold long_confirms += 1 # Determine direction based on asymmetry AND directional filters direction = TradeDirection.HOLD confidence = 0 entry = 0 stop_loss = 0 take_profit = 0 # SHORT BIAS FIRST (based on backtest: 100% winners were SHORT) if asymmetry < 0.85 and attention > 0.7: if not use_directional_filters or short_confirms >= 2: direction = TradeDirection.SHORT entry = current_price # Tighter stop loss using ATR stop_loss = current_price + atr * 1.5 # Take profit based on prediction with ATR buffer take_profit = current_price - pred_low * 0.8 # Confidence boosted by confirmations base_conf = min(2 / asymmetry, 1.0) * min(attention, 1.5) / 1.5 conf_boost = 1.0 + (short_confirms * 0.1) if use_directional_filters else 1.0 confidence = min(base_conf * conf_boost, 1.0) # LONG BIAS - Much stricter requirements elif asymmetry > 1.4 and attention > 1.0: # Higher thresholds for LONG if not use_directional_filters or long_confirms >= 3: # Need 3+ confirmations direction = TradeDirection.LONG entry = current_price stop_loss = current_price - atr * 1.5 take_profit = current_price + pred_high * 0.8 base_conf = min(asymmetry / 2, 1.0) * min(attention, 1.5) / 1.5 conf_boost = 1.0 + (long_confirms * 0.1) if use_directional_filters else 1.0 confidence = min(base_conf * conf_boost, 1.0) if direction == TradeDirection.HOLD: return TradeDirection.HOLD, 0, 0, 0, 0 # Different minimum confidence thresholds by direction min_conf_short = min_confidence min_conf_long = min_confidence + 0.15 # Higher bar for LONG required_conf = min_conf_long if direction == TradeDirection.LONG else min_conf_short if confidence < required_conf: return TradeDirection.HOLD, 0, 0, 0, 0 return direction, entry, stop_loss, take_profit, confidence # ============================================================================== # Backtester Engine # ============================================================================== class LLMStrategyBacktester: """ Backtester that simulates LLM-guided trading with ML predictions. """ def __init__( self, risk_config: RiskConfig = None, model_dir: str = 'models/reduced_features_models' ): self.risk_config = risk_config or RiskConfig() self.risk_manager = RiskManager(self.risk_config) self.signal_generator = MLSignalGenerator(model_dir) self.trades: List[Trade] = [] self.equity_curve: List[EquityPoint] = [] def run_backtest( self, df: pd.DataFrame, symbol: str, timeframe: str, start_date: str = None, end_date: str = None, min_confidence: float = 0.6, horizon: int = 3 ) -> BacktestResult: """ Run backtest on historical data. Args: df: OHLCV DataFrame symbol: Trading symbol timeframe: Timeframe start_date: Start date for backtest end_date: End date for backtest min_confidence: Minimum signal confidence horizon: Prediction horizon in bars Returns: BacktestResult with all metrics """ logger.info(f"\n{'='*60}") logger.info(f"Starting Backtest: {symbol} {timeframe}") logger.info(f"Capital: ${self.risk_config.initial_capital:,.2f}") logger.info(f"{'='*60}") # Generate features features = generate_reduced_features(df) # Get predictions predictions = self.signal_generator.get_prediction( features, symbol, timeframe, horizon ) if not predictions: logger.error("No predictions available") return self._create_empty_result(symbol, timeframe) # Compute attention weights config = VolatilityAttentionConfig(factor_window=100, w_max=3.0) attention_weights = compute_attention_weights(df, config) # Reset state self.trades = [] self.equity_curve = [] self.risk_manager = RiskManager(self.risk_config) open_trades: Dict[str, Trade] = {} trade_id = 0 # Main backtest loop for i in range(horizon, len(df) - horizon): current_time = df.index[i] current_price = df['close'].iloc[i] high_price = df['high'].iloc[i] low_price = df['low'].iloc[i] # Check and close existing trades trades_to_close = [] for tid, trade in open_trades.items(): closed, exit_price, status = self._check_trade_exit( trade, high_price, low_price, current_price, i, df, horizon ) if closed: trade.exit_price = exit_price trade.exit_time = current_time trade.status = status trade.calculate_pnl(exit_price) # Update risk manager self.risk_manager.update_equity(trade.pnl) self.risk_manager.open_positions.remove(trade) trades_to_close.append(tid) self.trades.append(trade) logger.debug(f"Closed trade {tid}: {trade.status.value}, PnL: ${trade.pnl:.2f}") for tid in trades_to_close: del open_trades[tid] # Generate new signal direction, entry, sl, tp, confidence = self.signal_generator.generate_signal( features, predictions, attention_weights, i, min_confidence ) # Check if we can open a trade if direction != TradeDirection.HOLD: can_trade, reason = self.risk_manager.can_open_trade() if can_trade: # Calculate position size position_size = self.risk_manager.calculate_position_size( entry, sl, symbol ) if position_size > 0: # Calculate R:R ratio risk = abs(entry - sl) reward = abs(tp - entry) rr_ratio = reward / risk if risk > 0 else 0 if rr_ratio >= self.risk_config.min_rr_ratio: trade_id += 1 trade = Trade( id=f"T{trade_id:04d}", symbol=symbol, direction=direction, entry_price=entry, stop_loss=sl, take_profit=tp, size=position_size, entry_time=current_time, confidence=confidence, predicted_high=predictions['high'][i], predicted_low=predictions['low'][i], attention_weight=attention_weights[i] ) open_trades[trade.id] = trade self.risk_manager.open_positions.append(trade) logger.debug(f"Opened trade {trade.id}: {direction.value} @ {entry:.2f}, " f"SL: {sl:.2f}, TP: {tp:.2f}, Conf: {confidence:.2f}") # Record equity point unrealized_pnl = sum( self._calculate_unrealized_pnl(t, current_price) for t in open_trades.values() ) equity = self.risk_manager.current_equity + unrealized_pnl self.equity_curve.append(EquityPoint( timestamp=current_time, equity=equity, balance=self.risk_manager.current_equity, drawdown=self.risk_manager.current_drawdown, open_positions=len(open_trades) )) # Reset daily counters at day change if i > 0 and df.index[i].date() != df.index[i-1].date(): self.risk_manager.reset_daily() # Close any remaining trades at end final_price = df['close'].iloc[-1] for trade in open_trades.values(): trade.exit_price = final_price trade.exit_time = df.index[-1] trade.status = TradeStatus.CLOSED_TIMEOUT trade.calculate_pnl(final_price) self.risk_manager.update_equity(trade.pnl) self.trades.append(trade) # Calculate final metrics return self._calculate_metrics(symbol, timeframe, df) def _check_trade_exit( self, trade: Trade, high: float, low: float, close: float, bar_idx: int, df: pd.DataFrame, horizon: int ) -> Tuple[bool, float, TradeStatus]: """Check if trade should be closed""" if trade.direction == TradeDirection.LONG: # Check stop loss if low <= trade.stop_loss: return True, trade.stop_loss, TradeStatus.CLOSED_SL # Check take profit if high >= trade.take_profit: return True, trade.take_profit, TradeStatus.CLOSED_TP else: # SHORT # Check stop loss if high >= trade.stop_loss: return True, trade.stop_loss, TradeStatus.CLOSED_SL # Check take profit if low <= trade.take_profit: return True, trade.take_profit, TradeStatus.CLOSED_TP # Check timeout (after horizon bars) bars_open = bar_idx - df.index.get_loc(trade.entry_time) if bars_open >= horizon * 2: return True, close, TradeStatus.CLOSED_TIMEOUT return False, 0, TradeStatus.OPEN def _calculate_unrealized_pnl(self, trade: Trade, current_price: float) -> float: """Calculate unrealized PnL for open trade""" if trade.direction == TradeDirection.LONG: return (current_price - trade.entry_price) * trade.size else: return (trade.entry_price - current_price) * trade.size def _calculate_metrics( self, symbol: str, timeframe: str, df: pd.DataFrame ) -> BacktestResult: """Calculate all backtest metrics""" if not self.trades: return self._create_empty_result(symbol, timeframe) # Basic stats total_trades = len(self.trades) winning_trades = [t for t in self.trades if t.pnl > 0] losing_trades = [t for t in self.trades if t.pnl < 0] win_count = len(winning_trades) loss_count = len(losing_trades) win_rate = win_count / total_trades if total_trades > 0 else 0 # PnL stats total_profit = sum(t.pnl for t in winning_trades) total_loss = abs(sum(t.pnl for t in losing_trades)) profit_factor = total_profit / total_loss if total_loss > 0 else float('inf') avg_winner = total_profit / win_count if win_count > 0 else 0 avg_loser = total_loss / loss_count if loss_count > 0 else 0 largest_winner = max((t.pnl for t in winning_trades), default=0) largest_loser = min((t.pnl for t in losing_trades), default=0) # Capital stats final_capital = self.risk_manager.current_equity total_return = final_capital - self.risk_config.initial_capital total_return_pct = total_return / self.risk_config.initial_capital * 100 # Drawdown equity_values = [e.equity for e in self.equity_curve] if equity_values: peak = equity_values[0] max_dd = 0 for eq in equity_values: if eq > peak: peak = eq dd = (peak - eq) / peak max_dd = max(max_dd, dd) else: max_dd = 0 # Sharpe ratio (simplified) if len(self.trades) > 1: returns = [t.pnl_pct for t in self.trades] avg_return = np.mean(returns) std_return = np.std(returns) sharpe = avg_return / std_return if std_return > 0 else 0 else: sharpe = 0 # Average trade duration durations = [] for t in self.trades: if t.exit_time and t.entry_time: duration = (t.exit_time - t.entry_time).total_seconds() / 3600 durations.append(duration) avg_duration = np.mean(durations) if durations else 0 return BacktestResult( symbol=symbol, timeframe=timeframe, start_date=str(df.index[0]), end_date=str(df.index[-1]), initial_capital=self.risk_config.initial_capital, final_capital=final_capital, total_return=total_return, total_return_pct=total_return_pct, total_trades=total_trades, winning_trades=win_count, losing_trades=loss_count, win_rate=win_rate, profit_factor=profit_factor, max_drawdown=max_dd * self.risk_config.initial_capital, max_drawdown_pct=max_dd * 100, sharpe_ratio=sharpe, avg_winner=avg_winner, avg_loser=avg_loser, largest_winner=largest_winner, largest_loser=largest_loser, avg_trade_duration=avg_duration, trades=self.trades, equity_curve=self.equity_curve ) def _create_empty_result(self, symbol: str, timeframe: str) -> BacktestResult: """Create empty result when no trades""" return BacktestResult( symbol=symbol, timeframe=timeframe, start_date="", end_date="", initial_capital=self.risk_config.initial_capital, final_capital=self.risk_config.initial_capital, total_return=0, total_return_pct=0, total_trades=0, winning_trades=0, losing_trades=0, win_rate=0, profit_factor=0, max_drawdown=0, max_drawdown_pct=0, sharpe_ratio=0, avg_winner=0, avg_loser=0, largest_winner=0, largest_loser=0, avg_trade_duration=0 ) # ============================================================================== # Report Generator for LLM # ============================================================================== class LLMReportGenerator: """Generates reports formatted for LLM analysis""" @staticmethod def generate_prediction_report( results: List[BacktestResult] ) -> str: """Generate comprehensive report for LLM to analyze""" report = """# INFORME DE PREDICCIONES ML PARA ESTRATEGIA DE TRADING ## Resumen Ejecutivo Este informe contiene los resultados del backtesting de los modelos ML para los 3 activos principales. El objetivo es que el agente LLM analice estos datos y genere una estrategia optimizada. ## Configuración del Backtest - **Capital Inicial:** $1,000.00 USD - **Riesgo por Operación:** 2% - **Máximo Drawdown Permitido:** 15% - **Posiciones Simultáneas:** Máximo 2 - **Ratio Riesgo:Beneficio Mínimo:** 1.5:1 --- ## Resultados por Activo """ for result in results: report += f""" ### {result.symbol} - {result.timeframe} | Métrica | Valor | |---------|-------| | Capital Final | ${result.final_capital:,.2f} | | Retorno Total | {result.total_return_pct:+.2f}% | | Total Trades | {result.total_trades} | | Trades Ganadores | {result.winning_trades} | | Trades Perdedores | {result.losing_trades} | | Win Rate | {result.win_rate:.1%} | | Profit Factor | {result.profit_factor:.2f} | | Max Drawdown | {result.max_drawdown_pct:.1f}% | | Sharpe Ratio | {result.sharpe_ratio:.2f} | | Promedio Ganador | ${result.avg_winner:.2f} | | Promedio Perdedor | ${result.avg_loser:.2f} | | Mayor Ganancia | ${result.largest_winner:.2f} | | Mayor Pérdida | ${result.largest_loser:.2f} | | Duración Promedio | {result.avg_trade_duration:.1f} horas | """ # Summary statistics total_trades = sum(r.total_trades for r in results) total_winners = sum(r.winning_trades for r in results) overall_win_rate = total_winners / total_trades if total_trades > 0 else 0 combined_return = sum(r.total_return for r in results) combined_return_pct = combined_return / 1000 * 100 # Assuming single 1000 USD report += f""" --- ## Resumen Consolidado | Métrica | Valor | |---------|-------| | Total Operaciones | {total_trades} | | Win Rate Global | {overall_win_rate:.1%} | | Retorno Combinado | ${combined_return:,.2f} ({combined_return_pct:+.2f}%) | --- ## Análisis por Activo """ # Rank assets by performance ranked = sorted(results, key=lambda x: x.total_return_pct, reverse=True) report += "### Ranking de Activos (por Retorno)\n\n" for i, r in enumerate(ranked, 1): status = "OPERAR" if r.total_return_pct > 0 and r.win_rate > 0.4 else "PRECAUCION" if r.total_return_pct > -5 else "EVITAR" report += f"{i}. **{r.symbol}**: {r.total_return_pct:+.2f}% - {status}\n" report += """ --- ## Recomendaciones para el Agente LLM Basándose en estos resultados, el agente LLM debe: 1. **Priorizar activos rentables** en las decisiones de trading 2. **Ajustar tamaño de posición** según el win rate histórico 3. **Aplicar gestión de riesgo estricta** especialmente en activos con alto drawdown 4. **Considerar la volatilidad** (attention weights) en las decisiones --- ## Datos para Fine-Tuning Los siguientes patrones fueron exitosos: """ for r in results: if r.trades: winning = [t for t in r.trades if t.pnl > 0] if winning: avg_confidence = np.mean([t.confidence for t in winning]) avg_attention = np.mean([t.attention_weight for t in winning]) report += f""" ### {r.symbol} - Patrones Exitosos - Confianza promedio en ganadores: {avg_confidence:.2f} - Attention weight promedio: {avg_attention:.2f} - Direcciones ganadoras: {sum(1 for t in winning if t.direction == TradeDirection.LONG)} LONG, {sum(1 for t in winning if t.direction == TradeDirection.SHORT)} SHORT """ return report @staticmethod def generate_trade_log(results: List[BacktestResult]) -> str: """Generate detailed trade log""" log = "# LOG DETALLADO DE OPERACIONES\n\n" for result in results: log += f"## {result.symbol} - {result.timeframe}\n\n" log += "| ID | Dirección | Entrada | SL | TP | Salida | PnL | Estado | Confianza |\n" log += "|-----|-----------|---------|-----|-----|--------|-----|--------|----------|\n" for trade in result.trades[:50]: # Limit to 50 trades per symbol exit_price_str = f"{trade.exit_price:.4f}" if trade.exit_price else "N/A" log += f"| {trade.id} | {trade.direction.value} | {trade.entry_price:.4f} | " log += f"{trade.stop_loss:.4f} | {trade.take_profit:.4f} | " log += f"{exit_price_str} | " log += f"${trade.pnl:+.2f} | {trade.status.value} | {trade.confidence:.2f} |\n" log += "\n" return log @staticmethod def save_reports( results: List[BacktestResult], 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') # Save prediction report report = LLMReportGenerator.generate_prediction_report(results) report_file = output_path / f"prediction_report_{timestamp}.md" with open(report_file, 'w') as f: f.write(report) logger.info(f"Saved prediction report to {report_file}") # Save trade log trade_log = LLMReportGenerator.generate_trade_log(results) log_file = output_path / f"trade_log_{timestamp}.md" with open(log_file, 'w') as f: f.write(trade_log) logger.info(f"Saved trade log to {log_file}") # Save JSON results results_dict = [] for r in results: results_dict.append({ 'symbol': r.symbol, 'timeframe': r.timeframe, 'start_date': r.start_date, 'end_date': r.end_date, 'initial_capital': r.initial_capital, 'final_capital': r.final_capital, 'total_return': r.total_return, 'total_return_pct': r.total_return_pct, 'total_trades': r.total_trades, 'winning_trades': r.winning_trades, 'losing_trades': r.losing_trades, 'win_rate': r.win_rate, 'profit_factor': r.profit_factor, 'max_drawdown_pct': r.max_drawdown_pct, 'sharpe_ratio': r.sharpe_ratio, 'avg_winner': r.avg_winner, 'avg_loser': r.avg_loser }) json_file = output_path / f"backtest_results_{timestamp}.json" with open(json_file, 'w') as f: json.dump(results_dict, f, indent=2, default=str) logger.info(f"Saved JSON results to {json_file}") return report_file, log_file, json_file # ============================================================================== # Main Execution # ============================================================================== def load_data_for_backtest( symbol: str, start_date: str = '2025-01-01', end_date: str = '2025-01-31', timeframe: str = '15m' ) -> pd.DataFrame: """Load data for backtesting""" try: sys.path.insert(0, str(Path(__file__).parent.parent / 'src')) from data.database import MySQLConnection db = MySQLConnection('config/database.yaml') # Get DB prefix prefixes = {'XAUUSD': 'C:', 'EURUSD': 'C:', 'BTCUSD': 'X:'} db_symbol = f"{prefixes.get(symbol, 'C:')}{symbol}" query = """ SELECT date_agg as time, open, high, low, close, volume FROM tickers_agg_data WHERE ticker = :symbol AND date_agg >= :start_date AND date_agg <= :end_date ORDER BY date_agg ASC """ df = db.execute_query(query, { 'symbol': db_symbol, 'start_date': start_date, 'end_date': end_date }) if df.empty: logger.warning(f"No data for {symbol}") return df df['time'] = pd.to_datetime(df['time']) 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() logger.info(f"Loaded {len(df)} records for {symbol} {timeframe}") return df except Exception as e: logger.error(f"Failed to load data: {e}") return pd.DataFrame() def run_full_backtest(): """Run full backtest for all symbols""" logger.info("=" * 60) logger.info("LLM STRATEGY BACKTESTER") logger.info("Account: $1,000 USD") logger.info("=" * 60) # Configuration risk_config = RiskConfig( initial_capital=1000.0, max_risk_per_trade=0.02, max_daily_loss=0.05, max_drawdown=0.15, max_positions=2, min_rr_ratio=1.5 ) backtester = LLMStrategyBacktester(risk_config) symbols = ['XAUUSD', 'EURUSD', 'BTCUSD'] timeframes = ['5m', '15m'] all_results = [] for symbol in symbols: for timeframe in timeframes: logger.info(f"\nBacktesting {symbol} {timeframe}...") # Load data df = load_data_for_backtest( symbol, start_date='2025-01-01', end_date='2025-01-31', timeframe=timeframe ) if df.empty: logger.warning(f"Skipping {symbol} {timeframe} - no data") continue # Run backtest result = backtester.run_backtest( df, symbol, timeframe, min_confidence=0.5, horizon=3 ) all_results.append(result) logger.info(f" Trades: {result.total_trades}") logger.info(f" Return: {result.total_return_pct:+.2f}%") logger.info(f" Win Rate: {result.win_rate:.1%}") # Generate reports logger.info("\nGenerating reports...") report_file, log_file, json_file = LLMReportGenerator.save_reports(all_results) # Print summary print("\n" + "=" * 60) print("BACKTEST SUMMARY") print("=" * 60) for r in all_results: print(f"\n{r.symbol} {r.timeframe}:") print(f" Capital: ${r.initial_capital:,.2f} -> ${r.final_capital:,.2f}") print(f" Return: {r.total_return_pct:+.2f}%") print(f" Trades: {r.total_trades} (Win: {r.winning_trades}, Loss: {r.losing_trades})") print(f" Win Rate: {r.win_rate:.1%}") print(f" Max Drawdown: {r.max_drawdown_pct:.1f}%") print("\n" + "=" * 60) print("Reports saved to:") print(f" - {report_file}") print(f" - {log_file}") print(f" - {json_file}") print("=" * 60) return all_results if __name__ == "__main__": # Setup logging logger.remove() logger.add(sys.stderr, level="INFO", format="{time:HH:mm:ss} | {level} | {message}") # Run backtest results = run_full_backtest()