232 lines
8.8 KiB
Python
232 lines
8.8 KiB
Python
# encoding: UTF-8
|
||
|
||
'''
|
||
本文件中实现了行情数据记录引擎，用于汇总TICK数据，并生成K线插入数据库。
|
||
|
||
使用DR_setting.json来配置需要收集的合约，以及主力合约代码。
|
||
'''
|
||
|
||
import json
|
||
import os
|
||
import copy
|
||
from collections import OrderedDict
|
||
from datetime import datetime, timedelta
|
||
from Queue import Queue
|
||
from threading import Thread
|
||
|
||
from eventEngine import *
|
||
from vtGateway import VtSubscribeReq, VtLogData
|
||
from drBase import *
|
||
from vtFunction import todayDate
|
||
|
||
|
||
########################################################################
|
||
class DrEngine(object):
|
||
"""数据记录引擎"""
|
||
|
||
settingFileName = 'DR_setting.json'
|
||
settingFileName = os.getcwd() + '/dataRecorder/' + settingFileName
|
||
|
||
#----------------------------------------------------------------------
|
||
def __init__(self, mainEngine, eventEngine):
|
||
"""Constructor"""
|
||
self.mainEngine = mainEngine
|
||
self.eventEngine = eventEngine
|
||
|
||
# 当前日期
|
||
self.today = todayDate()
|
||
|
||
# 主力合约代码映射字典，key为具体的合约代码（如IF1604），value为主力合约代码（如IF0000）
|
||
self.activeSymbolDict = {}
|
||
|
||
# Tick对象字典
|
||
self.tickDict = {}
|
||
|
||
# K线对象字典
|
||
self.barDict = {}
|
||
|
||
# 负责执行数据库插入的单独线程相关
|
||
self.active = False # 工作状态
|
||
self.queue = Queue() # 队列
|
||
self.thread = Thread(target=self.run) # 线程
|
||
|
||
# 载入设置，订阅行情
|
||
self.loadSetting()
|
||
|
||
#----------------------------------------------------------------------
|
||
def loadSetting(self):
|
||
"""载入设置"""
|
||
with open(self.settingFileName) as f:
|
||
drSetting = json.load(f)
|
||
|
||
# 如果working设为False则不启动行情记录功能
|
||
working = drSetting['working']
|
||
if not working:
|
||
return
|
||
|
||
if 'tick' in drSetting:
|
||
l = drSetting['tick']
|
||
|
||
for setting in l:
|
||
symbol = setting[0]
|
||
vtSymbol = symbol
|
||
|
||
req = VtSubscribeReq()
|
||
req.symbol = setting[0]
|
||
|
||
# 针对LTS和IB接口，订阅行情需要交易所代码
|
||
if len(setting)>=3:
|
||
req.exchange = setting[2]
|
||
vtSymbol = '.'.join([symbol, req.exchange])
|
||
|
||
# 针对IB接口，订阅行情需要货币和产品类型
|
||
if len(setting)>=5:
|
||
req.currency = setting[3]
|
||
req.productClass = setting[4]
|
||
|
||
self.mainEngine.subscribe(req, setting[1])
|
||
|
||
drTick = DrTickData() # 该tick实例可以用于缓存部分数据（目前未使用）
|
||
self.tickDict[vtSymbol] = drTick
|
||
|
||
if 'bar' in drSetting:
|
||
l = drSetting['bar']
|
||
|
||
for setting in l:
|
||
symbol = setting[0]
|
||
vtSymbol = symbol
|
||
|
||
req = VtSubscribeReq()
|
||
req.symbol = symbol
|
||
|
||
if len(setting)>=3:
|
||
req.exchange = setting[2]
|
||
vtSymbol = '.'.join([symbol, req.exchange])
|
||
|
||
if len(setting)>=5:
|
||
req.currency = setting[3]
|
||
req.productClass = setting[4]
|
||
|
||
self.mainEngine.subscribe(req, setting[1])
|
||
|
||
bar = DrBarData()
|
||
self.barDict[vtSymbol] = bar
|
||
|
||
if 'active' in drSetting:
|
||
d = drSetting['active']
|
||
|
||
# 注意这里的vtSymbol对于IB和LTS接口，应该后缀.交易所
|
||
for activeSymbol, vtSymbol in d.items():
|
||
self.activeSymbolDict[vtSymbol] = activeSymbol
|
||
|
||
# 启动数据插入线程
|
||
self.start()
|
||
|
||
# 注册事件监听
|
||
self.registerEvent()
|
||
|
||
#----------------------------------------------------------------------
|
||
def procecssTickEvent(self, event):
|
||
"""处理行情推送"""
|
||
tick = event.dict_['data']
|
||
vtSymbol = tick.vtSymbol
|
||
|
||
# 转化Tick格式
|
||
drTick = DrTickData()
|
||
d = drTick.__dict__
|
||
for key in d.keys():
|
||
if key != 'datetime':
|
||
d[key] = tick.__getattribute__(key)
|
||
drTick.datetime = datetime.strptime(' '.join([tick.date, tick.time]), '%Y%m%d %H:%M:%S.%f')
|
||
|
||
# 更新Tick数据
|
||
if vtSymbol in self.tickDict:
|
||
self.insertData(TICK_DB_NAME, vtSymbol, drTick)
|
||
|
||
if vtSymbol in self.activeSymbolDict:
|
||
activeSymbol = self.activeSymbolDict[vtSymbol]
|
||
self.insertData(TICK_DB_NAME, activeSymbol, drTick)
|
||
|
||
# 发出日志
|
||
self.writeDrLog(u'记录Tick数据%s，时间:%s, last:%s, bid:%s, ask:%s'
|
||
%(drTick.vtSymbol, drTick.time, drTick.lastPrice, drTick.bidPrice1, drTick.askPrice1))
|
||
|
||
# 更新分钟线数据
|
||
if vtSymbol in self.barDict:
|
||
bar = self.barDict[vtSymbol]
|
||
|
||
# 如果第一个TICK或者新的一分钟
|
||
if not bar.datetime or bar.datetime.minute != drTick.datetime.minute:
|
||
if bar.vtSymbol:
|
||
newBar = copy.copy(bar)
|
||
self.insertData(MINUTE_DB_NAME, vtSymbol, newBar)
|
||
|
||
if vtSymbol in self.activeSymbolDict:
|
||
activeSymbol = self.activeSymbolDict[vtSymbol]
|
||
self.insertData(MINUTE_DB_NAME, activeSymbol, newBar)
|
||
|
||
self.writeDrLog(u'记录分钟线数据%s，时间:%s, O:%s, H:%s, L:%s, C:%s'
|
||
%(bar.vtSymbol, bar.time, bar.open, bar.high,
|
||
bar.low, bar.close))
|
||
|
||
bar.vtSymbol = drTick.vtSymbol
|
||
bar.symbol = drTick.symbol
|
||
bar.exchange = drTick.exchange
|
||
|
||
bar.open = drTick.lastPrice
|
||
bar.high = drTick.lastPrice
|
||
bar.low = drTick.lastPrice
|
||
bar.close = drTick.lastPrice
|
||
|
||
bar.date = drTick.date
|
||
bar.time = drTick.time
|
||
bar.datetime = drTick.datetime
|
||
bar.volume = drTick.volume
|
||
bar.openInterest = drTick.openInterest
|
||
# 否则继续累加新的K线
|
||
else:
|
||
bar.high = max(bar.high, drTick.lastPrice)
|
||
bar.low = min(bar.low, drTick.lastPrice)
|
||
bar.close = drTick.lastPrice
|
||
|
||
#----------------------------------------------------------------------
|
||
def registerEvent(self):
|
||
"""注册事件监听"""
|
||
self.eventEngine.register(EVENT_TICK, self.procecssTickEvent)
|
||
|
||
#----------------------------------------------------------------------
|
||
def insertData(self, dbName, collectionName, data):
|
||
"""插入数据到数据库（这里的data可以是CtaTickData或者CtaBarData）"""
|
||
self.queue.put((dbName, collectionName, data.__dict__))
|
||
|
||
#----------------------------------------------------------------------
|
||
def run(self):
|
||
"""运行插入线程"""
|
||
while self.active:
|
||
try:
|
||
dbName, collectionName, d = self.queue.get(block=True, timeout=1)
|
||
self.mainEngine.dbInsert(dbName, collectionName, d)
|
||
except Empty:
|
||
pass
|
||
#----------------------------------------------------------------------
|
||
def start(self):
|
||
"""启动"""
|
||
self.active = True
|
||
self.thread.start()
|
||
|
||
#----------------------------------------------------------------------
|
||
def stop(self):
|
||
"""退出"""
|
||
if self.active:
|
||
self.active = False
|
||
self.thread.join()
|
||
|
||
#----------------------------------------------------------------------
|
||
def writeDrLog(self, content):
|
||
"""快速发出日志事件"""
|
||
log = VtLogData()
|
||
log.logContent = content
|
||
event = Event(type_=EVENT_DATARECORDER_LOG)
|
||
event.dict_['data'] = log
|
||
self.eventEngine.put(event)
|
||
|