金 融 量 化 分 析 • JoinQuant • 第 七 篇

• 一、回 测 是 什 么 ？
• 二、回 测 与 实 盘 交 易
• 三、开源双均线策略回测框架搭建（pyalgotrade）
• • （一）、模块安装
  • （二）、双 均 线 回 测 框 架 搭 建
  • （三）、分 析 结 果

回测是一种反向测试，从各种评价指标和统计学层面上，以大数据的方式评估策略，验证设定的交易规则是否有效。测试里是数据的预测模型，用于估计策略或模型在过去一段时间的表现，需要提供足够的细节模拟过去的条件，条件可以是收盘价为基准。

⚠️ 回测数据与实盘交易数据存在差异性
⚠️ 回测结果 并 不能代表策略百分百能赚钱：注意收益盈亏几点位（是否都是按策略交易）

影响的原因：
1）未来函数的差异
2）滑点

pip install pyalgotrade

# # 调用pyalgotrade回测框架模块
# 调用pyalgotrade回测模块：计算指标、回测数据
from pyalgotrade import strategy
# 调用pyalgotrade数据整理模块：提供数据
from pyalgotrade.barfeed import quandlfeed
# 调用pyalgotrade_tushare模块的导出股票行情函数
from pyalgotrade_tushare import tools, barfeed
# 调用pyalgotrade_tushare模块的  移动平均线函数
from pyalgotrade.technical import ma
# 调用pyalgotrade_tushare模块的  相对强弱指标函数
from pyalgotrade.technical import rsi
# 调用pyalgotrade的画图模块
from pyalgotrade import plotter

from pyalgotrade.stratanalyzer import returns

 



def safe_round(value, digits):
    if value is not None:
        value = round(value, digits)
    return value
# 设置回测类
class MyStrategy(strategy.BacktestingStrategy):
    # feed:数据集，instrument:股票代码 ，smaPeriod:设定N天的均价
    def __init__(self, feed, instrument, smaPeriod1,smaPeriod2):
        # super调用副类回测函数strategy.BacktestingStrategy(数据，投资金额)
        super(MyStrategy, self).__init__(feed,1e5)
        # 设置持有股数为空
        self.__position = None
        # 传递股票代码
        self.__instrument = instrument
         
        # 调整后的输盘价（复权后的收盘价）
        # self.setUseAdjustedValues(True)  

        # # 传递计算移动平均线
        """
        feed[instrument].getCloseDataSeries()：调取股票数据的收盘价数据
        ma.SMA()：计算移动平均线的函数
        smaPeriod：代表N天 
        """
        # 计算均线1
        self.__sma1 = ma.SMA(feed[instrument].getCloseDataSeries(),smaPeriod1)
        # 计算均线2
        self.__sma2 = ma.SMA(feed[instrument].getCloseDataSeries(),smaPeriod2)
        # # 传递计算相对强弱指标
        # self.__rsi = rsi.RSI(feed[instrument].getCloseDataSeries(), 14)
    
    # 定义移动平均线1模块
    def getSMA1(self):
        # 返回移动平均线数据
        return self.__sma1
    
     # 定义移动平均线2模块
    def getSMA2(self):
        # 返回移动平均线数据
        return self.__sma2

    # 买入成交模块
    def onEnterOk(self, position):
        # 设置买入订单信息的执行信息：持有股数 按照 获取入口订单 获取执行信息 实施
        execInfo = position.getEntryOrder().getExecutionInfo()
        # # 打印买入执行价格
        # self.info("BUY at $%.2f" % (execInfo.getPrice()))
    
    # 买入取消模块
    def onEnterCanceled(self, position):
        # 持有股数为空
        self.__position = None

    # 卖出成交模块
    def onExitOk(self, position):
        # 设置卖出订单信息的执行信息：持有股数 按照 获取入口订单 获取执行信息 实施
        execInfo = position.getExitOrder().getExecutionInfo()
        # # 打印卖出执行价格
        # self.info("SELL at $%.2f" % (execInfo.getPrice()))
        # 持有股数为空
        self.__position = None
    
    # 卖出取消模块
    def onExitCanceled(self, position):
        # 如果是卖出取消，则持有股数再次执行卖出执行的逻辑
        self.__position.exitMarket()
    
    # 行情数据模块
    def onBars(self, bars):
        # 如果足够的行情数据可用于计算移动平均线1则输出。
        # if self.__sma1[-1] is None:
        #     # 则输出。
        #     return
        # 如果足够的行情数据可用于计算移动平均线2则输出。
        if self.__sma2[-1] is None:
            # 则输出。
            return
        # 设置股票行情数据
        bar = bars[self.__instrument]
        # # 打印股票对象收盘价的信息
        # # bar.getClose():获取收盘价、self.__rsi：相对强弱指标、self.__sma：移动平均线[15天]
        # self.info("%s %s %s" % (bar.getClose(), safe_round(self.__rsi[-1], 2), safe_round(self.__sma[-1], 2)))
        
        # # 定义策略
        # 如果当前没有买入股票
        if self.__position is None:
            # 如果当前行情大于当前移动平均线值
            # if bar.getPrice() > self.__sma[-1]:
            # 如果当前短期均价大于当前长期期均价
            if self.__sma1[-1] > self.__sma2[-1]:
                """  
                self.enterLong():执行买入操作
                """
                # 输入购买100股股票的市场指令，订单取消前一直有效。(持有股数按照每手100股买入)
                self.__position = self.enterLong(self.__instrument, 100, True)
        # 否则检查是否必须退出该位置。(如果卖出：如果当前行情小于当前移动平均线值 且 持有股数不是撤单的情况)
        # elif bar.getPrice() < self.__sma[-1] and not self.__position.exitActive():
        # 否则检查是否必须退出该位置。(如果卖出：如果当前短期均价小于当前长期均价 且 持有股数不是撤单的情况)
        elif self.__sma1[-1] < self.__sma2[-1] and not self.__position.exitActive():
            # 持有股票数执行卖出逻辑
            self.__position.exitMarket()
        


