量化交易之回测篇 - 添加获取天勤合约数据api

from tqsdk import TqApi, TqAuth
from public_module.tqz_extern.tools.pandas_operator.pandas_operator import pandas  # noqafrom server_api.tqz_object import BarData, ExchangeTIME_GAP = 8 * 60 * 60 * 1000000000class TQZTianQinClient:"""天勤接口 每次只能拉取单一合约的数据！"""__tq_symbols = Nonedef __init__(self, account: str = "account", pass_word: str = "password"):self.api = TqApi(auth=TqAuth(account, pass_word))if TQZTianQinClient.__tq_symbols is None:TQZTianQinClient.__tq_symbols = self.api.query_quotes(ins_class="FUTURE", expired=False)def query_quote(self, tq_symbol: str) -> dict:result = self.api.get_quote(symbol=tq_symbol)self.api.close()return result  # noqadef query_history_bars(self, tq_symbol: str, tq_duration_seconds: int, tq_data_length: int = 8964) -> list:assert tq_symbol in TQZTianQinClient.__tq_symbols, f'bad tq_symbol: {tq_symbol}'tq_result = self.api.get_kline_serial(symbol=tq_symbol, duration_seconds=tq_duration_seconds, data_length=tq_data_length)self.api.close()tq_result["datetime"] = pandas.to_datetime(tq_result["datetime"] + TIME_GAP)tq_result['datetime'] = tq_result['datetime'].apply(lambda x: x.strftime('%Y-%m-%d %H:%M:%S'))  # %f是毫秒symbol, exchange = TQZTianQinClient.__get_symbol_exchange(tq_symbol=tq_symbol)history_bars = []if tq_result is not None:for ix, row in tq_result.loc[tq_result["id"] >= 0].iterrows():history_bars.append(BarData(symbol=symbol,exchange=exchange,interval='any_interval',  # noqadatetime=row["datetime"],open_price=row["open"],high_price=row["high"],low_price=row["low"],close_price=row["close"],volume=row["volume"],open_interest=row.get("open_oi", 0),gateway_name="TQ",))return history_barsdef query_index_history_bars(self, tq_index_symbol: str, tq_duration_seconds: int, tq_data_length: int = 8964) -> list:tq_result = self.api.get_kline_serial(symbol=tq_index_symbol, duration_seconds=tq_duration_seconds, data_length=tq_data_length)self.api.close()tq_result["datetime"] = pandas.to_datetime(tq_result["datetime"] + TIME_GAP)tq_result['datetime'] = tq_result['datetime'].apply(lambda x: x.strftime('%Y-%m-%d %H:%M:%S'))  # %f是毫秒symbol, exchange = f'{tq_index_symbol.split(".")[2]}000', Exchange.TQZ_INDEXhistory_bars = []if tq_result is not None:for ix, row in tq_result.loc[tq_result["id"] >= 0].iterrows():history_bars.append(BarData(symbol=symbol,exchange=exchange,interval='any_interval',  # noqadatetime=row["datetime"],open_price=row["open"],high_price=row["high"],low_price=row["low"],close_price=row["close"],volume=row["volume"],open_interest=row.get("open_oi", 0),gateway_name="TQ",))return history_barsdef query_main_history_bars(self, tq_main_symbol: str, tq_duration_seconds: int, tq_data_length: int = 8964) -> list:tq_result = self.api.get_kline_serial(symbol=tq_main_symbol, duration_seconds=tq_duration_seconds, data_length=tq_data_length)self.api.close()tq_result["datetime"] = pandas.to_datetime(tq_result["datetime"] + TIME_GAP)tq_result['datetime'] = tq_result['datetime'].apply(lambda x: x.strftime('%Y-%m-%d %H:%M:%S'))  # %f是毫秒symbol, exchange = f'{tq_main_symbol.split(".")[2]}888', Exchange.TQZ_MAINhistory_bars = []if tq_result is not None:for ix, row in tq_result.loc[tq_result["id"] >= 0].iterrows():history_bars.append(BarData(symbol=symbol,exchange=exchange,interval='any_interval',  # noqadatetime=row["datetime"],open_price=row["open"],high_price=row["high"],low_price=row["low"],close_price=row["close"],volume=row["volume"],open_interest=row.get("open_oi", 0),gateway_name="TQ",))return history_barsdef load_all_tq_symbols(self):self.api.close()return TQZTianQinClient.__tq_symbols# --- private part ---@classmethoddef __get_symbol_exchange(cls, tq_symbol: str) -> (str, Exchange):exchange_str, symbol = tq_symbol.split('.')[0], tq_symbol.split('.')[1]if exchange_str in [Exchange.SHFE.value]:exchange = Exchange.SHFEelif exchange_str in [Exchange.INE.value]:exchange = Exchange.INEelif exchange_str in [Exchange.DCE.value]:exchange = Exchange.DCEelif exchange_str in [Exchange.CZCE.value]:exchange = Exchange.CZCEelif exchange_str in [Exchange.CFFEX.value]:exchange = Exchange.CFFEXelse:assert False, f'bad exchange_str: {exchange_str}'return symbol, exchangeclass TQZTianQinDataManager:"""小时线: 一天6根1493个交易日30分钟线: 一天12根746个交易日10分钟线: 一天36根248个交易日"""__load_history_bars_map: dict = None__load_history_index_bars_map: dict = None__load_history_main_bars_map: dict = None@classmethoddef load_history_bars_map(cls, tq_symbols: list, tq_duration_seconds: int, tq_data_length: int = 8964) -> dict:"""Load history bars of multi symbols from tianqin, and only load once time before back tester.:param tq_symbols: list of multi symbols:param tq_duration_seconds: just tq_duration_seconds:param tq_data_length: data length:return: history bars map"""load_history_bars_map = {}if cls.__load_history_bars_map is None:for tq_symbol in tq_symbols:if tq_symbol not in load_history_bars_map.keys():load_history_bars_map[tq_symbol] = TQZTianQinClient().query_history_bars(tq_symbol=tq_symbol,tq_duration_seconds=tq_duration_seconds,tq_data_length=tq_data_length)cls.__load_history_bars_map = load_history_bars_mapreturn cls.__load_history_bars_map@classmethoddef load_history_index_bars_map(cls, tq_index_symbols: list, tq_duration_seconds: int, tq_data_length: int = 8964) -> dict:"""Load history bars of multi index symbols from tianqin, and only load once time before back tester.:param tq_index_symbols: list of multi symbols:param tq_duration_seconds: just tq_duration_seconds:param tq_data_length: data length:return: history index bars map"""load_history_index_bars_map = {}if cls.__load_history_index_bars_map is None:for tq_index_symbol in tq_index_symbols:if tq_index_symbol not in load_history_index_bars_map.keys():load_history_index_bars_map[tq_index_symbol] = TQZTianQinClient().query_index_history_bars(tq_index_symbol=tq_index_symbol,tq_duration_seconds=tq_duration_seconds,tq_data_length=tq_data_length)cls.__load_history_index_bars_map = load_history_index_bars_mapreturn cls.__load_history_index_bars_map@classmethoddef load_history_main_bars_map(cls, tq_main_symbols: list, tq_duration_seconds: int, tq_data_length: int = 8964) -> dict:"""Load history bars of multi main symbols from tianqin, and only load once time before back tester.:param tq_main_symbols: list of multi symbols:param tq_duration_seconds: just tq_duration_seconds:param tq_data_length: data length:return: history main bars map"""load_history_main_bars_map = {}if cls.__load_history_main_bars_map is None:for tq_main_symbols in tq_main_symbols:if tq_main_symbols not in load_history_main_bars_map.keys():load_history_main_bars_map[tq_main_symbols] = TQZTianQinClient().query_main_history_bars(tq_main_symbol=tq_main_symbols,tq_duration_seconds=tq_duration_seconds,tq_data_length=tq_data_length)cls.__load_history_main_bars_map = load_history_main_bars_mapreturn cls.__load_history_main_bars_mapif __name__ == '__main__':""" contract datacontent = TQZTianQinDataManager.load_history_bars_map(tq_symbols=["CZCE.SM206", "SHFE.rb2205"], tq_duration_seconds=60 * 60)print("content: " + str(content))"""""" index datacontent = TQZTianQinDataManager.load_history_index_bars_map(tq_index_symbols=["KQ.i@SHFE.bu"], tq_duration_seconds=60 * 60, tq_data_length=3)print("content: " + str(content))"""""" main datacontent = TQZTianQinDataManager.load_history_main_bars_map(tq_main_symbols=["KQ.m@SHFE.ag"], tq_duration_seconds=60 * 60, tq_data_length=15)print("content: " + str(content))"""pass

