miniqmt、backtesting实现双均线策略例子

# -*- coding: utf-8 -*-
import pandas as pd
import numpy as np
import time
from backtesting import Backtest, Strategy
from backtesting.lib import crossover
from xtquant import xtdata


class DualMovingAverageStrategy(Strategy):
    """
    双均线交叉策略回测类
    基于MiniQMT的xtdata数据源
    """

    # 策略参数：短期均线窗口和长期均线窗口
    short_window = 5
    long_window = 20

    def init(self):
        """
        初始化函数：在回测开始前计算技术指标
        """
        # 计算短期移动平均线
        self.short_ma = self.I(
            lambda x: pd.Series(x).rolling(self.short_window).mean(),
            self.data.Close
        )

        # 计算长期移动平均线
        self.long_ma = self.I(
            lambda x: pd.Series(x).rolling(self.long_window).mean(),
            self.data.Close
        )

    def next(self):
        """
        每个交易日执行的主要交易逻辑
        """
        # 确保有足够的数据计算均线
        if len(self.data.Close) < self.long_window:
            return

        # 双均线金叉信号：短期均线上穿长期均线，买入
        if crossover(self.short_ma, self.long_ma) and not self.position:
            self.buy(size=0.95)  # 使用95%资金买入

        # 双均线死叉信号：短期均线下穿长期均线，卖出
        elif crossover(self.long_ma, self.short_ma) and self.position:
            self.position.close()


def get_miniqmt_data(stock_code, start_time, end_time, period='1d'):
    """
    使用miniQMT的xtdata接口获取股票历史数据

    参数:
        stock_code: 股票代码，如 '000001.SZ'
        start_time: 开始时间，如 '20240101'
        end_time: 结束时间，如 '20240930'
        period: K线周期，默认为日线 '1d'

    返回:
        data_df: 格式化后的DataFrame，包含OHLCV数据
    """
    print(f"正在获取 {stock_code} 从 {start_time} 到 {end_time} 的{period}数据...")

    try:
        # 1. 下载历史数据到本地缓存
        print("步骤1: 下载历史数据...")
        xtdata.download_history_data2(
            stock_list=[stock_code],
            period=period,
            start_time=start_time,
            end_time=end_time
        )

        # 等待数据下载完成
        time.sleep(3)

        # 2. 获取数据
        print("步骤2: 从本地缓存读取数据...")
        raw_data =  xtdata.get_market_data_ex(
            field_list=['open', 'high', 'low', 'close', 'volume', 'amount', 'time'],  # 明确指定需要的字段
            stock_list=[stock_code],
            period=period,
            start_time=start_time,
            end_time=end_time
        )

        if not raw_data or stock_code not in raw_data:
            print(f"获取数据失败，请检查股票代码 {stock_code} 是否正确或数据是否存在")
            return None

        # 3. 数据格式转换
        stock_data = raw_data[stock_code]

        # 确保数据是DataFrame格式
        if not isinstance(stock_data, pd.DataFrame):
            print("获取的数据格式不符合预期")
            return None

        # 重命名列以符合backtesting.py的要求
        column_mapping = {
            'open': 'Open',
            'high': 'High',
            'low': 'Low',
            'close': 'Close',
            'volume': 'Volume'
        }

        # 应用列名映射
        stock_data = stock_data.rename(columns=column_mapping)

        # 确保包含所有必需的列
        required_columns = ['Open', 'High', 'Low', 'Close', 'Volume']
        missing_columns = [col for col in required_columns if col not in stock_data.columns]

        if missing_columns:
            print(f"数据缺少必要的列: {missing_columns}")
            print(f"可用的列: {list(stock_data.columns)}")
            return None

        # 选择所需的列并确保数据类型正确
        data_df = stock_data[required_columns].copy()
        data_df = data_df.apply(pd.to_numeric, errors='coerce')
        data_df = data_df.dropna()

        # 确保索引是日期时间类型
        if not isinstance(data_df.index, pd.DatetimeIndex):
            try:
                data_df.index = pd.to_datetime(data_df.index)
            except:
                print("无法将索引转换为日期时间格式，将使用默认索引")

        print(f"数据获取成功！共获取 {len(data_df)} 条记录")
        print(data_df.head())

        return data_df

    except Exception as e:
        print(f"获取数据过程中发生错误: {e}")
        return None


