2022年1月2日新版策略自动生成回测文件功能代码解析

admin · 发表于 2022-1-2 06:24:35

VNPY精华帖子列表：https://q.vnpy.cn/comm/thread-18-1-1.html
项目地址： https://gitee.com/vnpypro/vnpy
VNPY官网：https://www.vnpy.cn
快速入门： https://q.vnpy.cn/comm/thread-13-1-1.html

其他参考文章
《快速入门教程》
https://q.vnpy.cn/comm/thread-13-1-1.html
《新版策略自动生成回测文件功能代码解析》
https://q.vnpy.cn/comm/thread-22-1-1.html
《VNPY3.0以后的版本为什么不用数据库设计架构？》
https://q.vnpy.cn/comm/thread-21-1-1.html
《VNPY3.0行情数据调用的5种方式》
https://q.vnpy.cn/comm/thread-24-1-1.html
《VNPY新手常见问题说明》
https://q.vnpy.cn/comm/thread-3-1-1.html

VNPY官网
https://www.vnpy.cn
VNTA 证券和期货方案
http://www.vnta.cn
《用户福利》
https://www.vnpy.cn/kaihu/

图1 回测效果图

图1：回测过程

2022年1月2日版本，实现了策略文件生成模块，本文就是介绍回测机制。

在VNPY3.0项目目录下有strategyfile和strategyfilebacktest目录。

strategyfile 是实盘策略文件目录
strategyfilebacktest 是量化回测策略文件目录

在VNPY主界面点击红圈中的“回测”按钮，即可打开图3界面

回测是由module_backtest.py文件实现的

图2

图2：VNPY窗口界面

图3

图3：进入到回测界面

其中我们开发策略的时候是修改strategyfile下的策略文件，回测时会自动生成回测专用的策略文件，存在strategyfilebacktest 目录下。

点击上图中的“开始回测”按钮就会触发转换函数

class DialogBackTest(QtWidgets.QDialog) 类的 GenerateBackTestFile(self)这个方法

复制代码

这样就由strategyfile下的talib_MA.py（实盘策略文件）生成了strategyfilebacktest 目录下的talib_MA.py（量化回测策略文件）

我们看一下这2个策略文件结构上有什么不同，我们以最简单的TALIB MA策略为例,来说明实盘策略代码和量化回测用的代码的区别。

（1）在def __init__(self, period, slippoint):方法内增加了以下代码：

self.TradingDay = []
self.klinetime = []
self.open = []
self.high = []
self.low = []
self.close = []
self.volume = []
self.money = []
self.open_interest = []
self.InstrumentID = []

复制代码

（2）将

def OnKline(self, mddata, arg, strategyname):

复制代码

改为

def OnKline(self, reportpath1, reportpath2, mddata, arg, strategyname):

复制代码

（3）在def OnKline(self, mddata, arg, strategyname):方法内增加了以下代码：

super(MyStrategy, self).OnKline(reportpath1, reportpath2, arg, mddata)

复制代码

通过以上3处代码自动修改生成了strategyfilebacktest下的同名文件，就实现了量化交易的虚拟账号类来进行仿真柜台结算了，
下面这句表示，该类的父类是module_backtest.py文件里的VirtualAccount类，该类功能就是实现了一个量化回测用的仿真柜台。

class MyStrategy(module_backtest.VirtualAccount, QtCore.QThread):

复制代码

这是实盘用的策略文件代码如下：

