数据获取与预处理：打好预测模型的地基

做碳交易价格预测，有个道理我吃了不少亏才明白——数据质量决定了模型的天花板。模型再花哨，喂进去的是垃圾，吐出来的只能是更精致的垃圾。这一章，咱们就聊聊怎么把数据这块地基夯实。

1. 碳排放数据源：你从哪里找“粮食”？

数据源这块，我把它分成三类。你想想看，碳价受什么影响？政策、供需、能源价格、宏观经济……所以数据源必须多元。

1.1 交易所官方数据

这是最核心的数据源。国内八大碳交易所（湖北、上海、北京、广东、深圳、天津、重庆、福建）都提供历史行情数据。但有个坑——各交易所的数据格式、字段命名、更新频率都不一样。

我记得第一次做全国碳市场分析时，光统一字段就花了三天。湖北的叫“DEA”，广东的叫“GDEA”，其实都是配额。

交易所	主要品种	数据频率	获取方式
湖北碳交所	HBEA	日频	官网CSV下载
上海环境能源交易所	SHEA	日频	API + 网页
广州碳交所	GDEA	日频/周频	API
北京绿色交易所	BEA	日频	网页爬虫

1.2 辅助数据源

光有碳价不够。我在项目中遇到过，碳价突然暴跌，光看历史数据根本解释不了。后来一查，原来是欧盟CBAM政策出了新草案。所以辅助数据必须跟上：

• 能源数据：煤炭、天然气、原油价格（来源：Wind、CEIC、国家统计局）
• 宏观经济数据：PMI、工业增加值、发电量（来源：国家统计局）
• 政策新闻：生态环境部公告、各交易所通知（来源：官网、新闻API）
• 气候数据：气温、降雨量（水电出力影响碳配额需求）

2. 交易所API对接：别被限流坑了

说到API对接，我踩过的坑能写本书。各交易所的API风格迥异，有的用RESTful，有的用WebSocket，有的干脆只提供Excel下载。

2.1 通用对接流程

不管哪个交易所，流程大致相同：

1. 注册开发者账号——有些交易所需要企业资质审核
2. 获取API Key和Secret——保管好，别硬编码在代码里
3. 阅读接口文档——注意限流规则、数据格式、认证方式
4. 编写对接代码——建议用Python的requests库
5. 异常处理——网络超时、数据缺失、格式变化都要考虑

2.2 代码示例：对接上海环境能源交易所API

嗯，这里要注意。上海环交所的API有个特点——需要先获取token，然后每次请求都要带上。我曾经因为token过期没处理，导致数据采集程序跑了一晚上全是401错误。

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

class SHEA_API:
    def __init__(self, api_key, api_secret):
        self.base_url = "https://api.cneeex.com/v1"
        self.api_key = api_key
        self.api_secret = api_secret
        self.token = None
        self.token_expiry = None
    
    def get_token(self):
        """获取访问令牌"""
        # 我习惯把token缓存起来，避免频繁请求
        if self.token and self.token_expiry > datetime.now():
            return self.token
        
        resp = requests.post(
            f"{self.base_url}/auth/token",
            json={
                "api_key": self.api_key,
                "api_secret": self.api_secret
            },
            timeout=10
        )
        resp.raise_for_status()
        data = resp.json()
        self.token = data['access_token']
        # token有效期通常24小时，我设置23小时刷新
        self.token_expiry = datetime.now() + timedelta(hours=23)
        return self.token
    
    def get_daily_data(self, symbol, start_date, end_date):
        """获取日频行情数据"""
        token = self.get_token()
        headers = {"Authorization": f"Bearer {token}"}
        
        params = {
            "symbol": symbol,
            "start_date": start_date,
            "end_date": end_date,
            "limit": 1000  # 每页最大1000条
        }
        
        all_data = []
        page = 1
        
        while True:
            params['page'] = page
            resp = requests.get(
                f"{self.base_url}/market/daily",
                headers=headers,
                params=params,
                timeout=30
            )
            
            if resp.status_code == 429:
                # 被限流了！我曾经在这里吃过亏
                print(f"触发限流，等待10秒...")
                time.sleep(10)
                continue
            
            resp.raise_for_status()
            data = resp.json()
            
            if not data['results']:
                break
                
            all_data.extend(data['results'])
            page += 1
            
            # 避免请求过快
            time.sleep(0.5)
        
        return pd.DataFrame(all_data)

