量化工具：Python实现库存-基差数据获取与清洗

做基差交易的朋友都知道，库存和价差的关系是核心。但数据从哪来？怎么处理？这才是真正的门槛。

我个人习惯用Python来处理这整套流程。为什么？因为Python生态里，数据获取、清洗、分析、回测一条龙都能搞定。今天我就带你走一遍，从数据源到干净可用的DataFrame，每一步我都会讲清楚。

数据源的选择与接口对接

先说说数据源。国内商品期货的库存数据，我主要用三个渠道：

• 交易所官网：上期所、大商所、郑商所每周五下午发布库存周报
• Wind/聚宽：付费数据源，接口稳定，但需要授权
• 第三方爬虫：自己写爬虫抓取交易所网页

我个人建议，初期用免费数据源练手，等策略跑通了再上付费接口。我在项目中遇到过一个问题——免费数据源偶尔会断更，所以一定要加异常处理。

基差数据的获取逻辑

基差 = 现货价格 - 期货价格。但这里有个坑——现货价格怎么定义？

我一般用两种方式：

1. 交易所公布的现货指数：比如螺纹钢的上海现货价
2. 主力合约的基差：用近月合约价格替代现货

你想想看，如果现货数据缺失，用主力合约替代其实也能凑合。但要注意，换月的时候基差会跳变，这个必须处理。

import pandas as pd
import numpy as np
from datetime import datetime, timedelta

# 模拟获取基差数据
def get_basis_data(contract_code, start_date, end_date):
    """
    获取基差数据
    contract_code: 合约代码，如 'RB.SHF'
    """
    # 这里假设从本地CSV读取
    df = pd.read_csv(f'data/{contract_code}_basis.csv', parse_dates=['date'])
    df = df[(df['date'] >= start_date) & (df['date'] <= end_date)]
    return df

# 示例：获取螺纹钢基差
basis_df = get_basis_data('RB.SHF', '2023-01-01', '2023-12-31')
print(basis_df.head())

库存数据的清洗流程

库存数据比基差数据脏得多。为什么？因为不同交易所的格式不一样，甚至同一交易所不同品种的字段名都不同。

我记得有一次，大商所的豆粕库存数据里混入了空值，导致我的模型直接报错。从那以后，我养成了一个习惯——拿到数据先做三件事：

清洗步骤	具体操作	代码示例
缺失值处理	检查空值，用前向填充或插值	df.fillna(method='ffill')
异常值检测	用3σ原则或IQR剔除	df[(df - df.mean()).abs() < 3*df.std()]
格式统一	日期转datetime，数值转float	pd.to_datetime()

数据对齐与合并

基差数据和库存数据的时间频率不一样。基差是日频，库存是周频。怎么对齐？

我的做法是：以基差数据为主表，用库存数据做左连接。库存数据只在每周五更新，那周五之后的交易日，库存值保持不变。

# 合并基差和库存数据
def merge_basis_inventory(basis_df, inventory_df):
    # 将库存数据按日期排序
    inventory_df = inventory_df.sort_values('date')
    # 使用merge_asof进行时间对齐
    merged = pd.merge_asof(
        basis_df.sort_values('date'),
        inventory_df.sort_values('date'),
        on='date',
        direction='backward'  # 用最近的库存数据填充
    )
    return merged

# 执行合并
final_df = merge_basis_inventory(basis_df, inventory_df)
print(f"合并后数据量: {len(final_df)} 行")

这里有个细节——merge_asof的direction='backward'参数，意思是取当前日期之前最近的库存数据。说白了，就是用上周的库存数据来匹配本周的交易数据。

数据质量检查清单

数据清洗完了，别急着用。我建议你跑一遍质量检查：

• ✅ 检查日期范围是否完整，有没有跳空
• ✅ 检查基差的正负号是否合理（比如正向市场基差为负）
• ✅ 检查库存数据是否有突变（比如一天内库存翻倍，那肯定是错的）
• ✅ 检查合并后的数据量是否匹配预期

核心逻辑流程图

下面这张图，是我做库存-基差数据处理的完整流程。你可以把它当作操作手册：

完整代码示例

最后，我给你一个可以直接跑的完整脚本。这个脚本整合了上面所有的步骤：

import pandas as pd
import numpy as np
from datetime import datetime

class BasisInventoryDataPipeline:
    def __init__(self, contract_code):
        self.contract_code = contract_code
        self.basis_data = None
        self.inventory_data = None
        self.merged_data = None
    
    def load_basis_data(self, filepath):
        """加载基差数据"""
        self.basis_data = pd.read_csv(filepath, parse_dates=['date'])
        # 清洗：去除空值
        self.basis_data = self.basis_data.dropna(subset=['basis'])
        print(f"基差数据加载完成，共 {len(self.basis_data)} 条")
    
    def load_inventory_data(self, filepath):
        """加载库存数据"""
        self.inventory_data = pd.read_csv(filepath, parse_dates=['date'])
        # 清洗：去除异常值（3σ原则）
        mean = self.inventory_data['inventory'].mean()
        std = self.inventory_data['inventory'].std()
        self.inventory_data = self.inventory_data[
            (self.inventory_data['inventory'] > mean - 3*std) &
            (self.inventory_data['inventory'] < mean + 3*std)
        ]
        print(f"库存数据加载完成，共 {len(self.inventory_data)} 条")
    
    def merge_data(self):
        """合并数据"""
        if self.basis_data is None or self.inventory_data is None:
            raise ValueError("请先加载数据")
        
        self.merged_data = pd.merge_asof(
            self.basis_data.sort_values('date'),
            self.inventory_data.sort_values('date'),
            on='date',
            direction='backward'
        )
        # 添加基差-库存比率特征
        self.merged_data['basis_inventory_ratio'] = (
            self.merged_data['basis'] / self.merged_data['inventory']
        )
        print(f"数据合并完成，共 {len(self.merged_data)} 条")
        return self.merged_data

# 使用示例
pipeline = BasisInventoryDataPipeline('RB.SHF')
pipeline.load_basis_data('data/rb_basis.csv')
pipeline.load_inventory_data('data/rb_inventory.csv')
final_data = pipeline.merge_data()
print(final_data.tail())

嗯，到这里，数据获取和清洗的流程就完整了。你拿到的数据，可以直接用于后续的基差交易策略开发。记住一句话：数据质量决定策略上限。花80%的时间在数据处理上，一点都不亏。