趋势跟随：如何在六成时间都在亏钱的情况下赚钱

多数趋势跟随系统在 55% 至 65% 的交易中亏钱。然而它们为理查德·丹尼斯、艾德·塞柯塔、比尔·邓恩这些传奇交易者赚下了数十亿美元。秘诀是什么？狠心止损，并让盈利的仓位一直跑到趋势结束。下面讲这套数学如何成立——以及为什么多数交易者受不了它。

什么是趋势跟随？

趋势跟随简单得有些欺骗性：

1. 识别新趋势何时启动 （突破、均线交叉等）
2. 按趋势方向 建立仓位
3. 使用追踪止损 在趋势延续时锁定收益
4. 趋势反转时离场 （触及止损或出现反转信号）

核心原则： 快速止损（每笔承担 2% 至 3% 的风险），让盈利奔跑（20%、50%、100% 以上）。你在多数交易上都会亏钱，但少数几笔大赢足以补回来还有余。

趋势跟随为何有效

市场的趋势比你以为的多

学术研究显示，市场有 30% 至 40% 的时间处于强趋势中。剩下的 60% 至 70% 呢？震荡、区间内的噪音。

趋势跟随者的优势：

• 抓住大级别行情： 2009-2020 年的牛市、2022 年的熊市、商品超级周期
• 在震荡中接受小亏： 区间行情中每笔失败交易吐回 1% 至 2%
• 最终结果： 大赢压倒众多小亏

行为层面的优势（它为何持续有效）

趋势跟随利用了人的三种偏差：

1. 锚定： 投资者锚定在近期价格上（"不可能再涨了"），而趋势仍在延续
2. 处置效应： 投资者过早卖出赢家（趋势跟随者继续骑着），过久持有输家（趋势跟随者快刀斩断）
3. 羊群效应： 趋势展开时更多投资者涌入，把行情推得更远

三种经过验证的趋势跟随方法

方法一：均线交叉（经典款）

50/200 日均线"金叉"：

买入信号： 50 日均线上穿 200 日均线（金叉）

卖出信号： 50 日均线下穿 200 日均线（死叉）

它为何有效： 50/200 交叉很慢。它错过趋势最初的 10% 至 20%，但吃到了中段最肥的 50% 至 70%。它离场也晚，从峰值吐回 10% 至 15%。但它在熊市里空仓，从而躲开了 40% 至 50% 的崩盘。

更好的替代：20/50 EMA（指数移动平均线）

• 对趋势变化反应更快（EMA 给近期价格更高权重）
• 更早捕捉趋势，但来回打脸也更多
• 回测显示年化 9.4%，最大回撤 -22.1%（2000-2024）
• 胜率降到 38%，但平均盈亏比改善到 3.8:1

方法二：唐奇安通道突破（海龟系统）

这套策略让理查德·丹尼斯的"海龟"们把 1,000 美元变成了超过 1 亿美元：

设置：

• 唐奇安通道： 过去 N 天的最高价与最低价（通常取 20 或 55 天）
• 买入信号： 价格突破 20 日高点（创新高 = 趋势启动）
• 卖出信号： 价格跌破 10 日低点（趋势反转）
• 做空信号： 价格跌破 20 日低点（用于下降趋势）

仓位管理（海龟的秘密武器）：

• 每笔风险： 组合的 2%
• 止损： 入场价下方 2 × ATR（平均真实波幅）
• 仓位规模： （组合 × 2%）÷（2 × ATR）

示例（SPY 突破）：

• 组合：100,000 美元
• SPY 价格：450 美元
• ATR（14 日）：6.00 美元
• 止损：2 × 6 = 入场价下方 12 美元 = 438 美元
• 每股风险：12 美元
• 目标风险：100,000 × 2% = 2,000 美元
• 仓位规模：2,000 ÷ 12 = 166 股（约 75,000 美元的仓位）

方法三：ATR 追踪止损（随波动率自适应）

固定止损的问题： 对低波动股票设 2% 止损，正常噪音就会把你扫出局；对高波动股票设 2% 止损，则完全不给交易留出空间。

解决方案：基于 ATR 的追踪止损

