第3章数据获取与清洗

3. 数据获取与清洗：交易所API接口、数据源选择、缺失值处理、异常值检测

做碳市场高频分析，第一步不是写策略，而是搞定数据。我见过太多人一上来就调API，结果拿到的数据全是坑。今天咱们就把这块掰开揉碎了讲。

3.1 数据源选择：别在源头就输了

碳市场的数据源，说白了就三类：交易所直连、第三方数据商、公开爬虫。我个人习惯是优先走交易所API，原因很简单——延迟最低，字段最全。

目前国内几个主要碳交易所，接口风格差异很大。我整理了一张表，你感受一下：

交易所	API协议	数据频率	历史数据长度	备注
上海环境能源交易所	WebSocket + REST	Tick级（约100ms）	2013年至今	需要机构认证
湖北碳排放权交易中心	RESTful	分钟级	2014年至今	公开接口较稳定
广州碳排放权交易所	自定义TCP协议	秒级	2016年至今	文档比较老旧
深圳排放权交易所	HTTP轮询	5秒快照	2015年至今	延迟较高

嗯，这里要注意。第三方数据商虽然省事，但有个致命问题——他们会对原始数据做「平滑处理」。我在项目中遇到过，某家知名数据商提供的碳价序列，居然把日内波动给抹平了。你想想看，做高频策略的人拿到这种数据，那不是白费功夫吗？

避坑指南： 我曾经因为贪图方便，直接用了某数据商的CSV文件做回测。结果策略在模拟盘跑得飞起，一上实盘就崩。后来排查了三天，发现是数据商把撮合数据里的「撤单」记录给过滤掉了。所以我的建议是——能用原始接口就别用二手数据。

3.2 交易所API接口：实战接入

咱们拿上海环境能源交易所的WebSocket接口举个例子。这是我最常用的接入方式，延迟低，数据实时性好。

import websocket
import json
import pandas as pd

# 我个人习惯把连接参数单独拎出来
WS_URL = "wss://api.carbon.sh.cn/v1/market/tick"
API_KEY = "your_api_key_here"

def on_message(ws, message):
    data = json.loads(message)
    # 解析tick数据
    tick = {
        'timestamp': pd.to_datetime(data['time'], unit='ms'),
        'price': data['last_price'],
        'volume': data['volume'],
        'bid': data['bid1'],
        'ask': data['ask1'],
        'bid_vol': data['bid_vol1'],
        'ask_vol': data['ask_vol1']
    }
    # 这里可以追加到DataFrame或者写入数据库
    print(f"收到tick: {tick['timestamp']} 价格:{tick['price']}")

def on_error(ws, error):
    print(f"连接异常: {error}")
    # 自动重连逻辑
    ws.close()
    ws.connect(WS_URL)

ws = websocket.WebSocketApp(WS_URL,
                            on_message=on_message,
                            on_error=on_error)
ws.run_forever()

这段代码看着简单，但有几个坑你得注意。第一，WebSocket断连是家常便饭，一定要加自动重连。第二，数据时间戳要确认是本地时间还是UTC，我吃过这个亏——有一次没做时区转换，策略在开盘前半小时就开始交易了。

小技巧： 建议在本地维护一个「数据健康检查」脚本，每5分钟检查一次API连接状态和数据延迟。如果延迟超过500ms，自动切换备用数据源。

3.3 缺失值处理：别让空值毁了你的策略

碳市场的数据缺失，原因五花八门。交易所系统维护、网络抖动、节假日休市……你想想看，如果不对缺失值做处理，回测结果能准吗？

我一般把缺失值分成三类：

随机缺失： 网络丢包、API限流导致。占比通常小于1%。
结构性缺失： 非交易时段、节假日。这类缺失是正常的，不要填充。
系统性缺失： 交易所系统升级、数据迁移。占比可能达到5%-10%，需要特别处理。

处理方式上，我的经验是：

import pandas as pd
import numpy as np

# 假设df是5分钟K线数据
df = pd.read_csv('carbon_5min.csv', index_col='time', parse_dates=True)

# 1. 先标记结构性缺失（非交易时段）
trading_hours = df.between_time('09:30', '15:00')
df['is_trading'] = df.index.isin(trading_hours.index)

# 2. 对交易时段内的缺失值做前向填充
# 为什么用前向填充？因为碳市场流动性低，价格变化慢
df['price_ffill'] = df['price'].fillna(method='ffill', limit=3)

# 3. 如果连续缺失超过3个，用插值法
df['price_interp'] = df['price'].interpolate(method='linear', limit=5)

# 4. 实在补不了的，直接丢弃
df_clean = df.dropna(subset=['price'])

