量化投资交易 vn.py

前言：

当初接触到vnpy，一开始当然是按照该项目在GitHub上的指南，开始安装，配置，阅读Wiki，但是作为一个python新手，并不能马上利用vnpy来写策略回测甚至实盘。所以我决定还是从源码看起，一点一点摸透整个框架的细节。虽然看源代码对于一个python初学者真的很困难，特别是期间得了干眼症，看显示器那叫一个难受，但还是坚持下来。

看了一遍之后，把自己对vnpy的一些理解发上来，一来，希望和大家多交流，毕竟自己编程方面不是高手，肯定有理解的不对的地方，希望大家指正，二来再阅读一次代码，看看之前有没有遗漏疏忽的地方，另外，我确实认为vnpy是一个非常好的项目，非常适合学习和使用，但很多初学量化的人都像我一样并不是计算机科班出身，写一篇详细的使用指南可以帮助初学者节约时间，并更好的使用vnpy。

需要强调的是，整篇文章还在持续更新，会根据需要修改文章，特别是希望能与大家多交流，不管是任何问题，如有指点，希望不吝赐教。

当然还要感谢

，vnpy真是太棒了！

废话不多说，Let's beginning！

从回测开始说起：

对于这么复杂的系统，从什么地方开始是一个问题，一开始比较心急，按照文件的顺序一个一个读，想一次性消化整个系统，后来发现效率很低，代码连不到一起，所以读了几个就放弃了。转而换了一个思路，在\examples\CtaBacktesting文件夹下有回测引擎的具体示例文件，分别是loadCsv.py，runBacktesting.py和runOptimization.py，就从这三个文件一步一步来看vnpy是如何进行回测的。

图示可以清楚看清loadCsv.py文件导入了哪些模块（忽略系统模块和一些第三方模块）

我们来一个一个看

vtFunction.py

这里面包括了5个开发中常用的函数，

safeUnicode()

todayDate()

loadIconPath()

getTempPath()

getJsonPath()

其中vtGlobal.py导入的是getJsonPath()方法，作用是获取JSON配置文件的路径，就vtGlobal.py而言，它获取的是VT_setting.json的路径，一般你可以在\vnpy\trader找到，打开VT_setting.json，可以看到里面包含了一些设置，后面会用到。

vtGlobal.py

该文件就是将VT_setting.json里面的配置变成python可读取和使用的字典形式，并赋值给globalSetting，将它作为全局配置的字典。

__init__.py，constant.py，text.py

\vnpy\trader\language文件夹中有两个文件夹chinese和english，以及__init__.py文件，__init__.py默认设置为chinese，假如你想使用english，就可以在VT_setting.json里面修改。constant.py包含了近百个常量定义，仔细看可以把它们都归类成交易相关的常量，后面会经常用到。而text.py也定义了很多常量，可以把这些归类为显示相关的常量。

vtConstant.py

从constant.py导入了常量，并把它们添加到vtConstant.py的局部字典中。

ctaBase.py

定义了很多常量以及一个StopOrder类，定义的常量里面就包含了loadCsv.py里面导入的MINUTE_DB_NAME = 'VnTrader_1Min_Db'，后面在数据库导入数据的时候会碰到。StopOrder类定义了一个本地停止单。

vndatayes.py

里面定义了一个DatayesApi类，用于从通联数据下载数据。

vtObject.py

定义了几种数据类，后面会经常用到。

ctaHistoryData.py

定义了CTA模块用的历史数据引擎，从里面定义的方法可以看出，主要是下载历史数据和将csv文件导入数据库

__init__()方法用到了vtGlobal.py里面导入的globalSetting，默认是localhost，创建了本地的数据库链接。另外一个就是通联数据下载的api，需要传入token参数。

暂且只关注loadMcCsv()方法，需要传入三个参数，filename就是历史数据文件名，dbName与ctaBase.py里面定义的常量有关，以IF0000_1min.csv为例，里面保存的是1分钟bar数据，就传入MINUTE_DB_NAME，同理tick数据就传入TICK_DB_NAME，日线数据就传入DAILY_DB_NAME。symbol就是标的的代码，例如IF0000。中间的代码按照csv文件保存数据的格式，把数据存入数据库。