运作方式：

1. 计算 ATR(14)： 14 天的平均真实波幅（衡量波动率）
2. 入场： 在突破时买入（唐奇安、均线交叉等）
3. 初始止损： 入场价 -（3 × ATR）
4. 追踪止损： 价格上涨时，把止损上移到：最高收盘价 -（3 × 当前 ATR）
5. 离场： 当价格收在追踪止损之下时

为什么是 3 × ATR？

• 2 × ATR： 太紧，正常波动就会止损出局（胜率约 30%）
• 3 × ATR： 最佳平衡（经 30 年检验，胜率约 40%）
• 4 × ATR： 太松，吐回太多利润（夏普更低）

不同市场环境下的表现

When 趋势跟随 Dominates

When 趋势跟随 Struggles

震荡横盘的市场（2015-2016、2018、2023）：

• 价格来回拉锯
• 每次突破都失败（反复被止损）
• 胜率降到 25% 至 30%
• 年度回报：-5% 到 +2%（令人抓狂！）

V 形反弹（2020 年新冠）：

• 市场急速崩盘（趋势跟随者转现金或做空 = 正确）
• 市场急速反转（趋势跟随者仍持现金 = 错过反弹最初的 20%）
• 在假启动中被反复打脸
• 最后迟迟入场，结果下一次回调就被止损

交易成本：沉默的杀手

趋势跟随系统交易频繁（每年 20 至 60 笔）。成本很重要。

成本拆解（每笔来回交易）：

• 佣金： 0 至 1 美元（在 Robinhood、盈透证券基本免费）
• 买卖价差： 0.02% 至 0.10%（SPY、QQQ 这类流动性好的 ETF）
• 滑点： 0.05% 至 0.15%（高波动股票上的市价单）
• 总成本： 每笔来回约 0.10% 至 0.25%

年度影响（每年 40 笔）：

• 40 笔 × 0.15% = 每年 6% 的拖累
• 一个毛回报 15% 的策略，净回报只剩 9%（缩水 40%！）

如何把它压到最低：

• 交易流动性好的品种： SPY/QQQ（价差 0.01%）对比个别小盘股（价差 0.5%）
• 使用限价单： 每笔省下 0.05% 至 0.10%（但可能错过入场）
• 在收盘时段交易： 避开盘中反复，降低交易频率
• 止损放宽： 3 × ATR 相比 2 × ATR，可把打脸次数减少约 30%

何时该放弃这套策略

趋势跟随并非永远有效。要知道什么时候该退到一边。

警告信号（该降低仓位或转为现金）：

1. 低波动环境（VIX < 12）：
   • 平稳缓慢的市场 = 频繁被打脸
   • 止损相对太宽（ATR 很小时用 3 × ATR = 风险回报比很差）
   • 行动： 把仓位减半，或等波动率回来
2. 被打脸过于频繁（连续 6 笔以上亏损）：
   • 说明市场在区间内震荡，而非趋势行情
   • 行动： 停止交易，等 4 至 8 周看环境是否切换
3. 回撤超过 25%：
   • 对趋势跟随来说属正常，但很痛苦
   • 行动： 把仓位减半，在艰难时期保住本金

Python 实现：50/200 日均线趋势跟随

下面是一套完整的趋势跟随系统，你可以拿去回测：

"""
Trend Following System: 50/200 SMA Golden Cross
Author: Plan My Retire Finance University
Date: 2026-02-22

Strategy:
- Buy when 50-day SMA crosses above 200-day SMA (golden cross)
- Sell when 50-day SMA crosses below 200-day SMA (death cross)
- Risk management: 2% max loss per trade using ATR-based stops
"""

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import yfinance as yf
from datetime import datetime

