4. 数据获取与清洗：从交易所获取历史行情数据、处理缺失值与异常值、数据重采样与对齐

做量化做市策略，第一步不是写策略逻辑，而是搞定数据。我见过太多人策略写得天花乱坠，结果数据一塌糊涂，回测结果全是幻觉。说白了，数据是地基，地基不稳，楼盖得再高也得塌。

这一章，我们就来聊聊怎么从交易所拿到靠谱的历史行情数据，怎么把脏数据洗干净，以及怎么让不同频率的数据对齐。嗯，这些都是我踩过坑之后才真正重视起来的东西。

4.1 从交易所获取历史行情数据

获取数据的方式有很多种，我个人最推荐的是直接通过交易所的官方API。为什么？因为数据源最可靠，延迟最低，而且字段最全。你想想看，从第三方数据商拿数据，中间多了一层转发，谁知道有没有丢包或者篡改？

以币安交易所为例，它的REST API可以获取历史K线数据。我习惯用Python的requests库来拉取，简单直接。

import requests
import pandas as pd
from datetime import datetime

def fetch_binance_klines(symbol, interval, start_time, end_time):
    """
    从币安获取历史K线数据
    :param symbol: 交易对，如 'BTCUSDT'
    :param interval: K线周期，如 '1m', '5m', '1h'
    :param start_time: 开始时间，毫秒级时间戳
    :param end_time: 结束时间，毫秒级时间戳
    """
    url = 'https://api.binance.com/api/v3/klines'
    params = {
        'symbol': symbol,
        'interval': interval,
        'startTime': start_time,
        'endTime': end_time,
        'limit': 1000  # 单次最多1000条
    }
    
    response = requests.get(url, params=params)
    data = response.json()
    
    # 解析数据
    klines = []
    for k in data:
        klines.append({
            'timestamp': datetime.fromtimestamp(k[0] / 1000),
            'open': float(k[1]),
            'high': float(k[2]),
            'low': float(k[3]),
            'close': float(k[4]),
            'volume': float(k[5])
        })
    
    return pd.DataFrame(klines)

# 获取2024年1月的1分钟K线数据
df = fetch_binance_klines('BTCUSDT', '1m', 
                          int(datetime(2024,1,1).timestamp()*1000),
                          int(datetime(2024,1,31).timestamp()*1000))
print(df.head())

这里有个坑要注意——不同交易所的API格式差异很大。币安用的是毫秒级时间戳，而有些交易所用的是秒级。我曾经在对接某二线交易所时，因为时间戳单位搞错了，拉回来的数据全是乱的，排查了半天才发现问题。

4.2 处理缺失值与异常值

数据拿到手，别急着用。先检查一下有没有缺失值和异常值。这是数据清洗的核心工作，也是最容易出问题的地方。

4.2.1 缺失值处理

缺失值在行情数据里很常见，比如交易所宕机、网络波动、API限流，都可能导致某根K线没拉回来。我处理缺失值的策略是这样的：

• 检查缺失： 用isnull()或isna()检查DataFrame中是否有空值
• 定位缺失： 查看缺失值的时间点，判断是单点缺失还是连续缺失
• 填充策略： 单点缺失用前向填充（ffill）或插值法；连续缺失超过3根K线，建议直接删除

# 检查缺失值
print(df.isnull().sum())

# 前向填充（用上一根K线的数据填充）
df.fillna(method='ffill', inplace=True)

# 或者用线性插值
df.interpolate(method='linear', inplace=True)

# 连续缺失超过3根，删除
df = df.dropna(thresh=len(df.columns) - 3)

4.2.2 异常值处理

异常值比缺失值更隐蔽。比如某根K线的最高价突然比前后高出10倍，这明显是数据错误。我一般用两种方法检测异常值：

• 3σ原则： 计算价格序列的均值和标准差，超出均值±3倍标准差的值视为异常
• IQR方法： 计算四分位距，超出Q1-1.5*IQR或Q3+1.5*IQR的值视为异常

