参考资料及鸣谢
- https://www.vnpy.com/forum/topic/1472-ju-kuan-shu-ju-jddatasdk-ji-cheng-fang-an-qi-chang-da-yi-nian?page=1
- https://www.vnpy.com/forum/topic/7724-vnpy2-6-0-ji-cheng-jqdatafang-an
实验完成时间:2022年3月10日
实验结果:对接成功(见下图)
几个重点注意事项:
1.2022年3月10日的提示信息显示,jqdata数据的price字段只给提供14天;
2.jqdata的数据未和rqdata等其他的数据源进行校对,数据质量不能确保,需要交叉验证。
具体步骤
- 安装jqdata——根据jqdata注册邮件的内容进行安装,pip install jqdatasdk,如果有问题请参考官方链接https://www.joinquant.com/view/community/detail/cdf86c624992fc86ed51d920ef8c637b
- 测试jqdata,确保安装成果且账户名密码正确——
安装成功后,打开代码编辑器,输入如下代码登录JQData
运行后如果显示“auth success”代表安装成功且用户名密码没问题了from jqdatasdk import * auth('账号','密码') #账号是申请时所填写的手机号;密码为聚宽官网登录密码
- 通过程序配置jqdata和vnpy的datafeed模式对接
- 在site-packages目录下新建文件夹vnpy_jqdata
- 在vnpy_jqdata目录下,新建init.py。python新手注意要新建不要复制rqdata文件夹再修改。
- 在vnpy_jqdata目录下,新建jqdata_datafeed.py
具体代码:init.py
import importlib_metadata
from .jqdata_datafeed import JqdataDatafeed as Datafeed
try:
__version__ = importlib_metadata.version("vnpy_jqdata")
except importlib_metadata.PackageNotFoundError:
__version__ = "dev"
具体代码jqdata_datafeed.py
import jqdatasdk as jq
from datetime import timedelta, datetime
import datetime
from typing import List
from vnpy.trader.constant import Exchange, Interval
# from vnpy.trader.mddata.dataapi import MdDataApi
from vnpy.trader.datafeed import BaseDatafeed
from vnpy.trader.object import BarData, HistoryRequest
from vnpy.trader.setting import SETTINGS
from pytz import timezone
CHINA_TZ = timezone("Asia/Shanghai")
INTERVAL_VT2JQ = {
Interval.MINUTE: '1m',
Interval.HOUR: '60m',
Interval.DAILY: '1d',
}
INTERVAL_ADJUSTMENT_MAP_JQ = {
Interval.MINUTE: timedelta(minutes=1),
Interval.HOUR: timedelta(hours=1),
Interval.DAILY: timedelta() # no need to adjust for daily bar
}
class JqdataDatafeed(BaseDatafeed):
"""聚宽JQData客户端封装类"""
def __init__(self):
""""""
self.username = SETTINGS["datafeed.username"]
self.password = SETTINGS["datafeed.password"]
self.inited = False
print("here")
def init(self, username="", password=""):
""""""
if self.inited:
return True
if username and password:
self.username = username
self.password = password
if not self.username or not self.password:
return False
try:
jq.auth(self.username, self.password)
except Exception as ex:
print("jq auth fail:" + repr(ex))
return False
self.inited = True
return True
def to_jq_symbol(self, symbol: str, exchange: Exchange):
"""
CZCE product of RQData has symbol like "TA1905" while
vt symbol is "TA905.CZCE" so need to add "1" in symbol.
"""
if exchange in [Exchange.SSE, Exchange.SZSE]:
if exchange == Exchange.SSE:
jq_symbol = f"{symbol}.XSHG" # 上海证券交易所
else:
jq_symbol = f"{symbol}.XSHE" # 深圳证券交易所
elif exchange == Exchange.SHFE:
jq_symbol = f"{symbol}.XSGE" # 上期所
elif exchange == Exchange.CFFEX:
jq_symbol = f"{symbol}.CCFX" # 中金所
elif exchange == Exchange.DCE:
jq_symbol = f"{symbol}.XDCE" # 大商所
elif exchange == Exchange.INE:
jq_symbol = f"{symbol}.XINE" # 上海国际能源期货交易所
elif exchange == Exchange.CZCE:
# 郑商所 的合约代码年份只有三位
for count, word in enumerate(symbol):
if word.isdigit():
break
# Check for index symbol
time_str = symbol[count:]
if time_str in ["88", "888", "99", "8888", "9999"]:
return f"{symbol}.XZCE"
# noinspection PyUnboundLocalVariable
product = symbol[:count]
year = symbol[count]
month = symbol[count + 1:]
if year == "9":
year = "1" + year
else:
year = "2" + year
jq_symbol = f"{product}{year}{month}.XZCE"
return jq_symbol.upper()
def query_bar_history(self, req: HistoryRequest):
"""
Query history bar data from JQData.
"""
symbol = req.symbol
exchange = req.exchange
interval = req.interval
start = req.start
end = req.end
jq_symbol = self.to_jq_symbol(symbol, exchange)
# if jq_symbol not in self.symbols:
# return None
jq_interval = INTERVAL_VT2JQ.get(interval)
if not jq_interval:
return None
# For adjust timestamp from bar close point (RQData) to open point (VN Trader)
# adjustment = INTERVAL_ADJUSTMENT_MAP_JQ.get(interval)
adjustment = INTERVAL_ADJUSTMENT_MAP_JQ[interval]
# For querying night trading period data
end += timedelta(1)
df = jq.get_price(
jq_symbol,
frequency=jq_interval,
fields=["open", "high", "low", "close", "volume"],
start_date=start,
end_date=end,
skip_paused=True,
)
data: List[BarData] = []
if df is not None:
for ix, row in df.iterrows():
bar = BarData(
symbol=symbol,
exchange=exchange,
interval=interval,
datetime=row.name.to_pydatetime() - adjustment,
open_price=row["open"],
high_price=row["high"],
low_price=row["low"],
close_price=row["close"],
volume=row["volume"],
gateway_name="JQ"
)
data.append(bar)
return data
- 修改vnpy的datafeed配置为jqdata
点击顶部菜单栏的“配置”,然后参考下图进行
保存后,按提示关闭trader,重新从vnpy的station打开trader,回看到成功的提示信息
- 测试下载数据并回测
点击左边栏的“CTA回测”,修改本地代码为“IF9999.CFFEX”或其他有效的本地代码(注意默认的本地代码是不行的,估计因为jqdata修改了数据规则)。然后点击“下载数据”,会看到类似下面的
最后会提示“历史数据加载完成,数据量:175200”诸如此类的信息
点击回测按钮,会看到类似下图的,代表完全成功了