量化24小时 - 第二部分 Python 量化回测框架原理以及基本思路

发表于 | 分类于 | |

资料

zipline、quantdigger、pyfolio 这些比较成熟的框架源码

OnePy–构建属于自己的量化回测框架

Pandas 教程
教程资料链接: https://pan.baidu.com/s/1UVT_S9BHJZS8ktjAc8Xpcw 密码: dr78

策略回测基本思路

  1. 双 MA 策略是什么 - Moving Average 移动平均线

长短周期的移动平均线, 交叉展现出买卖信号

  1. 不考虑单日时间内的双向操作
  • 收盘价格 Close, 计算 MA5 MA20 以及等买卖信号
  • 持仓量 Position
  • 价格变动 Change (当天收盘价格 - 昨天收盘价格)
  • 损益 PNL (Profit & Loss, 当天持仓量 * 价格变动)
  • 手续费 Fee (持仓变动 * 手续费率)
  • 当日净损益 NetPNL (Net Profit & Loss, 损益 - 手续费)
  • 总损益 CumPNL (Cumulative Profit & Loss, 所有之前净损益之和)
  • 回撤率 = （之前最高 CumPNL 值 — 当前 CumPNL 值）/ 之前最高 CumPNL 值
  • 最大回撤率 = 所有时间点的回撤率, 取最大值
  • 回报率 = 损益 / 本金 * 100%

注意: 未来函数, 以当天的 Close 等收盘时才有的参数指标, 计算出来的持仓量的变化, 只能更改下一天的持仓量(因为当天 close 价格出现之后就无法再进行交易)

绘图类

币对, 最大回撤率,

1. 日期:
2. 总损益: 画线
3. 是否交易(方向, 价格, 交易额, 手续费, 其他信息): 描点
4. 其他值: 收盘价, 价格变动率, 持仓量, 持仓损益

四种回测方法

  1. Excel

  2. Python 向量化

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
import matplotlib.pylab as plt
import matplotlib.pyplot as plt
import tushare as ts
import pandas as pd
import numpy as np
import math

# 获取数据
data = ts.get_hist_data("510050")
data = data.sort_index()

# 计算每日涨跌
df = pd.DataFrame()
df['close'] = data['close']
df['change'] = df['close'] - df['close'].shift(1)

# 计算指标
df['ma5'] = df['close'].rolling(window=5, center=False).mean()
df['ma20'] = df['close'].rolling(window=20, center=False).mean()

df = df.dropna()

# ma5 > ma20 持有多头, 反之持有空头
df['pos'] = 0
df['pos'][df['ma5']>df['ma20']] = 1e4
df['pos'][df['ma5']==df['ma20']] = 1e4
df['pos'][df['ma5']<df['ma20']] = -1e4
# 这里为了防止未来函数, 持仓量后移
df['pos'] = df['pos'].shift(1).fillna(0)

# 当前持仓, 根据当日价格与昨日差价变化的当日损益
df['pnl'] = df['change'] * df['pos']
df['fee'] = df['close'] * (df['pos']-df['pos'].shift(1)).abs() * 3e-4
# 净损益
df['netpnl'] = df['pnl'] - df['fee']
# 损益
df['cumpnl'] = df['netpnl'].cumsum()

df = df.dropna()

df['cumpnl'].plot()
plt.show()
  1. Python 事件
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
import matplotlib.pylab as plt
#import matplotlib.pyplot as plt
import tushare as ts
import pandas as pd
import numpy as np
import math

# 获取数据
data = ts.get_hist_data("510050")
data = data.sort_index()

# 计算每日涨跌
df = pd.DataFrame()  # 注意数据结构
df['open'] = data['open']
df['close'] = data['close']
df['change'] = df['close'] - df['close'].shift(1)

# 以上部分内容为构建数据, 与向量化回测相同

close5_arr = np.zeros(5)
close20_arr = np.zeros(20)
last_signal = 0
last_pos = 0

dr_list = []

class DailyResult:
  """每日盈亏结果类"""

  def __init__(self):
    """构造函数, 初始化成员变量"""
    # 基础数据
    self.date = ''
    self.open = 0
    self.close = 0

    self.change = 0
    self.pos = 0
    self.last_pos = 0

    # 盈亏数据
    self.pnl = 0
    self.fee = 0
    self.netpnl = 0

  def caculate(self, date, open_price, close_price, change, last_signal, last_pos):
    """计算每日盈亏"""

    # 基础数据赋值
    self.date = date
    self.open = open_price
    self.close = close_price
    self.change = change
    self.last_signal = last_signal
    self.last_pos = last_pos

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


for idx, row in enumerate(df.iterrows()):
  date = row[0]
  open_price = row[1]['open']
  close_price = row[1]['close']
  change = row[1]['change']

  print(date, open_price, close_price, change)

  # 0:4 不包括4
  close5_arr[0:4] = close5_arr[1:5]
  close5_arr[-1] = close_price
  close20_arr[0:19] = close20_arr[1:20]
  close20_arr[-1] = close_price

  if (idx < 20):
    continue

  dr = DailyResult()
  dr.caculate(date, open_price, close_price, change, last_signal, last_pos)
  dr_list.append(dr)

  last_pos = dr.pos
  ma5 = close5_arr.mean()
  ma20 = close20_arr.mean()

  if (ma5 >= ma20):
    last_signal = 10000
  else:
    last_signal = -10000