loadCsv.py

所以整个代码完成的就是将csv历史数据的导入数据库。

图示可以清楚看清runBacktesting.py文件导入了哪些模块（忽略系统模块和一些第三方模块）

eventEngine.py

里面定义了三个类，EventEngine，EventEngine2，Event，以及一个测试函数。EventEngine，EventEngine2两个类的代码内容差不多，我们只看EventEngine

EventEngine定义了事件驱动引擎，理解这个引擎是理解vnpy工作原理的重要一步。关于导入的Queue模块和threading模块，可以百度一下它们的用法

__init__()方法：

self.__queue = Queue() 实例化事件队列

self.__active = False 事件引擎开关，默认为False

self.__thread = Thread(target = self.__run) 创建Thread类的实例，传给它一个函数，当线程启动，该函数运行

self.__timer = QTimer() 计时器，用于触发计时器事件

self.__timer.timeout.connect(self.__onTimer) 将timeout信号和self.__onTimer方法绑定，当触发timeout信号，self.__onTimer方法运行

self.__handlers = defaultdict(list) 这里的__handlers是一个字典，用来保存对应的事件调用关系其中每个键对应的值是一个列表，列表中保存了对该事件进行监听的函数功能

self.__generalHandlers __generalHandlers是一个列表，用来保存通用回调函数（所有事件均调用）

下面是类中定义的方法，我们不以定义的顺序来看，而是按照事件的传递顺序来看。

start(self, timer=True):

引擎启动，timer表示是否要启动计时器，默认为True。

self.__active = True 将引擎设为启动

self.__thread.start() 启动事件处理线程

self.__timer.start(1000) 启动计时器，计时器事件间隔默认设定为1秒,start()时间参数的单位是毫秒，意思是1000毫秒后触发timeout，而timeout与self.__onTimer绑定，故self.__onTimer被调用

当start()方法执行，事件处理线程和计时器同时启动

事件处理线程self.__thread启动，而self.__thread = Thread(target = self.__run)，也就是说，__run()方法执行

__run(self):

self.__active在start()方法中已经设置为True

event = self.__queue.get(block = True, timeout = 1) 获取事件的阻塞时间设为1秒，调用队列对象的get()方法从队头删除并返回一个项目。可选参数为block，默认为True。如果队列为空且block为True，get()就使调用线程暂停，直至有项目可用。如果队列为空且block为False，队列将引发Empty异常。

self.__process(event) 假设队列里面有项目，则执行__process()方法

__process(self, event):

事件处理方法，优先检查是否存在对该事件进行监听的处理函数，然后调用通用处理函数进行处理

计时器启动，self.__timer.timeout.connect(self.__onTimer)触发执行self.__onTimer

__onTimer(self):

创建计时器事件，调用put()方法向队列中存入计时器事件

put(self, event):

self.__queue.put(event) 调用队列对象的put()方法在队尾插入一个项目。

从上面可以看出，整个事件传递的过程是这样的：

调用start()方法，事件处理线程和计时器同时启动，计时器每隔一秒调用__onTimer()方法，创建计时器事件，调用put()方法在队尾插入一个事件，事件处理线程每隔一秒获取事件，若存在事件调用__process()方法，对事件进行处理。

stop(self):

停止引擎，事件处理线程和计时器

剩下的四个方法用于注册注销事件和通用事件处理函数监听

可以用test()函数自己验证一下

eventType.py

本文件仅用于存放对于事件类型常量的定义

vtEvent.py

基于vnpy.event.eventType，并添加更多字段

vtGateway.py

定义了VtGateway类作为交易接口，类方法都是关于事件的推送。

注意到一个细节，以onTick(self, tick)为例，参数tick是不是传入的是类VtTickData的实例，因为后面ctaBacktesting.py里面from vnpy.trader.vtGateway import VtOrderData, VtTradeData，而不是fromvtObject.py import VtOrderData, VtTradeData

ctaTemplate.py

写策略至关重要的部分，里面包含4个类，CtaTemplate，TargetPosTemplate，BarManager，ArrayManager，后面的例子没有用到TargetPosTemplate，我们暂时只看其他三个类

CtaTemplate

CTA策略模板，开发策略时需要继承CtaTemplate类

