4、数据获取模块：获取主力合约列表、获取历史日线数据、处理复权数据、数据缓存与增量更新

做期货价差分析，第一步就是搞数据。这事儿听起来简单，但坑特别多。我刚开始做的时候，以为从交易所拉个K线就完事了，结果发现数据对不上、复权搞错、每次跑脚本都要重新下载……嗯，今天咱们就把这块彻底讲透。

4.1 获取主力合约列表

期货和股票不一样，每个品种有多个合约同时在交易。比如螺纹钢，有RB2401、RB2405、RB2410……你总不能全拿来做价差分析吧？所以我们要找「主力合约」——也就是成交量最大、持仓量最大的那个。

我个人习惯用「持仓量」来判断主力合约。为什么？因为成交量可能被刷单干扰，但持仓量是实打实的资金沉淀。

代码实现其实不复杂，我贴一段我常用的逻辑：

def get_main_contracts(exchange='SHFE'):
    """
    获取指定交易所所有品种的主力合约
    返回: {'RB': 'RB2410', 'HC': 'HC2401', ...}
    """
    # 1. 获取所有合约列表
    all_contracts = get_all_contracts(exchange)
    
    # 2. 按品种分组
    from collections import defaultdict
    grouped = defaultdict(list)
    for contract in all_contracts:
        symbol = extract_symbol(contract)  # 提取品种代码
        grouped[symbol].append(contract)
    
    # 3. 每个品种选持仓量最大的
    main_contracts = {}
    for symbol, contracts in grouped.items():
        max_oi = 0
        main = None
        for c in contracts:
            oi = get_open_interest(c)  # 获取持仓量
            if oi > max_oi:
                max_oi = oi
                main = c
        main_contracts[symbol] = main
    
    return main_contracts

4.2 获取历史日线数据

拿到主力合约列表后，下一步就是拉历史K线。这里有个问题：主力合约是会变的。比如1月份主力是RB2405，到了5月份就变成RB2410了。如果你只拉一个合约的连续数据，那中间会有跳空。

所以，正确的做法是——拼接。把不同时间段的主力合约K线拼成一条连续的曲线。我管它叫「主力连续合约」。

举个例子：

时间段	主力合约
2024-01 ~ 2024-04	RB2405
2024-05 ~ 2024-08	RB2410
2024-09 ~ 2024-12	RB2501

拼接的时候要注意：切换日那天，前后两个合约的价格可能差很多。这时候就需要复权处理了。

4.3 处理复权数据

复权，说白了就是消除合约切换带来的价格跳空。我见过有人直接拿原始数据做分析，结果价差图上一堆莫名其妙的尖峰——其实就是合约切换造成的。

常用的复权方法有两种：

• 前复权： 调整历史价格，让切换日前后价格连续。适合看历史走势。
• 后复权： 调整当前价格，保持历史数据不变。适合做回测。

我个人做价差分析时，更倾向于用后复权。为什么？因为价差分析关注的是「相对关系」，后复权不会改变历史价差的形态。

复权计算的代码示例：

def adjust_price(df, method='backward'):
    """
    复权处理
    df: 包含主力合约切换标记的DataFrame
    method: 'forward' 前复权, 'backward' 后复权
    """
    df = df.copy()
    
    # 找到所有切换点
    switch_points = df[df['is_switch'] == True].index
    
    if method == 'backward':
        # 后复权：从后往前调整
        for i in range(len(switch_points)-1, -1, -1):
            idx = switch_points[i]
            ratio = df.loc[idx, 'close_old'] / df.loc[idx, 'close_new']
            df.loc[:idx, 'close'] *= ratio
            df.loc[:idx, 'open'] *= ratio
            df.loc[:idx, 'high'] *= ratio
            df.loc[:idx, 'low'] *= ratio
    
    return df

4.4 数据缓存与增量更新

每次跑脚本都重新下载全部数据？太慢了。尤其当你分析几十个品种、几年数据的时候，下载一次可能要十几分钟。

我的做法是：本地存一份，每天只更新最新的数据。这就是「增量更新」。

数据缓存的结构我一般这样设计：

• 合约列表缓存： 存成JSON文件，每天更新一次
• 日线数据缓存： 按品种+合约存成CSV或Parquet，每天追加最新K线
• 复权数据缓存： 存成HDF5或Feather，每周重新计算一次

增量更新的核心逻辑：

def update_daily_data(symbol, start_date=None):
    """
    增量更新日线数据
    """
    cache_file = f'data/{symbol}.parquet'
    
    # 1. 检查本地缓存
    if os.path.exists(cache_file):
        local_data = pd.read_parquet(cache_file)
        last_date = local_data.index[-1]
        print(f'本地已有数据，最后日期：{last_date}')
    else:
        last_date = None
        local_data = pd.DataFrame()
    
    # 2. 只下载缺失的数据
    if last_date:
        new_data = download_kline(symbol, start=last_date + timedelta(days=1))
    else:
        new_data = download_kline(symbol, start=start_date or '2020-01-01')
    
    # 3. 合并并去重
    if not new_data.empty:
        combined = pd.concat([local_data, new_data])
        combined = combined[~combined.index.duplicated(keep='last')]
        combined.sort_index(inplace=True)
        combined.to_parquet(cache_file)
        print(f'更新完成，共 {len(combined)} 条数据')
    else:
        print('数据已是最新，无需更新')

4.5 本章知识体系

下面这张图，把数据获取模块的整体流程串起来了。我建议你保存下来，写代码的时候对照着看：

整个流程走下来，你会发现数据获取其实是个「体力活」——逻辑不复杂，但细节特别多。我个人建议你先把主力合约列表和日线数据跑通，再慢慢加上复权和缓存。一步到位容易出bug，我当年就是这么过来的。

---