mql5/python/ai_backtest.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
AI交易策略回测脚本

支持使用历史数据测试AI交易策略的表现，输出详细的回测报告。
"""

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import matplotlib.dates as mdates
from datetime import datetime, timedelta
import logging
import os
import sys

# 添加项目根目录到Python路径
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from data_processor import MT5DataProcessor
from ai_client_factory import initialize_ai_clients
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('ai_backtest.log'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

class AIBacktester:
    """
    AI交易策略回测器
    """
    
    def __init__(self, initial_capital=100000.0, risk_per_trade=1.0):
        """
        初始化回测器
        
        Args:
            initial_capital (float): 初始资金
            risk_per_trade (float): 每笔交易风险百分比
        """
        self.initial_capital = initial_capital
        self.current_capital = initial_capital
        self.risk_per_trade = risk_per_trade
        self.position = 0  # 持仓状态：1=多头，-1=空头，0=空仓
        self.entry_price = 0.0
        self.trades = []
        self.equity_curve = []
        self.data_processor = MT5DataProcessor()
        
        # 初始化AI客户端 - 使用硅基流动API服务，基于ValueCell的模型工厂模式
        ai_clients = initialize_ai_clients()
        
        self.deepseek_client = ai_clients.get('deepseek')
        self.qwen_client = ai_clients.get('qwen')
        
        # 验证客户端初始化成功
        if not self.deepseek_client or not self.qwen_client:
            logger.error("AI客户端初始化失败，使用模拟数据进行回测")
        
        logger.info(f"AIBacktester初始化完成，初始资金: {initial_capital}, 每笔风险: {risk_per_trade}%")
    
    def get_historical_data(self, symbol, start_date, end_date, timeframe='H1'):
        """
        获取历史数据
        
        Args:
            symbol (str): 交易品种
            start_date (datetime): 开始日期
            end_date (datetime): 结束日期
            timeframe (str): 时间周期
        
        Returns:
            pd.DataFrame: 历史数据
        """
        logger.info(f"获取{symbol}的历史数据，时间范围: {start_date} 至 {end_date}")
        df = self.data_processor.get_historical_data(symbol, None, start_date, end_date)
        logger.info(f"获取到{len(df)}条历史数据")
        return df
    
    def get_ai_signal(self, df):
        """
        获取AI交易信号
        
        Args:
            df (pd.DataFrame): 市场数据
        
        Returns:
            str: 交易信号 (buy, sell, none)
            int: 信号强度 (0-100)
        """
        # 检查AI客户端是否初始化成功
        if not self.deepseek_client or not self.qwen_client:
            logger.error("AI客户端未初始化，无法生成交易信号")
            return "none", 50
        
        # 生成特征
        df_with_features = self.data_processor.generate_features(df)
        
        # 准备模型输入
        df_tail = df_with_features.tail(20)
        df_tail_reset = df_tail.reset_index()
        df_tail_reset['time'] = df_tail_reset['time'].astype(str)
        model_input = df_tail_reset.to_dict(orient='records')
        
        try:
            # 使用DeepSeek分析市场结构
            deepseek_analysis = self.deepseek_client.analyze_market_structure(model_input)
            
            # 使用Qwen3优化策略
            optimized_strategy = self.qwen_client.optimize_strategy_logic(deepseek_analysis, model_input)
            
            # 生成信号
            signal = "none"
            if optimized_strategy["signal_strength"] > 70:
                if df_with_features['ema_fast'].iloc[-1] > df_with_features['ema_slow'].iloc[-1]:
                    signal = "buy"
                else:
                    signal = "sell"
            
            return signal, optimized_strategy["signal_strength"]
        except Exception as e:
            logger.error(f"AI信号生成失败: {e}")
            return "none", 50
    
    def calculate_position_size(self, atr):
        """
        计算仓位大小
        
        Args:
            atr (float): 平均真实波动幅度
        
        Returns:
            float: 仓位大小
        """
        if atr <= 0:
            return 0.0
        
        # 计算风险金额
        risk_amount = self.current_capital * (self.risk_per_trade / 100.0)
        
        # 假设每手价值为100000（根据实际情况调整）
        # 这里简化处理，实际应该根据交易品种的点值计算
        position_size = risk_amount / (atr * 100000)
        
        # 限制最小和最大仓位
        position_size = max(0.01, min(1.0, position_size))
        
        return round(position_size, 2)
    
    def run_backtest(self, symbol, start_date, end_date):
        """
        运行回测
        
        Args:
            symbol (str): 交易品种
            start_date (datetime): 开始日期
            end_date (datetime): 结束日期
        """
        logger.info(f"开始回测 {symbol}, 时间范围: {start_date} 至 {end_date}")
        
        # 获取历史数据
        df = self.get_historical_data(symbol, start_date, end_date)
        
        # 生成特征
        df = self.data_processor.generate_features(df)
        
        # 初始化回测
        self.current_capital = self.initial_capital
        self.position = 0
        self.entry_price = 0.0
        self.trades = []
        self.equity_curve = [self.current_capital]
        
        # 遍历数据进行回测
        for i in range(20, len(df)):
            # 获取当前数据
            current_data = df.iloc[i]
            
            # 计算ATR
            atr = current_data['atr']
            
            # 获取AI信号
            window_df = df.iloc[i-20:i+1]
            signal, signal_strength = self.get_ai_signal(window_df)
            
            # 处理交易信号
            if signal == "buy" and self.position != 1:
                # 平仓现有仓位
                if self.position == -1:
                    self.close_position(current_data['close'])
                
                # 开多仓
                self.position = 1
                self.entry_price = current_data['close']
                position_size = self.calculate_position_size(atr)
                
                self.trades.append({
                    'date': current_data.name,
                    'signal': 'buy',
                    'price': self.entry_price,
                    'position_size': position_size,
                    'signal_strength': signal_strength,
                    'status': 'open'
                })
                
                logger.info(f"开多仓: {symbol}, 价格: {self.entry_price:.5f}, 仓位: {position_size}, 信号强度: {signal_strength}")
            
            elif signal == "sell" and self.position != -1:
                # 平仓现有仓位
                if self.position == 1:
                    self.close_position(current_data['close'])
                
                # 开空仓
                self.position = -1
                self.entry_price = current_data['close']
                position_size = self.calculate_position_size(atr)
                
                self.trades.append({
                    'date': current_data.name,
                    'signal': 'sell',
                    'price': self.entry_price,
                    'position_size': position_size,
                    'signal_strength': signal_strength,
                    'status': 'open'
                })
                
                logger.info(f"开空仓: {symbol}, 价格: {self.entry_price:.5f}, 仓位: {position_size}, 信号强度: {signal_strength}")
            
            # 更新权益曲线
            self.update_equity_curve(current_data['close'])
        
        # 回测结束，平仓所有仓位
        if self.position != 0:
            self.close_position(df.iloc[-1]['close'])
        
        logger.info("回测完成")
        
        # 生成回测报告
        self.generate_report(symbol, start_date, end_date)
    
    def close_position(self, exit_price):
        """
        平仓
        
        Args:
            exit_price (float): 平仓价格
        """
        if self.position == 0:
            return
        
        # 获取当前持仓的交易记录
        for trade in reversed(self.trades):
            if trade['status'] == 'open':
                # 计算盈亏
                if trade['signal'] == 'buy':
                    profit = (exit_price - trade['price']) * trade['position_size'] * 100000
                else:
                    profit = (trade['price'] - exit_price) * trade['position_size'] * 100000
                
                # 更新交易记录
                trade['exit_date'] = datetime.now()
                trade['exit_price'] = exit_price
                trade['profit'] = profit
                trade['status'] = 'closed'
                
                # 更新资金
                self.current_capital += profit
                
                logger.info(f"平仓: 信号: {trade['signal']}, 入场: {trade['price']:.5f}, 出场: {exit_price:.5f}, 盈亏: {profit:.2f}, 资金: {self.current_capital:.2f}")
                break
        
        # 重置持仓状态
        self.position = 0
        self.entry_price = 0.0
    
    def update_equity_curve(self, current_price):
        """
        更新权益曲线
        
        Args:
            current_price (float): 当前价格
        """
        # 如果持仓，计算浮动盈亏
        if self.position != 0:
            if self.position == 1:
                unrealized_profit = (current_price - self.entry_price) * self.trades[-1]['position_size'] * 100000
            else:
                unrealized_profit = (self.entry_price - current_price) * self.trades[-1]['position_size'] * 100000
            
            equity = self.current_capital + unrealized_profit
        else:
            equity = self.current_capital
        
        self.equity_curve.append(equity)
    
    def generate_report(self, symbol, start_date, end_date):
        """
        生成回测报告
        
        Args:
            symbol (str): 交易品种
            start_date (datetime): 开始日期
            end_date (datetime): 结束日期
        """
        logger.info("生成回测报告")
        
        # 计算回测指标
        total_trades = len(self.trades)
        winning_trades = sum(1 for trade in self.trades if trade['profit'] > 0)
        losing_trades = sum(1 for trade in self.trades if trade['profit'] < 0)
        total_profit = sum(trade['profit'] for trade in self.trades if trade['status'] == 'closed')
        max_drawdown = self.calculate_max_drawdown()
        
        # 计算胜率
        win_rate = winning_trades / total_trades * 100 if total_trades > 0 else 0
        
        # 计算平均盈亏比
        avg_win = sum(trade['profit'] for trade in self.trades if trade['profit'] > 0) / winning_trades if winning_trades > 0 else 0
        avg_loss = abs(sum(trade['profit'] for trade in self.trades if trade['profit'] < 0) / losing_trades) if losing_trades > 0 else 1
        risk_reward_ratio = avg_win / avg_loss
        
        # 计算总收益率
        total_return = (self.current_capital - self.initial_capital) / self.initial_capital * 100
        
        # 输出回测报告
        report = f"""===== AI交易策略回测报告 =====
