当前位置：首页>python>只用 ADX 和 CCI,这个 Python 策略 25 年跑赢大盘两倍

只用 ADX 和 CCI,这个 Python 策略 25 年跑赢大盘两倍

2026-02-06 17:20:12

欢迎加入专注于财经数据与量化投研的【数据科学实战】知识星球！在这里，您将获取持续更新的《财经数据宝典》和《量化投研宝典》，这两部宝典相辅相成，为您在量化投研道路上提供明确指引。我们提供了精选的国内外量化投研的 250+ 篇高质量文章，并每日更新最新研究成果，涵盖策略开发、因子分析、风险管理等核心领域。无论您是量化投资新手还是经验丰富的研究者，星球社区都能帮您少走弯路，事半功倍，共同探索数据驱动的投资世界！

引言

你有没有想过，一个简单的交易策略，在一只股票上坚持运行 25 年，能带来怎样的回报？

今天要分享的这个策略，从 2000 年到 2025 年，在波士顿科学公司（BSX）这只股票上实现了 1746% 的总回报，远超同期买入持有策略。更重要的是，它的核心逻辑非常简洁：趋势强时入场，动量衰竭时离场。

这篇文章将带你一步步拆解这个策略的实现原理，并用 Python 代码完整呈现。无论你是量化交易的初学者，还是想用 Python 做点有趣项目的开发者，都能从中收获不少。

策略核心理念

这个策略的哲学很简单：只在市场真正形成趋势时交易。

为了判断趋势是否真实存在，策略使用了 ADX（平均趋向指数）。ADX 不关心价格是涨还是跌，它只回答一个问题：市场现在有没有明确的趋势？

当 ADX 高于 20 时，答案是"有"。这是策略唯一允许入场的时机。

而离场的信号则交给 CCI（商品通道指数）。当 CCI 相比 5 天前开始下降时，说明动量正在衰减，是时候离场保护资金了。

简单来说：

ADX 负责让你进场
CCI 负责让你离场

数据获取

首先，我们需要下载 BSX 的历史数据。这里使用 yfinance 库来获取数据：

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

# -------------------------
# 下载数据
# -------------------------
symbol = "BSX"# 股票代码
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("BSX_clean.csv", index=False)

入场信号：ADX 指标计算

ADX 指标用于衡量趋势的强度。当 ADX 大于 20 时，表明市场存在明显的趋势，适合入场：

ADX_LEVEL = 20# ADX 阈值
ADX_PERIOD = 14# ADX 计算周期

defcalculate_adx(df, period=14):
"""
    计算 ADX 及相关指标
    """
# 真实波幅（True Range）
    df['H-L'] = df['High'] - df['Low']
    df['H-PC'] = abs(df['High'] - df['Close'].shift(1))
    df['L-PC'] = abs(df['Low'] - df['Close'].shift(1))
    df['TR'] = df[['H-L', 'H-PC', 'L-PC']].max(axis=1)

# 方向移动指标（Directional Movement）
    df['+DM'] = np.where(
        (df['High'] - df['High'].shift(1)) > (df['Low'].shift(1) - df['Low']),
        np.maximum(df['High'] - df['High'].shift(1), 0),
0
    )
    df['-DM'] = np.where(
        (df['Low'].shift(1) - df['Low']) > (df['High'] - df['High'].shift(1)),
        np.maximum(df['Low'].shift(1) - df['Low'], 0),
0
    )

# 平滑处理
    df['TRn'] = df['TR'].rolling(window=period).sum()
    df['+DMn'] = df['+DM'].rolling(window=period).sum()
    df['-DMn'] = df['-DM'].rolling(window=period).sum()

# 计算 +DI 和 -DI
    df['+DI'] = 100 * (df['+DMn'] / df['TRn'])
    df['-DI'] = 100 * (df['-DMn'] / df['TRn'])

# 计算 DX 和 ADX
    df['DX'] = 100 * abs(df['+DI'] - df['-DI']) / (df['+DI'] + df['-DI'])
    df['ADX'] = df['DX'].rolling(window=period).mean()

return df

defadx_is_higher_than(df, period=ADX_PERIOD, level=ADX_LEVEL):
"""
    判断 ADX 是否高于指定阈值
    """
    df = calculate_adx(df, period)
return df['ADX'] > level

# 生成入场信号
df["ADX_Higher20"] = adx_is_higher_than(df)
entry_conditions = ['ADX_Higher20']
df['entry_signal'] = df[entry_conditions].all(axis=1)

离场信号：CCI 指标计算

CCI 指标用于衡量价格相对于平均价格的偏离程度。当 CCI 开始下降时，说明动量正在减弱：

CCI_PERIOD = 14# CCI 计算周期
CCI_SHIFT = 5# 比较的时间间隔

defcalculate_cci(df, period=CCI_PERIOD):
"""
    计算商品通道指数（CCI）
    """
    df = df.copy()

# 典型价格（Typical Price）
    df['TP'] = (df['High'] + df['Low'] + df['Close']) / 3

# 典型价格的简单移动平均
    df['SMA_TP'] = df['TP'].rolling(period).mean()

# 平均偏差
    df['MeanDev'] = df['TP'].rolling(period).apply(
lambda x: np.mean(np.abs(x - np.mean(x))), raw=True
    )

# 计算 CCI
    df['CCI'] = (df['TP'] - df['SMA_TP']) / (0.015 * df['MeanDev'])

return df

defcci_falling(df, period=CCI_PERIOD, shifts=CCI_SHIFT):
"""
    判断 CCI 是否相比 N 天前在下降
    """
    df['CCI_main'] = calculate_cci(df, period)['CCI']
return df['CCI_main'] < df['CCI_main'].shift(shifts)

# 生成离场信号
df["CCI_Falling"] = cci_falling(df)
exit_conditions = ['CCI_Falling']
df['exit_signal'] = df[exit_conditions].all(axis=1)

回测执行

使用 vectorbt 库进行回测，模拟真实交易环境中的手续费和滑点：

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

# 将信号向前移动一天，避免使用未来数据
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(pf.stats())  # 打印统计数据
pf.plot().show()   # 绘制资金曲线

与买入持有策略对比

为了评估策略的有效性，我们还需要看看同期买入持有的表现：

# 买入持有策略回测
df_holding = df['Open']
pf_holding = vbt.Portfolio.from_holding(
    df_holding,
    init_cash=100_000,
    fees=0.001,
    slippage=0.002
)
print(pf_holding.stats())