量化掘金数据源应该是最近比较好的数据源了,免费版提供了5天的tick,6个月的分钟bar和无限的日线。
这个比之前聚宽的数据源还要简单,主要就是要在本地运行掘金的本地客户端,有三点注意:
- 分钟bar提供了bob和boe,bar开始时间和bar结束数据,bob是和vnpy一致
- 掘金的合约名称是 exchange.symbol 和vnpy的 symbol.exchange对调一下就可以,比起聚宽简单。
- 掘金的时间已经带了时区数据更加简单。
下面是代码,只有分钟的,tick其实基本差不多,是在vnpy2.x的版本用的。,直接替换或者模仿值聚宽的也可以。还有就是rqdata.username用来做掘金的token填入,注意。
from datetime import datetime
from typing import List
from gm.api import *
from tzlocal import get_localzone
from vnpy.trader.constant import Exchange, Interval
from vnpy.trader.mddata.dataapi import MdDataApi
from vnpy.trader.object import BarData, HistoryRequest
from vnpy.trader.setting import SETTINGS
INTERVAL_VT2GM = {
Interval.MINUTE: "60s",
Interval.HOUR: "3600s",
Interval.DAILY: "1d",
}
class GmdataClient(MdDataApi):
"""掘金GMData客户端封装类"""
def __init__(self):
""""""
# 加载配置
self.username = SETTINGS["rqdata.username"]
self.inited = False
self.inited = False
def init(self, username="", password=""):
""""""
if self.inited:
return True
try:
set_token(username)
except Exception as ex:
print("gm auth fail:" + repr(ex))
return False
self.inited = True
return True
def to_vn_symbol(symbol_exchange):
"""
gm symbol list
['CZCE.CY501', 'DCE.eb2411', 'DCE.m2501'', 'CFFEX.TL2412', 'DCE.c2501']
"""
exchange, symbol = symbol_exchange.split(".")
vt_symbol = f"{symbol}.{exchange}"
return vt_symbol
def to_gm_symbol(symbol: str, exchange: Exchange):
"""
"""
gm_symbol = f"{exchange.value}.{symbol}"
return gm_symbol
def get_dominant_future(self, symbol: str):
return "gm主力合约查找有待开发"
def query_history(self, req: HistoryRequest):
"""
Query history bar data from GMData.
"""
symbol = req.symbol
exchange = req.exchange
interval = req.interval
start = req.start
end = req.end
gm_symbol = self.to_gm_symbol(symbol, exchange)
gm_interval = INTERVAL_VT2GM.get(interval)
if not gm_interval:
return None
# adjustment = INTERVAL_ADJUSTMENT_MAP_GM.get(interval)
if start > end:
return
now = datetime.now(get_localzone())
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']
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)
data: List[BarData] = []
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"
)
data.append(bar)
return data
gmdata_client = GmdataClient()