迅投QMT实盘和回测模式下获取handlebar当前K线价格的方法
- 前端开发
- 1天前
- 11热度
- 0评论
迅投 QMT 主要通过 init
方法初始化,当 k 线行情更新时会触发 handlebar
方法回调执行。这是其量化策略执行的主要的逻辑之一。
在 handlebar
方法中提供了 ContextInfo
上下文对象参数,该对象包含许多属性和方法,但并不包含 K 线的当前价格。
在实盘情况下,可以通过 ContextInfo.get_full_tick([stockcode])
方法获取实时行情数据得到当前价格,但回测时该方法不可用。
ContextInfo.get_market_data_ex
行情获取函数可根据指定的时间、证券代码等参数信息返回行情数据。
在回测模式下可以基于 ContextInfo.get_market_data_ex
函数获取指定时间、证券代码的行情数据,从而获取到当前价格。
回测模型取本地数据遍历,不需要向服务器订阅实时行情,应使用 get_market_data_ex
函数,并指定 subscribe
参数为 False
来读取本地行情数据。
此外需注意的是,本地行情数据需提前下载(可手动通过界面下载,或在 init
方法中调用 download_history_data
方法下载),否则会获取不到行情。
于是我们可以得到一个获取当前价格的方法,参考如下:
# 创建一个空的全局数据缓存对象
class a: pass
A = a()
def init(ContextInfo):
A.stock_code = ContextInfo.stockcode + '.' + ContextInfo.market
if ContextInfo.do_back_test:
# 回测模式
A.isBacktesting = True
A.acctId = 'tests'
A.acctType = 'STOCK'
# 回测模式下需要提前下载历史数据
download_history_data(A.stock_code, ContextInfo.period, '20220101', '')
else:
A.acctId = account # 账号为模型交易界面选择账号
A.acctType= accountType # 账号类型为模型交易界面选择账号
# ContextInfo.set_account(account) # 启用资金账号成交实时主推函数调用 https://dict.thinktrader.net/innerApi/callback_function.html
def handlebar(ContextInfo):
# 获取当前价格
current_price = get_price(ContextInfo, ContextInfo.stockcode)
print('current_price:', current_price)
def get_price(ContextInfo, stock_code):
'''
获取当前价格,支持实盘和回测模式
'''
if ContextInfo.do_back_test:
timetag = ContextInfo.get_bar_timetag(ContextInfo.barpos)
endtime = timetag_to_datetime(timetag, '%Y%m%d%H%M%S')
# period 为 tick 时行情会包含 lastPrice 数据,否则没有该数据
df = ContextInfo.get_market_data_ex(['close', 'lastPrice'], [stock_code], end_time=endtime, count=1, period=ContextInfo.period, dividend_type=ContextInfo.dividend_type, fill_data=True, subscribe=False)
if df[stock_code].empty:
print('df is None', stock_code, ContextInfo.period, startdate, endtime)
return
# for col in df[stock_code].columns:
# print(col, df[stock_code][col].iloc[0])
if not df[stock_code]['lastPrice'].empty:
current_price = df[stock_code]['lastPrice'].iloc[-1]
if math.isnan(current_price) and not df[stock_code]['close'].empty:
current_price = df[stock_code]['close'].iloc[-1]
else:
tick = ContextInfo.get_full_tick([stock_code])
tick_data = tick.get(stock_code)
if not tick_data:
print("【错误】获取行情数据失败")
return
current_price = tick_data["lastPrice"]
return current_price
参考链接: