Python 量化交易实战：一个 25 年胜率 73% 的动量策略拆解

2026年重磅升级已全面落地！欢迎加入专注财经数据与量化投研的【数据科学实战】知识星球！您将获取持续更新的《财经数据宝典》与《量化投研宝典》，双典协同提供系统化指引；星球内含 350 篇以上独有高质量文章，深度覆盖策略开发、因子分析、风险管理等核心领域，内容基本每日更新；同步推出的「量化因子专题教程」系列（含完整可运行代码与实战案例），系统详解因子构建、回测与优化全流程，并实现日更迭代。我们持续扩充独家内容资源，全方位赋能您的投研效率与专业成长。无论您是量化新手还是资深研究者，这里都是助您少走弯路、事半功倍的理想伙伴，携手共探数据驱动的投资未来！

引言

在量化交易的世界里，很多人追求复杂的模型和高频交易，但有时候，简单而有纪律的策略反而能带来惊人的回报。今天我们来拆解一个基于 MACD 和 QQE 指标的动量吸收策略，这个策略在 2000 年至 2025 年的回测中，对 C.H. Robinson Worldwide（CHRW）股票实现了 4888% 的总回报率，同时最大回撤仅为 34%。

这篇文章将带你了解这个策略的核心逻辑，并提供完整的 Python 代码实现。

策略核心思想

这个策略的哲学很简单：在弱势中买入，在强势确认时卖出。

具体来说，策略结合了两个动量指标：

1. MACD：用于识别熊市动量，当 MACD 低于 0 时入场
2. QQE：用于确认强势回归，当 QQE 的平滑 RSI 线上穿 50 时离场

这种方法的精妙之处在于：不追涨杀跌，而是耐心等待市场情绪低迷时入场，然后等待动量真正恢复时才离场。

环境准备

首先，我们需要安装必要的 Python 库：

pip install pandas numpy yfinance vectorbt

完整代码实现

第一步：下载数据

import pandas as pd
import numpy as np
import yfinance as yf
import vectorbt as vbt

# -------------------------
# 下载股票数据
# -------------------------
symbol = "CHRW"  # 股票代码
start_date = "2000-01-01"  # 开始日期
end_date = "2026-01-01"  # 结束日期
interval = "1d"  # 日线级别




    
# 从 Yahoo Finance 下载数据
df = yf.download(symbol, start=start_date, end=end_date, interval=interval)

# 保存数据到本地 CSV 文件
df.to_csv("CHRW_clean.csv", index=False)

第二步：计算 MACD 指标

# MACD 参数设置
MACD_FAST = 12  # 快线周期
MACD_SIGNAL = 9  # 信号线周期
MACD_SLOW = 26  # 慢线周期

def calculate_macd(df, fast=MACD_FAST, slow=MACD_SLOW, signal=MACD_SIGNAL):
    """
    计算 MACD 和信号线
    
    参数：
        df: 包含收盘价的 DataFrame
        fast: 快速 EMA 周期
        slow: 慢速 EMA 周期
        signal: 信号线 EMA 周期
    
    返回：
        包含 MACD 相关列的 DataFrame
    """
    df = df.copy()
    
    # 计算快速和慢速指数移动平均线
    df["EMA_fast"] = df["Close"].ewm(span=fast, adjust=False).mean()
    df["EMA_slow"] = df["Close"].ewm(span=slow, adjust=False).mean()
    
    # MACD 线 = 快线 - 慢线
    df["MACD"] = df["EMA_fast"] - df["EMA_slow"]
    
    # 信号线 = MACD 的 EMA
    df["Signal"] = df["MACD"].ewm(span=signal, adjust=False).mean()
    
    # 保存前一天的值，用于判断交叉
    df["MACD_prev"] = df["MACD"].shift(1)
    df["Signal_prev"] = df["Signal"].shift(1)
    
    return df

def macd_lower_0(df):
    """
    判断 MACD 是否低于 0
    当 MACD < 0 时，表示短期趋势弱于长期趋势
    """
    df = calculate_macd(df)
    return df["MACD"] < 0

第三步：计算 QQE 指标

# QQE 参数设置
QQE_FACTOR = 4.236  # QQE 系数
QQE_LEVEL = 50  # 判断强弱的阈值
QQE_PERIOD = 14  # RSI 周期
QQE_SMOOTH = 5  # 平滑周期