这里有个细节。碳市场不像股票市场那么活跃，有时候连续十几分钟都没有成交。这种情况下，前向填充比插值更合理。为什么？因为碳价在无交易时段基本保持不变，插值反而会引入虚假波动。

  核心原则： 缺失值处理没有银弹。我的做法是——先理解数据缺失的原因，再选择处理方法。不要一上来就fillna(0)，那是最偷懒也最危险的做法。

3.4 异常值检测：揪出那些离谱的数据

异常值在碳市场里特别常见。我记得有一次，某交易所的API突然返回了一个价格——2000元/吨，而当时市场价才80元。这种数据如果不处理，你的策略会瞬间爆仓。

我常用的检测方法有三种：

基于统计的Z-score法： 适合正态分布的数据，碳价收益率基本符合。
基于距离的DBSCAN聚类： 适合发现离群点，但计算量大。
基于规则的硬阈值： 比如价格超过前收盘价的±20%，直接标记。

from scipy import stats
import numpy as np

def detect_outliers_zscore(df, column='price', threshold=3):
    """
    Z-score异常检测
    我一般用3倍标准差作为阈值，碳市场可以放宽到4倍
    """
    z_scores = np.abs(stats.zscore(df[column].dropna()))
    outliers = z_scores > threshold
    return outliers

def detect_outliers_iqr(df, column='price'):
    """
    四分位距法，对极端值更鲁棒
    """
    Q1 = df[column].quantile(0.25)
    Q3 = df[column].quantile(0.75)
    IQR = Q3 - Q1
    lower_bound = Q1 - 1.5 * IQR
    upper_bound = Q3 + 1.5 * IQR
    return (df[column] < lower_bound) | (df[column] > upper_bound)

# 实战中我会组合使用
df['is_outlier_z'] = detect_outliers_zscore(df)
df['is_outlier_iqr'] = detect_outliers_iqr(df)

# 标记为异常的数据，我会先人工复核
# 因为有些「异常」其实是市场突发事件
df['flag'] = df['is_outlier_z'] & df['is_outlier_iqr']
print(f"检测到 {df['flag'].sum()} 个异常值")

嗯，这里要特别说一句。不是所有异常值都要删除。比如政策发布瞬间的价格跳变，那其实是有效信息。我一般会建一个「异常事件库」，把已知的政策发布、拍卖公告等时间点记录下来，这些数据要保留，但单独标记。

警告： 千万不要在回测中「事后」删除异常值。你想想看，如果策略在实盘中遇到同样的异常情况，它该怎么应对？正确的做法是——在策略逻辑里加入异常处理机制，而不是在数据层面抹掉问题。

3.5 数据清洗流水线：把流程固化下来

说了这么多，最后分享一个我实际在用的数据清洗流水线。说白了，就是把上面这些步骤串起来，做成一个自动化脚本。

class CarbonDataCleaner:
    def __init__(self, source='api'):
        self.source = source
        self.raw_data = None
        self.clean_data = None

    def fetch_data(self, start_date, end_date):
        # 从API或本地文件读取
        pass

    def remove_duplicates(self):
        # 去重，按时间戳排序后保留第一条
        self.raw_data = self.raw_data.drop_duplicates(subset='timestamp')

    def handle_missing(self, method='ffill'):
        # 缺失值处理
        pass

    def detect_outliers(self, method='zscore'):
        # 异常值检测
        pass

    def run_pipeline(self):
        print("开始数据清洗...")
        self.remove_duplicates()
        self.handle_missing()
        self.detect_outliers()
        print(f"清洗完成，原始数据{len(self.raw_data)}条，清洗后{len(self.clean_data)}条")
        return self.clean_data

这个流水线我跑了快两年，中间迭代了十几个版本。最开始只有去重和缺失值处理，后来慢慢加了异常检测、事件标记、数据质量评分。你刚开始做的话，建议先跑通基础版本，再根据实际数据特征慢慢优化。

我的习惯： 每次清洗完数据，我都会生成一份「数据质量报告」，包含缺失率、异常率、延迟分布等指标。这样下次再跑的时候，可以快速对比数据质量的变化。

这张图把整个流程串起来了。从数据源选择开始，到API接入，再到清洗流水线，最后产出可用的数据。你每次做数据清洗的时候，都可以对照这个流程检查一下，看看哪一步漏了。

数据获取与清洗，说白了就是「垃圾进，垃圾出」的反面。你花80%的时间把数据搞干净了，后面20%的策略开发才能事半功倍。我见过太多人在这上面偷懒，结果回测漂亮、实盘惨淡。嗯，希望你能少走这些弯路。

📚 课程目录

3. 数据获取与清洗：交易所API接口、数据源选择、缺失值处理、异常值检测

3.1 数据源选择：别在源头就输了

3.2 交易所API接口：实战接入

3.3 缺失值处理：别让空值毁了你的策略

3.4 异常值检测：揪出那些离谱的数据

3.5 数据清洗流水线：把流程固化下来