from tqsdk import TqApi, TqAuth
from tqsdk.tafunc import time_to_datetime
import datetime
from datetime import datetime
import pandas
import re
import os
import warnings
warnings.filterwarnings('ignore')
pandas.set_option('display.max_columns', None)
pandas.set_option('display.max_rows', None)
from tqz_extern.json_operator import TQZJsonOperator
TIME_GAP = 8 * 60 * 60 * 1000000000
class TQZTianQinClient:
"""
天勤接口 每次只能拉取单一合约的数据!
"""
__tq_futures = None
def __init__(self, account: str = "xxxxxxx", pass_word: str = "xxxxxxx"):
self.api = TqApi(auth=TqAuth(account, pass_word))
if TQZTianQinClient.__tq_futures is None:
TQZTianQinClient.__tq_futures = self.api.query_quotes(ins_class="FUTURE", expired=False)
def query_history_ticks(self, tq_future: str, tq_data_length: int = 8964):
assert tq_future in TQZTianQinClient.__tq_futures, f'bad tq_future: {tq_future}'
tq_result = self.api.get_tick_serial(symbol=tq_future, data_length=tq_data_length)
self.api.close()
tq_result['datetime_format'] = tq_result['datetime'].apply(time_to_datetime)
return tq_result
def query_single_quote(self, tq_future: str) -> dict:
result = self.api.get_quote(symbol=tq_future)
self.api.close()
return result # noqa
def query_history_bars(self, tq_future: str, tq_duration_seconds: int, tq_data_length: int = 8964):
assert tq_future in TQZTianQinClient.__tq_futures, f'bad tq_future: {tq_future}'
tq_result = self.api.get_kline_serial(symbol=tq_future, 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是毫秒
return tq_result
def load_all_tq_futures(self):
self.api.close()
return TQZTianQinClient.__tq_futures
def load_all_tq_main_futures(self):
tq_main_contracts = self.api.query_quotes(ins_class="CONT")
main_vt_symbols = []
[main_vt_symbols.append(
self.api.get_quote(
symbol=main_contract
).underlying_symbol
) for main_contract in tq_main_contracts]
self.api.close()
return main_vt_symbols
@staticmethod
def record_data():
api = TqApi(auth=TqAuth('xxxxxxx', 'xxxxxxx'))
instruments = ["SHFE.cu2305", "DCE.i2309"]
quote_map = {}
data_map = {} # 之后考虑写共享内存里
for instrument in instruments:
quote_map[instrument] = api.get_quote(symbol=instrument)
data_map[instrument] = pandas.DataFrame()
while api.wait_update():
for instrument in instruments:
if api.is_changing(quote_map[instrument]):
# print(quote_map[instrument])
data_map[instrument].append(vars(quote_map[instrument]), ignore_index=True)
if datetime.datetime.now().hour == 15:
today = str(datetime.date.today()).replace("-", "")
for instrument in instruments:
csv_path = f'./config/{instrument.replace(".", "_")}_{today}.csv'
if os.path.exists(path=csv_path) is False:
data_map[instrument].to_csv(
csv_path,
index=False
)
break # 当前脚本, 每天晚盘前需要手工启动脚本
def __multi_symbols_day_bar(self):
klines = self.api.get_kline_serial(["SHFE.rb2305", "SHFE.rb2310"], duration_seconds=60 * 60 * 24, data_length=5)
self.api.close()
klines_filter = klines.loc[klines['id'] >= 0]
klines_filter['datetime_format'] = klines_filter['datetime'].apply(time_to_datetime)
print("klines_filter: " + str(klines_filter))
class TQZAutoMakeConfig:
__trading_time_config = './config/trading_time.json'
@classmethod
def auto_make_session_json(cls):
all_tq_main_futures = TQZTianQinClient().load_all_tq_main_futures()
ret = {}
for tq_main_future in all_tq_main_futures:
_tq_future = f'{tq_main_future.split(".")[0]}.{re.match(r"^[a-zA-Z]{1,3}", tq_main_future.split(".")[1]).group()}'
tq_main_future_data = TQZTianQinClient().query_single_quote(tq_future=tq_main_future)
ret[_tq_future] = {
'day': vars(tq_main_future_data['trading_time'])['day'],
'night': vars(tq_main_future_data['trading_time'])['night']
}
for instrument, data in ret.items():
for session_data in data.values():
for i in range(len(session_data)):
for j in range(len(session_data[i])):
hour = session_data[i][j].split(':')[0]
if hour >= '24':
new_hour = str(int(hour) - int('24')).zfill(2)
session_data[i][j] = f'{new_hour}:{session_data[i][j].split(":")[1]}:{session_data[i][j].split(":")[2]}'
TQZJsonOperator.tqz_write_jsonfile(content=ret, target_jsonfile=cls.__trading_time_config)
@classmethod
def auto_make_main_contracts_excel(cls):
"""
根据所有合约的日线收盘数据, 更新当日的所有品种的主力合约
"""
""" eg:
date | SHFE.rb | SHFE.ag | ...
20230426 | SHFE.rb2310 | SHFE.ag2306 | ...
20230427 | SHFE.rb2310 | SHFE.ag2306 | ...
20230428 | SHFE.rb2401 | SHFE.ag2312 | ...
"""
pass
if __name__ == '__main__':
# ret_df = TQZTianQinClient().query_history_bars(tq_future="SHFE.cu2305", tq_duration_seconds=60)
# print("ret_df: " + str(ret_df))
# ret_df = TQZTianQinClient().query_history_ticks(tq_future="SHFE.rb2310")
# print("ret_df: " + str(ret_df))
# TQZTianQinClient.record_data()
# TQZAutoMakeConfig.auto_make_main_contracts_excel()
pass
标签:main,datetime,self,tq,future,stable,Piece,data,天勤
From: https://blog.51cto.com/u_9527606/6236678