__init__():

初始化使用的ctaEngine，比如用回测引擎，可以在回测引擎类方法initStrategy中，有self.strategy = strategyClass(self, setting)，传入的self参数代表BacktestingEngine（原来还可以这么传参数，学到了）

setting是设置策略的参数，示例是空字典。

由于CtaTemplate是用来继承，方法的具体应用将在后面用具体的策略说明。

BarManager

K线合成器

updateTick(self, tick)：

用于将tick数据合成1分钟bar。

updateBar

用于将1分钟bar数据合成x分钟bar。

ArrayManager

K线序列管理工具，负责：1. K线时间序列的维护 2. 常用技术指标的计算

strategyKingKeltner.py

以具体的策略为例，看看如何使用上面的模板

首先设置策略的参数和变量，并把它们添加进列表

__init__()：

类KkStrategy是继承自CtaTemplate的子类，所以初始化先调用CtaTemplate的__init__()方法。

按照策略是否需要，创建BarManager和ArrayManager的实例（因为我看到有的策略并没有调用BarManager和ArrayManager的类方法，而是根据策略自己写了另外的k线处理方法）

以KkStrategy为例：

self.bm = BarManager(self.onBar, 5, self.onFiveBar)，从传入额参数可以看出这是一个基于5分钟k线的策略，第一个参数是1分钟k线回调函数，最后一个是5分钟回调函数。

onInit():

初始化策略

writeCtaLog是继承自CtaTemplate的方法，在CtaTemplate中能看到，该方法再次调用ctaEngine的writeCtaLog方法，用于记录日志。

initDays代表初始化需要的天数，本例中为十天，那么initData就是保存的十天的1分钟k线数据。然后调用onBar方法处理1分钟k线数据

回测中可以忽略putEvent()方法

onBar()：

调用updateBar()方法，如果策略用的是1分钟k线数据，那么这个函数就是用于实现整个策略的主体部分。

从updateBar()方法可以看出，首先更新推送进来的数据，合成5分钟k线，若当前时间能否被5整除，则调用onXminBar方法，本例就是onFiveBar

onFiveBar():

本例策略的思想就在这里实现。首先要撤销之前发出的尚未成交的委托，再来就是保证指标可以计算，当inited为True时，表示当Array里面缓存的数据长度大于等于规定的size，也就是说可以计算相关指标了。然后计算指标数值。下面的代码都是开仓平仓的条件判断，就不详细说明了。

sendOcoOrder()：

自定义的委托函数，用于突破时入场

onTrade():

用于成交后撤销委托

onStop():

停止策略

onTick():

处理tick数据，本例中没有用到，所以不调用

ctaBacktesting.py

里面定义了四个类BacktestingEngine，TradingResult，DailyResult，OptimizationSetting

BacktestingEngine

定义了回测引擎类，使用的策略代码和实盘一样

__init__(self):

需要设置回测的初始化参数都在里面，一般来说需要设置的有

self.strategy = None 回测的策略

self.mode = self.BAR_MODE 回测的模式，默认为bar

self.startDate = '' 回测起始时间，默认为空

self.initDays = 0 回测需要初始化的数据天数，即前面用于预先载入的历史数据的天数

self.endDate = '' 回测结束时间，默认为空

self.capital = 1000000 初始化本金，默认为100W

self.slippage = 0 回测的滑点，默认为0

self.rate = 0 回测的佣金比率，默认为0

self.size = 1 合约大小，默认为1

self.priceTick = 0 价格最小变动，默认为0

self.dbName = '' 回测的数据库名

self.symbol = '' 回测的标的名

self.dataStartDate = None 格式化后的回测起始时间

self.dataEndDate = None 格式化后的回测结束时间

self.strategyStartDate = None 策略开始时间，即回测开始时间加上初始化数据的天数

（跳过通用功能）

根据需要，调用下面的类方法设置参数

setStartDate(self, startDate='20100416', initDays=10):

用于设置策略的开始时间。

setEndDate(self, endDate=''):

用于设置策略的结束时间。

setBacktestingMode(self, mode):

设置回测模式，有tick和bar可选

setDatabase(self, dbName, symbol):

设置用到额数据库以及标的名称

