Keltner Channel策略

一、交易策略解释

核心思想

Keltner Channel策略是一种基于通道突破的趋势跟踪系统，核心思想是在价格突破由均线和波动率构成的通道时进行交易，该策略基于市场的趋势行为和价格回归特性，认为价格突破通道边界后，如果有趋势确认，则很可能会继续向突破方向运行，形成显著的价格移动；而当价格回归到通道中线或突破反向通道边界时，意味着趋势可能已经结束，应当及时平仓。

理论基础

Keltner Channel由Chester W. Keltner于20世纪60年代首次提出，后经Linda Bradford Raschke改进完善。其理论基础主要包含以下几个方面：

波动率适应理论：市场波动率不是恒定的，而是周期性变化的。Keltner Channel通过ATR动态调整通道宽度，使策略能够适应不同的市场环境。
突破持续性：根据道氏理论和动量原理，价格突破重要技术位置后往往会有延续性，特别是在趋势确认的情况下。
均值回归原理：价格在大幅偏离均值后，往往会有回归均值的趋势，这也是策略中使用EMA作为通道中线和平仓依据的理论支持。

策略适用场景

趋势明显的市场：黄金、原油等大宗商品在宏观经济变化或地缘政治事件影响下，往往形成持续的趋势行情，是该策略的理想应用场景。
波动率适中的市场：波动率既不过低导致无法突破，也不过高导致频繁虚假突破的市场环境最适合该策略。

流动性充足的品种：流动性好的产品滑点小，有利于策略的精确执行。

策略原理

核心指标计算：

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万元

回测结果

Keltner Channel策略 - 图片 1

上述回测累计收益走势图

Keltner Channel策略 - 图片 2

代码

#!/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}")

投资思想 · 其他文章 (273)