前言
上次我们介绍了一下vnpy量化框架的搭建，今天我们来说说TuShare数据源的接入。因为公司之前一直是从一些金融网站或者证券服务商获取的，公司最近决定改变策略通过TuShare获取。其实在VeighNa开源网站上官方也提供了一些数据服务商。本着能白嫖绝不花钱的准则，就试着接入下。接下来我们开始进入正题吧
账号注册
-
通过
TuShare官网进入，注册并登陆个人账户，进入个人主页
) -
积分查询与获取
新用户的积分是100，通过更新
个人资料，可以增加20的积分
-
积分其他获取方式
-
分享
通过分享推荐链接，成功注册一个有效用户(指真正会使用tushare数据的用户)可获得50积分，虚假用户带来的积分会被定期回收！
-
捐助（付费）
通过捐助社区捐助一定的金额可提高个人的积分。捐助积分与频次关系图如下：
积分数 每分钟频次 每天总量上限 可以访问的接口 捐助（元） 120 50 8000次 股票基础信息、股票非复权日线行情，其他接口无法调取 0 2000以上 200 100000次/个API tushare.pro 60%的API可以调取，可参考每个接口的积分要求 200 5000以上 500 常规数据无上限 tushare.pro 90%的API可以调取，可参考每个接口的积分要求 500 10000以上 1000 常规数据无上限，特色数据300次每分钟 特色数据权限，包括盈利预测数据、每日筹码和胜率、筹码分布、券商每月金股等数据 1000 15000以上 1000 特色数据无总量限制 特色数据专属权限 1500 10000积分以上可以有更高的API频次和权限，比如股票特色数据
此外，分钟和港美股数据权限不在积分范畴内，各类分钟单独开权限
类型 包含数据 历史起始 捐助（元） 频次 沪深A股 1、5、15、30、60分钟 2009年 1000 每分钟500次，每次8000行数据，总量不限制 期货 同上 2010年 1000 同上 期权 同上 2010年 2000，包含股指和商品期权 同上 港股 日线 全历史 1000 每分钟500次，每次3000行，总量不限制 美股 日线，包含估值指标、换手率等 全股票全历史 1000 每分钟500次，每次6000行，总量不限制 数字货币 1~60分钟 2020年开始 1000 每分钟500次，每次8000行，总量不限制 新闻资讯 快讯、长篇新闻、新闻联播、公司公告 3年以上 1000 每分钟400次，总量不限制 -
高校师生，确认身份后，可以免费获得2000积分（根据实际情况可以调整）
-
参与社区贡献，比如提交数据问题、参与数据贡献、编写文章发给群主或积分管理员
-
-
接口调用与测试
在本地编写
Python测试脚本import tushare as ts ts.set_token('你的token') pro = ts.pro_api() df = pro.fut_daily(ts_code='CU1811.SHF', start_date='20180101', end_date='20181113') print(df)
-
接入
vnpy-
安装
pip install vnpy_tushare -
在
vnpy项目根目录下创建vnpy_tushare\__init__.py
import importlib_metadata from .tushare_datafeed import TushareDatafeed as Datafeed try: __version__ = importlib_metadata.version("vnpy_tushare") except importlib_metadata.PackageNotFoundError: __version__ = "dev"vnpy_tushare\tushare_datafeed.py
from datetime import timedelta, datetime from typing import Dict, List, Optional from copy import deepcopy import pandas as pd from pandas import DataFrame import tushare as ts from tushare.pro.client import DataApi from vnpy.trader.setting import SETTINGS from vnpy.trader.datafeed import BaseDatafeed from vnpy.trader.constant import Exchange, Interval from vnpy.trader.object import BarData, HistoryRequest from vnpy.trader.utility import round_to, ZoneInfo # 数据频率映射 INTERVAL_VT2TS: Dict[Interval, str] = { Interval.MINUTE: "1min", Interval.HOUR: "60min", Interval.DAILY: "D", } # 股票支持列表 STOCK_LIST: List[Exchange] = [ Exchange.SSE, Exchange.SZSE, Exchange.BSE, ] # 期货支持列表 FUTURE_LIST: List[Exchange] = [ Exchange.CFFEX, Exchange.SHFE, Exchange.CZCE, Exchange.DCE, Exchange.INE, ] # 交易所映射 EXCHANGE_VT2TS: Dict[Exchange, str] = { Exchange.CFFEX: "CFX", Exchange.SHFE: "SHF", Exchange.CZCE: "ZCE", Exchange.DCE: "DCE", Exchange.INE: "INE", Exchange.SSE: "SH", Exchange.SZSE: "SZ", } # 时间调整映射 INTERVAL_ADJUSTMENT_MAP: Dict[Interval, timedelta] = { Interval.MINUTE: timedelta(minutes=1), Interval.HOUR: timedelta(hours=1), Interval.DAILY: timedelta() } # 中国上海时区 CHINA_TZ = ZoneInfo("Asia/Shanghai") def to_ts_symbol(symbol, exchange) -> Optional[str]: """将交易所代码转换为tushare代码""" # 股票 if exchange in STOCK_LIST: ts_symbol: str = f"{symbol}.{EXCHANGE_VT2TS[exchange]}" # 期货 elif exchange in FUTURE_LIST: if exchange is not Exchange.CZCE: ts_symbol: str = f"{symbol}.{EXCHANGE_VT2TS[exchange]}".upper() else: for count, word in enumerate(symbol): if word.isdigit(): break year: str = symbol[count] month: str = symbol[count + 1:] if year == "9": year = "1" + year else: year = "2" + year product: str = symbol[:count] ts_symbol: str = f"{product}{year}{month}.ZCE".upper() else: return None return ts_symbol def to_ts_asset(symbol, exchange) -> Optional[str]: """生成tushare资产类别""" # 股票 if exchange in STOCK_LIST: if exchange is Exchange.SSE and symbol[0] == "6": asset: str = "E" elif exchange is Exchange.SZSE and symbol[0] == "0" or symbol[0] == "3": asset: str = "E" else: asset: str = "I" # 期货 elif exchange in FUTURE_LIST: asset: str = "FT" else: return None return asset class TushareDatafeed(BaseDatafeed): """TuShare数据服务接口""" def __init__(self): """""" self.username: str = SETTINGS["datafeed.username"] self.password: str = SETTINGS["datafeed.password"] self.inited: bool = False def init(self) -> bool: """初始化""" if self.inited: return True ts.set_token(self.password) self.pro: Optional[DataApi] = ts.pro_api() self.inited = True return True def query_bar_history(self, req: HistoryRequest) -> Optional[List[BarData]]: """查询k线数据""" if not self.inited: self.init() symbol: str = req.symbol exchange: Exchange = req.exchange interval: Interval = req.interval start: datetime = req.start.strftime("%Y-%m-%d %H:%M:%S") end: datetime = req.end.strftime("%Y-%m-%d %H:%M:%S") ts_symbol: str = to_ts_symbol(symbol, exchange) if not ts_symbol: return None asset: str = to_ts_asset(symbol, exchange) if not asset: return None ts_interval: str = INTERVAL_VT2TS.get(interval) if not ts_interval: return None adjustment: timedelta = INTERVAL_ADJUSTMENT_MAP[interval] try: d1: DataFrame = ts.pro_bar( ts_code=ts_symbol, start_date=start, end_date=end, asset=asset, freq=ts_interval ) except IOError: return [] df: DataFrame = deepcopy(d1) while True: if len(d1) != 8000: break tmp_end: str = d1["trade_time"].values[-1] d1 = ts.pro_bar( ts_code=ts_symbol, start_date=start, end_date=tmp_end, asset=asset, freq=ts_interval ) df = pd.concat([df[:-1], d1]) bar_keys: List[datetime] = [] bar_dict: Dict[datetime, BarData] = {} data: List[BarData] = [] # 处理原始数据中的NaN值 df.fillna(0, inplace=True) if df is not None: for ix, row in df.iterrows(): if row["open"] is None: continue if interval.value == "d": dt: str = row["trade_date"] dt: datetime = datetime.strptime(dt, "%Y%m%d") else: dt: str = row["trade_time"] dt: datetime = datetime.strptime(dt, "%Y-%m-%d %H:%M:%S") - adjustment dt = dt.replace(tzinfo=CHINA_TZ) turnover = row.get("amount", 0) if turnover is None: turnover = 0 open_interest = row.get("oi", 0) if open_interest is None: open_interest = 0 bar: BarData = BarData( symbol=symbol, exchange=exchange, interval=interval, datetime=dt, open_price=round_to(row["open"], 0.000001), high_price=round_to(row["high"], 0.000001), low_price=round_to(row["low"], 0.000001), close_price=round_to(row["close"], 0.000001), volume=row["vol"], turnover=turnover, open_interest=open_interest, gateway_name="TS" ) bar_dict[dt] = bar bar_keys: list = bar_dict.keys() bar_keys = sorted(bar_keys, reverse=False) for i in bar_keys: data.append(bar_dict[i]) return data-
使用
在VeighNa中使用TuShare时，需要在全局配置中填写以下字段信息：
名称 含义 必填 举例 datafeed.name 名称 是 tushare datafeed.username 用户名 否 token datafeed.password 密码 是 c3a110417f08f26d2c221edc0c50d4a8a5001502eea89cf5
-
-
测试
-
查看配置项
这里的数据库选择默认的sqlite
-
填写参数
周期没有tick级的
-
下载数据
数据库查询
-
开始回测
-