setCapital(self, capital):

设置本金

setSlippage(self, slippage):

设置滑点

setSize(self, size):

设置合约大小

setRate(self, rate):

设置佣金比率

setPriceTick(self, priceTick):

设置最小价格变动

initStrategy(self, strategyClass, setting=None):

设置回测的策略

以上就是回测开始前的准备工作，下面就是如何利用历史数据进行回测

loadHistoryData(self):

用于载入历史数据，代码主要涉及pymongo的使用，可自行百度

crossLimitOrder(self):

基于最新数据撮合限价单

crossStopOrder(self):

基于最新数据撮合停止单

上面两个用于撮合成交的类方法代码逻辑类似，源代码的解释很详细，用文字解释反而麻烦多余。

sendOrder，cancelOrder，sendStopOrder，cancelStopOrder，cancelAll

都是策略接口，用于处理订单

newBar(self, bar):

传入bar数据，首先撮合订单，然后调用策略的onBar()方法处理数据，并更新每日收盘价

newTick(self, tick):

与上面类似

runBacktesting(self):

运行回测，逻辑很清晰，载入数据，选择数据类，初始化策略，启动策略，回放数据，结束。

后面的类方法都是依据回测中发生的交易计算结果，不在赘述。

到这里，整个回测的框架就很清楚了，现在根据runBacktesting.py，看看如何运用上面的框架来回测。

runBacktesting.py

现在是要回测策略strategyKingKeltner在IF0000的历史数据上的表现，之前已经通过loadCsv.py把数据导入了数据库。

首先from vnpy.trader.app.ctaStrategy.ctaBacktesting import BacktestingEngine, MINUTE_DB_NAME，用来创建BacktestingEngine的实例，以及连接刚才导入的数据库中的数据

from vnpy.trader.app.ctaStrategy.strategy.strategyKingKeltner import KkStrategy 导入策略

engine = BacktestingEngine()创建回测引擎，然后下一步通过里面的类方法设置你需要的初始化参数，本例中，回测模式为bar模式，然后设置开始时间，滑点等等，接着调用initStrategy方法，在引擎中建立策略的实例。

开始回测，要想了解回测过程中的具体细节，最好的方法是利用pycharm在每个运行到的地方设置断点，一步一步的看，走完整个过程（本来想用文字描述，感觉效率太低，还是请读者自己运行一遍）。

回测结束，看看结果吧。

从策略编写说起

其实到这里已经可以根据前面的内容写策略了，下面就举一个简单的例子。

交易螺纹钢，初始资金1W，只交易一手，最多持仓一手，策略是利用布林通道，上穿买入，下穿卖出，600分钟定时退出，1分钟k线。

第一步：导入数据

vnpy给的示例已经导入了rb0000。

第二步：编写策略

可以模仿vnpy给的示例策略，大致可以摸索出一个策略模板，代码添加了更详细的注释

from __future__ import division

from vnpy.trader.vtObject import VtBarData
from vnpy.trader.vtConstant import EMPTY_STRING
from vnpy.trader.app.ctaStrategy.ctaTemplate import (CtaTemplate,
BarManager,
ArrayManager)
#可以导入自己需要的包

class strategyname(CtaTemplate): #strategyname改成自己命名的策略名称，下面的strategyname同样替换

