量化研究---推出强大西蒙斯量化交易回测系统2.0

*******0079 · 发表于前天 13:26

工欲善其事必先利其器，有一个强大的回测框架检验策略还是很必要，我目前利用qmt，ptrade回测，感觉还是有很大的不足，我就开发了自己的回测系统

底层的源代码不亚于我开发的实盘交易系统

同时我把qmt实盘也兼容过来了支持实盘交易

看一下目前免费的回测框架

Python 的策略回测框架种类繁多，适用于不同场景和需求。以下是目前主流的 Python 回测框架及其特点：

1. Backtrader

特点：
- 高度模块化，支持多资产（股票、期货、外汇等）和多时间周期回测。
- 内置技术分析指标（如 Talib），支持自定义策略和数据分析。
- 本地数据兼容性强，可直接加载 CSV、Pandas DataFrame 或数据库数据。
- 支持实盘交易对接（如 Interactive Brokers）。
适用场景：适合需要深度定制策略的量化交易者710。

2. Zipline

特点：
- 由 Quantopian 开发，事件驱动模型，精确模拟订单执行、滑点与手续费。
- 适用于美股市场，但可通过插件扩展支持 A 股等市场。
- 提供 Quantopian 生态集成，支持社区策略共享。
适用场景：适合美股策略回测及社区协作28。

3. RQAlpha（米筐）

特点：
- 支持 A 股、期货、ETF 等市场，提供完整的回测、模拟交易、实盘交易解决方案。
- 模块化架构，易于扩展，适合结合机器学习进行量化研究。
- 提供滑点、手续费、换手率等真实交易环境模拟。
适用场景：适合中国市场的量化策略开发2。

4. PyAlgoTrade

特点：
- 轻量级，运行速度快，适合快速验证中小型策略。
- 支持实时交易接口（如 OANDA、Robinhood）。
- 内置 TALib 技术指标，但不深度整合 Pandas。
适用场景：适合高频策略回测及实盘交易410。

5. QuantConnect

特点：
- 云端一体化平台，支持 Python 和 C#，提供全球化数据（包括新加坡交易所）。
- 内置风险管理工具，如波动率监测、最大回撤预警。
- 订阅制收费，适合资源有限的团队。
适用场景：适合需要全托管服务的量化团队37。

6. vn.py

特点：
- 国内开源框架，侧重实盘交易，支持 A 股、期货、期权等市场。
- 提供 GUI 环境（VN Station），内置 CTA 策略、价差交易等功能。
- 社区活跃，中文文档完善。
适用场景：适合中国市场的专业量化交易员86。

7. Hikyuu

特点**：**
- 基于 C++/Python 的高性能框架，百万级 K 线回测仅需 1~2 秒。
- 支持 A 股全市场计算，内置多种技术指标优化。
- 提供 pyecharts 可视化支持。
适用场景**：适合超大规模数据回测及高频策略研究9。**

8. bt (Backtesting Toolkit)

特点：
- 基于 Pandas 和 NumPy，支持树状策略结构（多层策略组合）。
- 提供多种算法（如等权重再平衡、动量策略）。
- 适合多资产组合优化。
适用场景：适合资产配置策略回测5。

如何选择？

美股策略：Zipline、QuantConnect
A 股/中国市场：RQAlpha、vn.py、Hikyuu
高频/实盘交易：PyAlgoTrade、Backtrader

目前就backtrader简单一点本地的回测框架，但是数据处理也是一个难题，我底层的回测框架支持自动加载数据，支持东方财富，qmt，通达信数据源，自动加载数据

1点击安装第三方库库，安装需要安装的库

点击例子回测可以设置回测参数

参数设置

回测的结果自动加载数据

开始回测交易

生成的个股交易买卖点方便分析

个股交易买卖点

点击策略报告文件夹看策略的报告

策略报告

和实盘差不多，年化20，最大回撤3.7，实盘有高频模块的加持回撤小一点，大概2%

实盘挂服务器24小时运行

明细的策略报告内容

源代码框架我全部上传了知识星球可以下载直接使用，最后几张五一优惠

