76.2 趋势跟踪策略设计

项目二：期货CTA策略

🎯 学习目标

  • 掌握趋势跟踪策略的原理
  • 学会常用趋势识别指标
  • 理解信号生成逻辑
  • 能够设计完整的趋势策略
趋势跟踪策略设计

趋势跟踪策略设计

趋势跟踪策略是CTA策略中最经典、应用最广泛的策略类型，核心思想是"截断亏损，让利润奔跑"。

📈 趋势识别指标

移动平均线

def moving_average_crossover(close, short=20, long=60):
    """
    均线交叉策略
    """
    ma_short = close.rolling(short).mean()
    ma_long = close.rolling(long).mean()

    # 金叉买入，死叉卖出
    signal = pd.DataFrame(index=close.index)
    signal['ma_diff'] = ma_short - ma_long
    signal['buy'] = (signal['ma_diff'] > 0) & (signal['ma_diff'].shift(1) <= 0)
    signal['sell'] = (signal['ma_diff'] < 0) & (signal['ma_diff'].shift(1) >= 0)

    return signal

布林带

def bollinger_bands(close, period=20, std_dev=2):
    """
    布林带策略
    """
    ma = close.rolling(period).mean()
    std = close.rolling(period).std()

    upper = ma + std_dev * std
    lower = ma - std_dev * std

    # 突破买入，跌破卖出
    signal = pd.DataFrame(index=close.index)
    signal['buy'] = close > upper
    signal['sell'] = close < lower

    return signal, {'upper': upper, 'middle': ma, 'lower': lower}

唐奇安通道

def donchian_channel(close, period=20):
    """
    唐奇安通道策略
    """
    upper = close.rolling(period).max()
    lower = close.rolling(period).min()

    # 突破20日高点买入，跌破20日低点卖出
    signal = pd.DataFrame(index=close.index)
    signal['buy'] = close > upper.shift(1)
    signal['sell'] = close < lower.shift(1)

    return signal, {'upper': upper, 'lower': lower}

⚙️ 趋势强度指标

def adx(high, low, close, period=14):
    """
    平均趋向指标（ADX）
    用于判断趋势强度
    """
    # 计算真实波幅
    tr1 = high - low
    tr2 = abs(high - close.shift(1))
    tr3 = abs(low - close.shift(1))
    tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1)

    # 计算+DM和-DM
    plus_dm = high.diff()
    minus_dm = -low.diff()
    plus_dm[plus_dm < 0] = 0
    plus_dm[plus_dm < minus_dm] = 0
    minus_dm[minus_dm < 0] = 0
    minus_dm[minus_dm < plus_dm] = 0

    # 计算平滑值
    atr = tr.ewm(alpha=1/period).mean()
    plus_di = 100 * (plus_dm.ewm(alpha=1/period).mean() / atr)
    minus_di = 100 * (minus_dm.ewm(alpha=1/period).mean() / atr)

    # 计算ADX
    dx = 100 * abs(plus_di - minus_di) / (plus_di + minus_di)
    adx = dx.ewm(alpha=1/period).mean()

    return adx, plus_di, minus_di

def trend_strength_filter(adx, threshold=25):
    """
    趋势强度过滤
    """
    return adx > threshold

🎯 信号生成逻辑

class TrendFollowingStrategy:
    """
    趋势跟踪策略
    """

    def __init__(self, params=None):
        self.params = params or {
            'ma_short': 20,
            'ma_long': 60,
            'adx_threshold': 25,
            'atr_period': 14,
            'atr_multiplier': 2
        }

    def generate_signals(self, data):
        """
        生成交易信号
        """
        close = data['close']
        high = data['high']
        low = data['low']

        # 计算指标
        signals = {}

        # 1. 均线交叉信号
        ma_short = close.rolling(self.params['ma_short']).mean()
        ma_long = close.rolling(self.params['ma_long']).mean()
        signals['ma_cross'] = (ma_short > ma_long).astype(int)

        # 2. ADX趋势强度
        adx, plus_di, minus_di = adx(high, low, close)
        signals['trend_strength'] = adx > self.params['adx_threshold']

        # 3. ATR止损位
        atr = self._calculate_atr(high, low, close, self.params['atr_period'])
        signals['atr_stop'] = atr * self.params['atr_multiplier']

        # 4. 综合信号
        signals['long'] = (
            (signals['ma_cross'] == 1) &
            signals['trend_strength']
        )
        signals['short'] = (
            (signals['ma_cross'] == 0) &
            signals['trend_strength']
        )

        return signals

    def _calculate_atr(self, high, low, close, period):
        """
        计算ATR（真实波幅）
        """
        tr1 = high - low
        tr2 = abs(high - close.shift(1))
        tr3 = abs(low - close.shift(1))
        tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1)
        return tr.rolling(period).mean()

📊 策略组合设计

class MultiTrendStrategy:
    """
    多周期趋势组合策略
    """

    def __init__(self):
        self.short_term = TrendFollowingStrategy({'ma_short': 5, 'ma_long': 20})
        self.medium_term = TrendFollowingStrategy({'ma_short': 20, 'ma_long': 60})
        self.long_term = TrendFollowingStrategy({'ma_short': 60, 'ma_long': 120})

    def generate_signals(self, data):
        """
        生成组合信号
        """
        # 各周期信号
        short_signals = self.short_term.generate_signals(data)
        medium_signals = self.medium_term.generate_signals(data)
        long_signals = self.long_term.generate_signals(data)

        # 信号融合
        combined = pd.DataFrame(index=data.index)

        # 趋势强度投票
        long_votes = (
            short_signals['long'].astype(int) +
            medium_signals['long'].astype(int) +
            long_signals['long'].astype(int)
        )
        short_votes = (
            short_signals['short'].astype(int) +
            medium_signals['short'].astype(int) +
            long_signals['short'].astype(int)
        )

        # 综合信号（至少2个周期一致）
        combined['long'] = long_votes >= 2
        combined['short'] = short_votes >= 2

        return combined
✨
趋势策略关键

趋势跟踪策略的成功要素：1）及时识别趋势；2）严格止损；3）合理加仓；4）控制回撤；5）多品种分散。

📝 本节小结

✅
  • • 掌握了趋势跟踪策略的核心原理
  • • 学会了常用趋势识别指标
  • • 实现了完整的信号生成逻辑
  • • 理解了多周期组合的方法
  • • 掌握了趋势强度过滤技巧
← 上一节 下一节 →