Backtesting.py 数据要求：单资产 OHLCV、数据源接入与质量自检

import pandas as pd
# 目标形态：index = DatetimeIndex，列 = Open/High/Low/Close[/Volume]
df = pd.DataFrame(raw, columns=['Open', 'High', 'Low', 'Close', 'Volume'])
df.index = pd.to_datetime(df.index)
df = df.sort_index()

1. 读入原始数据

   先用 pandas 把数据读进来看看长什么样：字段名、时间列位置、是否有代码列。先观察再改名，能省掉后面反复对不上的时间。

   import pandas as pd
   raw = pd.read_csv('600519.csv')
   print(raw.head())
   print(raw.columns.tolist())

   预期输出：前几行数据与列名清单。若时间列是字符串、或混着股票代码列，下一步统一处理。

2. 统一列名与时间索引

   把列名重命名为 Open/High/Low/Close（Volume 可选），并把时间列设为 DatetimeIndex——这一步决定了后面所有指标与信号能不能算。

   df = raw.rename(columns={'日期': 'index', '开盘': 'Open', '高点': 'High',
                            '低点': 'Low', '收盘': 'Close', '成交量': 'Volume'})
   df['index'] = pd.to_datetime(df['index'])
   df = df.set_index('index').sort_index()
   print(df.dtypes)

   预期输出：各列为数值类型、索引为时间类型。若价格为字符串，需要先转数值再继续。

3. 做一次质量自检

   在把数据交给任何回测之前先跑一遍检查：缺失值、重复时间、价格是否为 0 或负、时间是否单调。数据问题越早发现，返工成本越低。

   print(df.index.is_monotonic_increasing, df.index.duplicated().sum())
   print(df[['Open', 'High', 'Low', 'Close']].isna().sum().to_dict())
   print(df.index.min(), df.index.max())

   预期输出：True / 0 / 各列缺失数全为 0 / 起止日期覆盖你要回测的区间。任一项不符先回到上一步修数据。

4. 确认单资产与区间

   这个库一次回测对应一条价格序列：多标的要先拆开分别回测，不要塞进一张宽表。同时确认区间足够长，能覆盖不同市场环境。

   price = df['Close']
   print(len(price), price.index[0], price.index[-1])

   预期输出：样本条数与首尾日期。样本过少时，任何指标结论都不可信（区间选择标准以你自己的研究设计为准）。

5. 交给 Backtesting.py 并核对

   把 DataFrame 按位置传给回测对象（它是回测的必要输入），跑一次最小回测确认数据能被正确读取，再进入策略与参数环节。

   from backtesting import Backtest
   from backtesting.test import SMA
   
   bt = Backtest(df, MyStrategy, cash=10_000, commission=.002)
   stats = bt.run()
   print(stats['# Trades'])

   预期输出：打印出成交笔数（具体数值随数据与策略变化）。若成交数为 0，先查列名、时间顺序与信号条件，而不是先怀疑策略逻辑。

def health_check(df, calendar=None, name=''):
    out = {}
    out['rows'] = len(df)
    out['dup_dates'] = int(df.index.duplicated().sum())
    out['monotonic'] = bool(df.index.is_monotonic_increasing)
    out['nan_rows'] = int(df.isna().any(axis=1).sum())
    out['bad_price'] = int((df[['Open', 'High', 'Low', 'Close']] <= 0).any(axis=1).sum())
    out['close_out_of_range'] = int(((df.Close > df.High) | (df.Close < df.Low)).sum())
    if calendar is not None:
        lo, hi = df.index[0], df.index[-1]
        out['missing_days'] = len(set(calendar[(calendar >= lo) & (calendar <= hi)]) - set(df.index))
    return out