428 lignes
14 Kio
Python
428 lignes
14 Kio
Python
#!/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()
|