交易品种: {symbol}
时间范围: {start_date} 至 {end_date}
初始资金: {self.initial_capital:.2f}
最终资金: {self.current_capital:.2f}
总收益率: {total_return:.2f}%
总交易次数: {total_trades}
盈利交易: {winning_trades}
亏损交易: {losing_trades}
胜率: {win_rate:.2f}%
平均盈利: {avg_win:.2f}
平均亏损: {avg_loss:.2f}
风险回报比: {risk_reward_ratio:.2f}
最大回撤: {max_drawdown:.2f}%
==========================="""
        
        print(report)
        logger.info(report)
        
        # 保存回测报告
        with open(f"ai_backtest_report_{symbol}_{start_date.strftime('%Y%m%d')}_{end_date.strftime('%Y%m%d')}.txt", 'w') as f:
            f.write(report)
        
        # 绘制权益曲线
        self.plot_equity_curve(symbol, start_date, end_date)
    
    def calculate_max_drawdown(self):
        """
        计算最大回撤
        
        Returns:
            float: 最大回撤百分比
        """
        if not self.equity_curve:
            return 0.0
        
        peak = self.equity_curve[0]
        max_drawdown = 0.0
        
        for equity in self.equity_curve:
            if equity > peak:
                peak = equity
            drawdown = (peak - equity) / peak * 100
            if drawdown > max_drawdown:
                max_drawdown = drawdown
        
        return max_drawdown
    
    def plot_equity_curve(self, symbol, start_date, end_date):
        """
        绘制权益曲线
        
        Args:
            symbol (str): 交易品种
            start_date (datetime): 开始日期
            end_date (datetime): 结束日期
        """
        logger.info("绘制权益曲线")
        
        plt.figure(figsize=(12, 6))
        plt.plot(self.equity_curve, label='Equity Curve')
        plt.title(f'AI Trading Strategy - Equity Curve ({symbol})')
        plt.xlabel('Time Steps')
        plt.ylabel('Equity')
        plt.grid(True)
        plt.legend()
        
        # 保存图表
        plt.savefig(f"ai_backtest_equity_{symbol}_{start_date.strftime('%Y%m%d')}_{end_date.strftime('%Y%m%d')}.png", dpi=300)
        plt.close()
        
        logger.info(f"权益曲线已保存")

def main():
    """
    主函数
    """
    # 配置回测参数
    symbol = "GOLD"
    start_date = datetime.now() - timedelta(days=30)  # 回测最近30天
    end_date = datetime.now()
    initial_capital = 100000.0
    risk_per_trade = 1.0
    
    # 创建回测器
    backtester = AIBacktester(initial_capital, risk_per_trade)
    
    # 运行回测
    backtester.run_backtest(symbol, start_date, end_date)

if __name__ == "__main__":
    main()