零滞后 Hull Moving Average (HMA) 均线策略

Hull Moving Average (HMA) 是一种由 Alan Hull 开发的移动平均线，旨在解决传统移动平均线（如 SMA 或 EMA）在平滑价格数据时产生的滞后问题。它通过使用加权移动平均线（WMA）的组合，实现了响应速度快（近乎零滞后）且曲线平滑的效果。

以下是在 QMT (Quant Trader) 平台中实现 HMA 策略的完整指南和代码。

1. HMA 计算原理

HMA 的计算步骤如下：

1. 计算周期为 $n/2$ 的加权移动平均线 (WMA)。
2. 计算周期为 $n$ 的加权移动平均线 (WMA)。
3. 计算差值序列：$Raw = 2 \times WMA(n/2) - WMA(n)$。
4. 对差值序列 $Raw$ 计算周期为 $\sqrt{n}$ 的 WMA，即为最终的 HMA。

2. 策略逻辑

本策略采用经典的 价格突破 HMA 逻辑：

• 买入信号：当收盘价 上穿 HMA 均线时，全仓买入。
• 卖出信号：当收盘价 下穿 HMA 均线时，清仓卖出。

3. QMT 策略代码实现

以下是完整的 Python 策略代码。代码中包含了一个自定义的 WMA 计算函数，不依赖 talib 库，确保在标准 QMT 环境中直接运行。

# -*- coding: gbk -*-
import pandas as pd
import numpy as np

def init(ContextInfo):
    """
    策略初始化函数
    """
    # 设置策略参数
    ContextInfo.hma_period = 20  # HMA 周期
    ContextInfo.account_id = '6000000000'  # 请替换为您的资金账号
    ContextInfo.account_type = 'STOCK'     # 账号类型：STOCK-股票，FUTURE-期货
    ContextInfo.stock_code = '600000.SH'   # 示例操作标的：浦发银行
    
    # 设置股票池
    ContextInfo.set_universe([ContextInfo.stock_code])
    
    # 设置交易账号
    ContextInfo.set_account(ContextInfo.account_id)
    
    print(f"策略初始化完成，HMA周期: {ContextInfo.hma_period}")

def weighted_moving_average(series, period):
    """
    计算加权移动平均线 (WMA)
    公式: (P1*1 + P2*2 + ... + Pn*n) / (1 + 2 + ... + n)
    """
    weights = np.arange(1, period + 1)
    # 使用 rolling 和 apply 计算 WMA
    # 注意：raw=True 可以提高计算速度
    wma = series.rolling(period).apply(lambda x: np.dot(x, weights) / weights.sum(), raw=True)
    return wma

def calc_hma(series, period):
    """
    计算 Hull Moving Average (HMA)
    HMA = WMA(2 * WMA(n/2) - WMA(n), sqrt(n))
    """
    # 1. 计算 WMA(n/2)
    half_length = int(period / 2)
    wma_half = weighted_moving_average(series, half_length)
    
    # 2. 计算 WMA(n)
    wma_full = weighted_moving_average(series, period)
    
    # 3. 计算 Raw HMA 序列: 2 * WMA(n/2) - WMA(n)
    raw_hma = 2 * wma_half - wma_full
    
    # 4. 计算最终 HMA: WMA(Raw, sqrt(n))
    sqrt_length = int(np.sqrt(period))
    hma = weighted_moving_average(raw_hma, sqrt_length)
    
    return hma

