QMT Python API
原生Python_策略回测_投研端
原生Python_策略回测
1
2
3
# 根据情况指定xtquant的路径4
import sys5
import numpy as np6
import pandas as pd7
from xtquant import xtdata8
import xtquant9
print(xtquant)10
print(xtdata.data_dir)11
# 指定获取投研端数据(可不指定，默认优先连接投研)12
# xtdata.reconnect(port=58613)13
# xtdata.download_sector_data()14
15
16
17
class G():18
pass19
20
21
g = G()22
23
24
25
26
def init(C):27
# ------------------------参数设定-----------------------------28
g.his_st = {}29
30
# g.s = C.get_stock_list_in_sector("沪深A股") # 获取沪深A股股票列表31
g.s = C.get_stock_list_in_sector("沪深300") # 获取沪深300股票列表32
# g.s = ['000001.SZ']33
g.day = 034
g.holdings = {i: 0 for i in g.s}35
g.weight = [0.1] * 1036
g.buypoint = {}37
g.money = 1000000 # C.capital38
g.accid = 'test'39
g.profit = 040
# 因子权重41
g.buy_num = 10 # 买排名前5的股票，在过滤中会用到42
g.per_money = g.money / g.buy_num * 0.9543
44
45
46
47
def after_init(C):48
# ------------------------量价数据获取-----------------------------49
data = xtdata.get_market_data_ex([], g.s, period='1d', dividend_type='front_ratio',50
fill_data=True)51
close_df = get_df_ex(data,"close")52
open_df = get_df_ex(data,"open")53
low_df = get_df_ex(data,"low")54
high_df = get_df_ex(data,"high")55
volume_df = get_df_ex(data,"volume")56
amount_df = get_df_ex(data,"amount")57
preclose_df = get_df_ex(data,"preClose")58
59
# ------------------------基础数据获取-----------------------------60
# 将 g.s 中的全部股票的 TotalVolume 都获取出来，组合成一个 DataFrame61
# 例如 g.s 中有 10 个股票，那么下面的代码就会返回一个 1 行 10 列的 DataFrame62
# 该 DataFrame 的 index 是股票代码，columns 是 TotalVolume63
# 该 DataFrame 的数据是每个股票的 TotalVolume，请给出代码：64
# C.get_instrumentdetail('600000.SH')['TotalVolume']65
# 使用字典推导来获取每个股票的TotalVolume66
total_volumes = {stock: C.get_instrumentdetail(stock)['TotalVolume'] for stock in g.s}67
# 将字典转换为DataFrame，但先转化为一个嵌套字典68
df_total_volume = pd.DataFrame({k: v for k, v in total_volumes.items()}, index=['TotalVolume'])69
70
71
# ------------------------财务数据获取-----------------------------72
73
# ------------------------因子1计算及处理--------------------------------74
# 1. 市值因子: 用市值因子 = 股票收盘价 * 股票总股本，要利用好，要求对应列名相乘75
factor = close_df * df_total_volume.loc['TotalVolume']76
77
# ------------------------因子2计算及处理--------------------------------78
# 判断 close_df 中的 code 上市时间大于120天79
stock_opendate_filter = filter_opendate_qmt(C, close_df, 120)80
81
# ------------------------上市日期过滤处理--------------------------------82
# 两个布尔值的 DataFrame 对应相乘过滤掉每个交易日上市不足120天的83
factor *= stock_opendate_filter.astype(int).replace(0, np.nan)84
85
# ------------------------排序处理-----------------------------------86
# 对 factor 每行在一行内进行排序87
factor_sorted = rank_filter(factor, 10, ascending=True, method='min', na_option='keep')88
89
90
# ------------------------因子组合得到布尔值信号--------------------------------91
92
# 确保没有未来数据的影响，将因子数据向后移动一天93
g.factor_df = factor_sorted.shift(1) #94
g.close_df = close_df.shift(1) # 为了计算收益率，将收盘价向后移动一天95
g.open_df = open_df96
g.stock_opendate_filter = stock_opendate_filter97
98
99
def handlebar(C):100
# 获取当前 K 线位置101
d = C.barpos102
# 获取当前 K 线时间103
backtest_time = timetag_to_datetime(C.get_bar_timetag(C.barpos), "%Y%m%d")104
factor_series = g.factor_df.loc[backtest_time]105
buy_list = daily_filter(factor_series, backtest_time)106
print(backtest_time, buy_list)107
108
# 获取持仓109
hold = get_holdings(g.accid, 'stock')110
need_sell = [s for s in hold if s not in buy_list]111
print('\t\t\t\t\t\t\t', backtest_time, 'sell list', need_sell)112
113
# 卖出114
for s in need_sell:115
price = g.open_df.loc[backtest_time, s]116
vol = hold[s]['持仓数量']117
passorder(24, 1101, g.accid, s, 11, price, vol, 1,"backtest","小市值",C)118
119
# 获取持仓120
hold = get_holdings(g.accid, 'stock')121
asset = get_trade_detail_data(g.accid, 'stock', 'account')122
cash = asset[0].m_dAvailable123
buy_num = g.buy_num - len(hold)124
buy_list = [s for s in buy_list if s not in hold]125
126
# 买入127
if buy_num > 0 and buy_list:128
buy_list = buy_list[:buy_num]129
# money = cash/buy_num130
print(backtest_time, 'buy list', buy_list)131
for s in buy_list:132
price = g.open_df.loc[backtest_time, s]133
if price > 0:134
passorder(23, 1102, g.accid, s, 11, float(price), g.per_money,1,"backtest","小市值",C)135
136
137
def daily_filter(factor_series, backtest_time):138
# 将 factor_series 中值 True 的index，转化成列表139
print(len(factor_series))140
sl = factor_series[factor_series].index.tolist()141
print(len(sl))142
# exit()143
# st过滤144
sl = [s for s in sl if not is_st(s, backtest_time)]145
sl = sorted(sl, key=lambda k: factor_series.loc[k])146
return sl[:g.buy_num]147
148
149
def is_st(s, date):150
# 判断某日在历史上是不是st *st151
st_dict = g.his_st.get(s, {})152
if not st_dict:153
return False154
else:155
st = st_dict.get('ST', []) + st_dict.get('*ST', [])156
for start, end in st:157
if start <= date <= end:158
return True159
160
161
def get_df(dt: dict, df: pd.DataFrame, values_name: str) -> pd.DataFrame:162
'''163
循环从字典里赋值矩阵164
values_name可选字段: ['time', 'stime', 'open', 'high', 'low', 'close', 'volume','amount', 'settelementPrice', 'openInterest', 'preClose', 'suspendFlag']165
'''166
df1 = df.copy()167
df1 = df1.apply(lambda x: dt[x.name][values_name])168
169
return df1170
171
def get_df_ex(data:dict,field:str) -> pd.DataFrame:172
173
'''174
ToDo:用于在使用get_market_data_ex的情况下，取到标准df175
176
Args:177
data: get_market_data_ex返回的dict178
field: ['time', 'open', 'high', 'low', 'close', 'volume','amount', 'settelementPrice', 'openInterest', 'preClose', 'suspendFlag']179
180
Return:181
一个以时间为index，标的为columns的df182
'''183
184
_index = data[list(data.keys())[0]].index.tolist()185
_columns = list(data.keys())186
df = pd.DataFrame(index=_index,columns=_columns)187
for i in _columns:188
df[i] = data[i][field]189
return df190
191
192
def rank_filter(df: pd.DataFrame, N: int, axis=1, ascending=False, method="max", na_option="keep") -> pd.DataFrame:193
"""194
Args:195
df: 标准数据的df196
N: 判断是否是前N名197
axis: 默认是横向排序198
ascending : 默认是降序排序199
na_option : 默认保留nan值,但不参与排名200
Return:201
pd.DataFrame:一个全是bool值的df202
"""203
_df = df.copy()204
205
_df = _df.rank(axis=axis, ascending=ascending, method=method, na_option=na_option)206
207
return _df <= N208
209
210
def filter_opendate_qmt(C, df: pd.DataFrame, n: int) -> pd.DataFrame:211
'''212
213
ToDo: 判断传入的df.columns中，上市天数是否大于N日，返回的值是一个全是bool值的df214
215
Args:216
C:contextinfo类217
df:index为时间，columns为stock_code的df,目的是为了和策略中的其他df对齐218
n:用于判断上市天数的参数，如要判断是否上市120天,则填写219
Return:pd.DataFrame220
221
'''222
local_df = pd.DataFrame(index=df.index, columns=df.columns)223
stock_list = df.columns224
stock_opendate = {i: C.get_instrument_detail(i)["OpenDate"] for i in stock_list}225
# print(type(stock_opendate["000001.SZ"]), stock_opendate["000001.SZ"])226
for stock, date in stock_opendate.items():227
local_df.at[date, stock] = 1228
df_fill = local_df.fillna(method="ffill")229
230
result = df_fill.expanding().sum() >= n231
232
return result233
234
235
def filter_opendate_xt(df: pd.DataFrame, n: int) -> pd.DataFrame:236
'''237
238
ToDo: 判断传入的df.columns中，上市天数是否大于N日，返回的值是一个全是bool值的df239
240
Args:241
C:contextinfo类242
df:index为时间，columns为stock_code的df,目的是为了和策略中的其他df对齐243
n:用于判断上市天数的参数，如要判断是否上市120天,则填写244
Return:pd.DataFrame245
246
'''247
local_df = pd.DataFrame(index=df.index, columns=df.columns)248
stock_list = df.columns249
stock_opendate = {i: xtdata.get_instrument_detail(i)["OpenDate"] for i in stock_list}250
for stock, date in stock_opendate.items():251
local_df.at[date, stock] = 1252
df_fill = local_df.fillna(method="ffill")253
254
result = df_fill.expanding().sum() >= n255
256
return result257
258
259
def get_holdings(accid, datatype):260
'''261
Arg:262
accondid:账户id263
datatype:264
'FUTURE'：期货265
'STOCK'：股票266
......267
return:268
{股票名:{'手数':int,"持仓成本":float,'浮动盈亏':float,"可用余额":int}}269
'''270
PositionInfo_dict = {}271
resultlist = get_trade_detail_data(accid, datatype, 'POSITION')272
for obj in resultlist:273
PositionInfo_dict[obj.m_strInstrumentID + "." + obj.m_strExchangeID] = {274
"持仓数量": obj.m_nVolume,275
"持仓成本": obj.m_dOpenPrice,276
"浮动盈亏": obj.m_dFloatProfit,277
"可用数量": obj.m_nCanUseVolume278
}279
return PositionInfo_dict280
281
282
283
if __name__ == '__main__':284
import sys285
from xtquant.qmttools import run_strategy_file286
287
# 参数定义方法一，如果使用方法二定义参数，run_strategy_file的param参数可不传288
param = {289
'stock_code': '000300.SH', # 驱动handlebar的代码,290
'period': '1d', # 策略执行周期 即主图周期291
'start_time': '2022-01-01 00:00:00', # 注意格式，不要写错292
'end_time': '2024-03-01 00:00:00', # 注意格式，不要写错293
'trade_mode': 'backtest', # 'backtest':回测294
'quote_mode': 'history',295
# handlebar模式，'realtime':仅实时行情（不调用历史行情的handlebar）,'history':仅历史行情, 'all'：所有，即history+realtime296
}297
# user_script = os.path.basename(__file__) # 当前脚本路径，相对路径，绝对路径均可,此处为相对路径的方法298
user_script = sys.argv[0] # 当前脚本路径，相对路径，绝对路径均可，此处为绝对路径的方法299
300
print(user_script)301
result = run_strategy_file(user_script, param=param)302
if result:303
print(result.get_backtest_index())304
print(result.get_group_result())305
306
xtdata.run()