class TrendFollowingSystem:
    """
    Moving average crossover trend following system

    Parameters:
    -----------
    fast_period : int
        Fast moving average period (default 50)
    slow_period : int
        Slow moving average period (default 200)
    atr_period : int
        ATR period for volatility (default 14)
    atr_multiplier : float
        Stop loss = ATR × multiplier (default 3.0)
    risk_per_trade : float
        Portfolio risk per trade (default 0.02 = 2%)
    """

    def __init__(self, fast_period=50, slow_period=200,
                 atr_period=14, atr_multiplier=3.0, risk_per_trade=0.02):
        self.fast_period = fast_period
        self.slow_period = slow_period
        self.atr_period = atr_period
        self.atr_multiplier = atr_multiplier
        self.risk_per_trade = risk_per_trade

    def calculate_atr(self, data):
        """Calculate Average True Range (ATR) for volatility"""
        high = data['High']
        low = data['Low']
        close = data['Close']

        tr1 = high - low
        tr2 = abs(high - close.shift())
        tr3 = abs(low - close.shift())

        tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1)
        atr = tr.rolling(window=self.atr_period).mean()

        return atr

    def generate_signals(self, data):
        """Generate buy/sell signals based on MA crossover"""
        df = data.copy()

        # Calculate moving averages
        df['SMA_Fast'] = df['Close'].rolling(window=self.fast_period).mean()
        df['SMA_Slow'] = df['Close'].rolling(window=self.slow_period).mean()

        # Calculate ATR for stops
        df['ATR'] = self.calculate_atr(df)

        # Generate signals
        df['Signal'] = 0
        df.loc[df['SMA_Fast'] > df['SMA_Slow'], 'Signal'] = 1  # Long
        df.loc[df['SMA_Fast'] < df['SMA_Slow'], 'Signal'] = -1  # Flat/Cash

        # Detect crossovers
        df['Position'] = df['Signal'].diff()
        df.loc[df['Position'] == 2, 'Buy'] = df['Close']  # Golden cross
        df.loc[df['Position'] == -2, 'Sell'] = df['Close']  # Death cross

        return df

    def backtest(self, data, initial_capital=100000, transaction_cost=0.001):
        """
        Run backtest with proper position sizing and transaction costs

        Parameters:
        -----------
        data : DataFrame
            Price data with OHLC columns
        initial_capital : float
            Starting portfolio value
        transaction_cost : float
            Cost per trade as percentage (0.001 = 0.1%)

        Returns:
        --------
        results : dict
            Performance metrics and equity curve
        """
        df = self.generate_signals(data)

        # Initialize tracking
        portfolio_value = initial_capital
        cash = initial_capital
        shares = 0
        equity_curve = []
        trades = []

        for i in range(len(df)):
            row = df.iloc[i]
            date = df.index[i]

            # Track portfolio value
            if shares > 0:
                portfolio_value = cash + (shares * row['Close'])
            else:
                portfolio_value = cash

            equity_curve.append({
                'Date': date,
                'Portfolio_Value': portfolio_value,
                'Position': 'Long' if shares > 0 else 'Cash'
            })

            # Execute trades
            if pd.notna(row.get('Buy')) and shares == 0:
                # Calculate position size based on ATR stop
                atr = row['ATR']
                if pd.notna(atr) and atr > 0:
                    # Stop loss = 3 × ATR below entry
                    stop_loss = row['Close'] - (self.atr_multiplier * atr)
                    risk_per_share = row['Close'] - stop_loss

                    # Position size = (Portfolio × Risk%) ÷ Risk per share
                    target_risk = portfolio_value * self.risk_per_trade
                    shares_to_buy = int(target_risk / risk_per_share)

                    # Don't use more than 95% of capital
                    max_shares = int((cash * 0.95) / row['Close'])
                    shares = min(shares_to_buy, max_shares)

                    if shares > 0:
                        cost = shares * row['Close']
                        commission = cost * transaction_cost
                        cash -= (cost + commission)

                        trades.append({
                            'Date': date,
                            'Type': 'BUY',
                            'Price': row['Close'],
                            'Shares': shares,
                            'Stop': stop_loss,
                            'Portfolio_Value': portfolio_value
                        })

            elif pd.notna(row.get('Sell')) and shares > 0:
                # Sell position
                proceeds = shares * row['Close']
                commission = proceeds * transaction_cost
                cash += (proceeds - commission)

                trades.append({
                    'Date': date,
                    'Type': 'SELL',
                    'Price': row['Close'],
                    'Shares': shares,
                    'Portfolio_Value': portfolio_value
                })

                shares = 0

        # Convert to DataFrames
        equity_df = pd.DataFrame(equity_curve).set_index('Date')
        trades_df = pd.DataFrame(trades)

        # Calculate performance metrics
        returns = equity_df['Portfolio_Value'].pct_change()

        # Calculate metrics
        total_return = (portfolio_value - initial_capital) / initial_capital
        years = (df.index[-1] - df.index[0]).days / 365.25
        cagr = (portfolio_value / initial_capital) ** (1 / years) - 1

        volatility = returns.std() * np.sqrt(252)
        sharpe_ratio = (cagr - 0.04) / volatility if volatility > 0 else 0

        # Drawdown analysis
        cumulative = (1 + returns).cumprod()
        running_max = cumulative.expanding().max()
        drawdown = (cumulative - running_max) / running_max
        max_drawdown = drawdown.min()

        # Win rate
        if len(trades_df) > 0:
            buy_trades = trades_df[trades_df['Type'] == 'BUY'].copy()
            sell_trades = trades_df[trades_df['Type'] == 'SELL'].copy()

            if len(buy_trades) == len(sell_trades):
                buy_trades = buy_trades.reset_index(drop=True)
                sell_trades = sell_trades.reset_index(drop=True)
                pnl = (sell_trades['Price'].values - buy_trades['Price'].values) / buy_trades['Price'].values
                win_rate = (pnl > 0).sum() / len(pnl)
                avg_win = pnl[pnl > 0].mean() if (pnl > 0).any() else 0
                avg_loss = abs(pnl[pnl < 0].mean()) if (pnl < 0).any() else 0
                win_loss_ratio = avg_win / avg_loss if avg_loss > 0 else 0
            else:
                win_rate = 0
                avg_win = 0
                avg_loss = 0
                win_loss_ratio = 0
        else:
            win_rate = 0
            avg_win = 0
            avg_loss = 0
            win_loss_ratio = 0

        results = {
            'final_value': portfolio_value,
            'total_return': total_return,
            'cagr': cagr,
            'volatility': volatility,
            'sharpe_ratio': sharpe_ratio,
            'max_drawdown': max_drawdown,
            'win_rate': win_rate,
            'avg_win': avg_win,
            'avg_loss': avg_loss,
            'win_loss_ratio': win_loss_ratio,
            'num_trades': len(trades_df),
            'equity_curve': equity_df,
            'trades': trades_df,
            'signals': df
        }

        return results

    def plot_results(self, results):
        """Visualize backtest results"""
        fig, axes = plt.subplots(3, 1, figsize=(14, 10))

        # Equity curve
        equity = results['equity_curve']
        axes[0].plot(equity.index, equity['Portfolio_Value'], linewidth=2, color='#2e7d32')
        axes[0].set_title('Portfolio Value Over Time', fontsize=14, fontweight='bold')
        axes[0].set_ylabel('Portfolio Value ($)', fontsize=12)
        axes[0].grid(alpha=0.3)
        axes[0].axhline(y=100000, color='gray', linestyle='--', alpha=0.5, label='Initial Capital')
        axes[0].legend()

        # Drawdown
        returns = equity['Portfolio_Value'].pct_change()
        cumulative = (1 + returns).cumprod()
        running_max = cumulative.expanding().max()
        drawdown = (cumulative - running_max) / running_max

        axes[1].fill_between(drawdown.index, drawdown * 100, 0, color='#c62828', alpha=0.3)
        axes[1].plot(drawdown.index, drawdown * 100, color='#c62828', linewidth=2)
        axes[1].set_title('Drawdown (%)', fontsize=14, fontweight='bold')
        axes[1].set_ylabel('Drawdown (%)', fontsize=12)
        axes[1].grid(alpha=0.3)

        # Price with signals
        signals = results['signals']
        axes[2].plot(signals.index, signals['Close'], label='Price', color='black', linewidth=1.5)
        axes[2].plot(signals.index, signals['SMA_Fast'], label=f'{self.fast_period}-day SMA',
                    color='#1976d2', linewidth=1.5, alpha=0.7)
        axes[2].plot(signals.index, signals['SMA_Slow'], label=f'{self.slow_period}-day SMA',
                    color='#f57c00', linewidth=1.5, alpha=0.7)

        # Plot buy/sell signals
        buy_signals = signals[signals['Buy'].notna()]
        sell_signals = signals[signals['Sell'].notna()]
        axes[2].scatter(buy_signals.index, buy_signals['Buy'], color='green',
                       marker='^', s=100, label='Buy', zorder=5)
        axes[2].scatter(sell_signals.index, sell_signals['Sell'], color='red',
                       marker='v', s=100, label='Sell', zorder=5)

        axes[2].set_title('Price with Moving Averages and Signals', fontsize=14, fontweight='bold')
        axes[2].set_ylabel('Price ($)', fontsize=12)
        axes[2].legend(loc='upper left')
        axes[2].grid(alpha=0.3)

        plt.tight_layout()
        plt.show()