def detect_outliers_iqr(series):
    Q1 = series.quantile(0.25)
    Q3 = series.quantile(0.75)
    IQR = Q3 - Q1
    lower_bound = Q1 - 1.5 * IQR
    upper_bound = Q3 + 1.5 * IQR
    return (series < lower_bound) | (series > upper_bound)

# 检测收盘价的异常值
outliers = detect_outliers_iqr(df['close'])
print(f"发现 {outliers.sum()} 个异常值")

# 将异常值替换为前后均值
df.loc[outliers, 'close'] = (df['close'].shift(1) + df['close'].shift(-1)) / 2

我个人更倾向于IQR方法，因为它不受极端值的影响，更稳健。3σ方法对正态分布假设依赖较强，而金融数据往往有厚尾特征，用3σ可能会漏掉一些异常。

4.3 数据重采样与对齐

做市策略通常需要多频率数据配合。比如，你可能用1分钟K线做高频信号，用1小时K线判断趋势方向。这时候就需要把不同频率的数据对齐到同一个时间轴上。

4.3.1 数据重采样

重采样就是把高频数据降采样到低频，或者把低频数据升采样到高频。Pandas的resample方法就是干这个的。

# 将1分钟数据重采样为5分钟
df_5min = df.resample('5T', on='timestamp').agg({
    'open': 'first',
    'high': 'max',
    'low': 'min',
    'close': 'last',
    'volume': 'sum'
}).dropna()

# 将1分钟数据重采样为1小时
df_1hour = df.resample('1H', on='timestamp').agg({
    'open': 'first',
    'high': 'max',
    'low': 'min',
    'close': 'last',
    'volume': 'sum'
}).dropna()

4.3.2 数据对齐

对齐是指让不同频率的数据在时间戳上保持一致。比如，你的1分钟数据和5分钟数据，时间戳必须能对应上。我常用的方法是：

• 左对齐： 以低频数据的时间戳为基准，将高频数据对齐到最近的低频时间点
• 右对齐： 以高频数据的时间戳为基准，将低频数据向前填充

# 左对齐：将1分钟数据对齐到5分钟时间点
df_1min_aligned = df.set_index('timestamp').resample('5T').ffill()

# 右对齐：将5分钟数据向前填充到1分钟时间点
df_5min_aligned = df_5min.reindex(df_1min.index, method='ffill')

这里有个细节要注意——对齐时一定要考虑未来信息。比如，你用5分钟数据去填充1分钟数据，如果用了未来的5分钟数据，那就引入了未来函数，回测结果会虚高。我建议统一用前向填充（ffill），只用过去的数据。

4.4 数据清洗的完整流程

说了这么多，我们来总结一下数据清洗的完整流程。我画了一张图，方便你理解整个链路。

这张图展示了我个人常用的数据清洗流程。从数据获取开始，经过缺失值处理、异常值检测与处理、数据重采样、数据对齐，最后到数据验证和保存。如果验证不通过，就返回前面的步骤重新处理。

4.5 数据存储的最佳实践

清洗好的数据怎么存？我推荐用HDF5或Parquet格式，而不是CSV。原因有三：

• 压缩率高： 同样的数据，Parquet文件大小只有CSV的1/5左右
• 读取速度快： Parquet支持列式存储，读取指定列时速度极快
• 支持元数据： 可以存储数据类型、时间戳等信息，避免类型转换问题

# 保存为Parquet格式
df.to_parquet('btcusdt_1m_202401.parquet', index=False)

# 读取Parquet文件
df = pd.read_parquet('btcusdt_1m_202401.parquet')

嗯，数据获取与清洗这部分就讲到这里。记住一句话：花80%的时间在数据上，花20%的时间在策略上。数据干净了，策略才能跑得稳。

---

公众号:蓝海数据掘金营，微信deep3321