[Add]新增RQData的Tick下载功能

examples/RQData/dataService.py

@@ -7,9 +7,12 @@ from datetime import datetime
 from time import time, sleep
 
 from pymongo import MongoClient, ASCENDING
+import pandas as pd
 
-from vnpy.trader.vtObject import VtBarData
+from vnpy.trader.vtObject import VtBarData, VtTickData
-from vnpy.trader.app.ctaStrategy.ctaBase import MINUTE_DB_NAME, DAILY_DB_NAME
+from vnpy.trader.app.ctaStrategy.ctaBase import (MINUTE_DB_NAME, 
+                                                 DAILY_DB_NAME, 
+                                                 TICK_DB_NAME)
 
 import rqdatac as rq
 
@@ -17,16 +20,13 @@ import rqdatac as rq
 config = open('config.json')
 setting = json.load(config)
 
-MONGO_HOST = setting['MONGO_HOST']
+mc = MongoClient()                  # Mongo连接
-MONGO_PORT = setting['MONGO_PORT']
+dbMinute = mc[MINUTE_DB_NAME]       # 数据库
-SYMBOLS = setting['SYMBOLS']
+dbDaily = mc[DAILY_DB_NAME]
+dbTick = mc[TICK_DB_NAME]
 
-mc = MongoClient(MONGO_HOST, MONGO_PORT)        # Mongo连接
+USERNAME = setting['rqUsername']
-db = mc[MINUTE_DB_NAME]                         # 数据库
+PASSWORD = setting['rqPassword']
-db2 = mc[DAILY_DB_NAME]
-
-USERNAME = setting['USERNAME']
-PASSWORD = setting['PASSWORD']
 rq.init(USERNAME, PASSWORD)
 
 FIELDS = ['open', 'high', 'low', 'close', 'volume']
@@ -49,12 +49,56 @@ def generateVtBar(row, symbol):
     
     return bar
 
+#----------------------------------------------------------------------
+def generateVtTick(row, symbol):
+    """生成K线"""
+    tick = VtTickData()
+    tick.symbol = symbol
+    tick.vtSymbol = symbol
+    
+    tick.lastPrice = row['last']
+    tick.volume = row['volume']
+    tick.openInterest = row['open_interest']
+    tick.datetime = row.name
+    tick.openPrice = row['open']
+    tick.highPrice = row['high']
+    tick.lowPrice = row['low']
+    tick.preClosePrice = row['prev_close']
+    tick.upperLimit = row['limit_up']
+    tick.lowerLimit = row['limit_down']
+    
+    tick.bidPrice1 = row['b1']
+    tick.bidPrice2 = row['b2']
+    tick.bidPrice3 = row['b3']
+    tick.bidPrice4 = row['b4']
+    tick.bidPrice5 = row['b5']
+    
+    tick.bidVolume1 = row['b1_v']
+    tick.bidVolume2 = row['b2_v']
+    tick.bidVolume3 = row['b3_v']
+    tick.bidVolume4 = row['b4_v']
+    tick.bidVolume5 = row['b5_v']    
+    
+    tick.askPrice1 = row['a1']
+    tick.askPrice2 = row['a2']
+    tick.askPrice3 = row['a3']
+    tick.askPrice4 = row['a4']
+    tick.askPrice5 = row['a5']
+    
+    tick.askVolume1 = row['a1_v']
+    tick.askVolume2 = row['a2_v']
+    tick.askVolume3 = row['a3_v']
+    tick.askVolume4 = row['a4_v']
+    tick.askVolume5 = row['a5_v']        
+    
+    return tick
+
 #----------------------------------------------------------------------
 def downloadMinuteBarBySymbol(symbol):
     """下载某一合约的分钟线数据"""
     start = time()
 
-    cl = db[symbol]
+    cl = dbMinute[symbol]
     cl.ensure_index([('datetime', ASCENDING)], unique=True)         # 添加索引
     
     df = rq.get_price(symbol, frequency='1m', fields=FIELDS)
@@ -68,14 +112,14 @@ def downloadMinuteBarBySymbol(symbol):
     end = time()
     cost = (end - start) * 1000
 
-    print(u'合约%s数据下载完成%s - %s，耗时%s毫秒' %(symbol, df.index[0], df.index[-1], cost))
+    print(u'合约%s的分钟K线数据下载完成%s - %s，耗时%s毫秒' %(symbol, df.index[0], df.index[-1], cost))
 
 #----------------------------------------------------------------------
 def downloadDailyBarBySymbol(symbol):
     """下载某一合约日线数据"""
     start = time()
 
-    cl = db2[symbol]
+    cl = dbDaily[symbol]
     cl.ensure_index([('datetime', ASCENDING)], unique=True)         # 添加索引
     
     df = rq.get_price(symbol, frequency='1d', fields=FIELDS, end_date=datetime.now().strftime('%Y%m%d'))
@@ -89,36 +133,28 @@ def downloadDailyBarBySymbol(symbol):
     end = time()
     cost = (end - start) * 1000
 
-    print(u'合约%s数据下载完成%s - %s，耗时%s毫秒' %(symbol, df.index[0], df.index[-1], cost))
+    print(u'合约%s的日K线数据下载完成%s - %s，耗时%s毫秒' %(symbol, df.index[0], df.index[-1], cost))
-
 
 #----------------------------------------------------------------------
-def downloadAllMinuteBar():
+def downloadTickBySymbol(symbol, date):
-    """下载所有配置中的合约的分钟线数据"""
+    """下载某一合约日线数据"""
-    print('-' * 50)
+    start = time()
-    print(u'开始下载合约分钟线数据')
-    print('-' * 50)
-    
-    # 添加下载任务
-    for symbol in SYMBOLS:
-        downloadMinuteBarBySymbol(str(symbol))
-    
-    print('-' * 50)
-    print(u'合约分钟线数据下载完成')
-    print('-' * 50)
 
-#----------------------------------------------------------------------
+    cl = dbTick[symbol]
-def downloadAllDailyBar():
+    cl.ensure_index([('datetime', ASCENDING)], unique=True)         # 添加索引
-    """下载所有配置中的合约的日数据"""
-    print('-' * 50)
-    print(u'开始下载合约日线数据')
-    print('-' * 50)
     
-    # 添加下载任务
+    df = rq.get_price(symbol, 
-    for symbol in SYMBOLS:
+                      frequency='tick', 
-        downloadDailyBarBySymbol(str(symbol))
+                      start_date=date, 
+                      end_date=date)
     
-    print('-' * 50)
+    for ix, row in df.iterrows():
-    print(u'合约日线数据下载完成')
+        tick = generateVtTick(row, symbol)
-    print('-' * 50)
+        d = tick.__dict__
-    
+        flt = {'datetime': tick.datetime}
+        cl.replace_one(flt, d, True)            
+
+    end = time()
+    cost = (end - start) * 1000
+
+    print(u'合约%sTick数据下载完成%s - %s，耗时%s毫秒' %(symbol, df.index[0], df.index[-1], cost))

examples/RQData/downloadData.py

@@ -8,7 +8,8 @@ from dataService import *
 
 
 if __name__ == '__main__':
-    downloadMinuteBarBySymbol('CU99')
+    #downloadMinuteBarBySymbol('CU99')
-    downloadDailyBarBySymbol('IF99')
+    #downloadDailyBarBySymbol('IF99')
-    downloadDailyBarBySymbol('TA99')
+    #downloadDailyBarBySymbol('TA99')
-    downloadDailyBarBySymbol('I99')
+    #downloadDailyBarBySymbol('I99')
+    downloadTickBySymbol('IF1901', '2018-12-21')

examples/RQData/runService.py

@@ -1,33 +0,0 @@
-# encoding: UTF-8
-
-"""
-定时服务，可无人值守运行，实现每日自动下载更新历史行情数据到数据库中。
-"""
-from __future__ import print_function
-
-import time
-import datetime
-
-from dataService import downloadAllMinuteBar
-
-
-if __name__ == '__main__':
-    taskCompletedDate = None
-    
-    # 生成一个随机的任务下载时间，用于避免所有用户在同一时间访问数据服务器
-    taskTime = datetime.time(hour=17, minute=0)
-    
-    # 进入主循环
-    while True:
-        t = datetime.datetime.now()
-        
-        # 每天到达任务下载时间后，执行数据下载的操作
-        if t.time() > taskTime and (taskCompletedDate is None or t.date() != taskCompletedDate):
-            downloadAllMinuteBar()
-            
-            # 更新任务完成的日期
-            taskCompletedDate = t.date()
-        else:
-            print(u'当前时间%s，任务定时%s' %(t, taskTime))
-    
-        time.sleep(60)