基于钱德动量震荡指标(CMO)的期货量化交易策略
一、交易策略解释
核心思想
钱德动量震荡指标(Chande Momentum Oscillator，简称CMO)期货量化交易策略的核心思想是利用价格变动的动量特性来识别市场趋势的强度和方向，从而捕捉价格的超买超卖状态以及潜在的反转点，指导交易决策。CMO作为一种动量指标，其取值范围在-100到+100之间，通过分析上涨和下跌幅度之间的相对关系，为期货交易提供具体的量化信号。该策略基于以下原则：
-
动量领先于价格：市场价格的变动往往先通过动量变化表现出来，因此CMO作为动量指标可以提供价格趋势变化的早期信号。
-
超买超卖状态识别：当CMO达到极值区域(如高于+50或低于-50)时，市场可能处于超买或超卖状态，预示着潜在的反转机会。
-
趋势方向确认：CMO穿越零轴可以确认趋势方向的变化，向上穿越零轴表明上升趋势开始形成，向下穿越零轴则表明下降趋势开始形成。
-
信号线交叉：通过将CMO与其移动平均线(信号线)进行比较，可以生成更灵敏的入场和出场信号，过滤掉部分噪音。
-
背离分析：价格与CMO之间的背离现象可以预示趋势可能即将结束，为交易者提供潜在的反转信号。
理论基础
动量理论：市场动量是指价格变动的速率和强度。根据动量理论，价格变动的速率在价格达到顶峰或谷底之前会开始减缓，这就是为什么动量指标通常被视为领先指标，可以预测潜在的价格反转。
行为金融学：CMO的有效性部分来自于投资者的心理行为。当市场情绪过度乐观(超买)或过度悲观(超卖)时，往往会产生反向修正。CMO通过量化这种情绪极端，帮助识别潜在的反转点。
相对强度分析：与RSI等其他动量指标不同，CMO在计算中直接比较上涨与下跌幅度的相对强度，而不是简单地计算价格变化的平均值。这种方法使CMO对短期价格波动更加敏感。
零轴理论：技术分析中的零轴理论认为，震荡指标从负区域穿越到正区域(或反之)可以确认趋势的转变。CMO的零轴交叉便是基于这一理论，用于确认买卖动能的方向变化。
此外，许多期货市场的定量研究表明，动量是期货价格行为中的一个关键因素。根据多项研究，期货市场中存在明显的趋势延续现象，使得基于动量的交易策略在期货市场中具有统计学上的优势。
策略适用场景
-
波动性适中的趋势市场：CMO在明显趋势但波动性适中的市场中表现最佳，此时CMO能够有效捕捉趋势的持续性和潜在的反转点。
-
快速变化的市场：由于CMO对价格变化较为敏感，它在快速变动的市场中能提供及时的信号，适合短期交易者捕捉市场动量的变化。
-
周期性波动市场：在具有明显周期性波动的期货市场(如部分商品期货)，CMO的超买超卖信号能够有效识别周期性高点和低点。
-
流动性较高的市场：CMO策略最适合应用于流动性充足的主要期货合约，如主要股指期货、国债期货和大宗商品期货等，这些市场价格变动更连续，指标表现更可靠。
二、天勤介绍
天勤平台概述
天勤（TqSdk）是一个由信易科技开发的开源量化交易系统，为期货、期权等衍生品交易提供专业的量化交易解决方案。平台具有以下特 点：
- 丰富的行情数据： 提供所有可交易合约的全部Tick和K线数据，基于内存数据库实现零延迟访问。
- 一站式的解决方案： 从历史数据分析到实盘交易的完整工具链，打通开发、回测、模拟到实盘的全流程。
- 专业的技术支持： 近百个技术指标源码，深度集成pandas和numpy，采用单线程异步模型保证性能。
策略开发流程
- 环境准备
- 安装Python环境（推荐Python 3.6或以上版本）
- 安装tqsdk包：pip install tqsdk
- 注册天勤账户获取访问密钥
- 数据准备
- 订阅近月与远月合约行情
- 获取历史K线或者Tick数据（用于分析与行情推进）
- 策略编写
- 设计信号生成逻辑（基于价差、均值和标准差）
- 编写交易执行模块（开仓、平仓逻辑）
- 实现风险控制措施（止损、资金管理）
- 回测验证
- 设置回测时间区间和初始资金
- 运行策略获取回测结果
- 分析绩效指标（胜率、收益率、夏普率等）
- 策略优化
- 调整参数（标准差倍数、窗口大小等）
- 添加过滤条件（成交量、波动率等）
- 完善风险控制机制
三、天勤实现策略
策略原理
钱德动量震荡指标(Chande Momentum Oscillator，简称CMO)的计算公式为：
CMO = 100 * ((Su - Sd) / (Su + Sd))
其中：
- Su（上涨之和）：指定周期内所有上涨日价格变动的总和
- Sd（下跌之和）：指定周期内所有下跌日价格变动绝对值的总和
CMO指标的范围在-100到+100之间，与其他振荡指标不同，它直接比较上涨与下跌力度的相对强度。
策略核心指标体系
- 基础CMO指标：使用6周期计算（比常规14周期更敏感），用于捕捉短期动量变化
- CMO信号线：4周期的CMO移动平均线，用于产生交叉信号
- CMO斜率：2周期的CMO变化率，评估动量变化速度
- 趋势过滤：10周期SMA作为趋势确认工具
- 超买超卖阈值：±50作为极值区间界定
交易逻辑
开仓信号系统
1. 多头开仓（满足以下任一条件）：
- 超卖反弹
- CMO从超卖区域(<-50)向上突破
- 价格在SMA上方（趋势确认）
- CMO斜率为正（动量加速）
- 信号线交叉
- CMO从下方向上穿越信号线
- 价格在SMA上方
- CMO值大于-30（避免极端区域反弹初期入场）
- 零轴交叉
- CMO从负值向上穿越零轴
- 价格在SMA上方
- 前一交易日CMO小于-10（确保有足够动量变化）
2. 空头开仓（满足以下任一条件）：
- 超买回落
- CMO从超买区域(>+50)向下突破
- 价格在SMA下方
- CMO斜率为负
- 信号线交叉
- CMO从上方向下穿越信号线
- 价格在SMA下方
- CMO值小于+30
- 零轴交叉
- CMO从正值向下穿越零轴
- 价格在SMA下方
- 前一交易日CMO大于+10
平仓信号系统
1. 多头平仓：
- 反向信号：出现任何空头开仓信号
- 动量减弱：CMO从上方向下穿越信号线且CMO值大于+30
2. 空头平仓
- 反向信号：出现任何多头开仓信号
- 动量减弱：CMO从下方向上穿越信号线且CMO值小于-30
风险管理体系
该策略采用了三层止损保护机制：
- 固定百分比止损：
- 多头：入场价格下方0.8%
- 空头：入场价格上方0.8%
- ATR动态止损：
- 多头：入场价 - (2 * ATR(14))
- 空头：入场价 + (2 * ATR(14))
- 在计算时取固定止损和ATR止损中更严格的一个
- 时间止损：持仓超过10个交易日自动平仓
- 获利了结：达到1.5%盈利目标自动平仓
回测
回测初始设置
- 测试周期： 2023 年 2 月 1 日 - 2023 年 7 月 31 日
- 交易品种： DCE.c2309
- 初始资金： 1000万元
回测结果
上述回测累计收益走势图
完整代码示例
#!/usr/bin/env python
# -*- coding: utf-8 -*-
__author__ = "Chaos"
from datetime import date
import numpy as np
import pandas as pd
from tqsdk import TqApi, TqAuth, TqBacktest, TargetPosTask, BacktestFinished
from tqsdk.ta import ATR
# ===== 全局参数设置 =====
SYMBOL = "DCE.c2309"
POSITION_SIZE = 500
START_DATE = date(2023, 2, 1) # 回测开始日期
END_DATE = date(2023, 7, 31) # 回测结束日期
# CMO参数设置
CMO_PERIOD = 6 # CMO计算周期
SIGNAL_PERIOD = 4 # CMO信号线周期
CMO_SLOPE_PERIOD = 2 # CMO斜率计算周期
OVERBOUGHT_THRESHOLD = 50 # 超买阈值
OVERSOLD_THRESHOLD = -50 # 超卖阈值
SMA_PERIOD = 10 # 趋势确认移动平均线周期
# 止损止盈参数
FIXED_STOP_LOSS_PCT = 0.008 # 固定止损百分比(0.8%)
TAKE_PROFIT_PCT = 0.015 # 止盈百分比(1.5%)
ATR_STOP_MULTIPLIER = 2.0 # ATR止损乘数
MAX_HOLDING_DAYS = 10 # 最大持仓天数
# ===== 全局变量 =====
current_direction = 0 # 当前持仓方向：1=多头，-1=空头，0=空仓
entry_price = 0 # 开仓价格
stop_loss_price = 0 # 止损价格
take_profit_price = 0 # 止盈价格
entry_date = None # 开仓日期
entry_position = 0 # 开仓数量
# ===== CMO指标计算函数 =====
def calculate_cmo(close_prices, period=14):
"""计算钱德动量震荡指标(CMO)"""
delta = close_prices.diff()
# 分离上涨和下跌
up_sum = np.zeros_like(delta)
down_sum = np.zeros_like(delta)
# 填充上涨和下跌数组
up_sum[delta > 0] = delta[delta > 0]
down_sum[delta < 0] = -delta[delta < 0] # 注意要取绝对值
# 计算上涨和下跌的滚动总和
up_rolling_sum = pd.Series(up_sum).rolling(period).sum()
down_rolling_sum = pd.Series(down_sum).rolling(period).sum()
# 计算CMO值
cmo = 100 * ((up_rolling_sum - down_rolling_sum) / (up_rolling_sum + down_rolling_sum))
return cmo
# ===== 策略开始 =====
print("开始运行钱德动量震荡指标(CMO)期货策略...")
# 创建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(CMO_PERIOD, SIGNAL_PERIOD, SMA_PERIOD) + 10:
continue
# 计算CMO及相关指标
klines['cmo'] = calculate_cmo(klines.close, CMO_PERIOD)
klines['cmo_signal'] = klines['cmo'].rolling(SIGNAL_PERIOD).mean() # CMO信号线
klines['cmo_slope'] = klines['cmo'].diff(CMO_SLOPE_PERIOD) # CMO斜率
klines['sma'] = klines.close.rolling(SMA_PERIOD).mean() # 趋势确认SMA
# 计算ATR用于动态止损
atr_data = ATR(klines, 14)
# 获取最新数据和前一个交易日数据
current_price = float(klines.close.iloc[-1])
current_datetime = pd.to_datetime(klines.datetime.iloc[-1], unit='ns')
current_cmo = float(klines.cmo.iloc[-1])
current_cmo_signal = float(klines.cmo_signal.iloc[-1])
current_cmo_slope = float(klines.cmo_slope.iloc[-1])
current_sma = float(klines.sma.iloc[-1])
current_atr = float(atr_data.atr.iloc[-1])
prev_price = float(klines.close.iloc[-2])
prev_cmo = float(klines.cmo.iloc[-2])
prev_cmo_signal = float(klines.cmo_signal.iloc[-2])
# 输出调试信息
print(f"日期: {current_datetime.strftime('%Y-%m-%d')}, 价格: {current_price}, CMO: {current_cmo:.2f}, 信号线: {current_cmo_signal:.2f}, 斜率: {current_cmo_slope:.2f}")
# ===== 交易逻辑 =====
# 空仓状态 - 寻找开仓机会
if current_direction == 0:
# 计算多头开仓信号
# 信号1: 超卖反弹
long_signal1 = prev_cmo < OVERSOLD_THRESHOLD and current_cmo > OVERSOLD_THRESHOLD and current_price > current_sma and current_cmo_slope > 0
# 信号2: 信号线交叉
long_signal2 = prev_cmo < prev_cmo_signal and current_cmo > current_cmo_signal and current_price > current_sma and current_cmo > -30
# 信号3: 零轴交叉
long_signal3 = prev_cmo < 0 and current_cmo > 0 and current_price > current_sma and prev_cmo < -10
# 计算空头开仓信号
# 信号1: 超买回落
short_signal1 = prev_cmo > OVERBOUGHT_THRESHOLD and current_cmo < OVERBOUGHT_THRESHOLD and current_price < current_sma and current_cmo_slope < 0
# 信号2: 信号线交叉
short_signal2 = prev_cmo > prev_cmo_signal and current_cmo < current_cmo_signal and current_price < current_sma and current_cmo < 30
# 信号3: 零轴交叉
short_signal3 = prev_cmo > 0 and current_cmo < 0 and current_price < current_sma and prev_cmo > 10
# 多头开仓条件
if long_signal1 or long_signal2 or long_signal3:
# 确定信号强度和头寸规模
signal_strength = 1
# 多个信号同时满足时增加头寸
if sum([long_signal1, long_signal2, long_signal3]) > 1:
signal_strength = 1.5
# 极端CMO值时减少头寸
if abs(current_cmo) > 80:
signal_strength = 0.7
# 设置持仓方向和规模
current_direction = 1
position_size = round(POSITION_SIZE * signal_strength)
entry_position = position_size
target_pos.set_target_volume(position_size)
# 记录开仓信息
entry_price = current_price
entry_date = current_datetime
# 设置止损价格
atr_stop = entry_price - ATR_STOP_MULTIPLIER * current_atr
fixed_stop = entry_price * (1 - FIXED_STOP_LOSS_PCT)
stop_loss_price = max(atr_stop, fixed_stop) # 取较严格的止损
# 设置止盈价格
take_profit_price = entry_price * (1 + TAKE_PROFIT_PCT)
# 记录信号类型
signal_type = ""
if long_signal1: signal_type += "超卖反弹 "
if long_signal2: signal_type += "信号线上穿 "
if long_signal3: signal_type += "零轴上穿 "
print(f"多头开仓: 价格={entry_price}, 手数={position_size}, 信号={signal_type}, 止损={stop_loss_price:.2f}, 止盈={take_profit_price:.2f}")
# 空头开仓条件
elif short_signal1 or short_signal2 or short_signal3:
# 确定信号强度和头寸规模
signal_strength = 1
# 多个信号同时满足时增加头寸
if sum([short_signal1, short_signal2, short_signal3]) > 1:
signal_strength = 1.5
# 极端CMO值时减少头寸
if abs(current_cmo) > 80:
signal_strength = 0.7
# 设置持仓方向和规模
current_direction = -1
position_size = round(POSITION_SIZE * signal_strength)
entry_position = position_size
target_pos.set_target_volume(-position_size)
# 记录开仓信息
entry_price = current_price
entry_date = current_datetime
# 设置止损价格
atr_stop = entry_price + ATR_STOP_MULTIPLIER * current_atr
fixed_stop = entry_price * (1 + FIXED_STOP_LOSS_PCT)
stop_loss_price = min(atr_stop, fixed_stop) # 取较严格的止损
# 设置止盈价格
take_profit_price = entry_price * (1 - TAKE_PROFIT_PCT)
# 记录信号类型
signal_type = ""
if short_signal1: signal_type += "超买回落 "
if short_signal2: signal_type += "信号线下穿 "
if short_signal3: signal_type += "零轴下穿 "
print(f"空头开仓: 价格={entry_price}, 手数={position_size}, 信号={signal_type}, 止损={stop_loss_price:.2f}, 止盈={take_profit_price:.2f}")
# 多头持仓 - 检查平仓条件
elif current_direction == 1:
# 计算持仓天数
holding_days = (current_datetime - entry_date).days
# 3. 基于CMO信号平仓
if (prev_cmo > prev_cmo_signal and current_cmo < current_cmo_signal and current_cmo > 30):
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}%, 持仓天数={holding_days}, 原因=动量减弱")
# 4. 反向信号产生
elif (prev_cmo > OVERBOUGHT_THRESHOLD and current_cmo < OVERBOUGHT_THRESHOLD and current_cmo_slope < 0) or \
(prev_cmo > prev_cmo_signal and current_cmo < current_cmo_signal and current_price < current_sma) or \
(prev_cmo > 0 and current_cmo < 0 and prev_cmo > 10):
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}%, 持仓天数={holding_days}, 原因=反向信号")
# 空头持仓 - 检查平仓条件
elif current_direction == -1:
# 计算持仓天数
holding_days = (current_datetime - entry_date).days
# 3. 基于CMO信号平仓
if (prev_cmo < prev_cmo_signal and current_cmo > current_cmo_signal and current_cmo < -30):
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}%, 持仓天数={holding_days}, 原因=动量减弱")
# 4. 反向信号产生
elif (prev_cmo < OVERSOLD_THRESHOLD and current_cmo > OVERSOLD_THRESHOLD and current_cmo_slope > 0) or \
(prev_cmo < prev_cmo_signal and current_cmo > current_cmo_signal and current_price > current_sma) or \
(prev_cmo < 0 and current_cmo > 0 and prev_cmo < -10):
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}%, 持仓天数={holding_days}, 原因=反向信号")
except BacktestFinished as e:
print("回测结束")
api.close()