# encoding: UTF-8 ''' 本文件中实现了CTA策略引擎，针对CTA类型的策略，抽象简化了部分底层接口的功能。 关于平今和平昨规则： 1. 普通的平仓OFFSET_CLOSET等于平昨OFFSET_CLOSEYESTERDAY 2. 只有上期所的品种需要考虑平今和平昨的区别 3. 当上期所的期货有今仓时，调用Sell和Cover会使用OFFSET_CLOSETODAY，否则 会使用OFFSET_CLOSE 4. 以上设计意味着如果Sell和Cover的数量超过今日持仓量时，会导致出错（即用户 希望通过一个指令同时平今和平昨） 5. 采用以上设计的原因是考虑到vn.trader的用户主要是对TB、MC和金字塔类的平台 感到功能不足的用户（即希望更高频的交易），交易策略不应该出现4中所述的情况 6. 对于想要实现4中所述情况的用户，需要实现一个策略信号引擎和交易委托引擎分开 的定制化统结构（没错，得自己写） ''' import json import os from collections import OrderedDict from datetime import datetime, timedelta from ctaBase import * from ctaSetting import STRATEGY_CLASS from eventEngine import * from vtConstant import * from vtGateway import VtSubscribeReq, VtOrderReq, VtCancelOrderReq, VtLogData from vtFunction import todayDate import logging ######################################################################## class CtaEngine(object): """CTA策略引擎""" # 策略配置文件 settingFileName = 'CTA_setting.json' settingFileName = os.getcwd() + '/ctaAlgo/' + settingFileName #---------------------------------------------------------------------- def __init__(self, mainEngine, eventEngine): """Constructor""" self.mainEngine = mainEngine self.eventEngine = eventEngine # 当前日期 self.today = todayDate() # 保存策略实例的字典 # key为策略名称，value为策略实例，注意策略名称不允许重复 self.strategyDict = {} # 保存vtSymbol和策略实例映射的字典（用于推送tick数据） # 由于可能多个strategy交易同一个vtSymbol，因此key为vtSymbol # value为包含所有相关strategy对象的list self.tickStrategyDict = {} # 保存vtOrderID和strategy对象映射的字典（用于推送order和trade数据） # key为vtOrderID，value为strategy对象 self.orderStrategyDict = {} # 本地停止单编号计数 self.stopOrderCount = 0 # stopOrderID = STOPORDERPREFIX + str(stopOrderCount) # 本地停止单字典 # key为stopOrderID，value为stopOrder对象 self.stopOrderDict = {} # 停止单撤销后不会从本字典中删除 self.workingStopOrderDict = {} # 停止单撤销后会从本字典中删除 # 持仓缓存字典 # key为vtSymbol，value为PositionBuffer对象 self.posBufferDict = {} # 注册事件监听 self.registerEvent() # ---------------------------------------------------------------------- def sendOrder(self, vtSymbol, orderType, price, volume, strategy): """发单""" contract = self.mainEngine.getContract(vtSymbol) req = VtOrderReq() req.symbol = contract.symbol # 合约代码 req.exchange = contract.exchange # 交易所 req.price = price # 价格 req.volume = volume # 数量 req.productClass = strategy.productClass req.currency = strategy.currency # 设计为CTA引擎发出的委托只允许使用限价单 req.priceType = PRICETYPE_LIMITPRICE # 价格类型 # CTA委托类型映射 if orderType == CTAORDER_BUY: req.direction = DIRECTION_LONG # 合约方向 req.offset = OFFSET_OPEN # 开/平 elif orderType == CTAORDER_SELL: req.direction = DIRECTION_SHORT # 只有上期所才要考虑平今平昨 if contract.exchange != EXCHANGE_SHFE: req.offset = OFFSET_CLOSE else: # 获取持仓缓存数据 posBuffer = self.posBufferDict.get(vtSymbol, None) # 如果获取持仓缓存失败，则默认平昨 if not posBuffer: req.offset = OFFSET_CLOSE # 否则如果有多头今仓，则使用平今 elif posBuffer.longToday: req.offset= OFFSET_CLOSETODAY # 其他情况使用平昨 else: req.offset = OFFSET_CLOSE elif orderType == CTAORDER_SHORT: req.direction = DIRECTION_SHORT req.offset = OFFSET_OPEN elif orderType == CTAORDER_COVER: req.direction = DIRECTION_LONG # 只有上期所才要考虑平今平昨 if contract.exchange != EXCHANGE_SHFE: req.offset = OFFSET_CLOSE else: # 获取持仓缓存数据 posBuffer = self.posBufferDict.get(vtSymbol, None) # 如果获取持仓缓存失败，则默认平昨 if not posBuffer: req.offset = OFFSET_CLOSE # 否则如果有空头今仓，则使用平今 elif posBuffer.shortToday: req.offset= OFFSET_CLOSETODAY # 其他情况使用平昨 else: req.offset = OFFSET_CLOSE vtOrderID = self.mainEngine.sendOrder(req, contract.gatewayName) # 发单 self.orderStrategyDict[vtOrderID] = strategy # 保存vtOrderID和策略的映射关系 self.writeCtaLog(u'策略%s发送委托，%s，%s，%s@%s' %(strategy.name, vtSymbol, req.direction, volume, price)) return vtOrderID # ---------------------------------------------------------------------- def cancelOrder(self, vtOrderID): """撤单""" # 查询报单对象 # 1.调用主引擎接口，查询委托单对象 order = self.mainEngine.getOrder(vtOrderID) # 如果查询成功 if order: # 2.检查是否报单（委托单）还有效，只有有效时才发出撤单指令 orderFinished = (order.status == STATUS_ALLTRADED or order.status == STATUS_CANCELLED) if not orderFinished: req = VtCancelOrderReq() req.symbol = order.symbol req.exchange = order.exchange req.frontID = order.frontID req.sessionID = order.sessionID req.orderID = order.orderID self.mainEngine.cancelOrder(req, order.gatewayName) else: if order.status == STATUS_ALLTRADED: self.writeCtaLog(u'委托单({0}已执行，无法撤销'.format(vtOrderID)) if order.status == STATUS_CANCELLED: self.writeCtaLog(u'委托单({0}已撤销，无法再次撤销'.format(vtOrderID)) # 查询不成功 else: self.writeCtaLog(u'委托单({0}不存在'.format(vtOrderID)) # ---------------------------------------------------------------------- def cancelOrders(self, symbol, offset=EMPTY_STRING): """撤销所有单""" # Symbol参数:指定合约的撤单； # OFFSET参数:指定Offset的撤单,缺省不填写时，为所有 l = self.mainEngine.getAllWorkingOrders() self.writeCtaLog(u'撤销所有订单,数量:{0}'.format(len(l))) for order in l: if offset == EMPTY_STRING: offsetCond = True else: offsetCond = order.offset == offset if order.symbol == symbol and offsetCond: req = VtCancelOrderReq() req.symbol = order.symbol req.exchange = order.exchange req.frontID = order.frontID req.sessionID = order.sessionID req.orderID = order.orderID self.writeCtaLog(u'撤单:{0}/{1},{2}{3}手' .format(order.symbol, order.orderID, order.offset, order.totalVolume-order.tradedVolume)) self.mainEngine.cancelOrder(req, order.gatewayName) # ---------------------------------------------------------------------- def sendStopOrder(self, vtSymbol, orderType, price, volume, strategy): """发停止单（本地实现）""" # 1.生成本地停止单ID self.stopOrderCount += 1 stopOrderID = STOPORDERPREFIX + str(self.stopOrderCount) # 2.创建停止单对象 so = StopOrder() so.vtSymbol = vtSymbol # 代码 so.orderType = orderType # 停止单类型 so.price = price # 价格 so.volume = volume # 数量 so.strategy = strategy # 来源策略 so.stopOrderID = stopOrderID # Id so.status = STOPORDER_WAITING # 状态 if orderType == CTAORDER_BUY: so.direction = DIRECTION_LONG so.offset = OFFSET_OPEN elif orderType == CTAORDER_SELL: so.direction = DIRECTION_SHORT so.offset = OFFSET_CLOSE elif orderType == CTAORDER_SHORT: so.direction = DIRECTION_SHORT so.offset = OFFSET_OPEN elif orderType == CTAORDER_COVER: so.direction = DIRECTION_LONG so.offset = OFFSET_CLOSE # 保存stopOrder对象到字典中 self.stopOrderDict[stopOrderID] = so # 字典中不会删除 self.workingStopOrderDict[stopOrderID] = so # 字典中会删除 self.writeCtaLog(u'发停止单成功，' u'Id:{0},Symbol:{1},Type:{2},Price:{3},Volume:{4}' u'.'.format(stopOrderID, vtSymbol, orderType, price, volume)) return stopOrderID # ---------------------------------------------------------------------- def cancelStopOrder(self, stopOrderID): """撤销停止单 Incense Li modified 20160124： 增加返回True 和 False """ # 1.检查停止单是否存在 if stopOrderID in self.workingStopOrderDict: so = self.workingStopOrderDict[stopOrderID] so.status = STOPORDER_CANCELLED # STOPORDER_WAITING =》STOPORDER_CANCELLED del self.workingStopOrderDict[stopOrderID] # 删除 self.writeCtaLog(u'撤销停止单:{0}成功.'.format(stopOrderID)) return True else: self.writeCtaLog(u'撤销停止单:{0}失败，不存在Id.'.format(stopOrderID)) return False # ---------------------------------------------------------------------- def processStopOrder(self, tick): """收到行情后处理本地停止单（检查是否要立即发出）""" vtSymbol = tick.vtSymbol # 1.首先检查是否有策略交易该合约 if vtSymbol in self.tickStrategyDict: # 2.遍历等待中的停止单，检查是否会被触发 for so in self.workingStopOrderDict.values(): if so.vtSymbol == vtSymbol: # 3. 触发标识判断 longTriggered = so.direction == DIRECTION_LONG and tick.lastPrice >= so.price # 多头停止单被触发 shortTriggered = so.direction == DIRECTION_SHORT and tick.lasatPrice <= so.price # 空头停止单被触发 # 4.触发处理 if longTriggered or shortTriggered: # 5.设定价格，买入和卖出分别以涨停跌停价发单（模拟市价单） if so.direction == DIRECTION_LONG: price = tick.upperLimit else: price = tick.lowerLimit # 6.更新停止单状态，触发 so.status = STOPORDER_TRIGGERED # 7.发单 self.sendOrder(so.vtSymbol, so.orderType, price, so.volume, so.strategy) # 8.删除停止单 del self.workingStopOrderDict[so.stopOrderID] # ---------------------------------------------------------------------- def procecssTickEvent(self, event): """处理行情推送事件""" # 1. 获取事件的Tick数据 tick = event.dict_['data'] # 2.收到tick行情后，优先处理本地停止单（检查是否要立即发出） self.processStopOrder(tick) # 3.推送tick到对应的策略对象进行处理 if tick.vtSymbol in self.tickStrategyDict: # 4.将vtTickData数据转化为ctaTickData ctaTick = CtaTickData() d = ctaTick.__dict__ for key in d.keys(): if key != 'datetime': d[key] = tick.__getattribute__(key) # 5.添加datetime字段 ctaTick.datetime = datetime.strptime(' '.join([tick.date, tick.time]), '%Y%m%d %H:%M:%S.%f') # 逐个推送到策略实例中 l = self.tickStrategyDict[tick.vtSymbol] for strategy in l: strategy.onTick(ctaTick) # ---------------------------------------------------------------------- def processOrderEvent(self, event): """处理委托推送事件""" # 1.获取事件的Order数据 order = event.dict_['data'] # order.vtOrderID 在gateway中，已经格式化为 gatewayName.vtOrderID # 2.判断order是否在映射字典中 if order.vtOrderID in self.orderStrategyDict: # 3.提取对应的策略 strategy = self.orderStrategyDict[order.vtOrderID] # 4.触发策略的委托推送事件方法 strategy.onOrder(order) # ---------------------------------------------------------------------- def processTradeEvent(self, event): """处理成交推送事件""" # 1.获取事件的Trade数据 trade = event.dict_['data'] # 2.判断Trade是否在映射字典中 if trade.vtOrderID in self.orderStrategyDict: # 3.提取对应的策略 strategy = self.orderStrategyDict[trade.vtOrderID] # 计算策略持仓 if trade.direction == DIRECTION_LONG: strategy.pos += trade.volume else: strategy.pos -= trade.volume # 4.触发策略的成交推送事件。 strategy.onTrade(trade) # 更新持仓缓存数据 if trade.vtSymbol in self.tickStrategyDict: posBuffer = self.posBufferDict.get(trade.vtSymbol, None) if not posBuffer: posBuffer = PositionBuffer() posBuffer.vtSymbol = trade.vtSymbol self.posBufferDict[trade.vtSymbol] = posBuffer posBuffer.updateTradeData(trade) #---------------------------------------------------------------------- def processPositionEvent(self, event): """处理持仓推送""" pos = event.dict_['data'] # 更新持仓缓存数据 if pos.vtSymbol in self.tickStrategyDict: posBuffer = self.posBufferDict.get(pos.vtSymbol, None) if not posBuffer: posBuffer = PositionBuffer() posBuffer.vtSymbol = pos.vtSymbol self.posBufferDict[pos.vtSymbol] = posBuffer posBuffer.updatePositionData(pos) #---------------------------------------------------------------------- def registerEvent(self): """注册事件监听""" # 注册行情数据推送（Tick数据到达）的响应事件 self.eventEngine.register(EVENT_TICK, self.procecssTickEvent) # 注册订单推送的响应事件 self.eventEngine.register(EVENT_ORDER, self.processOrderEvent) # 注册成交推送（交易）的响应时间 self.eventEngine.register(EVENT_TRADE, self.processTradeEvent) # 注册持仓更新事件 self.eventEngine.register(EVENT_POSITION, self.processPositionEvent) # ---------------------------------------------------------------------- def insertData(self, dbName, collectionName, data): """插入数据到数据库（这里的data可以是CtaTickData或者CtaBarData）""" self.mainEngine.dbInsert(dbName, collectionName, data.__dict__) # ---------------------------------------------------------------------- def loadBar(self, dbName, collectionName, days): """从数据库中读取Bar数据，startDate是datetime对象""" startDate = self.today - timedelta(days) d = {'datetime':{'$gte':startDate}} cursor = self.mainEngine.dbQuery(dbName, collectionName, d) l = [] if cursor: for d in cursor: bar = CtaBarData() bar.__dict__ = d l.append(bar) return l # ---------------------------------------------------------------------- def loadTick(self, dbName, collectionName, days): """从数据库中读取Tick数据，startDate是datetime对象""" startDate = self.today - timedelta(days) d = {'datetime':{'$gte':startDate}} cursor = self.mainEngine.dbQuery(dbName, collectionName, d) l = [] if cursor: for d in cursor: tick = CtaTickData() tick.__dict__ = d l.append(tick) return l # ---------------------------------------------------------------------- def getToday(self): """获取代表今日的datetime对象""" today = datetime.today() today = today.replace(hour=0, minute=0, second=0, microsecond=0) return today # ---------------------------------------------------------------------- def writeCtaLog(self, content): """快速发出CTA模块日志事件""" log = VtLogData() log.logContent = content event = Event(type_=EVENT_CTA_LOG) event.dict_['data'] = log self.eventEngine.put(event) # 写入本地log日志 logging.info(content) # ---------------------------------------------------------------------- def loadStrategy(self, setting): """载入策略""" try: name = setting['name'] className = setting['className'] except Exception, e: self.writeCtaLog(u'载入策略出错：%s' %e) return # 获取策略类 strategyClass = STRATEGY_CLASS.get(className, None) if not strategyClass: self.writeCtaLog(u'找不到策略类：%s' %className) return # 防止策略重名 if name in self.strategyDict: self.writeCtaLog(u'策略实例重名：%s' %name) else: # 1.创建策略对象 strategy = strategyClass(self, setting) self.strategyDict[name] = strategy # 2.保存Tick映射关系（symbol <==> Strategy[] ) if strategy.vtSymbol in self.tickStrategyDict: l = self.tickStrategyDict[strategy.vtSymbol] else: l = [] self.tickStrategyDict[strategy.vtSymbol] = l l.append(strategy) # 3.订阅合约 contract = self.mainEngine.getContract(strategy.vtSymbol) if contract: # 4.构造订阅请求包 req = VtSubscribeReq() req.symbol = contract.symbol req.exchange = contract.exchange # 对于IB接口订阅行情时所需的货币和产品类型，从策略属性中获取 req.currency = strategy.currency req.productClass = strategy.productClass # 5.调用主引擎的订阅接口 self.mainEngine.subscribe(req, contract.gatewayName) else: self.writeCtaLog(u'%s的交易合约%s无法找到' %(name, strategy.vtSymbol)) #---------------------------------------------------------------------- def initStrategy(self, name, force = False): """初始化策略""" if name in self.strategyDict: strategy = self.strategyDict[name] if not strategy.inited or force == True: strategy.onInit(force=force) strategy.inited = True else: self.writeCtaLog(u'请勿重复初始化策略实例：%s' %name) else: self.writeCtaLog(u'策略实例不存在：%s' %name) # --------------------------------------------------------------------- def startStrategy(self, name): """启动策略""" # 1.判断策略名称是否存在字典中 if name in self.strategyDict: # 2.提取策略 strategy = self.strategyDict[name] # 3.判断策略是否运行 if strategy.inited and not strategy.trading: # 4.设置运行状态 strategy.trading = True # 5.启动策略 strategy.onStart() else: self.writeCtaLog(u'策略实例不存在：%s' %name) # ---------------------------------------------------------------------- def stopStrategy(self, name): """停止策略运行""" # 1.判断策略名称是否存在字典中 if name in self.strategyDict: # 2.提取策略 strategy = self.strategyDict[name] # 3.停止交易 if strategy.trading: # 4.设置交易状态为False strategy.trading = False # 5.调用策略的停止方法 strategy.onStop() # 6.对该策略发出的所有限价单进行撤单 for vtOrderID, s in self.orderStrategyDict.items(): if s is strategy: self.cancelOrder(vtOrderID) # 7.对该策略发出的所有本地停止单撤单 for stopOrderID, so in self.workingStopOrderDict.items(): if so.strategy is strategy: self.cancelStopOrder(stopOrderID) else: self.writeCtaLog(u'策略实例不存在：%s' %name) # ---------------------------------------------------------------------- def saveSetting(self): """保存策略配置""" with open(self.settingFileName, 'w') as f: l = [] # 逐一循环：name，策略实例名称，strategy，策略 for strategy in self.strategyDict.values(): # 配置串 setting = {} for param in strategy.paramList: setting[param] = strategy.__getattribute__(param) l.append(setting) jsonL = json.dumps(l, indent=4) f.write(jsonL) # ---------------------------------------------------------------------- def loadSetting(self): """读取策略配置""" with open(self.settingFileName) as f: l = json.load(f) for setting in l: self.loadStrategy(setting) # ---------------------------------------------------------------------- def getStrategyVar(self, name): """获取策略当前的变量字典""" if name in self.strategyDict: # 获取策略实例 strategy = self.strategyDict[name] varDict = OrderedDict() for key in strategy.varList: varDict[key] = strategy.__getattribute__(key) return varDict else: self.writeCtaLog(u'策略实例不存在：' + name) return None # ---------------------------------------------------------------------- def getStrategyParam(self, name): """获取策略的参数字典""" if name in self.strategyDict: # 获取策略实例 strategy = self.strategyDict[name] paramDict = OrderedDict() for key in strategy.paramList: paramDict[key] = strategy.__getattribute__(key) return paramDict else: self.writeCtaLog(u'策略实例不存在：' + name) return None #---------------------------------------------------------------------- def putStrategyEvent(self, name): """触发策略状态变化事件（通常用于通知GUI更新）""" event = Event(EVENT_CTA_STRATEGY+name) self.eventEngine.put(event) ######################################################################## class PositionBuffer(object): """持仓缓存信息（本地维护的持仓数据）""" #---------------------------------------------------------------------- def __init__(self): """Constructor""" self.vtSymbol = EMPTY_STRING # 多头 self.longPosition = EMPTY_INT self.longToday = EMPTY_INT self.longYd = EMPTY_INT # 空头 self.shortPosition = EMPTY_INT self.shortToday = EMPTY_INT self.shortYd = EMPTY_INT #---------------------------------------------------------------------- def updatePositionData(self, pos): """更新持仓数据""" if pos.direction == DIRECTION_LONG: self.longPosition = pos.position self.longYd = pos.ydPosition self.longToday = self.longPosition - self.longYd else: self.shortPosition = pos.position self.shortYd = pos.ydPosition self.shortToday = self.shortPosition - self.shortYd #---------------------------------------------------------------------- def updateTradeData(self, trade): """更新成交数据""" if trade.direction == DIRECTION_LONG: # 多方开仓，则对应多头的持仓和今仓增加 if trade.offset == OFFSET_OPEN: self.longPosition += trade.volume self.longToday += trade.volume # 多方平今，对应空头的持仓和今仓减少 elif trade.offset == OFFSET_CLOSETODAY: self.shortPosition -= trade.volume self.shortToday -= trade.volume # 多方平昨，对应空头的持仓和昨仓减少 else: self.shortPosition -= trade.volume self.shortYd -= trade.volume else: # 空头和多头相同 if trade.offset == OFFSET_OPEN: self.shortPosition += trade.volume self.shortToday += trade.volume elif trade.offset == OFFSET_CLOSETODAY: self.longPosition -= trade.volume self.longToday -= trade.volume else: self.longPosition -= trade.volume self.longYd -= trade.volume