# 使用示例
api = SHEA_API("your_api_key", "your_api_secret")
df = api.get_daily_data("SHEA", "2024-01-01", "2024-12-31")
print(f"获取到 {len(df)} 条数据")

3. 数据清洗与缺失值处理：脏活累活但必须干

说实话，数据清洗是我最不喜欢的环节，但又是最关键的。你拿到的原始数据，大概率有这些问题：

• 缺失值：节假日没交易、数据采集失败、交易所系统维护
• 异常值：价格突然跳变、成交量异常放大（可能是数据录入错误）
• 重复值：同一笔交易被记录了两次
• 格式不一致：日期格式有的是"2024-01-01"，有的是"2024/01/01"

3.1 缺失值处理策略

碳交易数据有个特点——非连续交易。不是每天都有交易，特别是流动性差的品种，可能一周才成交一笔。这时候不能简单用前向填充。

缺失类型	处理方法	适用场景
交易日无交易	保留NaN，不填充	价格预测模型
短期缺失（1-3天）	线性插值	技术指标计算
中期缺失（1-2周）	前向填充 + 市场状态标记	波动率分析
长期缺失（>1月）	删除该时间段	模型训练

import pandas as pd
import numpy as np

def clean_carbon_data(df):
    """碳交易数据清洗函数"""
    # 1. 处理重复值
    df = df.drop_duplicates(subset=['date', 'symbol'])
    
    # 2. 日期格式统一
    df['date'] = pd.to_datetime(df['date'])
    df = df.sort_values('date')
    
    # 3. 处理异常价格（3倍标准差之外）
    price_mean = df['close'].mean()
    price_std = df['close'].std()
    df = df[
        (df['close'] > price_mean - 3 * price_std) &
        (df['close'] < price_mean + 3 * price_std)
    ]
    
    # 4. 缺失值处理
    # 对于非交易日，保留NaN
    # 对于交易日内的缺失，用线性插值
    df['close'] = df['close'].interpolate(method='linear', limit=3)
    df['volume'] = df['volume'].fillna(0)  # 成交量缺失视为0
    
    # 5. 标记市场状态
    df['is_trading_day'] = ~df['close'].isna()
    
    return df

# 使用示例
df_clean = clean_carbon_data(df_raw)
print(f"清洗前: {len(df_raw)} 条, 清洗后: {len(df_clean)} 条")

4. 特征工程基础：从数据到信息的跃迁

特征工程说白了，就是把原始数据变成模型能理解的“语言”。碳价预测的特征工程，我一般分三个层次：

4.1 基础统计特征

• 价格相关：开盘价、收盘价、最高价、最低价、均价
• 成交量相关：成交量、成交额、换手率
• 波动率：日内波动率（最高-最低）/ 收盘价

4.2 技术指标特征

从股票技术分析借鉴过来的，但要注意碳市场有自己的特点：

• 移动平均线（MA5, MA20, MA60）：反映不同周期的趋势
• 相对强弱指标（RSI）：判断超买超卖
• 布林带：衡量价格波动区间
• 成交量加权平均价（VWAP）：机构交易者常用的参考价

4.3 领域特定特征

这部分是我觉得最有价值的。碳价受政策影响极大，所以需要构造一些“政策因子”：

• 履约周期标记：距离下次履约截止日期的天数
• 政策事件哑变量：是否有新政策发布
• 配额供需比：市场配额发放量与实际排放量的比值
• 能源价格联动：碳价与煤炭/天然气价格的比值

def create_features(df):
    """构建碳价预测特征"""
    df = df.copy()
    
    # 技术指标
    df['ma5'] = df['close'].rolling(5).mean()
    df['ma20'] = df['close'].rolling(20).mean()
    df['rsi'] = calculate_rsi(df['close'], 14)
    
    # 波动率
    df['volatility'] = (df['high'] - df['low']) / df['close']
    
    # 成交量变化
    df['volume_change'] = df['volume'].pct_change()
    
    # 领域特征
    df['days_to_deadline'] = (df['deadline_date'] - df['date']).dt.days
    df['policy_event'] = df['date'].isin(policy_dates).astype(int)
    
    # 滞后特征（碳价有自相关性）
    for lag in [1, 2, 3, 5, 10]:
        df[f'close_lag_{lag}'] = df['close'].shift(lag)
    
    return df

知识体系总览

说了这么多，我画了张图帮你理清思路。数据获取与预处理这个环节，说白了就是一条流水线：从多源数据采集，到清洗加工，再到特征提取，最后才能喂给模型。