掘金量化数据的提供历史bar和tick历史股票、期货、基金等数据接口，包含免费和付费bar和tick数据读取函数，其中bar可以提供以秒为单位的，tick数据一次最多可以读取5天的，付费用户能够提供更长时间的数据读取。免费的已经够用已经可以满足我们量化策略对历史数据的基本需求。
1、安装掘金量化数据的python SDK
这一步是必须的，安装完成后，您的lib\site_packages会多出一个gm子目录，其中包含的是掘金量化数据的python SDK，这是我们实现。
如果未使用uv等包管理器，使用下面命令：
pip install gm -i https://mirrors.aliyun.com/pypi/simple/ -U如果使用uv包管理器，使用下面命令：
uv pip install gm -i https://mirrors.aliyun.com/pypi/simple/ -U
2、创建vnpy_gmdata
- 创建vnpy_gmdata文件夹
- 添加init.py到vnpy_gmdata目录下，内容如下：
# The MIT License (MIT)
#
# Copyright (c) 2015-present, Xiaoyou Chen
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
import importlib_metadata
from .gm_datafeed import GmDatafeed as Datafeed
try:
__version__ = importlib_metadata.version("vnpy_gmdata")
except importlib_metadata.PackageNotFoundError:
__version__ = "dev"
- 添加gm_datafeed.py到vnpy_gmdata目录下，内容如下：
from datetime import datetime
from typing import List, Callable
from gm.api import *
from tzlocal import get_localzone
from vnpy.trader.utility import ZoneInfo # hxxjava add
from vnpy.trader.constant import Exchange, Interval
from vnpy.trader.datafeed import BaseDatafeed
from vnpy.trader.object import BarData, TickData, HistoryRequest
from vnpy.trader.setting import SETTINGS
from datetime import timedelta
INTERVAL_VT2GM = {
Interval.MINUTE: "60s",
Interval.HOUR: "3600s",
Interval.DAILY: "1d",
}
def to_gm_symbol(symbol: str, exchange: Exchange) -> str:
"""
把合约名称与交易所构建成为掘金的格式：‘交易所大写.合约名称小写’
"""
gm_symbol = f"{exchange.value.upper()}.{symbol.lower()}"
return gm_symbol
class GmDatafeed(BaseDatafeed):
"""掘金GMData客户端封装类"""
def __init__(self):
""""""
# 加载配置
self.username = SETTINGS["datafeed.username"]
self.inited = False
self.init();
# print(f"【gm 1 token: {self.username}, inited = {self.inited} 】")
def init(self, output: Callable = print) -> bool:
""""""
if self.inited:
return True
try:
set_token(self.username)
output("gmdata inited !")
except Exception as ex:
output("gm auth fail:" + repr(ex))
return False
self.inited = True
return True
def to_vn_symbol(self, EXCHANGE_symbol:str) -> str:
"""
gm symbol list
['CZCE.CY501', 'DCE.eb2411', 'DCE.m2501'', 'CFFEX.TL2412', 'DCE.c2501']
"""
EXCHANGE, symbol = EXCHANGE_symbol.split(".")
vt_symbol = f"{symbol.lower()}.{EXCHANGE.upper()}"
return vt_symbol
def get_dominant_future(self, symbol: str):
return "gm主力合约查找有待开发"
def query_bar_history(self, req: HistoryRequest, output: Callable = print) -> list[BarData]:
"""
Query history bar data from GMData.
"""
bars: List[BarData] = []
shanghai_tz = ZoneInfo("Asia/Shanghai") # hxxjava
# print(f"[1 req={req}]")
symbol = req.symbol
exchange = req.exchange
interval = req.interval
start = req.start.replace(tzinfo=shanghai_tz) # hxxjava
end = req.end.replace(tzinfo=shanghai_tz) # hxxjava
# print(f"【gm 2 start= {start} end={end}")
gm_symbol = to_gm_symbol(symbol, exchange)
# print(f"[2 gm_symbol={gm_symbol}]")
gm_interval = INTERVAL_VT2GM.get(interval)
if not gm_interval:
return bars
# adjustment = INTERVAL_ADJUSTMENT_MAP_GM.get(interval)
if start > end:
return bars
# now = datetime.now(get_localzone())
now = datetime.now().replace(tzinfo=shanghai_tz) # hxxjava
if end >= now:
end = now
elif end.year == now.year and end.month == now.month and end.day == now.day:
end = now
fields = ['open', 'close', 'low', 'high', 'volume', 'amount', "position",'bob']
try:
df = history(symbol = gm_symbol, frequency = gm_interval, start_time = start, end_time = end, fields=fields , skip_suspended=True,
fill_missing=None, adjust=ADJUST_NONE, adjust_end_time='', df=True)
except Exception as ex:
output(f"[3 df={ex}]")
return bars
if df is not None:
for ix, row in df.iterrows():
dt = row["bob"].to_pydatetime()
# dt = CHINA_TZ.localize(dt)
bar = BarData(
symbol=symbol,
exchange=exchange,
interval=interval,
datetime=dt,
open_price=row["open"],
high_price=row["high"],
low_price=row["low"],
close_price=row["close"],
open_interest= row["position"],
volume=row["volume"],
turnover= row['amount'],
gateway_name="GM"
)
bars.append(bar)
# print(f"【gm 3 bars = {bars}") # hxxjava
return bars
def query_tick_history(self, req: HistoryRequest, output: Callable = print) -> list[TickData]:
"""查询Tick数据"""
if not self.inited:
n: bool = self.init(output)
if not n:
return []
symbol: str = req.symbol
exchange: Exchange = req.exchange
start: datetime = req.start
end: datetime = req.end
# 股票期权不添加交易所后缀
gm_symbol: str = to_gm_symbol(symbol,exchange)
fields = [
'symbol','open','high','low','price',
'cum_volume','cum_amount','iopv','last_amount','last_volume','quotes',
'cum_position','trade_type','flag','created_at'
]
# print(f"gm_symbol={gm_symbol},start={start},end={end},fields={fields}")
if (end > start + timedelta(days=180)):
print(f"读取的时间太长，读取失败!")
return []
try:
history_data : list = history(symbol=gm_symbol, frequency='tick', start_time=start, end_time = end, fields=fields, skip_suspended=True,
fill_missing=None, adjust=ADJUST_NONE, adjust_end_time='', df=False)
except Exception as ex:
output(f"读取历史tick错误！{ex}")
return ticks
ticks: List[TickData] = []
for td in history_data:
dt: datetime = td["created_at"]
# dt: datetime = dt.replace(tzinfo=CHINA_TZ)
tick : TickData = TickData(
symbol=symbol,
exchange=exchange,
datetime=dt,
open_price = td['open'],
high_price = td['high'],
low_price = td['low'],
last_price = td['price'],
turnover = td['last_amount'],
last_volume= td['last_volume'],
bid_price_1 = td['quotes'][0]['bid_p'],
bid_volume_1 = td['quotes'][0]['bid_v'],
ask_price_1 = td['quotes'][0]['ask_p'],
ask_volume_1 = td['quotes'][0]['ask_v'],
open_interest=td['cum_position'],
gateway_name="GM"
)
ticks.append(tick)
return ticks
3、修改用户目录下C:\users\<用户名>\ .vntrader\vt_setting.json的全局配置项目
方法1：
如下图所示，vt_setting.json文件中有关datafeed的配置项为"datafeed.name": "gmdata", "datafeed.username": "your token"，"datafeed.password": ""，其中your token为你的掘金账户的token：
{
"font.family": "微软雅黑",
"font.size": 12,
"log.active": true,
"log.level": 50,
"log.console": true,
"log.file": true,
"email.server": "smtp.qq.com",
"email.port": 465,
"email.username": "xxxxx@qq.com",
"email.password": "xxxxx",
"email.sender": "xxxxx@qq.com",
"email.receiver": "xxxxxx@qq.com",
"datafeed.name": "gmdata",
"datafeed.username": "your token",
"datafeed.password": "",
"database.timezone": "Asia/Shanghai",
"database.name": "mysql",
"database.database": "vnpy",
"database.host": "localhost",
"database.port": 3306,
"database.user": "root",
"database.password": "xxxxx"
}
方法2
在vntrader的菜单的配置中有关datafeed的选项做如下配置，其中datafeed.name=rqdata，保证datafeed实现匹配vnpy_gmdata模块，datafeed.username="xxxxx....xxxxxx"，为用户申请的掘金数据接口API的token。
掘金数据接口API的token获取方法：从https:www.myquant.cn下载掘金3基础版本，进入系统设置界面，复制其中的“密钥管理（Token）”，填写到我们的datafeed.username中：
4、测试掘金datafeed是否成功
启动vntrader，从主界面中点击数据管理图标进入数据管理界面，然后点击“下载数据”按钮，填写要选择的合约名称、交易所、周期为Minute和开始时间，等待一段时间后就可以完成指定合约的历史数据。数据完成后点击“刷新”按钮刷新已经下载的历史数据，就可以查询到已经下载数据。
至此，可以证明用掘金数据替代米宽的datafeed已经替换成功。