className = 'strategyname'
author = '' #随意输入# 策略参数，添加需要的参数# 策略变量，添加需要的变量# 参数列表
paramList = ['name','className','author','vtSymbol',]# 变量列表
varList = ['inited','trading','pos']# 列表中已有的都是继承自CtaTemplate
#----------------------------------------------------------------------
def __init__(self, ctaEngine, setting):"""Constructor"""super(strategyname, self).__init__(ctaEngine, setting)  #必须要有的语句self.bm = BarManager(self.onBar, xmin=0, onXminBar=None)        # 创建K线合成器对象,后面两个参数根据需要传入SELF.AM = ArrayManager()# 如果里面的指标不够用需要自己添加# 如果是多合约实例的话，变量需要放在__init__里面，可以参考github的说明#----------------------------------------------------------------------
def onInit(self):  #这里的代码不用更改，直接使用即可"""初始化策略（必须由用户继承实现）"""self.writeCtaLog(u'%s策略初始化' %self.name)# 载入历史数据，并采用回放计算的方式初始化策略数值initData = self.loadBar(self.initDays)for bar in initData:self.onBar(bar)self.putEvent()#----------------------------------------------------------------------
def onStart(self):  #这里的代码不用更改，直接使用即可"""启动策略（必须由用户继承实现）"""self.writeCtaLog(u'%s策略启动' %self.name)self.putEvent()#----------------------------------------------------------------------
def onStop(self):  #这里的代码不用更改，直接使用即可"""停止策略（必须由用户继承实现）"""self.writeCtaLog(u'%s策略停止' %self.name)self.putEvent()#----------------------------------------------------------------------
def onTick(self, tick):  # 如果是tick策略，则策略主体在这里，若不是，实盘时利用下面的类方法合成k线"""收到行情TICK推送（必须由用户继承实现）""" self.bm.updateTick(tick)#----------------------------------------------------------------------
def onBar(self, bar):  # 如果是1分钟k线策略，则策略主体在这里pass#----------------------------------------------------------------------
def onXminbar（self, bar):  # 如果是X分钟k线策略，则策略主体在这里pass#----------------------------------------------------------------------
def onOrder(self, order):"""收到委托变化推送（必须由用户继承实现）"""pass#----------------------------------------------------------------------
def onTrade(self, trade):# 发出状态更新事件self.putEvent()#----------------------------------------------------------------------
def onStopOrder(self, so):"""停止单推送"""pass#----------------------------------------------------------------------
def customized_function(self, *args): #  定制自己的类方法，例如strategyKingKeltner.py中定义的sendOcoOrder()pass</code></pre></div><p>依据上面的内容，这个策略就这样写</p><div class="highlight"><pre><code class="language-text">from __future__ import division

from vnpy.trader.vtObject import VtBarData
from vnpy.trader.vtConstant import EMPTY_STRING
from vnpy.trader.app.ctaStrategy.ctaTemplate import (CtaTemplate,
BarManager,
ArrayManager)

class bollinger(CtaTemplate): #strategyname改成自己命名的策略名称，下面的strategyname同样替换

className = 'bollinger'
author = u'尔鸫' #随意输入# 策略参数，添加需要的参数
boll_window = 600                     # 布林通道窗口数 
boll_dev = 2                          # 布林通道的偏差
leaving_window = 600                  # 定时离开的窗口数
init_days = 10                        # 初始化数据所用的天数
fixed_size = 1                        # 每次交易的数量# 策略变量，添加需要的变量
upper_band = 0                        # 布林通道上轨
lower_band = 0                        # 布林通道下轨
count_num = 0                         # 用于记录成交的k线与当前推送的bar距离多少# 参数列表
paramList = ['name','className','author','vtSymbol','boll_window','boll_dev','leaving_window','init_days','fixed_size']# 变量列表
varList = ['inited','trading','pos','upper_band','lower_band','count_num']#----------------------------------------------------------------------
def __init__(self, ctaEngine, setting):"""Constructor"""super(bollinger, self).__init__(ctaEngine, setting)  self.bm = BarManager(self.onBar, xmin=0, onXminBar=None)        SELF.AM = ArrayManager(1000)#----------------------------------------------------------------------
def onInit(self):  """初始化策略（必须由用户继承实现）"""self.writeCtaLog(u'%s策略初始化' %self.name)# 载入历史数据，并采用回放计算的方式初始化策略数值initData = self.loadBar(self.init_days)for bar in initData:self.onBar(bar)self.putEvent()#----------------------------------------------------------------------
def onStart(self):  #这里的代码不用更改，直接使用即可"""启动策略（必须由用户继承实现）"""self.writeCtaLog(u'%s策略启动' %self.name)self.putEvent()#----------------------------------------------------------------------
def onStop(self):  #这里的代码不用更改，直接使用即可"""停止策略（必须由用户继承实现）"""self.writeCtaLog(u'%s策略停止' %self.name)self.putEvent()#----------------------------------------------------------------------
def onTick(self, tick):  # 如果是tick策略，则策略主体在这里，若不是，利用下面的类方法合成k线"""收到行情TICK推送（必须由用户继承实现）""" self.bm.updateTick(tick)#----------------------------------------------------------------------
def onBar(self, bar):  # 如果是1分钟k线策略，则策略主体在这里# 全撤之前发出的委托self.cancelAll()# 保存K线数据am = SELF.AMam.updateBar(bar)if not am.inited:return# 计算指标数值self.upper_band, self.lower_band = am.boll(self.boll_window, self.boll_dev)self.count_num += 1if self.pos == 0:if bar.close &gt; self.upper_band:self.buy(bar.close+5, self.fixed_size)self.count_num = 0elif bar.close &lt; self.lower_band:self.short(bar.close-5, self.fixed_size)self.count_num = 0if self.pos &gt; 0:if self.count_num == self.leaving_window:self.sell(bar.close-10,abs(self.pos))elif self.pos &lt; 0:if self.count_num == self.leaving_window:self.cover(bar.close+10,abs(self.pos))self.putEvent()#----------------------------------------------------------------------
def onXminbar(self, bar):  # 如果是X分钟k线策略，则策略主体在这里pass#----------------------------------------------------------------------
def onOrder(self, order):"""收到委托变化推送（必须由用户继承实现）"""pass#----------------------------------------------------------------------
def onTrade(self, trade):# 发出状态更新事件self.putEvent()#----------------------------------------------------------------------
def onStopOrder(self, so):"""停止单推送"""pass#----------------------------------------------------------------------
def customized_function(self, *args): #  定制自己的类方法，例如strategyKingKeltner.py中定义的sendOcoOrder()pass</code></pre></div><p>第三步：编写runbacktesting</p><div class="highlight"><pre><code class="language-text">from __future__ import division

