当前位置：首页>python>用 Python 实现布林带 + ADX 量化交易策略:从压力测试到前向验证

用 Python 实现布林带 + ADX 量化交易策略:从压力测试到前向验证

2026-01-20 18:36:49

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

引言

你是否曾想过，如何用 Python 构建一个真正稳健的量化交易策略？很多初学者在学习量化交易时，往往只关注单一参数的优化，却忽略了策略的稳定性和适应性。

今天这篇文章将带你深入了解一个基于布林带（Bollinger Bands）扩张和 ADX 衰减的交易策略。更重要的是，我们将学习如何通过压力测试（Stress Test）和前向优化（Walk-Forward Optimization）来验证策略的稳健性，而不是简单地追求历史回测中的最佳参数。

策略核心逻辑

这个策略的思路非常直观：

入场条件：当布林带宽度扩张时（波动率突破信号），表明市场可能进入趋势行情
出场条件：当 ADX 指标下穿某个阈值时（趋势强度减弱信号），及时止盈离场

简单来说，就是在波动率扩大时进场，在趋势减弱时离场。

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

与其寻找一个"最优"参数组合，不如测试策略在大量参数组合下的整体表现。以下代码展示了如何进行压力测试：

import pandas as pdimport numpy as npimport yfinance as yfimport itertoolsimport vectorbt as vbtimport 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, 24ADX_PERIODS = list(range(10, 21, 2))  # ADX 周期范围：10, 12, 14, 16, 18, 20BB_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 指标的计算函数：

defcalculate_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  # 返回中轨、上轨、下轨defbb_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 天前的带宽defcalculate_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 adxdefadx_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）是量化交易中验证策略稳健性的黄金标准。其核心思想是：

将数据分成训练窗口和测试窗口
在训练窗口中寻找最优参数
将找到的参数应用于测试窗口
滚动前进，重复以上步骤

defrun_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 = Nonefor 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