量化投资交易 vn.py
前言：
當初接觸到vnpy，一開始當然是按照該項目在GitHub上的指南，開始安裝，配置，閱讀Wiki，但是作為一個python新手，并不能馬上利用vnpy來寫策略回測甚至實盤。所以我決定還是從源碼看起，一點一點摸透整個框架的細節。雖然看源代碼對于一個python初學者真的很困難，特別是期間得了干眼癥，看顯示器那叫一個難受，但還是堅持下來。
看了一遍之后，把自己對vnpy的一些理解發上來，一來，希望和大家多交流，畢竟自己編程方面不是高手，肯定有理解的不對的地方，希望大家指正，二來再閱讀一次代碼，看看之前有沒有遺漏疏忽的地方，另外，我確實認為vnpy是一個非常好的項目，非常適合學習和使用，但很多初學量化的人都像我一樣并不是計算機科班出身，寫一篇詳細的使用指南可以幫助初學者節約時間，并更好的使用vnpy。
需要強調的是，整篇文章還在持續更新，會根據需要修改文章，特別是希望能與大家多交流，不管是任何問題，如有指點，希望不吝賜教。
當然還要感謝
@用python的交易員 ，vnpy真是太棒了！廢話不多說，Let's beginning！
從回測開始說起：
對于這么復雜的系統，從什么地方開始是一個問題，一開始比較心急，按照文件的順序一個一個讀，想一次性消化整個系統，后來發現效率很低，代碼連不到一起，所以讀了幾個就放棄了。轉而換了一個思路，在\examples\CtaBacktesting文件夾下有回測引擎的具體示例文件，分別是loadCsv.py，runBacktesting.py和runOptimization.py，就從這三個文件一步一步來看vnpy是如何進行回測的。
圖示可以清楚看清loadCsv.py文件導入了哪些模塊（忽略系統模塊和一些第三方模塊）
<img src="https://pic4.zhimg.com/v2-c318403bd3931057bf959af0b2185dd7_b.jpg" data-caption="" data-size="normal" data-rawwidth="1763" data-rawheight="564" class="origin_image zh-lightbox-thumb" width="1763" data-original="https://pic4.zhimg.com/v2-c318403bd3931057bf959af0b2185dd7_r.jpg"/>我們來一個一個看
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文件導入了哪些模塊（忽略系統模塊和一些第三方模塊）
<img src="https://pic1.zhimg.com/v2-92d15cddd874f0aa5ab95c7717b5f560_b.jpg" data-caption="" data-size="normal" data-rawwidth="1817" data-rawheight="771" class="origin_image zh-lightbox-thumb" width="1817" data-original="https://pic1.zhimg.com/v2-92d15cddd874f0aa5ab95c7717b5f560_r.jpg"/>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 divisionfrom 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 divisionfrom 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 > self.upper_band:self.buy(bar.close+5, self.fixed_size)self.count_num = 0elif bar.close < self.lower_band:self.short(bar.close-5, self.fixed_size)self.count_num = 0if 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</code></pre></div><p>第三步：編寫runbacktesting</p><div class="highlight"><pre><code class="language-text">from __future__ import divisionfrom vnpy.trader.app.ctaStrategy.ctaBacktesting import BacktestingEngine, MINUTE_DB_NAME
if name == ‘main’:
from bollinger import bollinger #
總結
以上是生活随笔為你收集整理的量化投资交易 vn.py的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： ImportError: libSM.s
- 下一篇： lambda函数+map函数的结合使用