print(']]]]]]]]]]]]]]]]]]]]]]]]]]]]]]]]]]]]]]]')
# 策略运行模块
def run_strategy(smaPeriod1,smaPeriod2):
    """ smaPeriod:移动平均周期 """
    # 设置股票代码
    codes = ["000001"]
    # 调取股票行情数据
    data = tools.build_feed(codes, 2018,2022, "histdata")
    # 实化策略模块
    myStrategy = MyStrategy(data, codes[0], smaPeriod1 = smaPeriod1, smaPeriod2 = smaPeriod2)
    # # 运行策略回测
    # myStrategy.run()
    # # 打印策略投资结果
    # print(smaPeriod1,smaPeriod2,"Final portfolio value: $%.2f" % myStrategy.getResult())



    # # 可视化
    # 实化returns模块（分析器）
    returnsAnalyzer = returns.Returns()
    # 将分析器附加到策略当中
    myStrategy.attachAnalyzer(returnsAnalyzer)

    # 将绘图模块连接到策略当中。
    plt = plotter.StrategyPlotter(myStrategy)
    # 将SMA包含在分析图的子图中，将策略与收盘价一起显示。
    plt.getInstrumentSubplot(codes[0]).addDataSeries("SMA1", myStrategy.getSMA1())
    plt.getInstrumentSubplot(codes[0]).addDataSeries("SMA2", myStrategy.getSMA2())
    # 绘制每次收益率曲线。
    plt.getOrCreateSubplot("returns").addDataSeries("Simple returns", returnsAnalyzer.getReturns())

    # 运行策略回测
    myStrategy.run()
    # 打印策略投资结果
    myStrategy.info("Final portfolio value: $%.2f" % myStrategy.getResult())

    # 制定策略。
    plt.plot()

# # 设置不同的周期进行回测
# for i in range(5,30):
#     # 回测运行
run_strategy(15,30)
# ma1 = [5,10,15]
# ma2 = [10,20,30]
# for i in ma1:
#     for j in ma2:
#         if i < j:
#             run_strategy(i,j)