# Example usage
if __name__ == "__main__":
    # Download S&P 500 data
    print("Downloading S&P 500 data...")
    ticker = 'SPY'
    data = yf.download(ticker, start='2000-01-01', end='2024-12-31', progress=False)

    # Initialize strategy
    strategy = TrendFollowingSystem(
        fast_period=50,
        slow_period=200,
        atr_period=14,
        atr_multiplier=3.0,
        risk_per_trade=0.02  # 2% risk per trade
    )

    # Run backtest
    print("\nRunning backtest...")
    results = strategy.backtest(data, initial_capital=100000, transaction_cost=0.001)

    # Print results
    print("\n" + "="*60)
    print("TREND FOLLOWING BACKTEST RESULTS (2000-2024)")
    print("="*60)
    print(f"Initial Capital:       ${100000:,.0f}")
    print(f"Final Value:           ${results['final_value']:,.0f}")
    print(f"Total Return:          {results['total_return']:.2%}")
    print(f"CAGR:                  {results['cagr']:.2%}")
    print(f"Volatility (annual):   {results['volatility']:.2%}")
    print(f"Sharpe Ratio:          {results['sharpe_ratio']:.2f}")
    print(f"Max Drawdown:          {results['max_drawdown']:.2%}")
    print(f"Number of Trades:      {results['num_trades']}")
    print(f"Win Rate:              {results['win_rate']:.2%}")
    print(f"Avg Win:               {results['avg_win']:.2%}")
    print(f"Avg Loss:              {results['avg_loss']:.2%}")
    print(f"Win/Loss Ratio:        {results['win_loss_ratio']:.2f}")
    print("="*60)

    # Calculate buy-and-hold comparison
    bh_return = (data['Close'][-1] - data['Close'][0]) / data['Close'][0]
    bh_cagr = (data['Close'][-1] / data['Close'][0]) ** (1 / 24.9) - 1
    print(f"\nBuy-and-Hold Comparison:")
    print(f"B&H CAGR:              {bh_cagr:.2%}")
    print(f"Strategy - B&H:        {(results['cagr'] - bh_cagr):.2%}")
    print("="*60)

    # Plot results
    strategy.plot_results(results)

    # Show recent trades
    if len(results['trades']) > 0:
        print("\nMost Recent Trades:")
        print(results['trades'].tail(10).to_string(index=False))

要点回顾

下一步

在用真金白银交易趋势跟随系统之前：

1. 彻底回测： 用 20 年以上的数据、多种资产、不同市场环境去测试
2. 模拟交易 6 至 12 个月： 实时记录每个信号（不许用后见之明作弊）
3. 从小做起： 第一年只用计划仓位的 25%，再慢慢放大
4. 记交易日志： 记录每笔交易、你的情绪，以及你为何遵守或打破了规则
5. 读经典：
   • 趋势跟随 作者：Michael Covel
   • Following the Trend 作者：Andreas Clenow
   • The Complete TurtleTrader 作者：Michael Covel

接下来： 回测方法论——避开那些坑 我们会讲如何正确地检验策略，避免曲线拟合、幸存者偏差和前视偏差。