def safe_get_stat(stats, key, default='N/A'):
    """
    安全获取统计值，避免KeyError
    """
    return stats.get(key, default)


def main():
    """
    主函数：执行完整的回测流程
    """
    # --- 配置参数 ---
    STOCK_CODE = '510300.SH'  # 股票代码
    START_TIME = '20250101'  # 开始时间
    END_TIME = '20250930'  # 结束时间
    INITIAL_CASH = 100000  # 初始资金
    COMMISSION = 0.002  # 手续费率

    # 短期和长期均线参数
    SHORT_WINDOW = 5
    LONG_WINDOW = 20

    print("=" * 60)
    print("MiniQMT双均线策略回测系统")
    print("=" * 60)

    # 1. 获取真实数据
    print("\n1. 从MiniQMT获取行情数据...")
    price_data = get_miniqmt_data(STOCK_CODE, START_TIME, END_TIME)

    if price_data is None or price_data.empty:
        print("数据获取失败，请检查网络连接、股票代码和时间参数")
        return

    # 2. 初始化回测环境
    print("\n2. 初始化回测环境...")
    bt = Backtest(
        price_data,
        DualMovingAverageStrategy,
        cash=INITIAL_CASH,
        commission=COMMISSION,
        exclusive_orders=True
    )

    # 3. 运行回测
    print("3. 运行双均线策略回测...")
    stats = bt.run(short_window=SHORT_WINDOW, long_window=LONG_WINDOW)

    # 4. 输出回测结果
    print("\n" + "=" * 60)
    print("回测结果统计")
    print("=" * 60)

    # 使用安全方法获取统计值
    end_value = safe_get_stat(stats, 'Equity Final [$]', 0)
    total_return = safe_get_stat(stats, 'Return [%]', 0)
    sharpe_ratio = safe_get_stat(stats, 'Sharpe Ratio', 0)
    max_drawdown = safe_get_stat(stats, 'Max. Drawdown [%]', 0)
    trades_count = safe_get_stat(stats, '# Trades', 0)
    win_rate = safe_get_stat(stats, 'Win Rate [%]', 0)

    print(f"标的股票: {STOCK_CODE}")
    print(f"回测期间: {START_TIME} 至 {END_TIME}")
    print(f"策略参数: {SHORT_WINDOW}日/{LONG_WINDOW}日均线")
    print(f"最终净值: ¥{end_value:,.2f}")
    print(f"总收益率: {total_return:.2f}%")
    print(f"夏普比率: {sharpe_ratio:.2f}")
    print(f"最大回撤: {max_drawdown:.2f}%")
    print(f"总交易次数: {trades_count}")
    print(f"胜率: {win_rate:.2f}%")

    # 5. 显示关键性能指标
    print("\n关键性能指标:")
    important_stats = [
        'Equity Final [$]', 'Return [%]', 'Sharpe Ratio',
        'Max. Drawdown [%]', '# Trades', 'Win Rate [%]',
        'Volatility (Ann.) [%]', 'Sortino Ratio', 'Calmar Ratio'
    ]

    for key in important_stats:
        if key in stats:
            print(f"{key}: {stats[key]}")

    # 6. 参数优化
    try:
        print("\n4. 进行参数优化...")
        optimization_result = bt.optimize(
            short_window=range(3, 15, 2),
            long_window=range(15, 50, 5),
            maximize='Sharpe Ratio'
        )

        print("最优参数组合:")
        print(f"短期均线: {optimization_result._strategy.short_window}")
        print(f"长期均线: {optimization_result._strategy.long_window}")

    except Exception as e:
        print(f"参数优化跳过或失败: {e}")

    # 7. 绘制回测图表
    try:
        print("\n5. 生成回测图表...")
        bt.plot()
    except Exception as e:
        print(f"图表生成失败: {e}")


if __name__ == '__main__':
    # 确保已启动MiniQMT并登录
    print("请注意: 运行前请确保已启动MiniQMT并成功登录!")
    main()