# MA策略
import talib
import module_backtest
from vnctptdType661 import *
from PyQt5 import QtCore
# CTP行情库
from vnctpmd import *
import numpy as np
import globalType
import globalvar
# 策略参数定义 '名称，开始值，结束值，步长'
parlist = [['short', 3, 10, 1], ['long', 10, 30, 1]]
class MyStrategy(module_backtest.VirtualAccount, QtCore.QThread):
def __init__(self, period, slippoint):
super(MyStrategy, self).__init__(period, slippoint)
def OnKline(self, mddata, arg, strategyname):
if arg[0] <= 0 or arg[1] <= 0:
return
# TradingDay = klinedata.TradingDay.decode()
# klinetime = klinedata.klinetime.decode()
self.InstrumentID = mddata.InstrumentID.decode()
# self.exchange=mddata.exchange.decode()
self.close.append(float(mddata.close))
try:
float_close = [float(x) for x in self.close]
except Exception as e:
pass
self.MA_A = talib.MA(np.array(float_close), arg[0])
self.MA_B = talib.MA(np.array(float_close), arg[1])
# print('结果1:' + str(self.MA5))
# print('结果2:' + str(self.MA20))
if self.MA_A[len(self.MA_A) - 1] > self.MA_B[len(self.MA_B) - 1]:
if self.sellvol + self.sellvol_history > 0:
self.InsertOrder(self.InstrumentID, '', THOST_FTDC_D_Buy, THOST_FTDC_OF_Close, VN_OPT_LimitPrice,
mddata.close + 1, 1)
if self.buyvol + self.buyvol_history < 10:
self.InsertOrder(self.InstrumentID, '', THOST_FTDC_D_Buy, THOST_FTDC_OF_Open, VN_OPT_LimitPrice,
mddata.close + 1, 1)
elif self.MA_A[len(self.MA_A) - 1] < self.MA_B[len(self.MA_B) - 1]:
if self.buyvol + self.buyvol_history > 0:
self.InsertOrder(self.InstrumentID, '', THOST_FTDC_D_Sell, THOST_FTDC_OF_Close, VN_OPT_LimitPrice,
mddata.close - 1, 1)
if self.sellvol + self.sellvol_history < 10:
self.InsertOrder(self.InstrumentID, '', THOST_FTDC_D_Sell, THOST_FTDC_OF_Open, VN_OPT_LimitPrice,
mddata.close - 1, 1)

复制代码

转换后的回测用的策略文件代码如下：

# MA策略
import talib
import module_backtest
from vnctptdType661 import *
from PyQt5 import QtCore
# CTP行情库
from vnctpmd import *
import numpy as np
import globalType
import globalvar
# 策略参数定义 '名称，开始值，结束值，步长'
parlist = [['short', 3, 10, 1], ['long', 10, 30, 1]]
class MyStrategy(module_backtest.VirtualAccount, QtCore.QThread):
def __init__(self, period, slippoint):
super(MyStrategy, self).__init__(period, slippoint)
self.TradingDay = []
self.klinetime = []
self.open = []
self.high = []
self.low = []
self.close = []
self.volume = []
self.money = []
self.open_interest = []
self.InstrumentID = []
def InsertOrder(self, InstrumentID, exchangeid, direction, offside, VN_OPT_LimitPrice, price, vol):
self.InsertOrder_backtest(InstrumentID, exchangeid, direction, offside, VN_OPT_LimitPrice, price, vol)
def OnKline(self, reportpath1, reportpath2, mddata, arg, strategyname):
super(MyStrategy, self).OnKline(reportpath1, reportpath2, arg, mddata)
if arg[0] <= 0 or arg[1] <= 0:
return
# TradingDay = klinedata.TradingDay.decode()
# klinetime = klinedata.klinetime.decode()
self.InstrumentID = mddata.InstrumentID.decode()
# self.exchange=mddata.exchange.decode()
self.close.append(float(mddata.close))
try:
float_close = [float(x) for x in self.close]
except Exception as e:
pass
self.MA_A = talib.MA(np.array(float_close), arg[0])
self.MA_B = talib.MA(np.array(float_close), arg[1])
# print('结果1:' + str(self.MA5))
# print('结果2:' + str(self.MA20))
if self.MA_A[len(self.MA_A) - 1] > self.MA_B[len(self.MA_B) - 1]:
if self.sellvol + self.sellvol_history > 0:
self.InsertOrder(self.InstrumentID, '', THOST_FTDC_D_Buy, THOST_FTDC_OF_Close, VN_OPT_LimitPrice,
mddata.close + 1, 1)
if self.buyvol + self.buyvol_history < 10:
self.InsertOrder(self.InstrumentID, '', THOST_FTDC_D_Buy, THOST_FTDC_OF_Open, VN_OPT_LimitPrice,
mddata.close + 1, 1)
elif self.MA_A[len(self.MA_A) - 1] < self.MA_B[len(self.MA_B) - 1]:
if self.buyvol + self.buyvol_history > 0:
self.InsertOrder(self.InstrumentID, '', THOST_FTDC_D_Sell, THOST_FTDC_OF_Close, VN_OPT_LimitPrice,
mddata.close - 1, 1)
if self.sellvol + self.sellvol_history < 10:
self.InsertOrder(self.InstrumentID, '', THOST_FTDC_D_Sell, THOST_FTDC_OF_Open, VN_OPT_LimitPrice,
mddata.close - 1, 1)

复制代码