class TQZTqClient:
"""
Client for querying main future contracts of current market from Tianqin.
"""
def __init__(self, account_name, account_password):
"""
init api with account_name & account_password
"""
self.main_vt_symbols = []
self.tq_api = TqApi(TqKq(), auth=f'{account_name},{account_password}')
def load_main_contracts(self, ins_class):
"""
Load main contracts from TqSdk
"""
try:
tq_main_contracts = self.tq_api.query_quotes(ins_class=ins_class)
[self.main_vt_symbols.append(
self.__get_vt_symbol(
tq_symbol=self.tq_api.get_quote(symbol=main_contract).underlying_symbol
)
) for main_contract in tq_main_contracts]
except:
pass
finally:
print("主力合约数据收取完成")
print("self.main_vt_symbols: " + str(self.main_vt_symbols))
self.__create_main_contracts(current_main_vt_symbols=self.main_vt_symbols)
self.tq_api.close()
# --- private part ---
def __update_today_main_contracts(self, market_main_vt_symbols):
"""
Update main contracts of current day to excel
"""
TQZMainContractsChangeFilePath.ensure_mainContractsFold_is_exist()
market_main_vt_symbols = sorted(market_main_vt_symbols)
if os.path.exists(path=TQZMainContractsChangeFilePath.main_contracts_excel()) is False:
self.__create_main_contracts(current_main_vt_symbols=market_main_vt_symbols)
else:
pre_main_contracts_dataframe = self.__init_pre_main_contracts_dataframe(market_main_vt_symbols=market_main_vt_symbols)
today_main_contracts_dataframe = self.__init_today_main_contracts_dataframe(
market_main_vt_symbols=market_main_vt_symbols,
pre_main_contracts_dataframe=pre_main_contracts_dataframe
)
self.__to_excel(
content_dataframe=today_main_contracts_dataframe,
path=TQZMainContractsChangeFilePath.main_contracts_excel(),
sheet_name=TQZMainContractsSheetType.CURRENT_FUTURE_MAIN_CONTRACT.value
)
@staticmethod
def __init_today_main_contracts_dataframe(market_main_vt_symbols, pre_main_contracts_dataframe):
today_main_contracts_dataframe = pandas.DataFrame(
columns=[
TQZMainContractsColumnType.MAIN_CONTRACT.value,
TQZMainContractsColumnType.ENTRY_PRICE.value
]
)
today_main_contracts_dataframe[TQZMainContractsColumnType.MAIN_CONTRACT.value] = market_main_vt_symbols
today_main_contracts_dataframe.set_index(TQZMainContractsColumnType.MAIN_CONTRACT.value, inplace=True)
today_main_contracts_dataframe[TQZMainContractsColumnType.ENTRY_PRICE.value] = pre_main_contracts_dataframe[
TQZMainContractsColumnType.ENTRY_PRICE.value]
today_main_contracts_dataframe.reset_index(inplace=True)
return today_main_contracts_dataframe
@staticmethod
def __init_pre_main_contracts_dataframe(market_main_vt_symbols):
pre_main_contracts_dataframe = pandas.read_excel(
io=TQZMainContractsChangeFilePath.main_contracts_excel(),
sheet_name=TQZMainContractsSheetType.CURRENT_FUTURE_MAIN_CONTRACT.value
)
not_main_contracts = []
current_main_contracts = pre_main_contracts_dataframe[
TQZMainContractsColumnType.MAIN_CONTRACT.value].values.tolist()
for current_main_contract in current_main_contracts:
if current_main_contract not in market_main_vt_symbols:
not_main_contracts.append(current_main_contract)
pre_main_contracts_dataframe.set_index(TQZMainContractsColumnType.MAIN_CONTRACT.value, inplace=True)
pre_main_contracts_dataframe.drop(not_main_contracts, inplace=True) # drop all non_main_contracts
return pre_main_contracts_dataframe
def __create_main_contracts(self, current_main_vt_symbols):
"""
Create main-contracts-excel when excel is not exsit.
"""
main_contracts_dataframe = pandas.DataFrame(
columns=[
TQZMainContractsColumnType.MAIN_CONTRACT.value,
TQZMainContractsColumnType.ENTRY_PRICE.value
]
)
main_contracts_dataframe[TQZMainContractsColumnType.MAIN_CONTRACT.value] = sorted(current_main_vt_symbols)
self.__to_excel(
content_dataframe=main_contracts_dataframe,
path=TQZMainContractsChangeFilePath.main_contracts_excel(),
sheet_name=TQZMainContractsSheetType.CURRENT_FUTURE_MAIN_CONTRACT.value
)
@staticmethod
def __to_excel(content_dataframe, path, sheet_name):
"""
Write content_dataframe to path/excel/sheet_name
"""
excel_writer = pandas.ExcelWriter(path=path)
content_dataframe.to_excel(
excel_writer,
sheet_name=sheet_name,
index=False,
freeze_panes=(1, 0)
)
excel_writer.save()
@staticmethod
def __get_vt_symbol(tq_symbol):
"""
Change tq_symbol format to vt_symbol format
"""
return f'{tq_symbol.split(".")[1]}.{tq_symbol.split(".")[0]}'