本文共 22690 字，大约阅读时间需要 75 分钟。
一般来说，交易策略的思路主要来源于两个方向：第一、实盘中的交易经验总结；第二、数据挖掘、统计分析得到的规律。当然两者也可以结合使用，例如现在流行的深度学习。
策略模板是具体交易策略的基础，一般把大部分策略都用到的方法和公共变量放到策略模板里，而具体策略继承该策略模板，进而增加个性方法和变量（如：入场价格、止损止盈）。一般我个人喜欢在最基础模板上，按照交易策略的类型衍生出交易类型模板（如：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参数配置字典来创建策略策略的初始数据状态。
回调函数
在创建实例后如有需要还可以用该方法进一步的初始化，例如加载历史数据计算策略变量状态。
策略启动方法，一般做一些启动提示
策略停止方法，如，撤单、平今转平昨，结算收盘等
当最新tick数据更新时，做一些计算，信号触发，当然也可以做撤单（一般不建议这么做）。收盘后一段时间后和开盘前一段时间前（一般15分钟）之间的这段时间会有垃圾数据（至少CTP是这样），建议登出账户或在策略或者引擎里自行处理。tick为实例，获取属性请使用对象方法，如，'tick.lastPrice'，下面order、trade、bar皆如此。
当发出单子后，就会收到单子状态的数据，无论是否成交，可以根据回报的信息来进行撤单追单等处理。
成交数据回报。主要用来计算持仓，也可以用来触发发单追单等。
一般该方法在onTick里被调用，当满足生成新K线的时候，触发该方法，进而触发基于K线的策略。
主动函数
开仓做多，默认合约为self.vtSymbol，stop为False直接发单，为True时发本地止损止赢单，默认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等方法就是对这个方法的二次封装，在这个方法里区别对待本地止损止赢单和直接发交易柜台的单子。一般用的较少。
def cancelOrder(self, vtOrderID): """撤单""" # 如果发单号为空字符串，则不进行后续操作 if not vtOrderID: return if STOPORDERPREFIX in vtOrderID: self.ctaEngine.cancelStopOrder(vtOrderID) else: self.ctaEngine.cancelOrder(vtOrderID)
根据发单编号进行撤单，该方法根据vtOrderID来判断撤本地单子还是交易所排队的单子
向数据库中插入tick数据。对ctaEngine函数的二次封装，方便使用，以下3个也是如此
向数据库中插入bar数据
根据天数读取tick数据
根据天数读取bar数据
def writeCtaLog(self, content): """记录CTA日志""" content = self.name + ':' + content self.ctaEngine.writeCtaLog(content)
对ctaEngine日志方法进行二次封装，主要加上策略的name,用于区分多策略日志
发出策略状态变化信号，主要是通知监控系统（目前是GUI，后面也可以web等）
区分引擎，满足不同情况的处理
委托类型在vnpy中默认支持的有两种：limitOrder和stopOrder。其中，利用limitOrder还可以实现市价单的效果。当然，也可以根据需要拓展交易所的市价单、FOK及FAK等指令。这里主要介绍默认支持的两种指令。
限价单，指定交易价格发单，以发单价格或者更优的价格成交，否则排队等待。vnpy默认发单为此类型。
buyPrice = 3000 buyVolume = 1 self.buy(buyPrice,buyVolume)
停止单，又名本地止损止盈单，当然也可以用来开仓。 简单理解为所有停止单发单，即保存在CTA引擎中，等最新行情到来时，符合条件者直接转化为普通发单指令发送出去，不符合者，继续保持监控。
buyPrice = 3000 buyVolume = 1 self.buy(buyPrice,buyVolume, stop= True)
只要把发单指令的stop参数设置为True,则该指令为本地停止单
首先，在发单指令前一定要获取到涨跌停板的价格
upLimit = tick.upperLimit downLimit = tick.lowerLimit
发送多头市价单
self.buy(upLimit,1)
发送空头价单
self.short(downLimit,1)
为了计算各种指标方便，这里介绍利用numpy的Array构建动态数组 先创建一个'dynamicArray.py'文件：
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
length:一般为你准备计算指标数据的长度，建议多设置20%左右
name:该数组记录的数据名称，方便区别不同的数组及数据的保存和加载
item_type：为准备记录数据的类型，具体的参考numpy的数据类型，一般建议价格用float(如果确定int更好)。
#---------------------------------------------------------------------- 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): """获取数据""" 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函数说明。