4、数据获取与清洗：从公开API获取碳价数据、数据清洗与缺失值处理、时间序列对齐

好，咱们进入实战环节。前面聊了那么多理论，现在该动手了。

数据是量化策略的命根子。这话我反复跟团队讲。没有干净的数据，再牛的模型也是白搭。今天咱们就手把手把碳价数据搞到手，再把它洗得干干净净。

4.1 数据源的选择：去哪搞碳价数据？

公开的碳价数据，其实没那么难找。我个人习惯用这几个渠道：

• ICE（洲际交易所）：EUA期货的主力战场，数据最全，但需要付费订阅。
• Wind / 东方财富：国内碳市场数据，接口比较友好，但需要机构账号。
• 公开API（免费）：比如 CarbonPriceAPI 或某些研究机构提供的REST接口。

咱们这节课用免费API演示。你想想看，如果连数据都拿不到，策略还怎么跑？

4.2 从API拉数据：代码实战

咱们用 requests 库来拉取数据。这里我选了一个模拟的碳价API端点，实际使用时替换成你的真实API Key就行。

import requests
import pandas as pd
from datetime import datetime, timedelta

# 模拟API端点（实际使用时替换）
API_URL = "https://api.carbonprices.example.com/v1/prices"
API_KEY = "your_api_key_here"

def fetch_carbon_prices(symbol="EUA", start_date="2024-01-01", end_date="2024-06-30"):
    """
    从公开API获取碳价数据
    """
    params = {
        "symbol": symbol,
        "start": start_date,
        "end": end_date,
        "apikey": API_KEY,
        "format": "json"
    }
    
    try:
        response = requests.get(API_URL, params=params, timeout=10)
        response.raise_for_status()  # 检查HTTP错误
        data = response.json()
        
        # 解析成DataFrame
        df = pd.DataFrame(data["prices"])
        df["date"] = pd.to_datetime(df["date"])
        df.set_index("date", inplace=True)
        df.sort_index(inplace=True)
        
        print(f"成功获取 {len(df)} 条数据")
        return df
    
    except requests.exceptions.RequestException as e:
        print(f"API请求失败: {e}")
        return None

# 拉取数据
df_raw = fetch_carbon_prices()
print(df_raw.head())

嗯，这里要注意。API返回的数据格式千奇百怪。有的返回嵌套JSON，有的直接给CSV。我建议统一转成DataFrame，后面处理起来方便。

4.3 数据清洗：把脏数据洗干净

数据拉回来了，但别急着用。先看看长什么样。

# 快速检查数据质量
print("数据基本信息：")
print(df_raw.info())
print("\n缺失值统计：")
print(df_raw.isnull().sum())
print("\n基本统计量：")
print(df_raw.describe())

你会发现，真实数据总是有问题的。我曾经遇到过一个数据集，连续三天价格都是0——后来发现是API那边数据源挂了，返回了默认值。

4.3.1 处理缺失值

碳价数据常见的缺失情况有两种：

• 交易日缺失：比如节假日，市场没开。
• 数据点缺失：API返回了NaN，或者某天数据没更新。

我的处理原则是：

1. 先看缺失比例。超过5%就要警惕了。
2. 交易日缺失，用前向填充（ffill）或插值。
3. 数据点缺失，用前后均值填充。

def clean_price_data(df):
    """
    数据清洗函数
    """
    df_clean = df.copy()
    
    # 1. 删除全为NaN的行
    df_clean.dropna(how='all', inplace=True)
    
    # 2. 价格列缺失值处理
    if 'close' in df_clean.columns:
        # 前向填充（适用于非交易日）
        df_clean['close'].fillna(method='ffill', inplace=True)
        # 如果开头还有缺失，用后向填充
        df_clean['close'].fillna(method='bfill', inplace=True)
    
    # 3. 成交量缺失处理（如果有）
    if 'volume' in df_clean.columns:
        df_clean['volume'].fillna(0, inplace=True)
    
    # 4. 检查是否有异常值（比如价格小于0）
    df_clean = df_clean[df_clean['close'] > 0]
    
    return df_clean

df_clean = clean_price_data(df_raw)
print(f"清洗后数据量: {len(df_clean)}")

4.4 时间序列对齐：让数据对上节奏

做多资产策略时，时间对齐是个大坑。比如EUA和CER的交易时间不同，节假日也不同。

说白了，就是让不同数据源的时间戳对齐到同一个时间轴上。

def align_time_series(df_list, freq='D'):
    """
    对齐多个时间序列
    df_list: 包含多个DataFrame的列表，每个都有datetime索引
    freq: 目标频率，'D'为日频，'H'为小时频
    """
    # 找到所有时间索引的并集
    all_dates = pd.DatetimeIndex([])
    for df in df_list:
        all_dates = all_dates.union(df.index)
    
    # 生成目标频率的完整时间轴
    full_index = pd.date_range(
        start=all_dates.min(),
        end=all_dates.max(),
        freq=freq
    )
    
    # 重新索引并对齐
    aligned_dfs = []
    for i, df in enumerate(df_list):
        df_aligned = df.reindex(full_index)
        # 用前向填充处理缺失
        df_aligned.fillna(method='ffill', inplace=True)
        aligned_dfs.append(df_aligned)
    
    # 合并成一个DataFrame
    result = pd.concat(aligned_dfs, axis=1)
    result.columns = [f'asset_{i}' for i in range(len(df_list))]
    
    return result

# 示例：对齐EUA和CER两个品种
# eua_df = fetch_carbon_prices(symbol="EUA")
# cer_df = fetch_carbon_prices(symbol="CER")
# aligned_df = align_time_series([eua_df, cer_df])

你想想看，如果EUA在周一有数据，CER在周二才有，直接计算价差就会出问题。对齐之后，两个序列在同一个时间点上都有值，才能做后续分析。

4.5 知识体系总览

下面这张图，把咱们这节的核心逻辑串起来了。

4.6 完整实战：从API到清洗对齐

最后，我把整个流程串成一个函数。以后你拿到新数据，直接调用就行。

def carbon_data_pipeline(symbols=["EUA"], start="2024-01-01", end="2024-06-30"):
    """
    碳价数据获取与清洗完整流水线
    """
    all_dfs = []
    
    for sym in symbols:
        print(f"正在获取 {sym} 数据...")
        raw = fetch_carbon_prices(symbol=sym, start_date=start, end_date=end)
        if raw is not None:
            clean = clean_price_data(raw)
            all_dfs.append(clean)
    
    if len(all_dfs) > 1:
        # 多品种对齐
        result = align_time_series(all_dfs)
    else:
        result = all_dfs[0]
    
    # 添加常用技术指标
    result['returns'] = result.iloc[:, 0].pct_change()
    result['log_returns'] = np.log(result.iloc[:, 0] / result.iloc[:, 0].shift(1))
    
    return result

# 运行
df_final = carbon_data_pipeline(symbols=["EUA", "CER"])
print(df_final.tail())

好了，数据到手了，也洗干净了。下一节咱们就用这些数据，开始构建真正的碳交易策略。

---

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