量化交易之回测篇 - 添加获取天勤合约数据api相关推荐

量化交易之回测篇 - 海龟交易策略（初版）
""" 写这个模型花了小半天, 还没来得及测试, 先开源出来; 等期货回测器的cta分支没问题后,再测试该模型; PS: 为了方便模型的回测, 价格设置上确实有不合理的地 ...
量化交易之回测篇 - 重写vnpy自带的双均线策略
"""双均线策略作为测试期货回测器的一个测试策略, 在测试品种ag时出现了一个问题:回测时ag净值曲线无最大回撤值,说明ag在双均线策略上无回撤,理论上这是不可能的; 排 ...
量化交易之回测篇 - 拉取合成历史沉淀资金数据（主连合约）
""" 1. 期货品种的沉淀资金是未平仓的期货合约(持仓量)所占用的保证金. 期货品种的沉淀资金越大说明这个品种非常受资金喜欢. 相反,如果期货品种的沉淀资金越小,说明这 ...
量化交易简易回测框架
量化交易简易回测框架 import datetime import osimport dateutil import matplotlib.pyplot as plt import pandas a ...
量化交易-Backtrader回测
目录一.基础篇-backtrader简介二.backtrader回测应用实例三.backtrader数据全解析 3.1 数据中必须包含 3.2 一切皆是数据源 3.3 参数 3.4 Lines线 ...
量化交易12-backtrader回测三日定律K线形态图
三日定律: 三日定律其实是来源与乔治·道格拉斯·泰勒的"预约法",在后来它就被演变成了LSS三日周期法.泰勒说,市场波动最开始就是从内部开始驱动的,有内部消息者或者聪明人最先买进, ...
量化交易9-backtrader回测十字星K线形态图
十字星形态图,在股市中的应用: 十字星是一种K线基本型态.十字星,是一种只有上下影线,没有实体的K线图.开盘价即是收盘价,表示在交易中,股价出现高于或低于开盘价成交,但收盘价与开盘价相等.其中:上影线 ...
Python数字货币量化交易开发——回测系统的优化与云服务器部署
前言随着回测系统的数据量越来越大,为了解放个人电脑,决定将回测系统部署到云服务器. _ _ Episode Sp. 回测系统的优化与云服务器部署之前说到个人回测系统建立出来后为了有效和一些云回测平 ...
量化交易15-backtrader回测南方三星、三个白兵、脱离、藏婴吞没、弃婴、捉腰带线、反击线等形态
介绍一下上一章节所写的形态,在股市理论的描述,笔者从各个地方整理而来,地址也在形态中描述了. CDL3STARSINSOUTH - 南方三星(筑底上涨) 南方三星K线组合应该算是很少见到的K线形态,但 ...

量化交易之回测篇 - 添加获取天勤合约数据api

量化交易之回测篇 - 添加获取天勤合约数据api相关推荐

最新文章

热门文章