def calculate_qqe(df, rsi_period=QQE_PERIOD, smooth=QQE_SMOOTH, factor=QQE_FACTOR):
    """
    计算 QQE（Quantitative Qualitative Estimation）指标
    基于 RSI 平滑和类似 ATR 的计算方法


    

    
    参数：
        df: 包含收盘价的 DataFrame
        rsi_period: RSI 计算周期
        smooth: 平滑窗口大小
        factor: QQE 乘数因子
    
    返回：
        包含 QQE_Value1 和 QQE_Value2 的 DataFrame
    """
    df = df.copy()
    
    # --- 计算 RSI ---
    delta = df['Close'].diff()  # 价格变化
    up = delta.clip(lower=0)  # 上涨部分
    down = -delta.clip(upper=0)  # 下跌部分
    
    # 使用指数移动平均计算平均涨跌幅
    roll_up = up.ewm(alpha=1/rsi_period, adjust=False).mean()
    roll_down = down.ewm(alpha=1/rsi_period, adjust=False).mean()
    
    # RSI 公式
    rsi = 100 - (100 / (1 + roll_up / roll_down))
    
    # --- 平滑 RSI ---
    rsi_ma = rsi.rolling(window=smooth).mean().fillna(method='bfill')
    
    # --- RSI 变化率和类 ATR 平滑 ---
    rsi_delta = rsi_ma.diff().abs().fillna(0)
    atr_rsi = rsi_delta.ewm(alpha=1/smooth, adjust=False).mean()
    
    # --- QQE 计算 ---
    value1 = rsi_ma.copy()
    value2 = pd.Series(index=df.index, dtype=float)
    value2.iloc[0] = value1.iloc[0]
    
    # 动态计算 QQE 轨道线
    for i in range(1, len(df)):
        prev_trail = value2.iloc[i-1]
        prev_value1 = value1.iloc[i-1]
        atr = atr_rsi.iloc[i]
        
        # 根据趋势方向调整轨道
        direction = 1 if prev_value1 > prev_trail else -1
        value2.iloc[i] = prev_trail + direction * factor * atr
    
    df['QQE_Value1'] = value1
    df['QQE_Value2'] = value2
    
    return df

def qqe_cross_above_level(df, level=QQE_LEVEL):
    """
    判断 QQE 的 Value1 是否上穿指定水平
    这表示动量从弱势转为强势
    """
    df = calculate_qqe(df)
    return (df['QQE_Value1'] > level) & (df['QQE_Value1'].shift(1) <= level)

第四步：生成交易信号

# -------------------------
# 入场条件：MACD 低于 0
# -------------------------
df["MACD_lower_0"] = macd_lower_0(df)

entry_conditions = [
    'MACD_lower_0',
]

# 所有入场条件都满足时产生入场信号



    
df['entry_signal'] = df[entry_conditions].all(axis=1)

# -------------------------
# 出场条件：QQE Value1 上穿 50
# -------------------------
df["QQE_Value1_CrossAbove_Level"] = qqe_cross_above_level(df)

exit_conditions = [
    'QQE_Value1_CrossAbove_Level',
]

# 所有出场条件都满足时产生出场信号
df['exit_signal'] = df[exit_conditions].all(axis=1)

第五步：运行回测

# -------------------------
# 执行回测
# -------------------------

# 信号延迟一天，确保使用下一天开盘价执行
# 这样可以避免使用未来数据，使回测更加真实
shift_entries = df['entry_signal'].shift(1).astype(bool).fillna(False).to_numpy()
shift_exits = df['exit_signal'].shift(1).astype(bool).fillna(False).to_numpy()

# 创建投资组合并运行回测
pf = vbt.Portfolio.from_signals(
    close=df['Open'],  # 使用开盘价执行交易
    entries=shift_entries,  # 入场信号
    exits=shift_exits,  # 出场信号
    init_cash=100_000,  # 初始资金 10 万美元
    fees=0.001,  # 手续费 0.1%
    slippage=0.002,  # 滑点 0.2%
    freq='1d'  # 日线频率
)

# -------------------------
# 输出统计结果和可视化
# -------------------------
print("策略表现统计：")
print(pf.stats())

# 绘制资金曲线
pf.plot().show()

第六步：与买入持有策略对比

# -------------------------
# 买入持有策略对比
# -------------------------
df_holding = df['Open']

# 创建买入持有的投资组合
pf_holding = vbt.Portfolio.from_holding(
    df_holding, 
    init_cash=100_000,  # 同样的初始资金
    fees=0.001  # 同样的手续费
)

print("\n买入持有策略表现：")
print(pf_holding.stats())