第三种量化思维：不预测涨跌，只跟随趋势——在震荡市中忍耐，在趋势市中吃肉
在前几期文章中，我们分别介绍了均值回归策略（跌深了赌反弹）和动量策略（强者恒强，买入最强ETF）。今天，我们来学习第三种经典策略——趋势跟踪。
与动量策略的“横截面比较”不同，趋势跟踪聚焦于单一资产自身的时间序列：当价格呈现上升趋势时买入，趋势反转时卖出。最经典的工具就是双均线系统——金叉买入，死叉卖出，简单粗暴，却历经数十年市场检验。
本文基于与前两篇完全相同的11只美股ETF数据，实现一个双均线趋势跟踪策略，并展示真实回测结果。你会发现，在合适的参数下，趋势跟踪可以取得比均值回归更高、比动量策略更稳的业绩。
一、策略核心逻辑
双均线系统是最基础的趋势跟踪工具：
快线（短期均线）：MA_short = close.rolling(window=short).mean()
慢线（长期均线）：MA_long = close.rolling(window=long).mean()
买入信号：快线上穿慢线（金叉），且当前无持仓 → 次日开盘买入
卖出信号：快线下穿慢线（死叉），且当前持仓 → 次日开盘卖出
只做多：不进行做空操作，仅当金叉时开仓，死叉时平仓。
参数选择：常见组合有 (5,10)、(20,60)、(50,200) 等。经过测试，我们选用 (5,10) 作为默认参数——短期均线反应更灵敏，能够更快捕捉趋势变化，同时通过合理的仓位管理控制风险。
二、完整Python代码
"""================================================================================趋势跟踪策略（美股ETF版）：双均线金叉/死叉- 只做多，每日调仓（每只ETF独立判断）- 快线上穿慢线 → 买入；快线下穿慢线 → 卖出- 数据来源：本地 CSV 文件（列名：date,open,high,low,close,volume）================================================================================"""import pandas as pdimport numpy as npfrom datetime import datetimeimport matplotlibmatplotlib.use('TkAgg')import matplotlib.pyplot as pltfrom dataclasses import dataclassfrom typing import Dict, List, Optional, Tupleimport warningsimport oswarnings.filterwarnings('ignore')plt.rcParams['font.sans-serif'] = ['SimHei', 'DejaVu Sans']plt.rcParams['axes.unicode_minus'] = False@dataclassclass TradeRecord: date: datetime code: str action: str shares: int price: float value: float pnl: float = 0.0 reason: str = ""class TrendFollowingStrategy: def __init__( self, start_date: str = "20190101", end_date: str = "20260410", # 双均线参数 short_window: int = 10, long_window: int = 30, # 资金与风控 initial_capital: float = 1_000_000, commission: float = 0.0001, slippage: float = 0.0002, local_data_path: str = r"你的数据路径", # 仓位管理：每只ETF分配资金的百分比 capital_per_etf_pct: float = 0.10, ): self.start_date = start_date self.end_date = end_date self.short_window = short_window self.long_window = long_window self.initial_capital = initial_capital self.commission = commission self.slippage = slippage self.local_data_path = local_data_path self.capital_per_etf_pct = capital_per_etf_pct # 美股 ETF 标的池（与前两篇相同） self.etf_pool = [ "XLE", "XLF", "XLK", "XLU", "XLV", "XLY", "XLP", "XLI", "XLB", "XLRE", "XLC" ] self.close_data: Dict[str, pd.Series] = {} self.open_data: Dict[str, pd.Series] = {} self.trades: List[TradeRecord] = [] self.positions: Dict[str, Dict] = {} self.daily_stats = [] self.current_capital = initial_capital self.peak_capital = initial_capital self.trading_dates = [] # -------------------- 数据读取 -------------------- def _read_local_etf_file(self, code: str) -> Optional[Tuple[pd.Series, pd.Series]]: file_path = os.path.join(self.local_data_path, f"{code}.csv") if not os.path.exists(file_path): print(f" ✗ {code}: 文件不存在") return None try: df = pd.read_csv(file_path, encoding='utf-8') except UnicodeDecodeError: df = pd.read_csv(file_path, encoding='gbk') df['date'] = pd.to_datetime(df['date']) df.set_index('date', inplace=True) close = df['close'].astype(float) open_ = df['open'].astype(float) return close, open_ def fetch_data(self): print("\n[数据加载]") for code in self.etf_pool: result = self._read_local_etf_file(code) if result: close, open_ = result self.close_data[code] = close self.open_data[code] = open_ all_dates = [set(d.index) for d in self.close_data.values()] self.trading_dates = sorted(set.intersection(*all_dates)) self.start_idx = self.long_window + 1 print(f"共同交易日: {len(self.trading_dates)} 天") # -------------------- 双均线信号生成 -------------------- def generate_signals(self, current_date: datetime) -> Dict[str, str]: signals = {} for code in self.close_data: price_series = self.close_data[code] hist = price_series[price_series.index <= current_date] if len(hist) < self.long_window + 1: continue ma_short = hist.rolling(window=self.short_window).mean() ma_long = hist.rolling(window=self.long_window).mean() curr_short = ma_short.iloc[-1] curr_long = ma_long.iloc[-1] prev_short = ma_short.iloc[-2] if len(ma_short) >= 2 else np.nan prev_long = ma_long.iloc[-2] if len(ma_long) >= 2 else np.nan if pd.isna(curr_short) or pd.isna(curr_long) or pd.isna(prev_short) or pd.isna(prev_long): continue is_holding = code in self.positions # 金叉：短期上穿长期 if not is_holding and (prev_short <= prev_long) and (curr_short > curr_long): signals[code] = 'BUY' # 死叉：短期下穿长期 elif is_holding and (prev_short >= prev_long) and (curr_short < curr_long): signals[code] = 'SELL' return signals # -------------------- 交易执行 -------------------- def execute_trades(self, date: datetime, signals: Dict[str, str], next_opens: Dict[str, float]): # 先平仓 for code, action in signals.items(): if action == 'SELL' and code in self.positions: pos = self.positions[code] price = next_opens.get(code) if price is None: continue gross = pos['shares'] * price fee = gross * (self.commission + self.slippage) net = gross - fee self.current_capital += net pnl = net - (pos['shares'] * pos['entry_price']) self.trades.append(TradeRecord(date, code, 'SELL', pos['shares'], price, net, pnl, '死叉卖出')) del self.positions[code] # 再开仓 buy_codes = [c for c, a in signals.items() if a == 'BUY'] for code in buy_codes: price = next_opens.get(code) if price is None: continue risk_capital = self.current_capital * self.capital_per_etf_pct shares = int(risk_capital / price) if shares < 1: continue cost = shares * price * (1 + self.commission + self.slippage) if cost > self.current_capital: shares = int(self.current_capital / (price * (1 + self.commission + self.slippage))) if shares < 1: continue cost = shares * price * (1 + self.commission + self.slippage) self.current_capital -= cost self.positions[code] = {'shares': shares, 'entry_price': price} self.trades.append(TradeRecord(date, code, 'BUY', shares, price, -cost, 0, '金叉买入')) # -------------------- 回测引擎 -------------------- def run_backtest(self): print("\n[回测运行]") for i in range(self.start_idx, len(self.trading_dates) - 1): curr = self.trading_dates[i] nxt = self.trading_dates[i+1] signals = self.generate_signals(curr) next_opens = {c: self.open_data[c].loc[nxt] for c in self.close_data if nxt in self.open_data[c].index} self.execute_trades(curr, signals, next_opens) total = self.current_capital for code, pos in self.positions.items(): if curr in self.close_data[code].index: total += pos['shares'] * self.close_data[code].loc[curr] self.peak_capital = max(self.peak_capital, total) dd = (self.peak_capital - total) / self.peak_capital if self.peak_capital > 0 else 0 self.daily_stats.append({'date': curr, 'total': total, 'dd': dd}) return pd.DataFrame(self.daily_stats) # -------------------- 绩效报告 -------------------- def report(self, stats_df): if stats_df.empty: print("无回测数据") return final = stats_df.iloc[-1]['total'] ret = (final - self.initial_capital) / self.initial_capital * 100 days = len(stats_df) years = days / 252 ann_ret = (final / self.initial_capital) ** (1/years) - 1 if years > 0 else 0 stats_df['ret'] = stats_df['total'].pct_change() vol = stats_df['ret'].std() * np.sqrt(252) * 100 sharpe = (ann_ret*100 - 3) / vol if vol > 0 else 0 max_dd = stats_df['dd'].max() * 100 closed = [t for t in self.trades if t.action == 'SELL'] wins = [t for t in closed if t.pnl > 0] win_rate = len(wins)/len(closed)*100 if closed else 0 print("\n" + "="*60) print("趋势跟踪策略（双均线金叉/死叉）- 美股ETF版") print("="*60) print(f"回测区间: {self.start_date} ~ {self.end_date} ({days}天)") print(f"初始资金: {self.initial_capital:,.0f} 期末: {final:,.0f}") print(f"总收益率: {ret:+.2f}% 年化: {ann_ret*100:+.2f}%") print(f"年化波动: {vol:.2f}% 夏普: {sharpe:.2f}") print(f"最大回撤: {max_dd:.2f}%") print(f"交易次数: {len(closed)} 笔 胜率: {win_rate:.1f}%") if closed: avg_pnl = np.mean([t.pnl for t in closed]) print(f"平均盈亏: {avg_pnl:,.0f}") print("="*60) # 绘图 fig, ax = plt.subplots(2, 1, figsize=(12, 8)) ax[0].plot(stats_df['date'], stats_df['total'], label='NAV', color='blue') ax[0].axhline(self.initial_capital, color='gray', linestyle='--') ax[0].set_title('Trend Following Strategy (Dual Moving Average) - US ETFs') ax[0].legend() ax[1].fill_between(stats_df['date'], 0, stats_df['dd']*100, color='red', alpha=0.3) ax[1].set_ylabel('Drawdown %') plt.tight_layout() plt.show() def run(self): self.fetch_data() stats = self.run_backtest() self.report(stats) if self.trades: pd.DataFrame([t.__dict__ for t in self.trades]).to_csv('trades_trend_following_us.csv', index=False, encoding='utf-8-sig') print("交易记录已保存至 trades_trend_following_us.csv")if __name__ == "__main__": strategy = TrendFollowingStrategy( local_data_path=r"你的数据路径", start_date="20190101", end_date="20260410", short_window=5, long_window=10, initial_capital=1_000_000, commission=0.0001, slippage=0.0002, capital_per_etf_pct=0.10, ) strategy.run()
完整代码已包含所有必要函数，复制保存为 .py 文件并修改数据路径即可运行。
三、真实回测结果
我们使用与前两篇完全相同的11只美股ETF（XLE, XLF, XLK, XLU, XLV, XLY, XLP, XLI, XLB, XLRE, XLC），时间区间 2019年1月1日 – 2026年4月10日，初始资金100万美元。双均线参数为 (5,10)，每只ETF分配10%资金。
真实回测结果如下：
趋势跟踪策略（双均线金叉/死叉）- 美股ETF版
============================================================
回测区间: 20190101 ~ 20260410 (1816天)
初始资金: 1,000,000 期末: 1,650,310
总收益率: +65.03% 年化: +7.20%
年化波动: 8.69% 夏普: 0.48
最大回撤: 10.91%
交易次数: 1059 笔 胜率: 48.4%
平均盈亏: 617
============================================================
核心亮点：
总收益率65.03%，年化7.20%，显著高于均值回归（27.02%），接近动量策略（71.91%）。
年化波动仅8.69%，远低于动量策略的23.06%，甚至低于均值回归的9.61%！
最大回撤仅10.91%，是三个策略中最低的，比均值回归（17.55%）和动量策略（40.45%）都要小得多。
夏普比率0.48，同样是三个策略中最高的，说明风险调整后收益最佳。
胜率48.4%，接近一半，平均盈亏617美元，盈亏比良好。
结论：在参数 (5,10) 下，双均线趋势跟踪策略实现了高收益、低波动、低回撤的优异表现，夏普比率0.48远超另外两个策略。这意味着在2019–2026年的美股环境中，短期均线系统能够有效捕捉趋势，同时通过分散持仓控制风险，成为三款策略中的“性价比之王”。
四、为什么参数(5,10)表现如此出色？
之前测试的 (20,60) 参数收益平庸（年化3.08%），而 (5,10) 年化提升到7.20%。原因在于：
反应更灵敏：5日均线比20日均线更快对价格变化做出反应，能够在趋势早期捕捉入场机会，离场也更及时，减少利润回吐。
适应震荡上涨行情：2019–2026年美股并非单边暴涨，而是多次震荡上行。较短的均线组合在震荡中更容易抓住波段，而长周期均线往往滞后。
交易次数增加：(5,10) 产生了1059笔交易，是 (20,60) 的5倍多。更多的交易机会让策略能够充分捕捉市场中的多次小趋势，积少成多。
当然，更灵敏的均线也会带来更多假信号。但通过分散到11只ETF，单只ETF的假信号被组合平滑，最终实现了优异的风险调整后收益。
五、三种策略完整对比
| 策略 | 年化收益 | 年化波动 | 最大回撤 | 胜率 | 平均盈亏 | 夏普 |
|---|
| 均值回归 | 3.60% | 9.61% | 17.55% | 62.9% | 244 | 0.06 |
| 趋势跟踪(10,30) | 7.20% | 8.69% | 10.91% | 48.4% | 617 | 0.48 |
| 动量策略 | 8.04% | 23.06% | 40.45% | 49.7% | 1,073 | 0.22 |
综合排名（按夏普）：
趋势跟踪（0.48）—— 风险调整后收益最佳
动量策略（0.22）
均值回归（0.06）
按绝对收益：
动量策略（8.04%）
趋势跟踪（7.20%）
均值回归（3.60%）
按风险控制：
趋势跟踪（最大回撤10.91%，波动8.69%）
均值回归（17.55%，9.61%）
动量策略（40.45%，23.06%）
趋势跟踪策略在本次回测中实现了收益与风险的黄金平衡——它比均值回归多赚一倍，却承担了更小的回撤；它比动量策略少赚0.8个百分点，但回撤只有后者的四分之一。对于大多数投资者而言，趋势跟踪可能是最睡得着觉的选择。
六、策略优缺点与改进方向
优点：
逻辑简单，易于理解和实现。
低波动、低回撤，持有体验好。
交易次数适中，佣金磨损可控。
分散到多只ETF，避免单资产黑天鹅。
缺点：
改进方向：
动态参数：根据市场波动率调整均线周期，高波动时缩短，低波动时延长。
添加过滤器：例如要求金叉时成交量放大，或要求价格突破前期高点。
结合ADX：只在ADX>25时启用趋势跟踪，否则空仓或切换至均值回归。
移动止损：设置吊灯止损（从最高点回撤一定比例平仓），保护浮盈。
七、趋势跟踪 vs 动量策略：你该选哪个？
| 如果你的偏好... | 推荐策略 |
|---|
| 追求最高绝对收益，能承受40%回撤 | 动量策略 |
| 追求收益与风险的平衡，希望回撤小、睡得安稳 | 趋势跟踪（双均线） |
| 极度保守，或主要做震荡市 | 均值回归 |
从实际回测看，趋势跟踪策略在2019–2026年表现最为均衡，夏普比率0.48显著优于其他两者。对于大多数个人投资者，趋势跟踪可能是最实用的起点。
八、代码获取
完整代码获取：关注公众号并回复关键词“趋势跟踪ETF”，即可下载完整Python脚本及示例数据。
声明：本文所有内容仅为量化策略教学与交流，不构成任何投资建议。过去收益不代表未来表现，市场有风险，投资需谨慎。
