第1章：数据获取与清洗——ICE/EEX交易所数据接口、Python数据抓取、缺失值处理、异常值检测

做碳市场期限结构量化套利，第一步就是搞数据。这活儿看着简单，其实坑特别多。我刚开始做碳金融研究那会儿，光数据清洗就折腾了两周——嗯，后来发现是接口参数传错了。

今天咱们就聊聊，怎么从ICE和EEX这两个主流碳交易所拿到干净的数据，以及拿到之后怎么处理那些烦人的缺失值和异常值。

1.1 ICE与EEX交易所数据接口

先说说这两个交易所。ICE（洲际交易所）和EEX（欧洲能源交易所）是碳配额期货的主要交易场所。EUA（欧盟配额）期货在这两个地方都有挂牌，但数据格式、接口方式不太一样。

核心区别：

  • ICE：提供RESTful API，数据以JSON格式返回，需要申请API Key
  • EEX：提供FTP下载和CSV文件，数据更新频率为每日收盘后
  • 共同点：都支持历史数据回溯，但免费额度有限

我个人习惯用ICE的API做实时数据，用EEX的CSV做历史数据校验。为什么？因为ICE的数据延迟低，但收费贵；EEX的数据免费，但只有日频。

1.2 Python数据抓取实战

这里我主要用两个库：pandas-datareader 和 requests。前者适合快速拉取标准金融数据，后者适合处理自定义API。

1.2.1 使用pandas-datareader获取ICE数据

先看个例子。拉取ICE EUA期货近月合约的收盘价：

import pandas as pd
import pandas_datareader.data as web
from datetime import datetime

# 设置时间范围
start = datetime(2023, 1, 1)
end = datetime(2024, 12, 31)

# 获取ICE EUA期货数据（代码示例，实际需要替换为真实代码）
# 注意：pandas-datareader对ICE的支持需要特定数据源
try:
    data = web.DataReader('EUA_FUTURES', 'ice', start, end)
    print(data.head())
except Exception as e:
    print(f"数据获取失败：{e}")

避坑指南： 我曾经因为没注意时区问题，导致获取的数据日期偏移了一天。ICE的数据时间戳是UTC，而EEX是CET（欧洲中部时间）。记得统一转换成北京时间或你需要的时区。

1.2.2 使用requests抓取EEX数据

EEX的数据通常通过FTP下载，但有些数据也提供HTTP接口。这里用requests模拟：

import requests
import pandas as pd
from io import StringIO

# EEX数据下载URL（示例）
url = "https://www.eex.com/data/eua_futures.csv"

# 发送请求
response = requests.get(url, timeout=10)

# 检查状态码
if response.status_code == 200:
    # 读取CSV数据
    csv_data = StringIO(response.text)
    df = pd.read_csv(csv_data, sep=';', encoding='utf-8')
    print(f"成功获取{len(df)}条数据")
else:
    print(f"请求失败，状态码：{response.status_code}")

你想想看，如果网络不稳定怎么办？我一般会加个重试机制，最多重试3次，每次间隔5秒。

1.3 缺失值处理

拿到数据后，第一件事就是检查缺失值。碳市场数据有个特点：节假日、周末、以及某些特殊事件日（比如拍卖暂停）会导致数据缺失。

我常用的方法有这三种：

方法 适用场景 代码实现
向前填充 短期缺失（1-2天） df.fillna(method='ffill')
线性插值 趋势明显的数据 df.interpolate(method='linear')
删除缺失行 缺失比例超过20% df.dropna()

注意： 千万别一上来就删数据！我在项目中遇到过，某次因为连续三天缺失数据，直接删除后导致期限结构曲线出现断层，套利信号全乱了。先看看缺失的原因再说。

1.4 异常值检测

异常值在碳市场里很常见。比如2022年俄乌冲突期间，EUA价格一天内波动超过20%。这些是真实的市场行为，不是数据错误。

我一般用两种方法检测：

  • Z-score方法：适用于正态分布的数据，Z值超过3的视为异常
  • IQR方法：适用于非正态分布，超出1.5倍四分位距的视为异常

看代码：

import numpy as np

def detect_outliers_zscore(data, threshold=3):
    """Z-score异常检测"""
    z_scores = np.abs((data - data.mean()) / data.std())
    return z_scores > threshold

def detect_outliers_iqr(data):
    """IQR异常检测"""
    Q1 = data.quantile(0.25)
    Q3 = data.quantile(0.75)
    IQR = Q3 - Q1
    lower_bound = Q1 - 1.5 * IQR
    upper_bound = Q3 + 1.5 * IQR
    return (data < lower_bound) | (data > upper_bound)

# 使用示例
price_data = df['close_price']
outliers_z = detect_outliers_zscore(price_data)
outliers_iqr = detect_outliers_iqr(price_data)

print(f"Z-score检测到{outliers_z.sum()}个异常值")
print(f"IQR检测到{outliers_iqr.sum()}个异常值")

说白了，异常值不一定要删除。如果是数据录入错误（比如价格多了一个零），那就修正；如果是市场极端事件，那就保留，但要在模型里做标记。

1.5 知识体系总览

下面这张图，是我自己总结的数据获取与清洗流程。每次做新项目，我都会先过一遍这个框架：

碳市场数据获取与清洗流程 数据源 ICE API (JSON) EEX FTP/CSV pandas-datareader requests + CSV解析 数据合并与对齐 缺失值处理 + 异常值检测

这个流程看起来简单，但每一步都有细节。比如数据合并时，ICE和EEX的日期格式可能不一样，一个用YYYY-MM-DD，一个用DD/MM/YYYY。我建议统一用datetime类型，省得后面出问题。

个人经验： 数据清洗占整个量化项目的时间，通常超过60%。别嫌麻烦，数据干净了，后面的模型才能跑得稳。我曾经因为一个字段类型没转换，导致回测结果差了10倍——嗯，从那以后我每次都会做数据完整性校验。

好了，数据获取与清洗这块就聊到这儿。记住：好数据是好模型的基础。下一章咱们聊聊期限结构的构建方法，到时候会用到今天清洗好的数据。

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

← 上一章 📖 返回目录 下一章 →