3、数据获取与存储：从交易所API获取订单簿快照，使用Parquet格式存储
好，咱们进入第三章。这一章要解决一个很实际的问题：数据从哪来，又该存成什么样。
我个人习惯把数据获取和存储放在一起讲。为什么？因为这两件事是绑定的。你从交易所拿到的原始数据，如果不做任何处理直接存，后面回放的时候会非常痛苦。我踩过这个坑，所以想让你一开始就走对路。
3.1 订单簿快照：你到底需要什么？
先搞清楚我们要拿什么。
订单簿快照，说白了就是交易所某一时刻的「挂单全景图」。它包含两个核心部分：买单队列和卖单队列。每个队列里，价格从高到低（买单）或从低到高（卖单）排列，每个价格后面跟着对应的数量。
举个例子，一个典型的订单簿快照长这样：
{
"bids": [
[100.50, 1.2], // 价格100.50，数量1.2个BTC
[100.40, 3.5],
[100.30, 0.8]
],
"asks": [
[100.60, 2.1],
[100.70, 1.5],
[100.80, 4.0]
],
"timestamp": 1700000000000,
"symbol": "BTCUSDT"
}
你想想看，这个结构其实很简单。但真正做回放的时候，你会发现一个问题：快照是瞬时的，而市场是连续的。如果你只存快照，中间的价格变动就全丢了。
核心原则：快照用于重建市场状态，增量更新（tick级数据）用于还原过程。两者结合，才能做真正的回放。
3.2 从交易所API获取数据
现在主流交易所都提供REST API和WebSocket两种方式。我的建议是：用REST API拿快照，用WebSocket收增量。为什么？
- REST API：请求一次，拿到完整的订单簿。适合定时抓取快照。
- WebSocket：实时推送，但只推送变化的部分。适合做增量更新。
嗯，这里要注意：不同交易所的API格式差别很大。我以币安为例，写一个简单的快照获取函数：
import requests
import json
def fetch_orderbook_snapshot(symbol="btcusdt", limit=100):
"""
从币安获取订单簿快照
:param symbol: 交易对，如 btcusdt
:param limit: 返回的深度，最大5000
:return: 字典格式的快照数据
"""
url = f"https://api.binance.com/api/v3/depth"
params = {
"symbol": symbol.upper(),
"limit": limit
}
try:
response = requests.get(url, params=params, timeout=5)
response.raise_for_status()
data = response.json()
# 加上我们自己的时间戳，精确到毫秒
data["fetch_timestamp"] = int(time.time() * 1000)
return data
except Exception as e:
print(f"获取快照失败: {e}")
return None
小技巧：我建议在返回数据里加上fetch_timestamp字段。因为交易所返回的时间戳是服务器时间，但网络延迟会导致你实际拿到数据的时间更晚。两个时间戳一对比，你就能知道数据的新鲜度。
3.3 为什么选Parquet？
好，数据拿到了。存成什么格式？
你可能会想：CSV不行吗？JSON不行吗？
行，但不好。我在项目中遇到过这样的场景：一天抓了上亿条tick数据，用CSV存，一个文件几十GB，读取慢得要命。后来换成Parquet，文件大小直接缩到原来的1/5，读取速度快了10倍不止。
Parquet的优势：
| 特性 | 说明 |
|---|---|
| 列式存储 | 只读取你需要的列，而不是整行数据 |
| 高压缩比 | 支持Snappy、Zstd等压缩算法，节省磁盘空间 |
| Schema自描述 | 文件里自带字段类型信息，不用额外定义 |
| 与大数据生态兼容 | Spark、Pandas、DuckDB都能直接读 |
说白了，Parquet就是为「分析型工作负载」设计的。你后面做回放、做统计，经常要按时间范围过滤数据，或者只取某几个字段。Parquet在这些场景下表现极好。
3.4 把快照存成Parquet
接下来是实操。我们用Python的pandas和pyarrow库来实现。
先安装依赖：
pip install pandas pyarrow requests
然后写一个存储函数：
import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq
import os
from datetime import datetime
def save_snapshot_to_parquet(snapshot, symbol, output_dir="data/orderbook"):
"""
将订单簿快照保存为Parquet文件
:param snapshot: 从API获取的快照字典
:param symbol: 交易对
:param output_dir: 输出目录
"""
# 创建目录
os.makedirs(output_dir, exist_ok=True)
# 解析快照数据
bids = snapshot.get("bids", [])
asks = snapshot.get("asks", [])
timestamp = snapshot.get("fetch_timestamp", 0)
# 将买单和卖单分别转为DataFrame
df_bids = pd.DataFrame(bids, columns=["price", "quantity"])
df_bids["side"] = "bid"
df_bids["timestamp"] = timestamp
df_bids["symbol"] = symbol
df_asks = pd.DataFrame(asks, columns=["price", "quantity"])
df_asks["side"] = "ask"
df_asks["timestamp"] = timestamp
df_asks["symbol"] = symbol
# 合并成一个DataFrame
df = pd.concat([df_bids, df_asks], ignore_index=True)
# 生成文件名：按日期和交易对组织
date_str = datetime.fromtimestamp(timestamp / 1000).strftime("%Y%m%d")
filename = f"{symbol}_{date_str}_{timestamp}.parquet"
filepath = os.path.join(output_dir, filename)
# 保存为Parquet
df.to_parquet(filepath, index=False, compression="snappy")
print(f"快照已保存: {filepath}")
return filepath
避坑指南：我曾经犯过一个错误——把所有快照都塞进一个Parquet文件里。结果文件越来越大，读取时内存直接爆了。正确的做法是按时间分片，比如每小时或每天一个文件。这样既方便管理，也方便按时间范围回放。
3.5 数据存储的组织结构
文件多了以后，怎么组织目录结构？我推荐这种分层方式：
data/
orderbook/
BTCUSDT/
2024/
01/
01/
BTCUSDT_20240101_1700000000000.parquet
BTCUSDT_20240101_1700005000000.parquet
...
02/
...
ETHUSDT/
...
这种结构的好处是：
- 按交易对分：不同品种的数据互不干扰
- 按年月日分：回溯某一天的数据，直接定位到对应目录
- 文件名带时间戳：一眼就能看出数据的时间范围
3.6 完整的数据获取流程
最后，我把整个流程串起来。下面这张图展示了从API到Parquet的完整链路：
3.7 读取与验证
存完之后，怎么确认数据是对的？写一个简单的读取函数：
def read_parquet_snapshot(filepath):
"""
读取Parquet文件并验证数据完整性
"""
df = pd.read_parquet(filepath)
print(f"数据条数: {len(df)}")
print(f"字段列表: {df.columns.tolist()}")
print(f"时间范围: {df['timestamp'].min()} ~ {df['timestamp'].max()}")
print(f"买卖分布: \n{df['side'].value_counts()}")
# 检查是否有空值
if df.isnull().sum().sum() > 0:
print("⚠️ 数据中存在空值，请检查")
else:
print("✅ 数据完整性检查通过")
return df
个人经验：我每次存完数据，都会立刻读回来验证一遍。别嫌麻烦，我曾经因为网络波动导致一批数据全是空值，如果没及时检查，后面回放时整个策略都会跑偏。
好了，这一章的内容就这些。从API获取快照，用Parquet存储，每一步都有坑，但走通了之后，你会发现后面的回放工作会顺畅很多。