Python量化交易策略实战：布林带与ADX结合的压力测试与前向验证

你是否思考过，如何用 Python 构建一个真正稳健的量化交易策略？许多初学者往往只盯着单一参数进行优化，却忽视了策略在更广泛条件下的稳定性和适应性。

本文将深入探讨一个结合布林带（Bollinger Bands）扩张与 ADX（平均趋向指标）衰减的交易策略。更重要的是，我们将学习如何通过压力测试和前向优化这两种方法来严格验证策略的稳健性，而不是简单地追逐历史回测中的“最佳”参数。

策略核心逻辑

这个策略的思路清晰而直接：

• 入场条件：当布林带宽度扩张时（表明市场波动率突破，可能开启趋势行情）。
• 出场条件：当 ADX 指标下穿某个阈值时（表明趋势强度正在减弱）。

简单来说，其核心理念是在市场波动率放大时进场捕捉趋势，并在趋势动力衰退时及时离场。

压力测试：验证策略稳健性

与其费尽心思寻找一个可能过拟合的“最优”参数，不如测试策略在大量不同参数组合下的整体表现。以下代码展示了如何进行压力测试：

import pandas as pd
import numpy as np
import yfinance as yf
import itertools
import vectorbt as vbt
import matplotlib.pyplot as plt

# -------------------------
# 下载历史数据
# -------------------------
symbol = "BALL" # 股票代码
start_date = "2000-01-01" # 起始日期
end_date = "2026-01-01" # 结束日期
interval = "1d" # 日线数据

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

# -------------------------
# 定义参数网格（用于压力测试）
# -------------------------
ADX_LEVELS = list(range(16, 25, 2))  # ADX 阈值范围：16, 18, 20, 22, 24
ADX_PERIODS = list(range(10, 21, 2))  # ADX 周期范围：10, 12, 14, 16, 18, 20
BB_PERIODS = list(range(16, 25, 2))  # 布林带周期范围
BB_SHIFTS = list(range(1, 10, 2))  # 布林带宽度对比的偏移量
BB_STDS = [1.5, 2, 2.5, 3]  # 布林带标准差倍数

# 生成所有参数组合
param_sets = list(itertools.product(
    ADX_LEVELS,
    ADX_PERIODS,
    BB_PERIODS,
    BB_SHIFTS,
    BB_STDS
))

print(f"总共需要测试 {len(param_sets)} 个参数组合")

指标计算函数

接下来，我们定义计算布林带和 ADX 指标所需的函数：

def calculate_bollinger_bands(df, period, std):
    """
    计算布林带指标

    参数：
        df: 包含价格数据的 DataFrame
        period: 移动平均周期
        std: 标准差倍数

    返回：
        中轨、上轨、下轨
    """
    ma = df['Close'].rolling(period).mean()  # 计算移动平均线（中轨）
    sd = df['Close'].rolling(period).std()   # 计算标准差
    return ma, ma + std * sd, ma - std * sd  # 返回中轨、上轨、下轨

def bb_expansion(df, period, std, shift):
    """
    判断布林带是否扩张

    参数：
        df: 价格数据
        period: 布林带周期
        std: 标准差倍数
        shift: 对比的偏移周期数

    返回：
        布尔序列，True 表示布林带正在扩张
    """
    ma, upper, lower = calculate_bollinger_bands(df, period, std)
    bandwidth = upper - lower  # 计算带宽
    return bandwidth > bandwidth.shift(shift)  # 当前带宽大于 shift 天前的带宽

def calculate_adx(df, period):
    """
    计算 ADX（平均趋向指标）

    参数：
        df: 包含 High、Low、Close 的 DataFrame
        period: ADX 计算周期

    返回：
        ADX 序列
    """
    high, low, close = df['High'], df['Low'], df['Close']

    # 计算真实波幅（True Range）
    tr = pd.concat([
        high - low,
        (high - close.shift()).abs(),
        (low - close.shift()).abs()
    ], axis=1).max(axis=1)

    # 计算方向移动（Directional Movement）
    plus_dm = np.where(
        (high - high.shift()) > (low.shift() - low),
        np.maximum(high - high.shift(), 0),
        0
    )
    minus_dm = np.where(
        (low.shift() - low) > (high - high.shift()),
        np.maximum(low.shift() - low, 0),
        0
    )

    # 计算方向指标
    tr_n = tr.rolling(period).sum()
    plus_di = 100 * pd.Series(plus_dm).rolling(period).sum() / tr_n
    minus_di = 100 * pd.Series(minus_dm).rolling(period).sum() / tr_n

    # 计算 DX 和 ADX
    dx = 100 * (plus_di - minus_di).abs() / (plus_di + minus_di)
    adx = dx.rolling(period).mean()

    return adx

