[Add]新增RQData数据服务

This commit is contained in:
vn.py 2018-11-01 11:10:05 +08:00
parent 5b949c0124
commit f0f52f035c
5 changed files with 149 additions and 0 deletions

View File

@ -1,5 +1,7 @@
# DataService说明
* RqdataDataServiceRQData数据服务A股、期货、基金
* ShcifcoDataService上海中期历史行情服务期货
* TqDataService天勤历史行情服务期货

View File

@ -0,0 +1,9 @@
{
"MONGO_HOST": "localhost",
"MONGO_PORT": 27017,
"SYMBOLS": ["510050", "510300"],
"USERNAME": "",
"PASSWORD": ""
}

View File

@ -0,0 +1,93 @@
# encoding: UTF-8
from __future__ import print_function
import sys
import json
from datetime import datetime
from time import time, sleep
from pymongo import MongoClient, ASCENDING
from vnpy.trader.vtObject import VtBarData
from vnpy.trader.app.ctaStrategy.ctaBase import MINUTE_DB_NAME
import rqdatac as rq
# 加载配置
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] # 数据库
USERNAME = setting['USERNAME']
PASSWORD = setting['PASSWORD']
rq.init(USERNAME, PASSWORD)
FIELDS = ['open', 'high', 'low', 'close', 'volume']
#print (rq.all_instruments(type='Future'))
#print(rq.get_price('IF99', frequency='1m', fields=FIELDS))
#----------------------------------------------------------------------
def generateVtBar(row, symbol):
"""生成K线"""
bar = VtBarData()
bar.symbol = symbol
bar.vtSymbol = symbol
bar.open = row['open']
bar.high = row['high']
bar.low = row['low']
bar.close = row['close']
bar.volume = row['volume']
bar.datetime = row.name
bar.date = bar.datetime.strftime("%Y%m%d")
bar.time = bar.datetime.strftime("%H:%M:%S")
return bar
#----------------------------------------------------------------------
def downMinuteBarBySymbol(symbol):
"""下载某一合约的分钟线数据"""
start = time()
cl = db[symbol]
cl.ensure_index([('datetime', ASCENDING)], unique=True) # 添加索引
df = rq.get_price(symbol, frequency='1m', fields=FIELDS)
for ix, row in df.iterrows():
bar = generateVtBar(row, symbol)
d = bar.__dict__
flt = {'datetime': bar.datetime}
cl.replace_one(flt, d, True)
end = time()
cost = (end - start) * 1000
print(u'合约%s数据下载完成%s - %s,耗时%s毫秒' %(symbol, df.index[0], df.index[-1], cost))
#----------------------------------------------------------------------
def downloadAllMinuteBar():
"""下载所有配置中的合约的分钟线数据"""
print('-' * 50)
print(u'开始下载合约分钟线数据')
print('-' * 50)
# 添加下载任务
for symbol in SYMBOLS:
downMinuteBarBySymbol(str(symbol))
print('-' * 50)
print(u'合约分钟线数据下载完成')
print('-' * 50)

View File

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

View File

@ -0,0 +1,33 @@
# 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)