证券

vnpy教程

这个写的有助于理解,不错的教程
前言:

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

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

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

当然还要感谢 @用python的交易员 ,vnpy真是太棒了!

废话不多说,Let’s beginning!

从回测开始说起:

对于这么复杂的系统,从什么地方开始是一个问题,一开始比较心急,按照文件的顺序一个一个读,想一次性消化整个系统,后来发现效率很低,代码连不到一起,所以读了几个就放弃了。转而换了一个思路,在examplesCtaBacktesting文件夹下有回测引擎的具体示例文件,分别是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的路径,一般你可以在vnpytrader找到,打开VT_setting.json,可以看到里面包含了一些设置,后面会用到。

vtGlobal.py

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

__init__.py,constant.py,text.py

vnpytraderlanguage文件夹中有两个文件夹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
依据上面的内容,这个策略就这样写

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.AM

am.updateBar(bar)

if not am.inited:
return

# 计算指标数值
self.upper_band, self.lower_band = am.boll(self.boll_window, self.boll_dev)

self.count_num += 1

if self.pos == 0:

if bar.close > self.upper_band:
self.buy(bar.close+5, self.fixed_size)
self.count_num = 0
elif bar.close < self.lower_band: self.short(bar.close-5, self.fixed_size) self.count_num = 0 if self.pos > 0:
if self.count_num == self.leaving_window:
self.sell(bar.close-10,abs(self.pos))
elif self.pos < 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 第三步:编写runbacktesting 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() 第四步:结果 最后的回撤是因为设置的600根k线退出,而数据结尾不够600。 这样,整个vnpy策略编写的指南大致成型,利用上面的内容基本可以进行自己的研究了。 好好研究下

vnpy教程已关闭评论