from vnpy.trader.app.ctaStrategy.ctaBacktesting import BacktestingEngine, MINUTE_DB_NAME

if name == ‘main’:
from bollinger import bollinger #

# 创建回测引擎
engine = BacktestingEngine()# 设置引擎的回测模式为K线
engine.setBacktestingMode(engine.BAR_MODE)# 设置回测用的数据起始日期
engine.setStartDate('20110104')# 设置产品相关参数
engine.setCapital(10000)
engine.setSize(10)
engine.setSlippage(1)     # 股指1跳
engine.setRate(3/10000)   # 万0.3 
engine.setPriceTick(1)    # 股指最小价格变动# 设置使用的历史数据库
engine.setDatabase(MINUTE_DB_NAME, 'rb0000') #[IF0000, rb0000]# 在引擎中创建策略对象
d = {}
engine.initStrategy(bollinger, d)# 开始跑回测
engine.runBacktesting()# 显示回测结果
engine.showBacktestingResult()</code></pre></div><p>第四步：结果</p><figure data-size="normal"><noscript>&lt;img src="https://pic1.zhimg.com/v2-435737a755dd9cde69b64bd9523b2c88_b.jpg" data-caption="" data-size="normal" data-rawwidth="1000" data-rawheight="1246" class="origin_image zh-lightbox-thumb" width="1000" data-original="https://pic1.zhimg.com/v2-435737a755dd9cde69b64bd9523b2c88_r.jpg"/&gt;</noscript><img src="https://pic1.zhimg.com/80/v2-435737a755dd9cde69b64bd9523b2c88_1440w.jpg" data-caption="" data-size="normal" data-rawwidth="1000" data-rawheight="1246" class="origin_image zh-lightbox-thumb lazy" width="1000" data-original="https://pic1.zhimg.com/v2-435737a755dd9cde69b64bd9523b2c88_r.jpg" data-actualsrc="https://pic1.zhimg.com/v2-435737a755dd9cde69b64bd9523b2c88_b.jpg" data-lazy-status="ok"></figure><p>最后的回撤是因为设置的600根k线退出，而数据结尾不够600。</p><p class="ztext-empty-paragraph"><br></p><p>这样，整个vnpy策略编写的指南大致成型，利用上面的内容基本可以进行自己的研究了。</p><p class="ztext-empty-paragraph"><br></p><p>当然，这篇文章还远没有结束，仍有地方未探索，我会继续把想法更新上来。</p><p class="ztext-empty-paragraph"><br></p><p>未完待续...</p></div>