def handlebar(ContextInfo):
    """
    K线周期运行函数
    """
    # 获取当前正在处理的标的
    stock_code = ContextInfo.stock_code
    
    # 获取当前周期 (例如 '1d')
    period_type = ContextInfo.period
    
    # 获取历史行情数据
    # 为了计算 HMA，我们需要足够的历史数据。
    # 预留长度建议为 period * 2 + 100，防止计算结果前段为 NaN
    data_count = ContextInfo.hma_period * 3 + 50
    
    # 使用 get_market_data_ex 获取数据 (推荐使用 ex 接口)
    # 返回格式: {code: DataFrame}
    market_data = ContextInfo.get_market_data_ex(
        ['close'], 
        [stock_code], 
        period=period_type, 
        count=data_count, 
        dividend_type='front' # 前复权
    )
    
    if stock_code not in market_data:
        return
        
    df = market_data[stock_code]
    
    # 如果数据量不足以计算 HMA，直接返回
    if len(df) < ContextInfo.hma_period:
        return

    # 计算 HMA
    # 注意：Series 的索引必须是时间序列，计算后会自动对齐
    close_prices = df['close']
    hma_series = calc_hma(close_prices, ContextInfo.hma_period)
    
    # 获取最新和上一个周期的收盘价与 HMA 值
    # -1 代表当前K线，-2 代表上一根K线
    current_close = close_prices.iloc[-1]
    prev_close = close_prices.iloc[-2]
    
    current_hma = hma_series.iloc[-1]
    prev_hma = hma_series.iloc[-2]
    
    # 检查 HMA 是否计算出有效值 (非 NaN)
    if np.isnan(current_hma) or np.isnan(prev_hma):
        return

    # --- 绘图部分 (仅在回测或界面运行时有效) ---
    # 在主图上绘制 HMA 曲线
    ContextInfo.paint('HMA', current_hma, -1, 0, 'yellow')

    # --- 交易逻辑 ---
    
    # 获取当前持仓
    positions = ContextInfo.get_trade_detail_data(ContextInfo.account_id, ContextInfo.account_type, 'POSITION')
    current_vol = 0
    for pos in positions:
        if pos.m_strInstrumentID == stock_code:
            current_vol = pos.m_nVolume
            break

    # 信号判断
    # 1. 买入信号: 收盘价上穿 HMA (当前价 > HMA 且 上一价 <= 上一HMA)
    buy_signal = (current_close > current_hma) and (prev_close <= prev_hma)
    
    # 2. 卖出信号: 收盘价下穿 HMA (当前价 < HMA 且 上一价 >= 上一HMA)
    sell_signal = (current_close < current_hma) and (prev_close >= prev_hma)
    
    # 执行交易
    # 注意：ContextInfo.is_last_bar() 确保只在最新的一根K线上发单，避免回测时重复发单或历史K线重复计算
    if ContextInfo.is_last_bar():
        if buy_signal and current_vol == 0:
            print(f"[{stock_code}] 触发买入信号: Close={current_close:.2f}, HMA={current_hma:.2f}")
            # 全仓买入示例 (按固定手数或金额下单)
            # 这里演示买入 1000 股
            passorder(23, 1101, ContextInfo.account_id, stock_code, 5, -1, 1000, ContextInfo)
            
        elif sell_signal and current_vol > 0:
            print(f"[{stock_code}] 触发卖出信号: Close={current_close:.2f}, HMA={current_hma:.2f}")
            # 清仓卖出
            passorder(24, 1101, ContextInfo.account_id, stock_code, 5, -1, current_vol, ContextInfo)

4. 代码关键点解析

1. WMA 实现 (weighted_moving_average):

   • QMT 的 Python 环境通常包含 numpy 和 pandas。
   • 我们利用 rolling().apply() 结合 np.dot 来实现加权平均。这种方法比循环快，且不需要安装额外的金融库。
   • 权重生成使用 np.arange(1, period + 1)，这意味着最近的价格权重最大（线性加权）。

2. HMA 实现 (calc_hma):

   • 严格遵循 HMA 的数学定义。
   • half_length 和 sqrt_length 均取整数 (int) 以符合窗口期要求。

3. 数据获取 (get_market_data_ex):

   • 我们请求的数据长度 count 设置为 hma_period * 3 + 50。这是因为 HMA 涉及多次平滑，需要比周期长得多的数据才能消除前端的 NaN 值，确保最新的 HMA 值是准确的。

4. 交易信号:

   • 使用了交叉逻辑：current_close > current_hma 且 prev_close <= prev_hma。这比单纯比较当前价格更能捕捉趋势反转的瞬间。

5. 绘图 (paint):

   • 使用 ContextInfo.paint 可以在 QMT 的 K 线图上直接画出黄色的 HMA 线，方便视觉验证策略效果。

5. 如何使用

1. 打开 QMT 客户端，进入 模型研究/策略编辑器。
2. 新建一个 Python 策略。
3. 将上述代码完整复制粘贴进去。
4. 修改 init 函数中的 ContextInfo.account_id 为您的真实或模拟账号。
5. 点击 编译，确保无误。
6. 在图表上运行该策略，或者进行回测。

Q&A

Q: 为什么 HMA 比普通均线好？
A: 普通均线（SMA）为了平滑噪音会产生较大滞后，导致买卖信号延迟。HMA 通过数学上的加权处理，在保持曲线平滑的同时，极大地减少了滞后，更适合捕捉短期趋势。

Q: 如何调整策略的灵敏度？
A: 修改 init 函数中的 ContextInfo.hma_period。

• 调小 (如 10)：更灵敏，信号更多，但假信号（噪音）也更多。
• 调大 (如 60)：更平滑，过滤噪音效果好，但信号会滞后，适合长线趋势。

Q: 可以在期货上使用吗？
A: 可以。只需将 ContextInfo.stock_code 改为期货合约代码（如 IF2312.IF），并将 passorder 中的操作类型代码（23/24）修改为期货对应的开平仓代码（如 0/1/3/4 或 组合指令）即可。