CTA策略模組策略開發 - twkk/cta_strategy GitHub Wiki
作者：明心
目錄
策略模板
一般來說，交易策略的思路主要來源於兩個方向：第一、實盤中的交易經驗總結；第二、數據挖掘、統計分析得到的規律。當然兩者也可以結合使用，例如現在流行的深度學習。
策略模板是具體交易策略的基礎，一般把大部分策略都用到的方法和公共變量放到策略模板裏，而具體策略繼承該策略模板，進而增加個性方法和變量（如：入場價格、止損止盈）。一般我個人喜歡在最基礎模板上，按照交易策略的類型衍生出交易類型模板（如：CTA、套利、對沖等），具體交易策略繼承衍生的交易類型模板進行開發。
一個常見的錯誤
在介紹具體的模板函數之前先看一個常見的錯誤，傳送門：點我
我通俗易懂的總結一下：Python中的對象分為可變對象（如dict、list等數據容器）以及不可變對象（如str、int等數據類型），在__init__之上定義的可變對象變量在實例初始化的時候會直接指向該類的同名變量，導致多個實例間共享了同一個數據容器，也就會導致各種詭異的出錯情況。
為了解決這個問題，請將所有可變變量的定義（尤其是list、dict等數據容器），放到__init__函數中（不要放在類的成員定義中）。
錯誤用法：
########################################################################
class TestStrategy(CtaTemplate):
"""CTA策略模板"""
...
# 這裏定義的變量是類成員，方便引擎在創建策略對象前了解一些信息
barList = [] # 若基於TestStrategy類創建多個策略對象，他們的barList都會指向同一個列表，導致出錯
lastPrice = 0 # 數字(int、float)在Python中屬於不可變對象，因此每個策略的lastPrice互不影響
#----------------------------------------------------------------------
def __init__(self, ctaEngine, setting):
super(TestStrategy, self).__init__(ctaEngine, setting)
正確用法：
########################################################################
class TestStrategy(CtaTemplate):
"""CTA策略模板"""
...
# 這裏定義的變量是類成員，方便引擎在創建策略對象前了解一些信息
barList = [] # 若基於TestStrategy類創建多個策略對象，他們的barList都會指向同一個列表，導致出錯
lastPrice = 0 # 數字(int、float)在Python中屬於不可變對象，因此每個策略的lastPrice互不影響
#----------------------------------------------------------------------
def __init__(self, ctaEngine, setting):
super(TestStrategy, self).__init__(ctaEngine, setting)
# 這裏定義的變量是對象創建後自身獨有的成員
self.barList = [] # 在這裏對barList重新初始化，指向一個新建的獨立列表
定義成員變量
上面已經提到了類成員變量的定義，這裏主要介紹兩個特殊的列表paramList和varList。
# 參數列表，保存了參數的名稱
paramList = ['name',
'className',
'author',
'vtSymbol']
這個變量主要記錄和保存策略參數，除了'author'外，其他3個必選，根據需要可以拓展，但必須要先定義，後方可添加到參數列表。
# 變量列表，保存了變量的名稱
varList = ['inited',
'trading',
'pos']
和前面一個變量類似，但這個是變量列表，區別主要在於，該列表中變量主要記錄策略交易過程中的一些狀態，當然你可以根據需要拓展，設置把這些變量放在參數列表裏（但建議不要這麽做）。
加載常量
from vnpy.trader.vtConstant import *
from vnpy.trader.app.ctaStrategy.ctaBase import *
ctaBase中定義了些CTA策略開發中會用的常量：交易方向、停止單狀態、數據庫的名稱、及引擎的標識等。
vtConstant 位於vn.trader目錄下，定義了一些整個VnTrader中通用的常量，具體的我就不一一列舉了，大家打開就很清楚了。
構造函數
__init__是策略對象在創建時會被首先調用的構造函數，這個函數前後都有兩個下劃線，傳入ctaEngine實例和setting參數配置字典來創建策略策略的初始數據狀態。
回調函數
- onInit
在創建實例後如有需要還可以用該方法進一步的初始化，例如加載歷史數據計算策略變量狀態。
- onStart
策略啟動方法，一般做一些啟動提示
- onStop
策略停止方法，如，撤單、平今轉平昨，結算收盤等
- onTick(self, tick)
當最新tick數據更新時，做一些計算，信號觸發，當然也可以做撤單（一般不建議這麽做）。收盤後一段時間後和開盤前一段時間前（一般15分鐘）之間的這段時間會有垃圾數據（至少CTP是這樣），建議登出賬戶或在策略或者引擎裏自行處理。tick為實例，獲取屬性請使用對象方法，如，'tick.lastPrice'，下面order、trade、bar皆如此。
- onOrder(self, order)
當發出單子後，就會收到單子狀態的數據，無論是否成交，可以根據回報的信息來進行撤單追單等處理。
- onTrade(self, trade)
成交數據回報。主要用來計算持倉，也可以用來觸發發單追單等。
- onBar(self, bar)
一般該方法在onTick裏被調用，當滿足生成新K線的時候，觸發該方法，進而觸發基於K線的策略。
主動函數
- buy(self, price, volume, stop=False)
開倉做多，默認合約為self.vtSymbol，stop為False直接發單，為True時發本地止損止贏單，默認False。
- sell(self, price, volume, stop=False)
平多
- short(self, price, volume, stop=False)
開空
- cover(self, price, volume, stop=False)
平空
- sendOrder(self, orderType, price, volume, stop=False)
def sendOrder(self, orderType, price, volume, stop=False):
"""發送委托"""
if self.trading:
# 如果stop為True，則意味著發本地停止單
if stop:
vtOrderID = self.ctaEngine.sendStopOrder(self.vtSymbol, orderType, price, volume, self)
else:
vtOrderID = self.ctaEngine.sendOrder(self.vtSymbol, orderType, price, volume, self)
return vtOrderID
else:
# 交易停止時發單返回空字符串
return ''
前面buy,sell等方法就是對這個方法的二次封裝，在這個方法裏區別對待本地止損止贏單和直接發交易櫃台的單子。一般用的較少。
- cancelOrder(self, vtOrderID)
def cancelOrder(self, vtOrderID):
"""撤單"""
# 如果發單號為空字符串，則不進行後續操作
if not vtOrderID:
return
if STOPORDERPREFIX in vtOrderID:
self.ctaEngine.cancelStopOrder(vtOrderID)
else:
self.ctaEngine.cancelOrder(vtOrderID)
根據發單編號進行撤單，該方法根據vtOrderID來判斷撤本地單子還是交易所排隊的單子
- insertTick(self, tick)
向數據庫中插入tick數據。對ctaEngine函數的二次封裝，方便使用，以下3個也是如此
- insertBar(self, bar)
向數據庫中插入bar數據
- loadTick(self, days)
根據天數讀取tick數據
- loadBar(self, days)
根據天數讀取bar數據
- writeCtaLog(self, content)
def writeCtaLog(self, content):
"""記錄CTA日志"""
content = self.name + ':' + content
self.ctaEngine.writeCtaLog(content)
對ctaEngine日志方法進行二次封裝，主要加上策略的name,用於區分多策略日志
- putEvent
發出策略狀態變化信號，主要是通知監控系統（目前是GUI，後面也可以web等）
- getEngineType
區分引擎，滿足不同情況的處理
委托類型
委托類型在vnpy中默認支持的有兩種：limitOrder和stopOrder。其中，利用limitOrder還可以實現市價單的效果。當然，也可以根據需要拓展交易所的市價單、FOK及FAK等指令。這裏主要介紹默認支持的兩種指令。
- limitOrder
限價單，指定交易價格發單，以發單價格或者更優的價格成交，否則排隊等待。vnpy默認發單為此類型。
buyPrice = 3000
buyVolume = 1
self.buy(buyPrice,buyVolume)
- stopOrder
停止單，又名本地止損止盈單，當然也可以用來開倉。 簡單理解為所有停止單發單，即保存在CTA引擎中，等最新行情到來時，符合條件者直接轉化為普通發單指令發送出去，不符合者，繼續保持監控。
buyPrice = 3000
buyVolume = 1
self.buy(buyPrice,buyVolume,stop=True)
只要把發單指令的stop參數設置為True,則該指令為本地停止單
- 利用limitOrder實現市價單效果
首先，在發單指令前一定要獲取到漲跌停板的價格
upLimit = tick.upperLimit
downLimit = tick.lowerLimit
發送多頭市價單
self.buy(upLimit,1)
發送空頭價單
self.short(downLimit,1)
時間序列
為了計算各種指標方便，這裏介紹利用numpy的Array構建動態數組 先創建一個'dynamicArray.py'文件：
- import numpy
import numpy as np
- DynamicArray
定義一個動態數組類，並初始化
########################################################################
class DynamicArray(object):
"""基於np擴展一個固定長度的動態數組"""
# ----------------------------------------------------------------------
def __init__(self, length, name = None, item_type = float):
""""""
self.name = name
self._data = np.zeros(length, dtype=item_type)
self._length = length
self._dataSize = 0
self._dtype = item_type
length:一般為你準備計算指標數據的長度，建議多設置20%左右
name:該數組記錄的數據名稱，方便區別不同的數組及數據的保存和加載
item_type：為準備記錄數據的類型，具體的參考numpy的數據類型，一般建議價格用float(如果確定int更好)。
- append(self, value)
#----------------------------------------------------------------------
def append(self, value):
"""動態添加數據"""
self._data[0:self._length - 1] = self._data[1:self._length]
# print self.name,value
self._data[-1] = value
self._dataSize += 1
self._dataSize = min(self._dataSize, self._length)
這個方法主要利用錯位覆制更新數據，並更新數據大小
- def getData(self)
#----------------------------------------------------------------------
def getData(self):
"""獲取數據"""
return self._data
這個方法看起來比較簡單，但用起來很方便，後面有使用例子。
另外，寫了兩個常用的函數方便使用，個人也可以根據自己需要拓展
#----------------------------------------------------------------------
def MA(self,m = None):
"""均價"""
if m is None or m >= self.getDataSize():
return self._data.mean()
else:
return self._data[0-m:].mean()
#----------------------------------------------------------------------
def STDDEV(self,m = None):
"""標準差"""
if m is None or m >= self.getDataSize():
return self._data.std()
else:
return self._data[0-m:].std()
其他的方法都比較簡單，我直接上完整代碼，大家斧正。
# encoding: UTF-8
import numpy as np
########################################################################
class DynamicArray(object):
"""基於np擴展一個固定長度的動態數組"""
# ----------------------------------------------------------------------
def __init__(self, length, name = None, item_type = float):
""""""
self.name = name
self._data = np.zeros(length, dtype=item_type)
self._length = length
self._dataSize = 0
self._dtype = item_type
#----------------------------------------------------------------------
def append(self, value):
"""動態添加數據"""
self._data[0:self._length - 1] = self._data[1:self._length]
# print self.name,value
self._data[-1] = value
self._dataSize += 1
self._dataSize = min(self._dataSize, self._length)
#----------------------------------------------------------------------
def update(self, value):
"""更新數據"""
self._data[-1] = value
#----------------------------------------------------------------------
def getData(self):
"""獲取數據"""
return self._data
#----------------------------------------------------------------------
def reinit(self):
"""清空數據"""
self._data = self._data*0
self._dataSize = 0
#----------------------------------------------------------------------
def getDataSize(self):
"""獲取數據的數量"""
return self._dataSize
#----------------------------------------------------------------------
def MA(self,m = None):
"""均價"""
if m is None or m >= self.getDataSize():
return self._data.mean()
else:
return self._data[0-m:].mean()
#----------------------------------------------------------------------
def STDDEV(self,m = None):
"""標準差"""
if m is None or m >= self.getDataSize():
return self._data.std()
else:
return self._data[0-m:].std()
- 舉例
策略邏輯：連續兩根Bar收陽，以最新價格價買入1手
定義數據
self.maxLength = 10
self._closeArray = DynamicArray(self.maxLength,'close')
self.close = self._closeArray.getData()
循環利用append加載本地歷史數據
dataList = [1,2,3,4,5,6]
for m in dataList:
self._closeArray.append(m)
保存到本地數據文件
直接保存self.close即可，這裏就不啰嗦了。
onBar 更新數據
self.minType = 10 #10分鐘周期bar
isInsertBar = False
close = tick.lastPrice
dateAndTime = datetime.strptime(' '.join([tick.date, tick.time]), '%Y%m%d %H:%M:%S.%f')
tickMinute = dateAndTime.minute
if tickMinute % self.minType == 0 and tickMinute!=self.tickMinute:
self.tickMinute = tickMinute
isInsertBar = True
if isInsertBar:
self._closeArray.append(close)
交易邏輯
if self.close[-1] >self.close[-2] and self.close[-2] >self.close[-3]:
self.buy(self.close[-1],1)
拓展為多周期
按照close數據的方法，新建一個數據，如
self.maxLength_2 = 15
self._closeArray_15 = DynamicArray(self.maxLength_2,'close_15')
self.close_15 = self._closeArray_15.getData()
其他方法應用和之前一樣，請參考上文。
計算時
if self.close[-1] > self.close_15[-1]:
pass
技術指標
在開發CTA策略的時候可能會用到一些經典指標，這裏簡單介紹一下talib。
- 安裝
群主寫的很好，傳送門點我
- 使用
import talib as ta
import numpy as np
data = np.array([3,5,7,9,11,8,6,4])
# 均線
ma5 = ta.MA(data,5)
# 標準差
stdTa = ta.STDDEV(data, timeperiod=5, nbdev=1)
# 返回的數據是一個numpy數組
# 舉例,onbar
if self.pos == 0:
if ma5[-1] > ma5[-2] and ma5[-2] < ma5[-3]:
self.buy(data[-1],1)
pass
具體的指標及參數請參考talib函數說明。