不懂的问我就可以，加我备注入群可以加入量化研究群

回测的源代码

'''
西蒙斯量化回测系统
作者:西蒙斯量化
微信:xg_quant
'''
from backtrader import backtrader
from user_def_data.user_def_data import user_def_data
import pandas as pd
from trader_tool import jsl_data
from trader_tool.dfcf_etf_data import dfcf_etf_data
from tqdm import tqdm
from trader_tool.tdx_trader_function import tdx_trader_function
import pandas as pd
class my_backtrader:
    '''
    分析模型
    '''
    def __init__(self,trader_tool='ths',data_api='dfcf',
                start_date='20240101',end_date='20500101',data_type='D',
                 starting_cash=300000,cash=300000,commission=0.001,index_stock='000300',is_week=False):
        self.trader_tool=trader_tool
        self.data_api=data_api
        self.start_date=start_date
        self.end_date=end_date
        self.data_type=data_type
        self.starting_cash=starting_cash
        self.commission=commission
        self.index_stock=index_stock
        #这里输入代码就可以
        #导入自定义股票池
        stock_df=pd.read_excel(r'自定义交易股票池.xlsx',dtype='object')
        stock_df['证券代码']=stock_df['证券代码'].astype(str)
        print(stock_df['证券代码'].tolist())
        print(stock_df['名称'].tolist())
        print(stock_df)
        stock_list=stock_df['证券代码'].tolist()
        #stock_list=['512890']
        stock_list=stock_list
        self.stock_list=stock_list
        self.amount=1000
        self.hold_limit=2000
        #采用目标数量交易
        self.buy_target_volume=10000
        self.sell_target_volume=0
        self.buy_target_value=10000
        self.sell_target_value=0
        self.tdx_trader_function=tdx_trader_function()
        self.is_week=is_week
        self.week_buy_n=2
        self.week_sell_n=1
        self.daily_buy_n=4
        self.daily_sell_n=3
        #一般是-3%为强制
        self.top_sell=-2
        self.trader=backtrader(start_date=self.start_date,end_date=self.end_date,
                               data_type=self.data_type,starting_cash=self.starting_cash,
                               commission=self.commission,cash=cash,index_stock=self.index_stock)
        self.data=user_def_data(trader_tool=self.trader_tool,
                        start_date=self.start_date,end_date=self.end_date,data_type=self.data_type)
    def add_all_data(self):
        '''
        多线程加载数据
        '''
        self.data.get_thread_add_data(stock_list=self.stock_list)
        self.hist=self.data.hist
        return self.hist
    def add_all_week_data(self):
        '''
        获取全部的周数据
        '''
        self.week_df=pd.DataFrame()
        for i in range(len(self.stock_list)):
            stock=self.stock_list[i]
            df=self.data.data.get_hist_data_em(stock=stock,start_date=self.start_date,end_date=self.end_date,data_type='W')
            df['stock']=stock
            self.week_df=pd.concat([self.week_df,df],ignore_index=True)
        print(self.week_df)
        return self.week_df
    #单线程加载数据，避免数据不成功
    def add_all_data_1(self):
        '''
        单线程加载数据，避免数据不成功
        '''
        for i in tqdm(range(len(self.stock_list))):
            stock=self.stock_list[i]
            stock=str(stock)
            self.data.get_add_data(stock=stock)
        self.hist=self.data.hist
        print(self.hist)
        return self.hist
    def connect_daily_week_data(self):
        '''
        合并日线周线数据
        '''
        self.week_df=self.add_all_week_data()
        self.hist=self.add_all_data_1()
        data=pd.DataFrame()
        for stock in self.stock_list:
            week=self.week_df[self.week_df['stock']==stock]
            week_close=dict(zip(week['date'],week['close']))
            week_open=dict(zip(week['date'],week['open']))
            week_low=dict(zip(week['date'],week['low']))
            week_high=dict(zip(week['date'],week['high']))
            hist=self.hist[self.hist['stock']==stock]
            hist['week_close']=hist['date'].apply(lambda x:week_close.get(x,None))
            hist['week_open']=hist['date'].apply(lambda x:week_open.get(x,None))
            hist['week_low']=hist['date'].apply(lambda x:week_low.get(x,None))
            hist['week_high']=hist['date'].apply(lambda x:week_high.get(x,None))
            hist=hist.fillna(method='ffill')
            data=pd.concat([data,hist])
        return data
    def get_cacal_all_indicators(self):
        '''
        计算全部的指标
        '''
        hist=self.connect_daily_week_data()
        trader_info=pd.DataFrame()
        #拆分数据
        for stock in self.stock_list:
            df=hist[hist['stock']==stock]
            week_df=df[['date','week_close','week_open','week_low','week_high']]
            week_df.columns=['date','close','open','low','high']
            signal_daily,markers=self.tdx_trader_function.six_pulse_excalibur_hist(df)
            signal_week,markers=self.tdx_trader_function.six_pulse_excalibur_hist(week_df)
            df['signal_daily']=signal_daily
            df['signal_week']=signal_week
            trader_info=pd.concat([trader_info,df],ignore_index=True)
        return trader_info
    def run_backtrader(self):
        '''
        运行回测
        '''
        trader_list=self.trader.get_trader_date_list()
        trader_info=self.get_cacal_all_indicators()
        for date in trader_list:
            df=trader_info[trader_info['date']==date]
            stock_list=df['stock'].tolist()
            for stock in stock_list:
                df1=df[df['stock']==stock]
                price=df1['close'].tolist()[-1]
                price=float(price)
                zdf=df1['涨跌幅'].tolist()[-1]
                signal_daily=df1['signal_daily'].tolist()[-1]
                signal_week=df1['signal_week'].tolist()[-1]
                if self.is_week:
                    if signal_week>=self.week_buy_n:
                        if signal_daily>=self.daily_buy_n:
                            result=self.trader.order_target_volume(date=date,stock=stock,amount=self.buy_target_volume,price=price)
                            if result==True:
                                pass
                            else:
                                self.trader.settle(date=date,stock=stock,price=price)
                        else:
                            pass
                    else:
                        print("时间{} 标的{} 不符合周线买入".format(date,stock))
                else:
                    if signal_daily>=self.daily_buy_n:
                        result=self.trader.order_target_volume(date=date,stock=stock,amount=self.buy_target_volume,price=price)
                        if result==True:
                            pass
                        else:
                            self.trader.settle(date=date,stock=stock,price=price)
                    else:
                        pass
                if self.is_week:
                    if signal_week<=self.week_sell_n:
                        result=self.trader.order_target_volume(date=date,stock=stock,amount=self.sell_target_volume,price=price)
                        if result==True:
                            pass
                        else:
                            self.trader.settle(date=date,stock=stock,price=price)
                    else:
                        print("时间{} 标的{} 不符合周线卖出".format(date,stock))
                else:
                    print("时间{} 标的{} 不开启周线卖出".format(date,stock))
                if signal_daily<self.daily_sell_n:
                    result=self.trader.order_target_volume(date=date,stock=stock,amount=self.sell_target_volume,price=price)
                    if result==True:
                        pass
                    else:
                        self.trader.settle(date=date,stock=stock,price=price)
                else:
                    print("时间{} 标的{} 不符合日线卖出".format(date,stock))
                if zdf<=self.top_sell:
                    result=self.trader.order_target_volume(date=date,stock=stock,amount=self.sell_target_volume,price=price)
                    if result==True:
                        pass
                    else:
                        self.trader.settle(date=date,stock=stock,price=price)
                else:
                    print("时间{} 标的{} 不符合止盈止损卖出".format(date,stock))
                
if __name__=='__main__':
    trader=my_backtrader(data_type='D',
                )
    trader.run_backtrader()
    #获取全部的交易报告
    trader.trader.get_poition_all_trader_report_html()
    #获取策略报告
    trader.trader.get_portfolio_trader_report_html()
    #显示个股的交易图
    trader.trader.get_plot_all_trader_data_figure(limit=1000)
    #显示策略数据
    df=trader.trader.get_portfolio_trader_data_figure(limit=100000)