数据获取与清洗：碳期货历史数据来源与实战处理
做量化交易，数据就是你的弹药库。没有干净的数据，再牛的策略也是白搭。今天咱们就聊聊碳期货数据怎么拿、怎么洗。
我个人习惯把数据获取分成两条路：一条是直接从交易所拿，另一条是通过第三方数据商。两条路我都走过，各有各的坑。
碳期货数据来源：ICE vs EEX
全球碳期货交易，说白了就两大阵地：ICE（洲际交易所）和EEX（欧洲能源交易所）。
ICE 是老大，EUA（欧盟配额）期货的主力合约基本都在这里。我当年第一次调ICE的API时，差点被它的认证机制搞疯——后来发现是文档没看仔细。
EEX 呢，欧洲本土的交易所，碳配额现货和期货都有。它的数据格式跟ICE不太一样，字段命名也任性。
| 交易所 | 主要产品 | 数据获取方式 | 费用 |
|---|---|---|---|
| ICE | EUA期货、CER期货 | ICE Data Services API | 付费（按数据量） |
| EEX | EUA现货、期货、碳配额拍卖 | EEX Market Data API | 部分免费+高级付费 |
API数据接口调用实战
嗯，这里要注意。API调用不是简单的发个请求就完事。我见过太多人卡在认证这一步。
拿ICE的API举例，它用的是OAuth2.0认证。你得先注册应用，拿到client_id和client_secret。然后每次请求前先换token。
import requests
import pandas as pd
from datetime import datetime, timedelta
# ICE API 认证示例
def get_ice_token(client_id, client_secret):
url = "https://api.ice.com/oauth/token"
payload = {
"grant_type": "client_credentials",
"client_id": client_id,
"client_secret": client_secret
}
response = requests.post(url, data=payload)
return response.json()["access_token"]
# 获取EUA期货历史数据
def fetch_eua_futures(token, contract_code, start_date, end_date):
url = f"https://api.ice.com/marketdata/v2/history"
headers = {"Authorization": f"Bearer {token}"}
params = {
"symbol": contract_code,
"start": start_date,
"end": end_date,
"interval": "daily"
}
response = requests.get(url, headers=headers, params=params)
data = response.json()
# 解析成DataFrame
df = pd.DataFrame(data["data"])
df["date"] = pd.to_datetime(df["date"])
return df
# 调用示例
token = get_ice_token("your_client_id", "your_client_secret")
eua_data = fetch_eua_futures(token, "EUA_DEC2024", "2024-01-01", "2024-06-30")
print(eua_data.head())
数据清洗与预处理实战
数据拿到手，别急着用。原始数据脏得很。我总结了一套「三步清洗法」：去重、补缺、标准化。
第一步：去重与异常值处理
交易所的数据偶尔会重复推送。比如网络波动时，同一根K线可能收到两次。
# 去重
df = df.drop_duplicates(subset=["date", "contract"])
# 异常值检测：价格不能为负，成交量不能为0（非交易日除外）
df = df[(df["close"] > 0) & (df["volume"] >= 0)]
# 我习惯用Z-score检测极端值
from scipy import stats
z_scores = stats.zscore(df["close"])
df = df[(abs(z_scores) < 3)] # 剔除3个标准差以外的数据
第二步：缺失值填充
碳期货市场有节假日，非交易日自然没数据。但有些时候，明明是交易日，数据却丢了——可能是交易所系统延迟。
我的做法是：先检查是不是真的缺失，还是数据还没推送。如果是真缺失，用前向填充。
# 检查缺失
print(df.isnull().sum())
# 前向填充（用上一个交易日的数据）
df["close"] = df["close"].ffill()
df["volume"] = df["volume"].fillna(0) # 成交量缺失视为0
# 如果连续缺失超过3天，我建议直接删除
df = df[df["close"].notna()] # 删除收盘价缺失的行
第三步：数据标准化
不同交易所的数据格式不一样。ICE用美式日期（MM/DD/YYYY），EEX用欧式（DD.MM.YYYY）。字段名也不同——ICE叫"settle_price"，EEX叫"settlement_price"。
我统一转成ISO格式，字段名也统一。
# 统一日期格式
df["date"] = pd.to_datetime(df["date"], format="%Y-%m-%d")
# 统一字段名
df = df.rename(columns={
"settle_price": "close",
"settlement_price": "close",
"volume_traded": "volume"
})
# 添加辅助字段
df["year"] = df["date"].dt.year
df["month"] = df["date"].dt.month
df["day_of_week"] = df["date"].dt.dayofweek # 0=周一
数据清洗流程图
下面这张图是我自己画的数据处理流程，每次做新项目都拿出来参考。
实战中的几个坑
做数据清洗这么多年，我总结了几条血泪教训：
- 时区问题：ICE的数据用UTC时间，EEX用CET。我一开始没注意，导致回测结果差了整整一天。后来统一转成UTC+0。
- 合约代码变化：ICE偶尔会改合约代码。比如EUA期货以前叫"EUA"，后来改成"EUA Futures"。你的代码得兼容历史数据。
- 数据延迟：EEX的免费数据有15分钟延迟。做日内策略的话，这个延迟会要命。我建议付费买实时数据。
好了，数据获取和清洗这块就聊到这儿。记住一句话：数据质量决定策略上限。花80%的时间在数据上，剩下20%做策略，这是值得的。