第三章 数据收集与清洗：数据源选择、采集API、清洗标准化与存储方案

做量化交易，数据就是你的石油。

这话一点不夸张。我见过太多人，策略写得天花乱坠，结果一跑实盘就崩——查到最后，往往是数据源头出了问题。今天咱们就把这块硬骨头啃下来。

3.1 数据源选择：交易所 vs OTC市场

先说说数据源怎么选。我个人习惯把数据源分成两大类：交易所数据和OTC市场数据。

交易所数据，说白了就是中心化交易所（CEX）的行情。像Binance、OKX、Coinbase这些。优点是数据规范、延迟低、历史数据完整。我在项目中遇到过最头疼的事，就是某小交易所的K线数据居然有30%的缺失——后来果断换成了主流交易所的数据源。

OTC市场数据，也就是场外交易数据。这类数据更分散，但能反映真实的大宗交易情况。比如一些机构级的OTC平台，成交数据往往比交易所的盘口更有参考价值。

你想想看，如果只用一个数据源，万一它出问题了呢？我曾经就因为单一数据源故障，导致一个套利策略在5分钟内亏了2%——从那以后，我所有策略都至少接两个独立数据源做交叉验证。

3.2 数据采集API使用

数据采集这块，我踩过的坑比走过的路还多。先说说最常用的几种API。

REST API：适合历史数据批量下载。比如Binance的GET /api/v3/klines，一次能拉500根K线。但要注意频率限制——我记得有一次写了个循环，忘了加sleep，结果IP直接被封了24小时。

WebSocket API：适合实时行情订阅。延迟低，但需要自己维护连接。我习惯用Python的websockets库，配合asyncio做异步处理。

下面是我常用的一个数据采集模板：

import requests
import time
from datetime import datetime

def fetch_klines(symbol, interval, limit=500):
    """
    从Binance获取K线数据
    """
    url = "https://api.binance.com/api/v3/klines"
    params = {
        "symbol": symbol,
        "interval": interval,
        "limit": limit
    }
    
    try:
        resp = requests.get(url, params=params, timeout=10)
        resp.raise_for_status()
        data = resp.json()
        
        # 清洗成标准格式
        klines = []
        for item in data:
            klines.append({
                "timestamp": item[0],
                "open": float(item[1]),
                "high": float(item[2]),
                "low": float(item[3]),
                "close": float(item[4]),
                "volume": float(item[5])
            })
        return klines
    
    except Exception as e:
        print(f"采集失败: {e}")
        return None

# 使用示例
data = fetch_klines("BTCUSDT", "1h")
if data:
    print(f"成功获取 {len(data)} 根K线")

3.3 数据清洗与标准化

数据采集回来，千万别直接用。我敢说，90%的原始数据都有问题。

常见问题：

• 时间戳不统一：有的用毫秒，有的用秒，有的甚至用字符串
• 缺失值：比如某根K线的成交量是0
• 异常值：价格突然跳变，明显是数据错误
• 重复数据：同一个时间点出现多条记录

我的清洗流程是这样的：

1. 时间标准化：统一转为UTC时间戳（毫秒级）
2. 去重：按时间戳去重，保留最后一条
3. 缺失值处理：用前一根K线的收盘价填充，或者直接删除
4. 异常值检测：用3-sigma法则剔除明显错误的数据
5. 格式统一：所有数值转为float，时间戳转为int

举个例子，我曾经处理过一组数据，其中某根K线的最高价居然是最低价的10倍——明显是交易所的bug。如果不剔除，策略回测结果会完全失真。

import pandas as pd
import numpy as np

def clean_klines(df):
    """
    清洗K线数据
    """
    # 1. 时间戳标准化
    df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
    
    # 2. 去重
    df = df.drop_duplicates(subset=['timestamp'], keep='last')
    
    # 3. 缺失值处理
    df = df.fillna(method='ffill')
    
    # 4. 异常值检测（3-sigma）
    for col in ['open', 'high', 'low', 'close']:
        mean = df[col].mean()
        std = df[col].std()
        df = df[(df[col] > mean - 3*std) & (df[col] < mean + 3*std)]
    
    # 5. 排序
    df = df.sort_values('timestamp')
    
    return df

# 使用示例
df = pd.DataFrame(data)
df_clean = clean_klines(df)
print(f"清洗前: {len(df)} 行, 清洗后: {len(df_clean)} 行")

3.4 数据存储方案

数据存哪里？这是个好问题。我试过很多方案，最后总结出三条路。

我个人习惯：

• 开发阶段用CSV，方便调试
• 回测阶段用SQLite，查询灵活
• 生产环境用ClickHouse，性能拉满

嗯，这里要注意一点：千万别把所有数据都塞到一个表里。我建议按交易对和时间分区存储。比如btcusdt_2024_01、btcusdt_2024_02这样。查询时只扫需要的分区，速度能快10倍以上。

data/
├── raw/          # 原始数据，不做任何修改
│   ├── binance/
│   └── okx/
├── clean/        # 清洗后的标准数据
│   ├── btcusdt/
│   └── ethusdt/
└── features/     # 衍生特征数据
    └── indicators/
  

为什么要保留原始数据？因为清洗逻辑可能会变。我曾经改过一次清洗规则，结果发现之前清洗过的数据全得重来——幸好原始数据还在，不然就抓瞎了。

3.5 本章知识体系

下面这张图，是我做数据工程时的核心框架。你照着这个来，基本不会出大问题。

这张图的核心逻辑很简单：数据从源头来，经过采集、清洗，最后存起来。但别忘了那个虚线箭头——数据质量反馈。我每次跑完策略，都会回头检查数据质量。如果策略表现异常，90%的概率是数据出了问题。