qmt量化--常见的3种交易触发算法解析

落花忆流年 · 发表于 7 小时前

今天继续安更新计划给大家更新量化的研究，给大家参考学习使用，更新计划

1 策略算法原理
2qmt使用教程
3策略算法原理
4量化研报
5 ptrad使用教程
6 因子分析
7策略算法

大qmt触发交易的方式有3中常见的我一般使用run time模块比较方便，方便代码的编写，扩展，3种方式不是绝对的，一般都是交叉使用

大QMT系统提供两大类（事件驱动与定时任务），共三种具体的运行机制来触发交易逻辑。它们的设计初衷是为了满足不同场景下的策略需求。

逐K线驱动 (handlebar)

分类：事件驱动触发原理：

回测/初始化阶段：策略启动时，系统加载所选周期的历史K线，从左到右，每一根K线触发一次handlebar(ContextInfo)函数的调用。

盘中运行阶段：当策略附着于行情主图运行时，主图品种每一个新的分笔行情数据（Tick）到达，都会触发一次handlebar函数调用。但其核心设计是模拟逐K线生效的效果，即交易信号通常在K线结束时才被最终判定和执行。

核心特点：旨在在实盘中完美复现回测的逐K线逻辑。默认情况下，在handlebar中调用的交易函数（如passorder）只在最后一根K线（即当前K线）的最后一个Tick被调用时才会真正触发下单。若要立即下单，需使用quickTrade参数。适用场景：所有需要在实盘中严格模拟历史回测“一根K线走完才产生信号”的传统策略，例如基于日线、小时线收盘价的均线、MACD等指标策略。

以下代码展示了经典的双均线金叉死叉策略在handlebar机制下的实现，这是一个完整的回测示例。代码参考官方教程

#coding:gbk
''' 双均线策略示例 '''
import talib
def init(ContextInfo):
    # 初始化交易账户和股票池
    ContextInfo.accID = 'test'  # 测试账户
    ContextInfo.set_universe(['001.SZ'])  # 设定交易品种为平安银行
    # 定义均线周期
    ContextInfo.short_period = 10
    ContextInfo.long_period = 20
def handlebar(ContextInfo):
    # 获取历史收盘价数据
    close_prices = ContextInfo.get_history_data(ContextInfo.long_period+1, ‘1d‘, ‘close‘)


    # 计算快慢均线
    short_ma = talib.MA(close_prices, timeperiod=ContextInfo.short_period)
    long_ma = talib.MA(close_prices, timeperiod=ContextInfo.long_period)


    # 获取当前持仓
    positions = get_position(ContextInfo.accID, ‘STOCK‘)
    has_position = any(pos for pos in positions if pos.m_strInstrumentID == ‘01.SZ‘)


    # 交易逻辑：金叉买入，死叉卖出
    if short_ma[-1] > long_ma[-1] and short_ma[-2] <= long_ma[-2]:  # 金叉
        if not has_position:
            # 买入80%仓位
            passorder(23, 1101, ContextInfo.accID, ‘01.SZ‘, 5, -1, 8000, ‘双均线策略‘, ContextInfo)
    elif short_ma[-1] < long_ma[-1] and short_ma[-2] >= long_ma[-2]:  # 死叉
        if has_position:
            # 卖出全部持仓
            passorder(24, 1101, ContextInfo.accID, ‘001.SZ‘, 5, -1, -1, ‘双均线策略‘, ContextInfo)

订阅推送 (subscribe)

分类：事件驱动

触发原理：在init()函数中，通过ContextInfo.subscribe_quote()函数主动订阅一个或多个指定品种的行情。此后，每当这些被订阅的品种有新的分笔数据到达，系统就会自动触发您预先定义的回调函数。

核心特点：响应速度极快，交易逻辑的执行与特定品种的行情变动严格同步。它不依赖于主图，可以独立监控特定的股票池。在回调函数中产生的交易信号可以立即下单。

适用场景：对行情响应延迟敏感的高频策略、Tick级策略、盘口分析、以及对特定一篮子股票进行实时监控的策略。以下代码展示了如何订阅两只股票的日线行情，并在回调函数中计算实时涨幅，满足条件时触发买入

#coding:gbk
class State():
    pass
State.bought_list = []  # 记录已买入股票，防止重复下单
account = ‘testaccount‘  # 交易账户
def init(ContextInfo):
    # 定义行情回调函数
    def quote_callback(data):
        # data 是字典，key为股票代码，value为该股票的行情快照
        for stock_code in data:
            current_price = data[stock_code][‘close‘]
            pre_close = data[stock_code][‘preClose‘]
            increase_ratio = current_price / pre_close - 1


            print(f“{stock_code} 当前涨幅: {increase_ratio:.2%}“)


            # 交易逻辑：涨幅大于0且未买入过，则买入
            if increase_ratio > 0 and stock_code not in State.bought_list:
                msg = f“涨幅 {increase_ratio:.2%} 大于0，执行买入“
                print(msg)
                # 安全提示：实际测试时需取消下行注释
                # passorder(23, 1101, account, stock_code, 5, -1, 100, ‘订阅下单示例‘, 2, msg, ContextInfo)
                State.bought_list.append(stock_code)


    # 订阅股票列表的行情，并绑定回调函数
    stock_list = [‘6.SH‘, ‘001.SZ‘]
    for stock in stock_list:
        ContextInfo.subscribe_quote(stock, period=‘1d‘, callback=quote_callback)
def handlebar(ContextInfo):
    # handlebar函数仍需保留，但主体可为空或处理其他逻辑
    pass

定时运行 (run_time)

分类：定时任务

触发原理：在init()函数中，通过ContextInfo.run_time()函数注册一个定时任务。您可以指定一个固定的时间间隔（如‘5nSecond’表示每5秒）。系统会周期性地、按照设定的时间节奏触发您指定的回调函数，与行情变化无关。

核心特点：执行节奏固定，适用于需要定期执行的任务。注意：在run_time注册的回调函数中调用passorder，必须将quickTrade参数设置为2，以确保立即触发下单，不受K线状态影响。

适用场景：定期轮询任务（如检查持仓、读取外部信号）、固定时间点操作（如集合竞价、尾盘交易）、等。

官方代码参考（实盘示例）

以下代码展示了如何在集合竞价期间（09:15-09:25），利用定时器每5秒检查一次时间，并在满足条件时以指定价格下单

#coding:gbk
import time
order_triggered = False  # 标记是否已下单
target_stock = ‘001.SZ‘  # 平行
def init(ContextInfo):
    # 设置定时器，每5秒执行一次my_timer_task函数
    # 起始时间设为过去时间，意味着定时器会立刻开始等待第一个周期，然后执行
    ContextInfo.run_time(“my_timer_task“, “5nSecond“, “2019-10-14 13:20:00“)
def my_timer_task(ContextInfo):
    global order_triggered
    current_time = time.strftime(‘%H%M%S‘)  # 获取当前时分秒


    # 判断是否在集合竞价时间段且未下过单
    if not order_triggered and ‘091500‘ <= current_time <= ‘092500‘:
        order_triggered = True
        print(f“{current_time} 触发集合竞价下单“)
        # 以指定价14.00元买入100股， quickTrade=2 确保立即下单
        passorder(23, 1101, account, target_stock, 11, 14.00, 100, ‘定时器示例‘, 2, ContextInfo)
def handlebar(ContextInfo):
    # handlebar函数仍需保留
    return

下面做一个简单的3个模式总结

我是小果，主做量化研究，算法策略的开发，多年开发经验，不懂的问我就可以

量化福利，专业的技术支持

留言

写留言