第4章数据模块

数据模块：历史行情数据获取、数据清洗与对齐、OHLCV数据结构设计、数据存储与读取

做市商策略回测，说白了就是拿历史数据模拟交易。数据要是出了问题，后面全白搭。我见过太多人花大把时间调策略参数，结果最后发现是数据没对齐——那种感觉，就像你跑完马拉松才发现鞋里有个石子。

这一章，咱们就把数据模块彻底讲透。从怎么拿到数据，到怎么洗干净，再到怎么存、怎么读，一条龙搞定。

历史行情数据获取：从哪里来，怎么来

数据源的选择，直接决定了你回测的可靠性。我个人习惯把数据源分成三类：

交易所官方API：比如Binance、OKX、Coinbase。数据最权威，但有限流，历史数据拉取慢。
第三方数据平台：比如Kaiko、CoinGecko、CryptoDataDownload。数据全，但收费。
本地CSV/Parquet文件：自己攒的数据仓库。最灵活，但需要维护。

我建议初学者先从交易所API入手。为什么？因为免费，而且能让你理解数据最原始的形态。下面是我常用的一个数据获取函数：

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

def fetch_ohlcv(symbol='BTC/USDT', timeframe='1m', limit=1000):
    """
    从交易所获取OHLCV数据
    """
    exchange = ccxt.binance({
        'rateLimit': 1200,
        'enableRateLimit': True,
    })
    
    since = exchange.parse8601('2024-01-01T00:00:00Z')
    all_candles = []
    
    while True:
        candles = exchange.fetch_ohlcv(
            symbol, 
            timeframe=timeframe,
            since=since,
            limit=limit
        )
        
        if len(candles) == 0:
            break
            
        all_candles += candles
        since = candles[-1][0] + 1  # 避免重复
        
        # 控制请求频率
        time.sleep(exchange.rateLimit / 1000)
    
    df = pd.DataFrame(
        all_candles, 
        columns=['timestamp', 'open', 'high', 'low', 'close', 'volume']
    )
    df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
    df.set_index('timestamp', inplace=True)
    
    return df

这里有个坑：交易所API返回的时间戳是毫秒级的Unix时间戳。我刚开始做的时候直接当整数用了，结果时间轴全乱套。嗯，记得转成datetime。

数据清洗与对齐：脏数据是回测的隐形杀手

拿到原始数据后，千万别直接扔进回测引擎。我曾在项目中遇到过这样的情况：某交易所某天的数据突然跳了一个大缺口，导致策略在那天开仓量暴增，回测结果直接翻倍。当时我还以为自己发现了圣杯，后来才发现是数据问题。

数据清洗，我一般做这几件事：

去重：同一个时间戳出现多条记录，保留最后一条。
补缺失：某些时间点没有数据，用前值填充或插值。
去异常值：价格突然跳变超过阈值，标记出来人工判断。
时间对齐：不同数据源的时间戳可能差几毫秒，统一对齐到整点。

核心原则：宁可少数据，不要错数据。回测中一个错误的价格点，可能让你误以为策略有效。

下面是我常用的清洗函数：

def clean_ohlcv(df):
    """
    清洗OHLCV数据
    """
    # 1. 去重：保留最后一条
    df = df[~df.index.duplicated(keep='last')]
    
    # 2. 排序
    df.sort_index(inplace=True)
    
    # 3. 补缺失：前向填充
    full_index = pd.date_range(
        start=df.index.min(), 
        end=df.index.max(), 
        freq='1min'
    )
    df = df.reindex(full_index, method='ffill')
    
    # 4. 去异常：价格变化超过20%标记
    price_change = df['close'].pct_change().abs()
    outliers = price_change > 0.2
    if outliers.any():
        print(f"发现 {outliers.sum()} 个异常点，已标记")
        df.loc[outliers, 'is_outlier'] = True
    
    return df

注意：前向填充（ffill）在长时间停盘时会有问题。比如周末没有交易，你用前一个周五的数据填充整个周末，回测时策略可能会在周末「交易」——这显然不对。我建议对停盘时段做特殊标记，或者直接删除。

OHLCV数据结构设计：别小看这六列数据

OHLCV，就是Open、High、Low、Close、Volume。看起来简单，但设计得好不好，直接影响后续所有模块的代码质量。

我个人习惯用Pandas DataFrame作为核心数据结构。为什么？因为Pandas的时间序列操作太方便了，滚动计算、重采样、合并，一行代码搞定。

下面是我推荐的数据结构：

字段	类型	说明
timestamp	datetime64[ns]	时间戳，作为索引
open	float64	开盘价
high	float64	最高价
low	float64	最低价
close	float64	收盘价
volume	float64	成交量
symbol	str	交易对（可选）
is_outlier	bool	异常标记（可选）

你想想看，为什么要把timestamp设成索引？因为回测时你经常需要按时间切片，比如「只回测2024年1月的数据」。用索引切片，效率高、代码也干净。

小技巧：如果你同时回测多个交易对，建议用MultiIndex，第一层是symbol，第二层是timestamp。这样合并数据、按品种筛选都非常方便。

数据存储与读取：别每次都重新拉数据

从交易所拉数据很慢，而且有限流。我建议把清洗好的数据存到本地，下次直接读文件。

存储格式我试过好几种：

CSV：通用，但慢，文件大。
Parquet：快，压缩率高，推荐。
HDF5：适合大规模数据，但依赖多。
SQLite：适合需要查询的场景。

我个人最常用Parquet。为什么？因为读写速度比CSV快10倍以上，而且文件大小只有CSV的1/3。下面是我的存储和读取代码：

def save_ohlcv(df, filepath):
    """
    存储OHLCV数据到Parquet文件
    """
    df.to_parquet(filepath, compression='snappy')
    print(f"数据已保存到 {filepath}")

def load_ohlcv(filepath):
    """
    从Parquet文件读取OHLCV数据
    """
    df = pd.read_parquet(filepath)
    return df

这里有个细节：存储时保留索引。Parquet默认会保存索引，但如果你用CSV，记得设置index=True，否则时间戳就丢了。

我的工作流：每天定时拉取最新数据 → 清洗 → 追加到本地Parquet文件。回测时直接读文件，几秒钟就能加载一年的分钟级数据。

知识体系总览

下面这张图，把数据模块的核心逻辑串起来了。你一看就明白：

这张图展示了数据从获取到最终被回测引擎使用的完整链路。你注意看，数据清洗和对齐是承上启下的关键环节——数据质量不过关，后面全是白费功夫。

好了，数据模块的核心内容就这些。记住一句话：数据是回测的基石，花再多时间打磨都不为过。我曾经因为数据没对齐，白白浪费了两周调参数，后来发现是数据源的时间戳差了1秒。嗯，从那以后，我每次回测前都会先跑一遍数据校验。

📚 课程目录

数据模块：历史行情数据获取、数据清洗与对齐、OHLCV数据结构设计、数据存储与读取

历史行情数据获取：从哪里来，怎么来

数据清洗与对齐：脏数据是回测的隐形杀手

OHLCV数据结构设计：别小看这六列数据

数据存储与读取：别每次都重新拉数据

知识体系总览