QMT Python API

场景化示例

盘前数据准备

qmt://docs/python

1from xtquant import xtdata

2from datetime import datetime,timedelta

3import time

4from tqdm import tqdm

6## 交易日切换判断

8def trading_date_is_change(market_list = ['SH', 'SZ']) -> bool:

9 """判断交易日是否切换，当全部交易市场都完成交易日切换后，返回True

11 Args:

12 market_list (list, optional): 交易市场，市场分类参考 /docs/python/variable_convention.html

13 Return:

14 bool

15 """

17 market_list = ['SH', 'SZ', 'BJ']

18 ls = [0,0,0]

19 now_date = datetime.now().date()

20 for i in range(len(market_list)):

21 market = market_list[i]

22 trading_dates = xtdata.get_trading_dates(market)

23 if trading_dates:

24 if datetime.fromtimestamp(trading_dates[-1] / 1000).date() == now_date:

25 print(f'trading date switch success, market: {market}, trading date: {now_date}')

26 ls[i] = 1

27 else:

28 print(f"{market} -- 交易日未切换")

30 if all(ls):

31 return True

32 else:

33 return False

35def update_sector_data() -> None:

36 """更新静态数据，包含板块数据

    静态数据是指在当天交易日内不会变化，或基本不会变化的数据（如合约信息可能会盘中少量更新，可忽略）

38 交易日列表合约列表（合约信息） <- 当日自动更新

39 复权数据 <- 当日自动更新

40 板块列表、板块成分 <- 需要每天执行下载

41 """

43 xtdata.download_sector_data() # 下载当日板块数据

44 【已废弃】xtdata.download_index_weight() # 更新指数权重数据

45 xtdata.download_history_contracts() # 下载过期合约数据，包含退市标的信息

def update_kline_data(sector_name:str ,period_list:list, start_time:str = "", end_time:str = "", incrementally:bool = True) -> None:

    """xtquant历史数据都是以压缩形式存储在本地的，get_market_data函数与get_market_data_ex函数将会自动拼接本地历史行情与服务器实时行情

51 Args:

52 sector_name (str): 板块名称，如"沪深A股"

        period_list (list): 支持的周期参考 /docs/python/data_function.html#下载指定合约代码指定周期对应时间范围的行情数据-download-history-data

54 start_time (str): 开始时间

55 格式为 YYYYMMDD 或 YYYYMMDDhhmmss 或 ''

56 例如：'20230101' '20231231235959'

57 空字符串代表全部，自动扩展到完整范围

58 end_time (str): 结束时间格式同开始时间

59 incrementally (bool): 是否增量下载

61 Raises:

62 KeyError: _description_

63 """

64 ls = xtdata.get_stock_list_in_sector(sector_name)

65 if len(ls) == 0:

66 raise KeyError("股票列表长度为0,板块可能不存在，或数据未更新")

67 for period in period_list:

68 for stock in tqdm(ls,f"{period} 数据"):

69 xtdata.download_history_data(stock,period,start_time,end_time,incrementally)

72def update_financial_data(sector_name:str) -> None:

73 """

74 Args:

75 sector_name (str): 板块名称，如"沪深A股"

76 """

77 ls = xtdata.get_stock_list_in_sector(sector_name)

78 for stock in tqdm(ls,f"更新财务数据"):

79 xtdata.download_financial_data(stock)

82def is_trade_time(trading_time_info) -> bool:

83 '''

84 Args:

85 trading_time_info:格式需要如下

86 stock : (["09:30:00","11:30:00"],["13:00:00","15:00:00"])

87 future : (["09:00:00","10:15:00"],["10:30:00","11:30:00"],["13:30:00","15:00:00"],["21:00:00","26:30:00"])

88 return:bool

89 '''

91 _now = int((datetime.datetime.now() - datetime.timedelta(hours=4)).strftime("%H%M%S"))

92 for _time_list_ in trading_time_info:

93 st_str = _time_list_[0]

94 _sp_st = (int(st_str.split(":")[0]) - 4) * 10000 + (int(st_str.replace(":", "")) % 10000)

95 et_str = _time_list_[1]

96 _sp_et = (int(et_str.split(":")[0]) - 4) * 10000 + (int(et_str.replace(":", "")) % 10000)

98 if _sp_st <= _now < _sp_et:

99 return True

100 return False

101

102def trade_logic():

103 """具体策略逻辑，写法参考 /docs/xtquant/code_examples.html#交易示例

104 """

105 pass

106

107if __name__ == "__main__":

108 while not trading_date_is_change(["SH","SZ"]):

109 print("休眠60s")

110 time.sleep(60)

111 # 更新板块信息

112 update_sector_data()

113 # 更新财务数据

114 update_financial_data("沪深A股")

115 # 更新历史K线数据

116 # 当前交易日的数据来自服务器，历史交易日的数据来自本地

117

    # 实时K线数据需要通过subscribe_quote订阅,get_market_data_ex函数将会自动拼接本地历史行情与服务器实时行情

118 update_kline_data("沪深A股",["1d"],"","",True)

119

120

121 ## 非交易时间过滤

122 while not is_trade_time((["09:30:00","11:30:00"],["13:00:00","15:00:00"])):

123 print("非交易时间")

124 time.sleep(3)

125

126 while 1:

127 # 固定3s判断一次交易逻辑

128 trade_logic()

129 time.sleep(3)

130