import datetime
import re
from tqsdk import TqApi, TqAuth
from tqsdk.tools import DataDownloader
from contextlib import closing
from public_module.tqz_extern.tools.pandas_operator.pandas_operator import pandas # noqa
from public_module.tqz_extern.tools.file_path_operator.file_path_operator import TQZFilePathOperator
from server_api.tqz_object import TickData, BarData, Exchange
import csv
TIME_GAP = 8 * 60 * 60 * 1000000000
class TQZTianQinClient:
"""
略...
"""
@staticmethod
def query_history_ticks_from_csv(tq_symbol: str, csv_path: str) -> list:
csv_read = csv.reader(open(csv_path))
tq_sym = tq_symbol.split('@')[1]
history_ticks = []
index = 0
for line in csv_read:
index += 1
if line[0] == 'datetime':
continue
if index == 20: # for test
break
std_datetime_string = re.match('(\d+)-(\d+)-(\d+) (\d+):(\d+):(\d+\.[\d]{4})', line[0]).group()
history_ticks.append(
TickData(
symbol=f'{tq_sym.split(".")[1]}888',
exchange=Exchange(tq_sym.split('.')[0]),
name=tq_symbol,
datetime=datetime.datetime.strptime(std_datetime_string, "%Y-%m-%d %H:%M:%S.%f"),
last_price=float(line[2]),
high_price=float(line[3]),
low_price=float(line[4]),
volume=float(line[5]),
open_interest=float(line[7]),
bid_price_1=float(line[8]),
bid_volume_1=float(line[9]),
ask_price_1=float(line[10]),
ask_volume_1=float(line[11]),
gateway_name="csv_ticks"
) # noqa
)
for tick in history_ticks[:20]:
print(tick)
return history_ticks
class TQZTianQinDataManager:
"""
小时线: 一天6根
1493个交易日
30分钟线: 一天12根
746个交易日
10分钟线: 一天36根
248个交易日
"""
@classmethod
def load_main_history_ticks_from_csv(cls, tq_m_symbol: str, csv_path: str):
return TQZTianQinClient.query_history_ticks_from_csv(tq_symbol=tq_m_symbol, csv_path=csv_path)
if __name__ == '__main__':
cu_file = TQZFilePathOperator.grandfather_path(source_path=__file__) + f'/history_ticks_csv/KQ_m@SHFE_cu.csv'
ticks_content = TQZTianQinDataManager.load_main_history_ticks_from_csv(tq_m_symbol="KQ.m@SHFE.cu", csv_path=cu_file)
print("here...")
|