result_df = pd.DataFrame()
result_df['netpnl'] = [dr.netpnl for dr in dr_list]
result_df.index = [dr.date for dr in dr_list]

result_df['cumpnl'] = result_df['netpnl'].cumsum()
result_df['cumpnl'].plot()
plt.show()
  1. vn.py

数据的获取与存储

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
import tushare as ts
import pymongo

from datetime import datetime

from vnpy.trader.vtObject import VtBarData
from vnpy.trader.app.ctaStrategy.ctaBase import DAILY_DB_NAME

symbol = '510050'
exchange = 'SSR'
vtSymbol = '.'.join([symbol, exchange])

data = ts.get_hist_data(symbol, '2017-01-01')
data = data.sort_index()

print('数据下载完毕')

client = pymongo.MongoClient('127.0.0.1', 27017)
collection = client[DAILY_DB_NAME][vtSymbol]
collection.create_index('datetime')

print('数据库连接成功')

for row in data.iterrows():
  date = row[0]
  data = row[1]

  bar = VtBarData()
  bar.vtSymbol = vtSymbol
  bar.symbol = symbol
  bar.exchange = exchange
  bar.date = date
  bar.datatime = datatime.strptime(date, '%Y-%m-%d')
  bar.open = data['open']
  bar.close = data['close']
  bar.high = data['high']
  bar.low = data['low']
  bar.volume = data['volume']

  cond = {'datetime': bar.datetime}
  collection.update_one(cond, {'$set': bar.__dict__}, upsert=True)

print('数据入库完成')

策略编写

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
import numpy as np

from vnpy.trader.app.ctaStrategy.ctaTemplate import CtaTemplate

class DoubleMaStrategy(CtaTemplate):
  """双均线策略"""
  className = "DoubleMaStrategy"
  author = "OnO<corn.mars@ono.lol>"

  initDays = 25

  barCount = 0
  closeArray = np.zeros(20)

  ma5 = 0
  ma20 = 0
  lastMa5 = 0
  lastMa20 = 0

  parmaList = ["name", "className", "author", "vtSymbol"]
  varList = ["inited", "trading", "pos"]

  def __init__(self, ctaEngine, settings):
    super(DoubleMaStrategy, self).__init__(ctaEngine, settings)
    # 注意, 所有可变对象必须在 init 的时候初始化一次, 以免数据出现问题
    self.closeArray = np.zeros(20)

  def onInit(self):
    self.writeLog("初始化双均线策略")

    initData = self.loadBar(self.initDays)
    for bar in initData:
      self.onBar(bar)

    self.putEvent()

  def onStart(self):
    self.writeLog("策略启动")
    self.putEvent()

  def onStop(self):
    self.writeLog("策略停止")
    self.putEvent()

  def onTick(self, tick):
    self.writeLog("接收到行情 tick 推送, 实盘需要根据 tick 手动合成 K线")
    self.writeLog(tick)
    self.putEvent()

  def onBar(self, bar):
    self.writeLog("接收到 K线 bar 推送")

    self.closeArray[0,19] = self.closeArray[1,20]
    self.closeArray[-1] = bar.close

    self.barCount += 1
    if self.barCount < self.initDays:
      return

    self.ma5 = self.closeArray[15:20].mean()
    self.ma20 = self.closeArray.mean()

    goldX = self.ma5 >= self.ma20 and self.lastMa5 < self.lastMa20
    deadX = self.ma5 <= self.ma20 and self.lastMa5 > self.lastMa20

    if goldX:
      if self.pos is 0:
        self.buy(bar.close * 1.05, 1e4)
      elif self.pos < 0:
        self.cover(bar.close * 1.05, 1e4)
        self.buy(bar.close * 1.05, 1e4)
    elif deadX:
      if self.pos is 0:
        self.sell(bar.close * 1.05, 1e4)
      elif self.pos > 0:
        self.sell(bar.close * 1.05, 1e4)
        self.short(bar.close * 1.05, 1e4)

    self.putEvent()

  def onOrder(self, order):
    self.writeLog("委托")
    self.putEvent()

  def onTrade(self, trade):
    self.writeLog("成交")
    self.putEvent()

  def.onStopOrder(self):
    self.writeLog("停止订单推送")
    self.putEvent()

调用策略

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
# encoding: utf8
import os
import sys

current_dir = os.path.abspath(os.path.dirname(__file__))
sys.path.append(current_dir)
sys.path.append("..")

from vnpy.trader.app.ctaStrategy.ctaBase import DAILY_DB_NAME
from vnpy.trader.app.ctaStrategy.ctaBacktesting import BacktestingEngine

import lib.DoubleMaStrategy as DoubleMaStrategy

symbol = '510050'
exchange = 'SSR'
vtSymbol = '.'.join([symbol, exchange])

engine = BacktestingEngine()
engine.setBacktestingMode(engine.BAR_MODE)
engine.setStartDate('20170101', initDays=20)

engine.setSlippage(0)      # 滑点
engine.setRate(1.5 / 1e4)  # 手续费
engine.setSize(1)          # ETF 都为1
engine.setPriceTick(1e-3)  # 最小价格变动
engine.setCapital(1)       # 初始资金

engine.setDatabase(DAILY_DB_NAME, vtSymbol)
engine.initStrategy(DoubleMaStrategy, {})

engine.runBacktesting()
engine.showDailyResult()
Donate - Support to make this site better.
捐助 - 支持我让我做得更好.