一、交易策略解释
核心思想
Hull 移动平均线(HMA)策略的核心思想是利用一种特殊设计的移动平均线来减少传统移动平均线的滞后性问题，同时保持曲线的平滑度，从而更快速、更准确地识别市场趋势和转折点。
HMA由艾伦·赫尔于2005年开发，其设计目标是解决传统移动平均线(如SMA和EMA)在信号生成上的两个主要问题：
滞后性：传统移动平均线通常在价格已经明显变动后才产生信号，导致交易时机滞后。
噪音敏感性：简单的短周期移动平均线虽然响应快，但容易受市场噪音影响产生虚假信号。
HMA通过一种特殊的计算方法，结合加权移动平均线(WMA)的多重应用和平方根函数，在保持曲线平滑度的同时显著减少了滞后，使交易者能更快速准确地识别趋势变化，从而在期货市场中把握更有利的入场和出场时机。
理论基础
HMA的设计基于几个关键的数学和统计学原理：
加权移动平均法：与简单移动平均线对所有价格赋予相同权重不同，加权移动平均线(WMA)对较新的数据赋予更高的权重。这一原理使得WMA能更快地响应最新的价格变化。
平方根函数的应用：HMA计算中使用周期的平方根作为最终WMA的周期，这种方法能在维持足够样本量的同时减少计算滞后。
差分放大原理：HMA计算过程中通过"2×短期WMA - 长期WMA"的方式创建一个"去滞后"的价格序列，这一操作实际上是对短期价格变动的放大，使指标对趋势变化更为敏感。
牛津策略(Oxford Strategy)的研究对包含42个不同期货市场的投资组合进行了长达36年(1980-2016)的HMA策略回溯测试，结果显示在优化参数的情况下，特别是采用较低交易频率(较长HMA周期)时，策略可以实现11.50%的复合年增长率和1.10的利润因子，表明HMA在长期期货交易中具有潜在的有效性。
策略适用场景
明确趋势市场：HMA在识别和跟踪持续性趋势方面表现出色，无论是上升趋势还是下降趋势。
波动性中等的市场：在波动适中的市场中，HMA能有效过滤噪音同时保持对价格变化的敏感性。
高流动性市场：HMA适用于主要期货合约等流动性较高的市场，这些市场价格变动更连续，指标表现更可靠。
跨市场应用：牛津策略的研究表明，HMA策略可以在多种期货市场中应用，包括商品、货币、利率和股指期货。
策略原理
Hull移动平均线(HMA)的计算公式如下：
HMA = WMA(2 * WMA(price, n/2) - WMA(price, n), sqrt(n))- EMA（指数移动平均线）： 作为Keltner Channel的中轨，EMA比SMA反应更快速，能更好地跟踪价格变化。
ema = pd.Series(close).ewm(span=EMA_PERIOD, adjust=False).mean().values其中：
- WMA是加权移动平均线
- n是回溯周期
- sqrt(n)是n的平方根 计算过程可分为三个步骤：
- 计算周期为n的WMA和周期为n/2的WMA
- 将周期为n/2的WMA乘以2，然后减去周期为n的WMA
- 对步骤2得到的结果计算一个周期为sqrt(n)的WMA 这种计算方法实现了三个关键效果：
- 通过加权方式优先考虑近期价格数据
- 通过"2×短期WMA - 长期WMA"放大近期价格变化的影响
- 通过最终的sqrt(n)周期WMA保持曲线平滑度
在天勤量化策略中，我们将构建以下基于HMA的指标：
- 双HMA交叉系统：结合短期HMA和长期HMA，用于生成更可靠的交易信号。当短期HMA上穿长期HMA时产生买入信号，下穿时产生卖出信号。
交易逻辑
开仓信号：
- 多头开仓：
- 短期HMA上穿长期HMA
- 当前价格高于长期HMA(确认上升趋势)
- 空头开仓：
- 短期HMA下穿长期HMA
- 当前价格低于长期HMA(确认下降趋势)
平仓信号：
多头平仓，满足下列任一条件：
短期HMA下穿长期HMA(反向交叉)
价格跌破动态止损线
达到预设的止盈价格(入场价格上涨3%)
空头平仓，满足下列任一条件：
短期HMA上穿长期HMA(反向交叉)
价格突破动态止损线
达到预设的止盈价格(入场价格下跌3%)
止损策略：
- 技术止损：
- 多头：止损位设在长期HMA下方2倍ATR距离
- 空头：止损位设在长期HMA上方2倍ATR距离
- 追踪止损：
- 随着交易盈利，止损位会基于短期HMA和ATR动态调整
- 多头：止损位为短期HMA减去2倍ATR，并且只上移不下移
- 空头：止损位为短期HMA加上2倍ATR，并且只下移不上移
止盈策略：
- 固定止盈：
- 多头：当价格达到入场价格上涨3%时平仓
- 空头：当价格达到入场价格下跌3%时平仓
- 追踪止盈：
- 记录入场后的最高价（多头）或最低价（空头）
- 当获利达到一定阈值(2%)时，启动追踪止盈机制
- 若价格回撤0.5%，则触发平仓以锁定利润
回测
回测初始设置
- 测试周期： 2022 年 11 月 1 日 - 2023 年 4 月 1 日
- 交易品种： SHFE.au2306
- 初始资金： 1000万元
回测结果
上述回测累计收益走势图
代码
#!/usr/bin/env python # -*- coding: utf-8 -*- __author__ = "Chaos" from datetime import date import numpy as np from tqsdk import TqApi, TqAuth, TqBacktest, TargetPosTask, BacktestFinished from tqsdk.ta import ATR # ===== 全局参数设置 ===== SYMBOL = "SHFE.au2306" # 黄金期货合约 POSITION_SIZE = 30 # 单次交易手数 START_DATE = date(2022, 11, 1) # 回测开始日期 END_DATE = date(2023, 4, 1) # 回测结束日期 # Hull Moving Average 参数 LONG_HMA_PERIOD = 30 # 长周期HMA，用于确定大趋势 SHORT_HMA_PERIOD = 5 # 短周期HMA，用于入场信号 # 止损止盈参数 ATR_PERIOD = 14 # ATR计算周期 STOP_LOSS_ATR = 2.0 # 止损为入场点的2倍ATR TAKE_PROFIT_ATR = 4.0 # 获利为入场点的4倍ATR TRAILING_STOP = True # 使用移动止损 # 新增止盈参数 FIXED_TAKE_PROFIT_PCT = 0.03 # 固定止盈比例（3%） USE_TRAILING_PROFIT = True # 是否使用追踪止盈 TRAILING_PROFIT_THRESHOLD = 0.02 # 触发追踪止盈的收益率阈值（2%） TRAILING_PROFIT_STEP = 0.005 # 追踪止盈回撤幅度（0.5%） # ===== 全局变量 ===== current_direction = 0 # 当前持仓方向：1=多头，-1=空头，0=空仓 entry_price = 0 # 开仓价格 stop_loss_price = 0 # 止损价格 take_profit_price = 0 # 止盈价格 highest_since_entry = 0 # 入场后的最高价（用于多头追踪止盈） lowest_since_entry = 0 # 入场后的最低价（用于空头追踪止盈） # ===== Hull移动平均线相关函数 ===== def wma(series, period): """计算加权移动平均线""" weights = np.arange(1, period + 1) return series.rolling(period).apply(lambda x: np.sum(weights * x) / weights.sum(), raw=True) def hma(series, period): """计算Hull移动平均线""" period = int(period) if period < 3: return series half_period = period // 2 sqrt_period = int(np.sqrt(period)) wma1 = wma(series, half_period) wma2 = wma(series, period) raw_hma = 2 * wma1 - wma2 return wma(raw_hma, sqrt_period) # ===== 策略开始 ===== print("开始运行Hull移动平均线期货策略...") # 创建API实例 api = TqApi(backtest=TqBacktest(start_dt=START_DATE, end_dt=END_DATE), auth=TqAuth("快期账号", "快期密码")) # 订阅合约的K线数据 klines = api.get_kline_serial(SYMBOL, 60 * 60 * 24) # 创建目标持仓任务 target_pos = TargetPosTask(api, SYMBOL) try: while True: # 等待更新 api.wait_update() # 如果K线有更新 if api.is_changing(klines.iloc[-1], "datetime"): # 确保有足够的数据计算指标 if len(klines) < max(SHORT_HMA_PERIOD, LONG_HMA_PERIOD, ATR_PERIOD) + 10: continue # 计算短期和长期HMA klines['short_hma'] = hma(klines.close, SHORT_HMA_PERIOD) klines['long_hma'] = hma(klines.close, LONG_HMA_PERIOD) # 计算ATR用于止损设置 atr_data = ATR(klines, ATR_PERIOD) # 获取最新数据 current_price = float(klines.close.iloc[-1]) current_short_hma = float(klines.short_hma.iloc[-1]) current_long_hma = float(klines.long_hma.iloc[-1]) current_atr = float(atr_data.atr.iloc[-1]) # 获取前一个周期的数据 prev_short_hma = float(klines.short_hma.iloc[-2]) prev_long_hma = float(klines.long_hma.iloc[-2]) # 输出调试信息 print(f"价格: {current_price}, 短期HMA: {current_short_hma:.2f}, 长期HMA: {current_long_hma:.2f}") # ===== 交易逻辑 ===== # 空仓状态 - 寻找开仓机会 if current_direction == 0: # 多头开仓条件 if (prev_short_hma <= prev_long_hma and current_short_hma > current_long_hma and current_price > current_long_hma): print(f"多头开仓信号: 短期HMA上穿长期HMA") # 设置持仓和记录开仓价格 current_direction = 1 target_pos.set_target_volume(POSITION_SIZE) entry_price = current_price # 设置止损价格 stop_loss_price = current_long_hma - STOP_LOSS_ATR * current_atr # 设置止盈价格 take_profit_price = entry_price * (1 + FIXED_TAKE_PROFIT_PCT) # 重置追踪止盈变量 highest_since_entry = entry_price print(f"多头开仓价格: {entry_price}, 止损: {stop_loss_price:.2f}, 止盈: {take_profit_price:.2f}") # 空头开仓条件 elif (prev_short_hma >= prev_long_hma and current_short_hma < current_long_hma and current_price < current_long_hma): print(f"空头开仓信号: 短期HMA下穿长期HMA") # 设置持仓和记录开仓价格 current_direction = -1 target_pos.set_target_volume(-POSITION_SIZE) entry_price = current_price # 设置止损价格 stop_loss_price = current_long_hma + STOP_LOSS_ATR * current_atr # 设置止盈价格 take_profit_price = entry_price * (1 - FIXED_TAKE_PROFIT_PCT) # 重置追踪止盈变量 lowest_since_entry = entry_price print(f"空头开仓价格: {entry_price}, 止损: {stop_loss_price:.2f}, 止盈: {take_profit_price:.2f}") # 多头持仓 - 检查平仓条件 elif current_direction == 1: # 更新入场后的最高价 if current_price > highest_since_entry: highest_since_entry = current_price # 计算固定止损 fixed_stop_loss = entry_price * (1 - STOP_LOSS_ATR) # 更新追踪止损（只上移不下移） new_stop = current_short_hma - STOP_LOSS_ATR * current_atr if new_stop > stop_loss_price: stop_loss_price = new_stop print(f"更新多头止损: {stop_loss_price:.2f}") # 平仓条件1: 短期HMA下穿长期HMA if prev_short_hma >= prev_long_hma and current_short_hma < current_long_hma: profit_pct = (current_price - entry_price) / entry_price * 100 target_pos.set_target_volume(0) current_direction = 0 print(f"多头平仓: 价格={current_price}, 盈亏={profit_pct:.2f}%") # 平仓条件2: 价格跌破止损 elif current_price < stop_loss_price or current_price < fixed_stop_loss: profit_pct = (current_price - entry_price) / entry_price * 100 target_pos.set_target_volume(0) current_direction = 0 print(f"多头止损: 价格={current_price}, 盈亏={profit_pct:.2f}%") # 平仓条件3: 价格达到止盈价格 elif current_price >= take_profit_price: profit_pct = (current_price - entry_price) / entry_price * 100 target_pos.set_target_volume(0) current_direction = 0 print(f"多头止盈: 价格={current_price}, 盈亏={profit_pct:.2f}%") # 空头持仓 - 检查平仓条件 elif current_direction == -1: # 更新入场后的最低价 if current_price < lowest_since_entry or lowest_since_entry == 0: lowest_since_entry = current_price # 计算固定止损 fixed_stop_loss = entry_price * (1 + STOP_LOSS_ATR) # 更新追踪止损（只下移不上移） new_stop = current_short_hma + STOP_LOSS_ATR * current_atr if new_stop < stop_loss_price or stop_loss_price == 0: stop_loss_price = new_stop print(f"更新空头止损: {stop_loss_price:.2f}") # 平仓条件1: 短期HMA上穿长期HMA if prev_short_hma <= prev_long_hma and current_short_hma > current_long_hma: profit_pct = (entry_price - current_price) / entry_price * 100 target_pos.set_target_volume(0) current_direction = 0 print(f"空头平仓: 价格={current_price}, 盈亏={profit_pct:.2f}%") # 平仓条件2: 价格突破止损 elif current_price > stop_loss_price or current_price > fixed_stop_loss: profit_pct = (entry_price - current_price) / entry_price * 100 target_pos.set_target_volume(0) current_direction = 0 print(f"空头止损: 价格={current_price}, 盈亏={profit_pct:.2f}%") # 平仓条件3: 价格达到止盈价格 elif current_price <= take_profit_price: profit_pct = (entry_price - current_price) / entry_price * 100 target_pos.set_target_volume(0) current_direction = 0 print(f"空头止盈: 价格={current_price}, 盈亏={profit_pct:.2f}%") except BacktestFinished as e: print("回测结束") api.close()