策略进阶
本篇文章未重写，代码仅供参考。
熟悉策略基础后我们再来提供一个市场上普遍编写的一些基于技术指标的程序化交易策略。下面的策略是基于肯特纳通道的技术指标（类似于布林带）。整体的策略逻辑是当某一合约K线周期的收盘价大于开盘价，并且收盘价大于该技术指标的上轨就买入开仓，同样的如果收盘价小于开盘价，并且收盘价小于该技术指标的下轨就开仓卖出；那么有人也可能想做反向的逻辑，所以我们需要通过一个参数来控制反向逻辑的情况。
-
首先我们要导入策略中要用到的 Python 模块，比如策略模板，相关的库。
Demo_KC.pyimport datetime from ctaBase import * from ctaTemplate import * -
然后我们需要定义自己的策略类（类名和文件名必须保持一致），为了方便我们通常需要继承策略模板。
Demo_KC.pyclass KC(CtaTemplate): """肯特纳通道指标策略""" -
然后我们需要定义策略的参数和状态变量，这次我们考虑的多一点，要实现切换不同日内分钟周期的K线，限制交易次数，提供两种对手价的报单方式和交易时间段。
Demo_KC.py# 参数映射表 paramMap = { 'techtype': '技术指标', 'period': 'K线周期', 'N': '指标参数', 'tradetype': '交易类型', 'pricetype': '价格优化', 'volume': '数量', 'istime': '是否选择时间段', 'startdt': '开始时间', 'enddt': '结束时间', 'istlimit': '是否限制次数', 'tlimit': '次数限制', 'exchange': '交易所', 'vtSymbol': '合约代码', 'investor': '投资者账号' } paramList = list(paramMap.keys()) # 参数列表，保存了参数的名称 # 变量映射表 varMap = { 'trading': '交易中', 'excTimes': '执行次数', 'bup': '肯特纳上轨', 'bdn': '肯特纳下轨' } varList = list(varMap.keys()) # 变量列表，保存了变量的名称 techtypes = ['keltner'] periods = [1, 5, 15, 30, 60, 90, 120] # 定义分钟周期 tradetypes = ['B', 'S'] # 定义两种逻辑分类 pricetypes = ['D1', 'D2'] # 定义两种对手价报单价格优化方式 -
上述参数是在对策略有整体认知和设计的基础上编写的，其实真正编写策略过程中，参数的定义是不断根据自己的想法来完善的。完成上述代码编写后，我们进行初始化操作。
Demo_KC.pydef __init__(self, ctaEngine=None, setting={}): """Constructor""" self.widgetClass = KLWidget self.widget = None # 唤起定义好的K线页面 # 策略默认参数 self.techtype = 'keltner' # 该参数只是展示用，不参与逻辑 self.period = 1 # 初始的K线周期，为 1 分钟，也可以为空，在客户端填写。 self.volume = 1 # 初始的报单手数，为 1 手，也可以为空，在客户端填写。 self.tradetype = 'B' # 初始交易逻辑模式，为 B，也可以为空，在客户端填写。 self.pricetype = 'D1' # 初始的价格优化模式，为 D1 手，也可以为空，在客户端填写。 self.N = 26 # 初始的技术指标参数周期。 self.istime = True # 是否选择交易时间段 self.startdt = '09:00:00' self.enddt = '15:00:00' # 交易时间段时间 self.istlimit = False # 是否限制交易次数 self.tlimit = 10 # 限制交易次数数量10 self.investor = '119016' # 填写的时候一定要加 ''，来确保该字段为 string，如 '119016' self.upPrice = 0 # 涨停价 self.lowPrice = 0 # 跌停价 self.askPrice1 = 0 # 卖盘价1 self.bidPrice1 = 0 # 买盘价1 self.bup = 0 self.bdn = 0 # 策略状态变量 self.refClose = 0 self.excTimes = 0 self.buySig = False self.sellSig = False # 策略信号参数 self.tick = None super().__init__(ctaEngine, setting) self.bm = BarManager(self.onBar, self.period, self.onBarX) # K线合成 self.am = ArrayManager() # K线数据缓存 # 启动界面相关参数 self.cost = 0 self.signal = 0 # 买卖标志 self.mainSigs = ['bup', 'bdn'] # 主图显示 self.subSigs = [] # 副图显示 self.getGui() # 唤起页面 -
接下来我们订阅合约获取合约的行情，由于需要历史分钟数据，所以我们还要加载历史分钟数据。下面分成 2 块来处理，一块是接收实时行情代码，另一块是处理合成K线数据相关代码。
Demo_KC.pydef onTick(self, tick): """收到行情 TICK 推送""" # 过滤涨跌停和集合竞价 super().onTick(tick) if (tick.lastPrice == 0) or (tick.askPrice1 == 0) or (tick.bidPrice1 == 0): return self.tick = tick # 将收到的 tick 赋值给全局变量 self.bm.updateTick(tick) # 更新 def onStart(self): """启动策略（必须由用户继承实现）""" self.bm = BarManager(self.onBar, self.period, self.onBarX) self.am = ArrayManager() if str(self.period).isdigit(): self.loadBar(30) elif self.period == 'D': # 这里暂不实现日线级别 self.loadDay(1) # self.output(self.loadDay(2)) self.sTime = datetime.datetime.strptime(self.startdt, '%H:%M:%S').time() # 获取策略启动时间 self.eTime = datetime.datetime.strptime(self.enddt, '%H:%M:%S').time() # 获取策略结束时间 self.excTimes = 0 self.signal = 0 self.cost = 0 self.refClose = 0 self.buySig = False self.sellSig = False self.tick = None super().onStart() # 查询持仓信息 pos_list = self.getInvestorPosition(self.investor) for pos in pos_list: if pos['InstrumentID'] == self.vtSymbol: self.pos[pos['InstrumentID']] = pos['Position'] self.tpos0L[pos['InstrumentID']] = pos['Position'] - pos['YdPositionClose'] self.getGui() # 唤起页面 def onBar(self, bar): self.bm.updateBar(bar) # 更新 1 分钟 bar 数据 if self.tradeDate != bar.date: self.tradeDate = bar.date # ---------------------------------------------------------------------- def onBarX(self, bar): self.bar = bar # 接收到 bar 数据推送，赋值给全局变量 # 记录数据 if not self.am.updateBar(bar): return self.getCtaIndictor(bar) # 计算指标 self.getCtaSignal(bar) # 计算信号 self.execSignal(self.volume) # 简易信号执行 self.refclose = bar.close # 前收盘 if self.widget and self.bar: data = { 'bar': self.bar, 'sig': self.signal, 'bup': self.bup, 'bdn': self.bdn } # 发出状态更新事件 self.widget.addBar(data) # 页面上绘制新k线的 bar if self.trading: self.putEvent() -
我们再实现一下 onStop 回调
Demo_KC.pydef onStop(self): """停止策略（必须由用户继承实现）""" # 订单和成本管理 super().onStop() -
处理完数据和页面相关的代码，接下来我们要编写策略的主要核心：计算技术指标数据，确定交易信号，执行交易信号三大块。首先我们先计算技术指标数据，因为是收到K线 bar 后运行指标的计算，所以我们要定义一个技术指标计算的方法，在这个方法里面做一些处理。
Demo_KC.pydef getCtaIndictor(self, bar): """计算指标数据""" # 使用默认参数计算指标数值 self.bup, self.bdn = self.am.keltner(self.N, 2) # keltner 的计算方法已经在策略模板中的 ArrayManager 中已经实现，所以我们直接调用。一些常见的技术指标在 talib 库中有计算，我们封装好了一些指标；您也可以参考里面技术指标实现的方式自己开发。 self.bup, self.bdn = round(self.bup, 2), round(self.bdn, 2) # 对指标的数值进行四舍五入 -
计算交易信号。
Demo_KC.pydef getCtaSignal(self, bar): """计算交易信号""" self.endOfDay = False # 定义尾盘，尾盘不交易并且空仓 if self.techtype == 'keltner': # 判断技术指标是否是 keltner self.getBBANDS() # 计算技术指标交易信号 # 计算交易价格 self.longPrice = bar.close # K线 bar 的收盘价 self.shortPrice = bar.close # 判断交易报单模式 if self.tick and (self.pricetype in self.pricetypes): if self.pricetype == 'D1': # D1 模式采用对手价1 self.longPrice = self.tick.askPrice1 self.shortPrice = self.tick.bidPrice1 elif self.pricetype == 'D2': # D2 模式采用对手价2 self.longPrice = self.tick.askPrice2 self.shortPrice = self.tick.bidPrice2 else: try: self.longPrice = float(self.pricetype) self.shortPrice = float(self.pricetype) except: pass def getBBANDS(self): if self.tradetype == 'B': # 交易模式如果是B则采用正向交易逻辑 self.buySig = (self.bar.close > self.bar.open) and (self.bar.close > self.bup) self.sellSig = (self.bar.close < self.bar.open) and (self.bar.close < self.bdn) else: # 交易模式如果不是 B 则采用逆向交易逻辑 self.buySig = (self.bar.close < self.bar.open) and (self.bar.close < self.bdn) self.sellSig = (self.bar.close > self.bar.open) and (self.bar.close > self.bup) self.shortSig = False # 平仓信号 self.coverSig = False -
交易信号执行，在执行交易信号的时候，我们做了很多判断，比如判断是否行情正常，报单次数是否超过限制，是否处于交易时间段，以及持有仓位时的情况。这些都是常规的判断，如果有其他特殊需求，可以自己按照需求编写。其次在平仓的是要注意，目前 PythonGO 没有封装全平的函数，您需要根据交易所的特点（优先平今或者优先平昨）采用
sendOrder方法去设计全平的函数。Demo_KC.pydef execSignal(self, volume): """简易交易信号执行""" if self.trading and self.tick is None: # 没接到行情判断处理 self.output('行情未初始化') return elif self.istlimit and self.excTimes > self.tLimit: # 报单次数超过限制处理 self.output('交易次数超限') return elif self.istime and self.tick and \ not (self.sTime < self.tick.datetime.time() < self.eTime): # 时间不满足设置的交易时间段 self.output('非交易时间段') return self.signal = 0 pos = self.pos.get(self.vtSymbol, 0) if not self.orderID is None: # 挂单未成交 self.cancelOrder(self.orderID) # 挂单未成交进行撤单处理 if pos > 0 and self.sellSig: # 持有多头仓位 触发平仓信号 self.signal = -self.shortPrice self.orderID = self.sell_y(self.shortPrice, pos) if self.trading: self.excTimes += 1 # 增加报单次数 return elif pos < 0 and self.coverSig: # 持有空头仓位 触发平仓信号 self.signal = self.longPrice self.orderID = self.cover(self.longPrice, -pos) if self.trading: self.excTimes += 1 # 增加报单次数 return elif pos == 0: # 当前无仓位 if self.shortSig: # 买开 self.signal = -self.shortPrice self.orderID = self.short(self.shortPrice, volume) if self.trading: self.excTimes += 1 elif self.buySig: # 卖开 self.signal = self.longPrice self.orderID = self.buy(self.longPrice, volume) if self.trading: self.excTimes += 1 -
至此已经完成了一个完整的技术指标策略，虽然还缺少动态的风控模块，但已具备技术指标策略的所有框架。如果您熟悉并完全掌握了上述策略的所有过程，那么您已具备独立自研策略，编写策略的能力。接下来就是对策略的优化和编程能力的提升。
完整策略代码
完整的代码都在无限易中自带，请自行修改。