4、自定义Bundle开发（上）：编写Bundle扩展文件（extension.py）、定义ingest函数、处理数据格式（OHLCV）、数据清洗与对齐

好，咱们进入正题。

上一章我们聊了Zipline的Bundle机制，说白了就是一套数据加载的规范。但Zipline官方只提供了美股和A股的部分数据源，如果你要接入自己的数据——比如某家券商的实时行情、自己爬取的日线数据，或者某个小众交易所的币圈数据——那就得自己写Bundle了。

我自己第一次写Bundle时，踩了不少坑。最头疼的就是数据格式不对齐，Zipline对OHLCV的字段名、时间戳精度、缺失值处理都有严格的要求。今天我们就从最基础的extension.py文件开始，一步步把自定义Bundle搭起来。

4.1 什么是Bundle扩展文件？

说白了，extension.py就是Zipline的“插件注册中心”。你在这个文件里定义好数据源、注册函数，Zipline启动时就会自动加载。它通常放在项目的~/.zipline/extension.py路径下，或者通过环境变量ZIPLINE_EXTENSION指定。

我习惯把extension.py放在项目根目录，然后软链接到用户目录，方便版本管理。嗯，这里要注意：extension.py里不能有语法错误，否则Zipline启动会直接报错，而且错误信息往往很隐晦。

4.2 编写extension.py的基本结构

一个最简单的extension.py长这样：

# extension.py
from zipline.data.bundles import register
from zipline.data.bundles.my_bundle import my_bundle_ingest

register(
    'my_bundle',          # bundle名称，后面ingest和run时要用
    my_bundle_ingest,     # 数据摄入函数
    calendar_name='NYSE', # 交易日历，美股用NYSE，A股用SSE
    minutes_per_day=390,  # 美股每天390分钟（6.5小时）
)

看到没？核心就三样东西：名字、函数、日历。但实际项目中，你还需要处理数据格式、清洗逻辑、对齐时间戳。这些都在my_bundle_ingest函数里实现。

4.3 定义ingest函数——数据摄入的核心

ingest函数是Bundle的灵魂。它的签名是固定的：

def my_bundle_ingest(environ,
                     asset_db_writer,
                     minute_bar_writer,
                     daily_bar_writer,
                     adjustment_writer,
                     calendar,
                     start_session,
                     end_session,
                     cache,
                     show_progress,
                     output_dir):
    # 你的数据获取和写入逻辑
    pass

参数很多，但别慌。我们只需要关注几个关键的：

• daily_bar_writer：写入日线数据
• minute_bar_writer：写入分钟数据（可选）
• asset_db_writer：写入资产元数据（股票代码、名称、上市退市日期）
• adjustment_writer：写入复权数据（分红、拆股）
• start_session / end_session：数据的时间范围

我个人习惯先把数据从原始源（比如CSV、数据库、API）拉下来，转成Pandas DataFrame，再交给这些Writer。这样逻辑清晰，调试也方便。

4.4 处理数据格式——OHLCV的“潜规则”

Zipline对OHLCV字段名有硬性要求：

字段	Zipline要求	常见坑
Open	float64，不能有NaN	有些数据源用'open'小写，Zipline不认
High	float64，不能有NaN	High必须 ≥ Open, Close, Low
Low	float64，不能有NaN	Low必须 ≤ Open, Close, High
Close	float64，不能有NaN	停牌日Close可能为0，需要填充
Volume	int64，不能有NaN	停牌日Volume为0，不能是NaN
Dividend	float64，可选	复权时需要
Split	float64，可选	拆股比例，1:2就是2.0

我在项目中遇到过最坑的事：某数据源的Volume字段是float64，但Zipline要求int64。直接写入会报类型错误。所以一定要做类型转换：

df['volume'] = df['volume'].astype('int64')

还有，时间戳必须是UTC的datetime类型，不能带时区信息。Zipline内部统一用UTC。

4.5 数据清洗与对齐——避坑指南

数据清洗是Bundle开发中最耗时、最容易出错的环节。我曾经因为一个数据源在节假日返回了空数据，导致整个回测结果偏移了5%。

这里分享几个我常用的清洗步骤：

1. 去除重复行：同一个交易日出现两次，取最后一次或平均值
2. 填充缺失值：停牌日用前一天的Close填充Open/High/Low/Close，Volume填0
3. 检查OHLC逻辑：High必须大于等于Open、Close、Low；Low必须小于等于Open、Close、High。如果不满足，我一般取三者极值修正
4. 对齐交易日历：只保留交易日的数据，非交易日剔除
5. 排序：按时间升序排列，Zipline要求数据是时间递增的

举个例子，假设你从CSV读到的数据是这样的：

date,open,high,low,close,volume
2023-01-01,100.0,102.0,99.0,101.0,1000000
2023-01-02,101.0,103.0,100.0,102.0,1200000
2023-01-03,102.0,104.0,101.0,103.0,1100000

但2023-01-01是元旦，非交易日。你需要用calendar过滤掉它：

from zipline.utils.calendars import get_calendar

cal = get_calendar('NYSE')
trading_days = cal.sessions_in_range(start_session, end_session)
df = df[df['date'].isin(trading_days)]

这样处理后，数据就干净了。

4.6 完整的ingest函数示例

把上面所有步骤串起来，一个完整的ingest函数大概长这样：

def my_bundle_ingest(environ,
                     asset_db_writer,
                     minute_bar_writer,
                     daily_bar_writer,
                     adjustment_writer,
                     calendar,
                     start_session,
                     end_session,
                     cache,
                     show_progress,
                     output_dir):

    # 1. 从CSV读取原始数据
    raw_df = pd.read_csv('my_data.csv', parse_dates=['date'])
    
    # 2. 数据清洗
    raw_df = raw_df.drop_duplicates(subset=['date', 'symbol'])
    raw_df = raw_df.sort_values('date')
    
    # 3. 对齐交易日历
    trading_days = calendar.sessions_in_range(start_session, end_session)
    raw_df = raw_df[raw_df['date'].isin(trading_days)]
    
    # 4. 处理OHLCV字段
    raw_df['volume'] = raw_df['volume'].fillna(0).astype('int64')
    raw_df['open'] = raw_df['open'].fillna(method='ffill')
    raw_df['high'] = raw_df['high'].fillna(method='ffill')
    raw_df['low'] = raw_df['low'].fillna(method='ffill')
    raw_df['close'] = raw_df['close'].fillna(method='ffill')
    
    # 5. 写入日线数据
    daily_bar_writer.write(raw_df)
    
    # 6. 写入资产元数据
    asset_db_writer.write(...)
    
    # 7. 写入复权数据（如果有）
    adjustment_writer.write(...)

嗯，这里省略了资产元数据和复权数据的细节，因为每个数据源不一样。但核心逻辑就是：先清洗，再写入。

4.7 本章知识体系

下面这张图帮你理清自定义Bundle开发的整体流程：

从图上可以看得很清楚：extension.py是入口，ingest函数是引擎，数据清洗是必经之路。每一步都环环相扣，缺一不可。

好了，这一章我们完成了自定义Bundle的骨架搭建。下一章会深入ingest函数的内部细节，包括如何高效处理分钟数据、如何管理资产元数据、以及如何应对数据源的各种“幺蛾子”。

---

公众号:蓝海资料掘金营，微信deep3321