蜗牛之路

事件回测的最大好处是基本上可以避免未来函数的出现。

代码如下：

# -*- coding: utf-8 -*-

from __future__ import division

# %matplotlib inline

import tushare as ts
import pandas as pd
import numpy as np

data = ts.get_hist_data("510050",start="2017-01-01",end="2017-09-08")
data = data.sort_values("date")
df = pd.DataFrame()
df['close'] = data['close']
df['change'] = df['close']- df['close'].shift(1)
df = df.dropna()  #将带有Na的去掉，返回一个新的对像

close5_array = np.zeros(5)    #缓存过去5天收盘价
close20_array = np.zeros(20)
last_signal = 0              #最近交易信号，初始化为0。因为双均线策略要使用前一天的收盘价，就需要将信号缓存下来，第2天计算仓位的时候使用
last_pos = 0                 #最近的持仓，初始化为0

dr_list = []   #每日盈亏结果列表

class DailyResult:
    """每日盈亏结果"""
    def _init__(self):
        self.date = ''
        self.close = 0
        self.change = 0
        self.pos = 0
        self.last_pos = 0     #昨日持仓

        self.pnl = 0
        self.fee = 0
        self.net_pnl = 0

    def calculate(self,date,close,change, last_signal, last_pos):
        """计算每日盈亏"""
        #赋值原始数据
        self.date = date
        self.close  = close
        self.change = change
        self.pos = last_signal
        self.last_pos = last_pos

        #计算结果数据
        self.pnl = self.change * self.pos
        self.fee = abs(self.pos-self.last_pos) * 1.5/10000
        self.net_pnl = self.pnl - self.fee

#运行回溯
#iterrows生成迭代器，enumerate在迭代过程中返回当前的变量的计数情况。df刚好有date, close, change三列。叠代过程中会以tuple的形式返回每一行，其中的第一个元素是日期，第二个元素是数值段。
for i, row in enumerate(df.iterrows()):
    date = row[0]
    close =row[1]['close']
    change = row[1]['change']

    #将数组中的数据平移一格
    close5_array[0:4] = close5_array[1:5]
    close20_array[0:19] = close20_array[1:20]

    #将新数据添加到数组末尾
    close5_array[-1] = close
    close20_array[-1] = close

    #如果尚未有20个数据点的缓存数量，则不执行后续逻辑
    if i < 20:
        continue

    #计算当日持仓盈亏
    dr = DailyResult()
    dr.calculate(date, close, change, last_signal,last_pos)

    dr_list.append(dr)

    #记录当日持仓
    last_pos = dr.pos

    #计算信号数据
    ma5 = close5_array.mean()
    ma20 = close20_array.mean()
    if ma5 > ma20:
        last_signal = 10000
    else:
        last_signal = -10000

#显示回测结果
result_df = pd.DataFrame()
result_df['net_pnl'] = [dr.net_pnl for dr in dr_list]    #将dailyresult列表中的数据转换为pandas的dataframe
result_df.index = [dr.date for dr in dr_list]              #添加日期索引

result_df['cum_pnl'] = result_df['net_pnl'].cumsum()    #累积求和
result_df['cum_pnl'].plot()

在jupyter运行结果如下：