迅投QMT实盘和回测模式下获取handlebar当前K线价格的方法

任侠
前端开发
23天前
88热度
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

参考链接：