[Add]新增天勤数据服务TqDataService

examples/TqDataService/config.json

@@ -0,0 +1,8 @@
+{
+	"MONGO_HOST": "localhost",
+	"MONGO_PORT": 27017,
+
+	"SYMBOLS": ["IF1710", "IF1711", "IF1712", "IF1803", 
+				"IH1710", "IH1711", "IH1712", "IH1803", 
+				"IC1710", "IC1711", "IC1712", "IC1803"]
+}

examples/TqDataService/dataService.py

@@ -0,0 +1,107 @@
+# encoding: UTF-8
+
+import sys
+import json
+from datetime import datetime
+from time import sleep
+
+from pymongo import MongoClient, ASCENDING
+
+from vnpy.data.tq.vntq import TqApi
+from vnpy.trader.vtObject import VtBarData
+from vnpy.trader.app.ctaStrategy.ctaBase import MINUTE_DB_NAME
+
+
+# 加载配置
+config = open('config.json')
+setting = json.load(config)
+
+MONGO_HOST = setting['MONGO_HOST']
+MONGO_PORT = setting['MONGO_PORT']
+SYMBOLS = setting['SYMBOLS']
+
+mc = MongoClient(MONGO_HOST, MONGO_PORT)        # Mongo连接
+db = mc[MINUTE_DB_NAME]                         # 数据库
+
+api = TqApi()   # 历史行情服务API对象
+api.connect()   # 连接
+taskList = []   # 下载任务列表
+
+
+#----------------------------------------------------------------------
+def generateVtBar(symbol, d):
+    """生成K线"""
+    bar = VtBarData()
+    
+    bar.symbol = symbol
+    bar.vtSymbol = symbol
+    bar.open = d['open']
+    bar.high = d['high']
+    bar.low = d['low']
+    bar.close = d['close']
+    bar.volume = d['volume']
+    bar.openInterest = d['open_oi']
+    bar.datetime = datetime.fromtimestamp(d['datetime']/1000000000)
+    bar.date = bar.datetime.strftime("%Y%m%d")
+    bar.time = bar.datetime.strftime("%H:%M:%S")
+    
+    return bar
+
+#----------------------------------------------------------------------
+def onChart(symbol, seconds):
+    """K线更新处理函数"""    
+    # 避免重复记录已经完成的任务
+    if symbol not in taskList:
+        return 
+    
+    serial = api.get_kline_serial(symbol, seconds)
+    
+    cl = db[symbol]                                                 # 集合
+    cl.ensure_index([('datetime', ASCENDING)], unique=True)         # 添加索引
+    
+    l = serial.values()
+    for d in l:
+        bar = generateVtBar(symbol, d)
+        d = bar.__dict__
+        flt = {'datetime': bar.datetime}
+        cl.replace_one(flt, d, True)
+    
+    start = datetime.fromtimestamp(l[0]['datetime']/1000000000)
+    end = datetime.fromtimestamp(l[-1]['datetime']/1000000000)
+    print u'合约%s下载完成%s - %s' %(symbol, start, end)
+    
+    # 移除已经完成的任务
+    if symbol in taskList:
+        taskList.remove(symbol)
+
+#----------------------------------------------------------------------
+def downMinuteBarBySymbol(symbol, num):
+    """下载某一合约的分钟线数据"""
+    api.subscribe_chart(symbol, 60, num, onChart)
+    
+#----------------------------------------------------------------------
+def downloadAllMinuteBar(num):
+    """下载所有配置中的合约的分钟线数据"""
+    print '-' * 50
+    print u'开始下载合约分钟线数据'
+    print '-' * 50
+    
+    # 添加下载任务
+    taskList.extend(SYMBOLS)
+    
+    for symbol in SYMBOLS:
+        downMinuteBarBySymbol(str(symbol), num)
+    
+    while True:
+        sleep(2)
+
+        # 如果任务列表为空，则说明数据已经全部下载完成
+        if not taskList:
+            print '-' * 50
+            print u'合约分钟线数据下载完成'
+            print '-' * 50
+            return       
+    
+
+
+    

examples/TqDataService/downloadData.py

@@ -0,0 +1,11 @@
+# encoding: UTF-8
+
+"""
+立即下载数据到数据库中，用于手动执行更新操作。
+"""
+
+from dataService import *
+
+
+if __name__ == '__main__':
+    downloadAllMinuteBar(1000)

examples/TqDataService/runService.py

@@ -0,0 +1,33 @@
+# encoding: UTF-8
+
+"""
+定时服务，可无人值守运行，实现每日自动下载更新历史行情数据到数据库中。
+"""
+
+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):
+            # 下载1000根分钟线数据，足以覆盖过去两天的行情
+            downloadAllMinuteBar(1000)
+            
+            # 更新任务完成的日期
+            taskCompletedDate = t.date()
+        else:
+            print u'当前时间%s，任务定时%s' %(t, taskTime)
+    
+        time.sleep(60)

vnpy/data/tq/test.py

@@ -1,5 +1,6 @@
 # encoding: UTF-8
 
+from time import sleep
 
 from vntq import TqApi
 
@@ -28,15 +29,23 @@ def onChart(symbol, seconds):
         serial = api.get_kline_serial(symbol, seconds)
         
     print serial
-
+    
 
 if __name__ == "__main__":
+    symbol = 'IF1710'
+    
     api = TqApi()
     
     api.connect()
     
-    api.subscribe_quote(["RM801"], onQuote)
-    api.subscribe_chart("RM801", 0, 100, onChart)
+    # 订阅Tick推送
+    #api.subscribe_quote([symbol], onQuote)
+    
+    # 订阅Tick图表
+    #api.subscribe_chart(symbol, 0, 100, onChart)
+    
+    # 订阅K线图表
+    api.subscribe_chart(symbol, 60, 1000, onChart)
     
     raw_input()