Keltner Channel策略
一、交易策略解释
核心思想
Keltner Channel策略是一种基于通道突破的趋势跟踪系统，核心思想是在价格突破由均线和波动率构成的通道时进行交易， 该策略基于市场的趋势行为和价格回归特性，认为价格突破通道边界后，如果有趋势确认，则很可能会继续向突破方向运行，形成显著的价格移动； 而当价格回归到通道中线或突破反向通道边界时，意味着趋势可能已经结束，应当及时平仓。
理论基础
Keltner Channel由Chester W. Keltner于20世纪60年代首次提出，后经Linda Bradford Raschke改进完善。其理论基础主要包含以下几个方面：
-
波动率适应理论：市场波动率不是恒定的，而是周期性变化的。Keltner Channel通过ATR动态调整通道宽度，使策略能够适应不同的市场环境。
-
突破持续性：根据道氏理论和动量原理，价格突破重要技术位置后往往会有延续性，特别是在趋势确认的情况下。
-
均值回归原理：价格在大幅偏离均值后，往往会有回归均值的趋势，这也是策略中使用EMA作为通道中线和平仓依据的理论支持。
策略适用场景
-
趋势明显的市场：黄金、原油等大宗商品在宏观经济变化或地缘政治事件影响下，往往形成持续的趋势行情，是该策略的理想应用场景。
-
波动率适中的市场：波动率既不过低导致无法突破，也不过高导致频繁虚假突破的市场环境最适合该策略。
-
流动性充足的品种：流动性好的产品滑点小，有利于策略的精确执行。
二、天勤介绍
天勤平台概述
天勤（TqSdk）是一个由信易科技开发的开源量化交易系统，为期货、期权等衍生品交易提供专业的量化交易解决方案。平台具有以下特 点：
- 丰富的行情数据： 提供所有可交易合约的全部Tick和K线数据，基于内存数据库实现零延迟访问。
- 一站式的解决方案： 从历史数据分析到实盘交易的完整工具链，打通开发、回测、模拟到实盘的全流程。
- 专业的技术支持： 近百个技术指标源码，深度集成pandas和numpy，采用单线程异步模型保证性能。
策略开发流程
- 环境准备
- 安装Python环境（推荐Python 3.6或以上版本）
- 安装tqsdk包：pip install tqsdk
- 注册天勤账户获取访问密钥
- 数据准备
- 订阅近月与远月合约行情
- 获取历史K线或者Tick数据（用于分析与行情推进）
- 策略编写
- 设计信号生成逻辑（基于价差、均值和标准差）
- 编写交易执行模块（开仓、平仓逻辑）
- 实现风险控制措施（止损、资金管理）
- 回测验证
- 设置回测时间区间和初始资金
- 运行策略获取回测结果
- 分析绩效指标（胜率、收益率、夏普率等）
- 策略优化
- 调整参数（标准差倍数、窗口大小等）
- 添加过滤条件（成交量、波动率等）
- 完善风险控制机制
三、天勤实现策略
策略原理
核心指标计算：
- Keltner Channel上下轨： 通过动态ATR乘数调整通道宽度，使策略能适应不同的市场环境。
dynamic_multiplier = ATR_MULTIPLIER
if trend_strength > 0.5: # 强趋势时使用更窄的通道
dynamic_multiplier = ATR_MULTIPLIER * 0.8
upper_band = ema + dynamic_multiplier * atr
lower_band = ema - dynamic_multiplier * atr
- EMA（指数移动平均线）： 作为Keltner Channel的中轨，EMA比SMA反应更快速，能更好地跟踪价格变化。
ema = pd.Series(close).ewm(span=EMA_PERIOD, adjust=False).mean().values
- ATR（真实波幅）： 用于测量市场波动率，是通道宽度的决定因素。
tr = np.maximum(high - low,
np.maximum(
np.abs(high - np.roll(close, 1)),
np.abs(low - np.roll(close, 1))
))
atr = pd.Series(tr).rolling(ATR_PERIOD).mean().values
- 趋势确认： 使用短期EMA与主EMA的交叉确认趋势方向，降低虚假突破风险。
trend_direction = 1 if ema_short[-1] > ema[-1] else -1 if ema_short[-1] < ema[-1] else 0
trend_strength = abs(ema_short[-1] - ema[-1]) / close[-1] * 100 # 趋势强度百分比
- 动态止损设置： 基于ATR设置初始止损，并根据价格走势进行移动止损调整。
# 初始止损
stop_loss = current_price * (1 - STOP_LOSS_PCT * current_atr / current_price) # 多头止损
# 移动止损
if TRAILING_STOP and high_since_entry > entry_price:
trailing_stop = high_since_entry * (1 - STOP_LOSS_PCT * current_atr / current_price)
stop_loss = max(stop_loss, trailing_stop)
交易逻辑
开仓信号：
- 多头开仓：价格突破上轨道且短期EMA位于长期EMA之上（趋势向上）
if current_price > current_upper and trend_direction > 0:
# 开多头逻辑
- 空头开仓：价格跌破下轨道且短期EMA位于长期EMA之下（趋势向下）
elif current_price < current_lower and trend_direction < 0:
# 开空头逻辑
平仓信号：
- 多头平仓：满足以下任一条件时平仓：
- 触及止损位
- 价格跌破通道中线（EMA）
- 价格跌破下轨且趋势转为向下
if (current_price <= stop_loss or
current_price <= current_ema or
(current_price < current_lower and trend_direction < 0)):
# 平多头逻辑
- 空头平仓：满足以下任一条件时平仓：
- 触及止损位
- 价格突破通道中线（EMA）
- 价格突破上轨且趋势转为向上
if (current_price >= stop_loss or
current_price >= current_ema or
(current_price > current_upper and trend_direction > 0)):
# 平空头逻辑
- 风险管理：
- 动态止损：初始止损基于入场价格和当前ATR设置
- 移动止损：随着价格向有利方向移动，止损线也相应调整，锁定部分利润
- 趋势过滤：使用短期EMA和长期EMA交叉确认趋势，减少虚假突破
- 动态通道宽度：根据趋势强度动态调整ATR乘数，在强趋势中使用更紧的通道获取更早的信号
回测
回测初始设置
- 测试周期： 2022 年 11 月 1 日 - 2023 年 4 月 30 日
- 交易品种： CFFEX.IC2306
- 初始资金： 1000万元
回测结果
上述回测累计收益走势图
完整代码示例
#!/usr/bin/env python
# -*- coding: utf-8 -*-
__author__ = "Chaos"
from tqsdk import TqApi, TqAuth, TqBacktest, TargetPosTask, BacktestFinished
from datetime import date
import numpy as np
import pandas as pd
# ===== 全局参数设置 =====
SYMBOL = "CFFEX.IC2306" # 中证500指数期货合约
POSITION_SIZE = 30 # 持仓手数（黄金的合适仓位）
START_DATE = date(2022, 11, 1) # 回测开始日期
END_DATE = date(2023, 4, 30) # 回测结束日期
# Keltner Channel参数
EMA_PERIOD = 8 # EMA周期
ATR_PERIOD = 7 # ATR周期
ATR_MULTIPLIER = 1.5 # ATR乘数
# 新增参数 - 趋势确认与止损
SHORT_EMA_PERIOD = 5 # 短期EMA用于趋势确认
STOP_LOSS_PCT = 0.8 # 止损百分比（相对于ATR）
TRAILING_STOP = True # 使用移动止损
print(f"开始回测 {SYMBOL} 的Keltner Channel策略...")
print(f"参数: EMA周期={EMA_PERIOD}, ATR周期={ATR_PERIOD}, ATR乘数={ATR_MULTIPLIER}")
print(f"额外参数: 短期EMA={SHORT_EMA_PERIOD}, 止损参数={STOP_LOSS_PCT}ATR, 移动止损={TRAILING_STOP}")
try:
api = TqApi(backtest=TqBacktest(start_dt=START_DATE, end_dt=END_DATE),
auth=TqAuth("快期账号", "快期密码"))
# 订阅K线数据
klines = api.get_kline_serial(SYMBOL, 60 * 60 * 24) # 日K线
# 订阅行情获取交易时间
quote = api.get_quote(SYMBOL)
target_pos = TargetPosTask(api, SYMBOL)
# 初始化交易状态
position = 0 # 当前持仓
entry_price = 0 # 入场价格
stop_loss = 0 # 止损价格
high_since_entry = 0 # 入场后的最高价（用于移动止损）
low_since_entry = 0 # 入场后的最低价（用于移动止损）
trend_strength = 0 # 趋势强度
# 记录交易信息
trades = []
while True:
api.wait_update()
if api.is_changing(klines):
# 确保有足够的数据
if len(klines) < max(EMA_PERIOD, ATR_PERIOD, SHORT_EMA_PERIOD) + 1:
continue
# 计算指标
close = klines.close.values
high = klines.high.values
low = klines.low.values
# 计算中轨（EMA）和短期EMA（用于趋势确认）
ema = pd.Series(close).ewm(span=EMA_PERIOD, adjust=False).mean().values
ema_short = pd.Series(close).ewm(span=SHORT_EMA_PERIOD, adjust=False).mean().values
# 计算趋势方向和强度
trend_direction = 1 if ema_short[-1] > ema[-1] else -1 if ema_short[-1] < ema[-1] else 0
trend_strength = abs(ema_short[-1] - ema[-1]) / close[-1] * 100 # 趋势强度百分比
# 计算ATR
tr = np.maximum(high - low,
np.maximum(
np.abs(high - np.roll(close, 1)),
np.abs(low - np.roll(close, 1))
))
atr = pd.Series(tr).rolling(ATR_PERIOD).mean().values
current_atr = float(atr[-1])
# 动态调整ATR乘数，根据趋势强度调整通道宽度
dynamic_multiplier = ATR_MULTIPLIER
if trend_strength > 0.5: # 强趋势时使用更窄的通道
dynamic_multiplier = ATR_MULTIPLIER * 0.8
# 计算通道上下轨
upper_band = ema + dynamic_multiplier * atr
lower_band = ema - dynamic_multiplier * atr
# 获取当前价格和指标值
current_price = float(close[-1])
current_upper = float(upper_band[-1])
current_lower = float(lower_band[-1])
current_ema = float(ema[-1])
current_time = quote.datetime # 使用quote的datetime获取当前时间
# 更新入场后的最高/最低价
if position > 0:
high_since_entry = max(high_since_entry, current_price)
# 更新移动止损
if TRAILING_STOP and high_since_entry > entry_price:
trailing_stop = high_since_entry * (1 - STOP_LOSS_PCT * current_atr / current_price)
stop_loss = max(stop_loss, trailing_stop)
elif position < 0:
low_since_entry = min(low_since_entry, current_price)
# 更新移动止损
if TRAILING_STOP and low_since_entry < entry_price:
trailing_stop = low_since_entry * (1 + STOP_LOSS_PCT * current_atr / current_price)
stop_loss = min(stop_loss if stop_loss > 0 else float('inf'), trailing_stop)
# 交易逻辑
if position == 0: # 空仓
# 确认趋势方向并突破通道
if current_price > current_upper and trend_direction > 0:
# 增加成交量过滤
position = POSITION_SIZE
entry_price = current_price
high_since_entry = current_price
low_since_entry = current_price
# 设置初始止损
stop_loss = current_price * (1 - STOP_LOSS_PCT * current_atr / current_price)
target_pos.set_target_volume(position)
print(f"开多仓: 价格={current_price:.2f}, 上轨={current_upper:.2f}, 止损={stop_loss:.2f}")
trades.append(("开多", current_time, current_price))
elif current_price < current_lower and trend_direction < 0:
position = -POSITION_SIZE
entry_price = current_price
high_since_entry = current_price
low_since_entry = current_price
# 设置初始止损
stop_loss = current_price * (1 + STOP_LOSS_PCT * current_atr / current_price)
target_pos.set_target_volume(position)
print(f"开空仓: 价格={current_price:.2f}, 下轨={current_lower:.2f}, 止损={stop_loss:.2f}")
trades.append(("开空", current_time, current_price))
elif position > 0: # 持有多头
# 止损、回落到中轨或趋势转向时平仓
if (current_price <= stop_loss or
current_price <= current_ema or
(current_price < current_lower and trend_direction < 0)):
profit_pct = (current_price / entry_price - 1) * 100
profit_points = current_price - entry_price
target_pos.set_target_volume(0)
print(f"平多仓: 价格={current_price:.2f}, 盈亏={profit_pct:.2f}%, {profit_points:.2f}点")
position = 0
entry_price = 0
stop_loss = 0
trades.append(("平多", current_time, current_price))
elif position < 0: # 持有空头
# 止损、回升到中轨或趋势转向时平仓
if (current_price >= stop_loss or
current_price >= current_ema or
(current_price > current_upper and trend_direction > 0)):
profit_pct = (entry_price / current_price - 1) * 100
profit_points = entry_price - current_price
target_pos.set_target_volume(0)
print(f"平空仓: 价格={current_price:.2f}, 盈亏={profit_pct:.2f}%, {profit_points:.2f}点")
position = 0
entry_price = 0
stop_loss = 0
trades.append(("平空", current_time, current_price))
except BacktestFinished as e:
print(f"回测完成: {e}")
api.close()