使用backtesting.py与pandas_ta编写布林带策略返回Null报错求助

问题场景 #

使用backtesting.py结合pandas_ta实现布林带策略回测时，初始化阶段触发ValueError，提示指标bbands(C,20,2,False)返回None，与数据长度（2148）不匹配。已确认self.data.Close有有效数据，单独调用ta.bbands(GOOG.Close, length=20, std=2, append=False)能正常输出结果，但在Strategy类的init方法中通过self.I()调用时出现异常。

错误信息 #

ValueError: Indicators must return (optionally a tuple of) numpy.arrays of same length as `data` (data shape: (2148,); indicator "bbands(C,20,2,False)"shape: , returned value: None)

问题原因 #

backtesting.py的self.I()方法要求传入的指标函数返回numpy数组（或数组元组），但pandas_ta的bbands函数默认返回pandas DataFrame；同时self.data.Close是backtesting封装的Data对象，直接传入pandas_ta可能存在数据类型适配问题，导致函数返回None。

解决方案 #

步骤1：转换数据类型并提取数组 #

在init方法中，先将self.data.Close转为pandas Series，再传入ta.bbands，然后提取返回DataFrame中的三个波段（中轨、上轨、下轨）对应的numpy数组，通过self.I()分别注册。

步骤2：调整next方法的索引方式 #

因为指标返回的是numpy数组，无法用列名索引，需直接通过数组下标访问对应波段数据。

修改后的完整代码 #

import pandas as pd
import numpy as np
import pandas_ta as ta
from backtesting import Backtest, Strategy
from backtesting.test import GOOG

class BBandsStrat(Strategy):
    time_period = 20
    dev = 2

    def init(self):
        # 将backtesting的Data对象转为pandas Series
        close_series = pd.Series(self.data.Close)
        # 计算布林带，得到DataFrame
        bbands_df = ta.bbands(close_series, length=self.time_period, std=self.dev, append=False)
        # 提取三个波段的numpy数组，通过self.I注册
        self.middle_band = self.I(lambda x: x, bbands_df['BBM_20_2.0'].values)
        self.upper_band = self.I(lambda x: x, bbands_df['BBU_20_2.0'].values)
        self.lower_band = self.I(lambda x: x, bbands_df['BBL_20_2.0'].values)

    def next(self):
        current_close = self.data.Close[-1]
        current_lower = self.lower_band[-1]
        current_upper = self.upper_band[-1]

        if self.position:
            if current_close > current_upper:
                self.position.close()
        else:
            if current_close < current_lower:
                self.buy()

cash = 10_000
bt = Backtest(GOOG, BBandsStrat, cash=cash)
stats = bt.run()
print(stats)
# 可选：绘制回测结果
# bt.plot()

补充说明 #

也可以将三个波段数组打包成元组一次性注册，示例如下：

def init(self):
    close_series = pd.Series(self.data.Close)
    bbands_df = ta.bbands(close_series, length=self.time_period, std=self.dev, append=False)
    self.bbands = self.I(lambda x: (x['BBM_20_2.0'].values, x['BBU_20_2.0'].values, x['BBL_20_2.0'].values), bbands_df)

此时在next方法中需通过self.bbands[0]（中轨）、self.bbands[1]（上轨）、self.bbands[2]（下轨）访问数据。

内容的提问来源于stack exchange，提问作者Boerboel1991