def adx_cross_below(df, period, level):
    """
    判断 ADX 是否下穿指定阈值

    参数：
        df: 价格数据
        period: ADX 周期
        level: 阈值

    返回：
        布尔序列，True 表示 ADX 下穿阈值
    """
    adx = calculate_adx(df, period)
    return (adx.shift(1) >= level) & (adx < level)

执行压力测试

# -------------------------
# 压力测试主循环
# -------------------------
equity_curves = {}  # 存储所有参数组合的权益曲线

for adx_level, adx_period, bb_period, bb_shift, bb_std in param_sets:
    df = df_raw.copy()

    # 计算入场和出场信号
    df['BB_Expansion'] = bb_expansion(df, period=bb_period, std=bb_std, shift=bb_shift)
    df['ADX_Exit'] = adx_cross_below(df, period=adx_period, level=adx_level)

    # 生成交易信号（延迟一天执行，避免未来函数）
    entries = df['BB_Expansion'].shift(1).astype(bool).fillna(False).to_numpy()
    exits = df['ADX_Exit'].shift(1).astype(bool).fillna(False).to_numpy()

    # 使用 vectorbt 进行回测
    pf = vbt.Portfolio.from_signals(
        close=df['Open'],     # 使用开盘价成交
        entries=entries,      # 入场信号
        exits=exits,          # 出场信号
        init_cash=100,        # 初始资金
        fees=0.001,           # 手续费 0.1%
        slippage=0.002,       # 滑点 0.2%
        freq='1d'
    )

    # 保存权益曲线
    label = f"ADX({adx_period},{adx_level}) BB({bb_period},{bb_std}) S({bb_shift})"
    equity_curves[label] = pf.value()

# -------------------------
# 绘制所有权益曲线
# -------------------------
plt.figure(figsize=(14, 7))
for label, curve in equity_curves.items():
    plt.plot(curve.index, curve.values, alpha=0.35, linewidth=1)

plt.title("权益曲线 - BALL 策略压力测试")
plt.xlabel("日期")
plt.ylabel("组合价值")
plt.grid(True)
plt.show()

与买入持有策略对比

压力测试的一个关键环节，是将我们的策略与最简单的“买入并持有”基准进行对比。

# -------------------------
# 构建买入持有基准
# -------------------------
df_holding = df_raw['Open']
pf_holding = vbt.Portfolio.from_holding(df_holding, init_cash=100, freq='1d')
buy_hold_curve = pf_holding.value()

# 计算所有策略的平均权益曲线
equity_df = pd.DataFrame(equity_curves)
average_equity = equity_df.mean(axis=1)

# -------------------------
# 绘制对比图
# -------------------------
plt.figure(figsize=(14, 7))

# 绘制所有压力测试曲线（浅灰色）
for label, curve in equity_curves.items():
    plt.plot(curve.index, curve.values, alpha=0.25, linewidth=1, color='gray')

# 绘制平均权益曲线（蓝色粗线）
plt.plot(average_equity.index, average_equity.values,
         color='blue', linewidth=2, label='策略平均表现')

# 绘制买入持有曲线（红色虚线）
plt.plot(buy_hold_curve.index, buy_hold_curve.values,
         color='red', linewidth=2, linestyle='--', label='买入持有')

plt.title(f"权益曲线对比 - {symbol}")
plt.xlabel("日期")
plt.ylabel("组合价值 (%)")
plt.legend()
plt.grid(True, alpha=0.3)
plt.tight_layout()
plt.show()

压力测试揭示了一个关键结论：平均权益曲线持续高于买入持有曲线。这强烈暗示，策略的优势来源于其内在逻辑结构，而不是某个偶然的优秀参数组合。

前向优化：更严格的验证

前向优化（Walk-Forward Optimization）被广泛认为是量化交易中验证策略稳健性的黄金标准。其核心思想模拟了实盘中的参数调整过程：

1. 将历史数据分割成连续的训练窗口和测试窗口。
2. 在训练窗口内寻找“当前”最优的参数。
3. 将找到的参数应用于紧随其后的测试窗口进行绩效评估。
4. 滚动窗口，重复以上步骤，避免使用未来数据。

