CCI均值回归策略
一、交易策略解释
核心思想
CCI均值回归策略的核心思想是利用价格在短期内的极端偏离往往会回归均值的市场特性。当价格显著偏离其移动平均值时，市场可能处于超买或超卖状态，此时价格往往会出现反转并向平均值回归。CCI指标通过衡量当前价格与其移动平均线之间的偏离程度，帮助交易者识别这些潜在的反转点。
具体而言，当CCI从低于-100上升至-100以上时，表明市场可能从超卖状态回归，产生买入信号；相反，当CCI从高于+100下降至+100以下时，表明市场可能从超买状态回归，产生卖出信号。
理论基础
CCI均值回归策略的理论基础主要来自于金融市场的均值回归特性和技术分析理论：
-
随机游走理论修正：虽然有效市场假说认为市场价格遵循随机游走，但实证研究表明，在短期内市场价格存在一定程度的可预测性，特别是在价格出现极端波动后。
-
行为金融学解释：投资者过度反应和羊群效应导致市场价格短期内偏离基本面价值，但随着时间推移，理性力量会将价格拉回合理水平。
-
统计显著性：多项研究表明，CCI指标在各类市场中具有一定的预测能力，尤其是在识别周期性商品的价格转折点方面。例如，Lambert在1980年的原始研究中，CCI在商品市场的周期性识别上显示出88%的准确率。
策略适用场景
-
震荡市场：价格在一定区间内上下波动的市场环境，是该策略的理想应用场景。
-
商品期货市场：由于商品价格受供需关系影响明显，往往呈现更规律的周期性波动，特别适合CCI策略。
-
中等波动性品种：波动太小则难以产生有效信号，波动过大则可能导致频繁止损。
-
流动性充足的品种：确保在信号触发时能够以合理价格执行交易
二、天勤介绍
天勤平台概述
天勤（TqSdk）是一个由信易科技开发的开源量化交易系统，为期货、期权等衍生品交易提供专业的量化交易解决方案。平台具有以下特 点：
- 丰富的行情数据： 提供所有可交易合约的全部Tick和K线数据，基于内存数据库实现零延迟访问。
- 一站式的解决方案： 从历史数据分析到实盘交易的完整工具链，打通开发、回测、模拟到实盘的全流程。
- 专业的技术支持： 近百个技术指标源码，深度集成pandas和numpy，采用单线程异步模型保证性能。
策略开发流程
- 环境准备
- 安装Python环境（推荐Python 3.6或以上版本）
- 安装tqsdk包：pip install tqsdk
- 注册天勤账户获取访问密钥
- 数据准备
- 订阅近月与远月合约行情
- 获取历史K线或者Tick数据（用于分析与行情推进）
- 策略编写
- 设计信号生成逻辑（基于价差、均值和标准差）
- 编写交易执行模块（开仓、平仓逻辑）
- 实现风险控制措施（止损、资金管理）
- 回测验证
- 设置回测时间区间和初始资金
- 运行策略获取回测结果
- 分析绩效指标（胜率、收益率、夏普率等）
- 策略优化
- 调整参数（标准差倍数、窗口大小等）
- 添加过滤条件（成交量、波动率等）
- 完善风险控制机制
三、天勤实现策略
策略原理
CCI指标的计算方法如下：
- 计算典型价格(Typical Price, TP)：
TP = (最高价 + 最低价 + 收盘价) / 3
- 计算典型价格的N日简单移动平均(SMA)：
SMA(TP, N) = (TP₁ + TP₂ + ... + TPₙ) / N
- 计算典型价格与其移动平均值之间的平均偏差(Mean Deviation, MD)：
MD = (|TP₁ - SMA| + |TP₂ - SMA| + ... + |TPₙ - SMA|) / N
- 计算CCI值：
CCI = (TP - SMA(TP, N)) / (0.015 * MD)
在天勤量化平台中，我们直接使用内置的CCI函数计算指标，设置周期参数为10日。相比传统的20日周期，10日周期能够更快捕捉市场变化，但同时也会产生更多交易信号。
交易逻辑
入场信号：
-
多头入场：当CCI从低于-100上升至-100以上时发出买入信号
-
空头入场：当CCI从高于+100下降至+100以下时发出卖出信号
出场信号：
-
多头平仓：当CCI上升至+100以上后回落至+100以下时
-
空头平仓：当CCI下降至-100以下后上升至-100以上时
止损设置：
-
多头止损：入场点下方设置固定比例止损，如1.5%
-
空头止损：入场点上方设置固定比例止损，如1.5%
回测
回测初始设置
- 测试周期： 2023 年 9 月 20 日 - 2024 年 2 月 28 日
- 交易品种： SHFE.au2406
- 初始资金： 1000万元
回测结果
上述回测累计收益走势图
完整代码示例
#!/usr/bin/env python
# -*- coding: utf-8 -*-
__author__ = "Chaos"
from datetime import date
from tqsdk import TqApi, TqAuth, TqBacktest, TargetPosTask, BacktestFinished
from tqsdk.tafunc import time_to_str
from tqsdk.ta import CCI
# ===== 全局参数设置 =====
SYMBOL = "SHFE.au2406" # 黄金期货主力合约
POSITION_SIZE = 50 # 每次交易手数
START_DATE = date(2023, 9, 20) # 回测开始日期
END_DATE = date(2024, 2, 28) # 回测结束日期
# CCI参数
CCI_PERIOD = 10 # CCI计算周期
CCI_UPPER = 100 # CCI上轨
CCI_LOWER = -100 # CCI下轨
STOP_LOSS_PERCENT = 0.01 # 止损比例
# ===== 全局变量 =====
current_direction = 0 # 当前持仓方向：1=多头，-1=空头，0=空仓
entry_price = 0 # 开仓价格
stop_loss_price = 0 # 止损价格
last_cci_state = 0 # 上一次CCI状态：1=高于上轨，-1=低于下轨，0=中间区域
# ===== 策略开始 =====
print("开始运行CCI均值回归策略...")
# 创建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) < CCI_PERIOD + 10:
continue
# 计算CCI
cci_series = CCI(klines, CCI_PERIOD)
cci = cci_series.iloc[-1].item() # 使用item()方法获取标量值
current_price = klines.close.iloc[-1].item() # 同样使用item()方法
# 确定当前CCI状态
current_cci_state = 0
if cci > CCI_UPPER:
current_cci_state = 1
elif cci < CCI_LOWER:
current_cci_state = -1
# 获取最新数据
current_timestamp = klines.datetime.iloc[-1]
current_datetime = time_to_str(current_timestamp)
# 打印当前状态
print(f"日期: {current_datetime}, 价格: {current_price:.2f}, CCI: {cci:.2f}")
# ===== 交易逻辑 =====
# 空仓状态 - 寻找开仓机会
if current_direction == 0:
# 多头开仓条件：CCI从下轨上穿
if last_cci_state == -1 and current_cci_state == 0:
current_direction = 1
target_pos.set_target_volume(POSITION_SIZE)
entry_price = current_price
stop_loss_price = entry_price * (1 - STOP_LOSS_PERCENT)
print(f"多头开仓: 价格={entry_price:.2f}, 止损价={stop_loss_price:.2f}")
# 空头开仓条件：CCI从上轨下穿
elif last_cci_state == 1 and current_cci_state == 0:
current_direction = -1
target_pos.set_target_volume(-POSITION_SIZE)
entry_price = current_price
stop_loss_price = entry_price * (1 + STOP_LOSS_PERCENT)
print(f"空头开仓: 价格={entry_price:.2f}, 止损价={stop_loss_price:.2f}")
# 多头持仓 - 检查平仓条件
elif current_direction == 1:
# 止盈条件：CCI从上轨下穿
if last_cci_state == 1 and current_cci_state == 0:
profit_pct = (current_price - entry_price) / entry_price * 100
target_pos.set_target_volume(0)
current_direction = 0
print(f"多头止盈平仓: 价格={current_price:.2f}, 盈亏={profit_pct:.2f}%")
# 止损条件：价格跌破止损价
elif current_price < stop_loss_price:
loss_pct = (entry_price - current_price) / entry_price * 100
target_pos.set_target_volume(0)
current_direction = 0
print(f"多头止损平仓: 价格={current_price:.2f}, 亏损={loss_pct:.2f}%")
# 空头持仓 - 检查平仓条件
elif current_direction == -1:
# 止盈条件：CCI从下轨上穿
if last_cci_state == -1 and current_cci_state == 0:
profit_pct = (entry_price - current_price) / entry_price * 100
target_pos.set_target_volume(0)
current_direction = 0
print(f"空头止盈平仓: 价格={current_price:.2f}, 盈亏={profit_pct:.2f}%")
# 止损条件：价格突破止损价
elif current_price > stop_loss_price:
loss_pct = (current_price - entry_price) / entry_price * 100
target_pos.set_target_volume(0)
current_direction = 0
print(f"空头止损平仓: 价格={current_price:.2f}, 亏损={loss_pct:.2f}%")
# 更新上一次CCI状态
last_cci_state = current_cci_state
except BacktestFinished as e:
print("回测结束")
api.close()