第一章：数据获取与清洗——打好量化地基

做量化交易，尤其是期限结构策略，最怕什么？

不是策略亏钱，而是数据出错。

我见过太多人，模型跑得飞起，结果一查，主力合约切换日期搞错了，或者交割日数据没处理干净。嗯，那真是欲哭无泪。所以第一章，咱们先把地基打牢。

1.1 数据源的选择与对比

做有色金属期限结构，你绕不开三个交易所：上期所（SHFE）、LME、COMEX。每个交易所的数据特点都不一样，我简单说说我的经验。

交易所	合约代码规则	数据频率	主力合约切换规则	常见坑点
上期所（SHFE）	CU2301（铜，2023年1月）	日频	通常持仓量最大	节假日调整、涨跌停板
LME	CA03M（铜，3个月远期）	日频	滚动合约，无固定主力	交割日规则复杂（3个月滚动）
COMEX	HGH23（铜，2023年3月）	日频	持仓量+成交量综合判断	电子盘与场内盘价差

我个人习惯，上期所用万得或天软，LME用路透或Bloomberg，COMEX用Quandl或直接交易所API。但不管用哪个源，清洗逻辑是通用的。

1.2 交割日处理——别让过期合约坑了你

交割日处理，说白了就是知道哪天合约到期，然后把它从数据里剔除。为什么？因为临近交割的合约，流动性差，价格容易失真。

我曾经在回测时发现，某铜合约在最后交易日突然暴跌5%，一查，原来是交割日数据没剔除，把异常值当成了信号。那真是血的教训。

处理逻辑很简单：

• 上期所：合约代码的最后两位数字代表年份和月份，比如CU2301，交割日是2023年1月的第15个交易日（具体看交易所公告）。
• LME：每个合约都有明确的到期日，通常是每个月的第三个星期三。
• COMEX：交割月前一个月的最后一个交易日。

代码实现上，我一般这样写：

import pandas as pd
from datetime import datetime

def get_delivery_date(contract_code, exchange):
    """
    根据合约代码和交易所，返回交割日期
    """
    if exchange == 'SHFE':
        # 假设合约代码格式：CU2301
        year = int('20' + contract_code[-4:-2])
        month = int(contract_code[-2:])
        # 这里简化处理，实际需要查询交易所日历
        return datetime(year, month, 15)  # 示例
    elif exchange == 'LME':
        # LME规则复杂，这里仅示意
        return None
    elif exchange == 'COMEX':
        # COMEX规则
        return None
    else:
        raise ValueError('Unknown exchange')

1.3 主力合约切换——别让流动性坑了你

主力合约切换，是期限结构策略的核心。你想想看，如果你一直用旧的主力合约，流动性越来越差，价差越来越大，策略信号能准吗？

我常用的切换规则有两种：

1. 持仓量最大法：每天比较所有同品种合约的持仓量，选最大的那个。优点是简单，缺点是切换可能滞后。
2. 成交量+持仓量综合法：给成交量和持仓量分别赋权，比如0.6和0.4，算综合得分。我个人更推荐这个，因为成交量反应更灵敏。

代码实现：

def get_main_contract(data, method='volume_position'):
    """
    获取主力合约
    data: DataFrame，包含合约代码、持仓量、成交量
    method: 'position' 或 'volume_position'
    """
    if method == 'position':
        # 持仓量最大
        main_contract = data.loc[data.groupby('date')['open_interest'].idxmax()]
    elif method == 'volume_position':
        # 综合得分
        data['score'] = 0.6 * data['volume'] / data.groupby('date')['volume'].transform('sum') + \
                        0.4 * data['open_interest'] / data.groupby('date')['open_interest'].transform('sum')
        main_contract = data.loc[data.groupby('date')['score'].idxmax()]
    return main_contract

1.4 异常值清洗——别让噪音坑了你

异常值清洗，说白了就是去掉那些明显不合理的数据。比如价格突然跳空10%，或者成交量突然为0。

我常用的方法：

• 3-sigma法则：超过均值±3倍标准差的数据，视为异常。
• IQR法则：超过Q1-1.5*IQR或Q3+1.5*IQR的数据，视为异常。
• 固定阈值法：比如涨跌幅超过±10%的，直接剔除。

我个人习惯用IQR法则，因为它对极端值不那么敏感。3-sigma容易被极端值拉偏。

def clean_outliers(data, column='close', method='iqr'):
    """
    异常值清洗
    """
    if method == 'iqr':
        Q1 = data[column].quantile(0.25)
        Q3 = data[column].quantile(0.75)
        IQR = Q3 - Q1
        lower_bound = Q1 - 1.5 * IQR
        upper_bound = Q3 + 1.5 * IQR
        data_clean = data[(data[column] >= lower_bound) & (data[column] <= upper_bound)]
    elif method == 'sigma':
        mean = data[column].mean()
        std = data[column].std()
        lower_bound = mean - 3 * std
        upper_bound = mean + 3 * std
        data_clean = data[(data[column] >= lower_bound) & (data[column] <= upper_bound)]
    return data_clean

1.5 整体流程框架

下面这张图，是我做数据清洗时的标准流程。你可以照着这个来。

这个流程看起来简单，但每一步都有坑。我刚开始做的时候，光主力合约切换就折腾了两周。后来发现，不同品种的切换规则其实不一样，比如铜和铝的流动性特征就不同。所以，别偷懒，每个品种单独写逻辑。