def run_walk_forward(df, train_years=4, test_years=1,
                     ADX_LEVEL=[20], ADX_PERIOD=[14],
                     BB_PERIOD=[20], BB_SHIFT=[2], BB_STD=[2]):
    """
    执行前向优化

    参数：
        df: 价格数据
        train_years: 训练窗口年数
        test_years: 测试窗口年数
        其他参数: 各指标的参数范围

    返回：
        包含每个窗口优化结果的列表
    """
    results = []
    max_lookback = max(max(ADX_PERIOD), max(BB_PERIOD)) + 10

    start_year = df.index[0].year
    end_year = df.index[-1].year

    # 滚动窗口遍历
    for train_start in range(start_year, end_year - train_years - test_years + 1):
        train_end = train_start + train_years - 1
        test_start = train_end + 1
        test_end = test_start + test_years - 1

        # 获取训练和测试数据的索引
        train_start_idx = df.index.get_loc(df[df.index.year == train_start].index[0])
        train_end_idx = df.index.get_loc(df[df.index.year == train_end].index[-1])
        test_start_idx = df.index.get_loc(df[df.index.year == test_start].index[0])
        test_end_idx = df.index.get_loc(df[df.index.year == test_end].index[-1])

        # 切分数据（包含足够的回溯期）
        train_slice = df.iloc[max(0, train_start_idx - max_lookback): train_end_idx + 1]
        test_slice = df.iloc[max(0, test_start_idx - max_lookback): test_end_idx + 1]

        # 在训练窗口中寻找最优参数
        best_perf = -np.inf
        best_params = None

        for al, ap, bp, sh, st in product(ADX_LEVEL, ADX_PERIOD, BB_PERIOD, BB_SHIFT, BB_STD):
            # 计算信号
            entry_raw = bb_expansion(train_slice.copy(), bp, st, sh).fillna(False)
            exit_raw = adx_cross_below(train_slice.copy(), ap, al).fillna(False)

            # 回测
            pf_train = vbt.Portfolio.from_signals(
                close=df['Open'].iloc[train_start_idx:train_end_idx+1],
                entries=entry_raw.shift(1).astype(bool).fillna(False).to_numpy()[-len(df['Open'].iloc[train_start_idx:train_end_idx+1]):],
                exits=exit_raw.shift(1).astype(bool).fillna(False).to_numpy()[-len(df['Open'].iloc[train_start_idx:train_end_idx+1]):],
                init_cash=100_000,
                fees=0.001,
                slippage=0.002,
                freq='1d'
            )

            # 记录最佳参数
            if pf_train.total_return() > best_perf:
                best_perf = pf_train.total_return()
                best_params = (al, ap, bp, sh, st)

        results.append({
            "train_period": f"{train_start}-{train_end}",
            "test_period": f"{test_start}-{test_end}",
            "params": best_params,
            "test_slice": test_slice
        })

    return results

前向优化结果

根据文中的前向优化测试，该策略在2004年至2025年期间的表现数据如下：

结果显示，策略不仅在绝对收益上显著超越了简单的买入持有策略，而且在风险控制（更低的最大回撤）方面也表现得更为出色。

总结

本文详细讲解了一个结合布林带扩张与ADX衰减的Python量化交易策略，并着重介绍了压力测试和前向优化这两种验证策略稳健性的核心方法。

策略逻辑本身简洁有力：利用波动率扩张捕捉趋势起点，凭借趋势强度衰减信号锁定利润离场。与追求单一“圣杯”参数的传统优化思路不同，压力测试通过评估策略在海量参数空间下的平均表现来证明其普适性；而前向优化则通过模拟实盘中的滚动参数调整过程，有效避免了“未来函数”偏差，提供了更为可靠的策略验证。这些方法都离不开对智能与数据云等领域工具和思想的运用。

对于希望通过 Python 进入量化领域的开发者而言，本案例完整展示了如何将 pandas、numpy 和 vectorbt 等库有机结合，完成从策略构思、代码实现到严谨验证的全流程。它传递了一个至关重要的理念：优秀的量化策略追求的并非历史数据上的极致曲线，而是在各种参数设定与市场环境中都能持续展现的结构性优势。

当然，必须牢记：所有历史回测结果均不代表未来表现，任何策略投入实盘前都必须经过更充分的研究与测试。希望这篇实战指南能为你带来启发，更多深入的技术探讨欢迎关注云栈社区。