From 4fa897c2b075dbba1e9048f56f269c3178bcb5c0 Mon Sep 17 00:00:00 2001 From: "vn.py" Date: Tue, 26 Sep 2017 09:35:02 +0800 Subject: [PATCH] =?UTF-8?q?[Add]=E6=96=B0=E5=A2=9E=E5=A4=A9=E5=8B=A4?= =?UTF-8?q?=E6=95=B0=E6=8D=AE=E6=9C=8D=E5=8A=A1TqDataService?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- examples/TqDataService/config.json | 8 ++ examples/TqDataService/dataService.py | 107 +++++++++++++++++++++++++ examples/TqDataService/downloadData.py | 11 +++ examples/TqDataService/runService.py | 33 ++++++++ vnpy/data/tq/test.py | 15 +++- 5 files changed, 171 insertions(+), 3 deletions(-) create mode 100644 examples/TqDataService/config.json create mode 100644 examples/TqDataService/dataService.py create mode 100644 examples/TqDataService/downloadData.py create mode 100644 examples/TqDataService/runService.py diff --git a/examples/TqDataService/config.json b/examples/TqDataService/config.json new file mode 100644 index 00000000..ff747220 --- /dev/null +++ b/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"] +} \ No newline at end of file diff --git a/examples/TqDataService/dataService.py b/examples/TqDataService/dataService.py new file mode 100644 index 00000000..0861fe22 --- /dev/null +++ b/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 + + + + \ No newline at end of file diff --git a/examples/TqDataService/downloadData.py b/examples/TqDataService/downloadData.py new file mode 100644 index 00000000..3cf6f3cc --- /dev/null +++ b/examples/TqDataService/downloadData.py @@ -0,0 +1,11 @@ +# encoding: UTF-8 + +""" +立即下载数据到数据库中，用于手动执行更新操作。 +""" + +from dataService import * + + +if __name__ == '__main__': + downloadAllMinuteBar(1000) \ No newline at end of file diff --git a/examples/TqDataService/runService.py b/examples/TqDataService/runService.py new file mode 100644 index 00000000..17d6d165 --- /dev/null +++ b/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) \ No newline at end of file diff --git a/vnpy/data/tq/test.py b/vnpy/data/tq/test.py index 2c8f254f..bbcb590f 100644 --- a/vnpy/data/tq/test.py +++ b/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()