3、数据层设计（上）：行情数据源接入（REST/WebSocket）、数据清洗与标准化、数据存储选型（InfluxDB/ClickHouse/PostgreSQL）、数据分片策略
行情数据，是量化系统的血液。没有它，你的策略再牛也是空中楼阁。这一章，我们聊聊数据层最核心的几个问题：数据从哪来？怎么洗干净？存哪里？怎么存才能扛住千万级写入？
我个人习惯把数据层拆成三个环节：接入层 → 清洗层 → 存储层。每个环节都有坑，咱们一个一个说。
3.1 行情数据源接入：REST vs WebSocket
行情源接入，说白了就是跟交易所或数据商打交道。目前主流就两种方式：REST 和 WebSocket。
3.1.1 REST 接口：适合历史数据与低频快照
REST 是请求-响应模式。你发一个 HTTP 请求，对方返回数据。简单、可靠，但延迟高。
适用场景：
- 拉取历史 K 线数据
- 获取交易品种列表
- 低频的快照数据（比如每分钟一次）
代码示例：
import requests
import time
def fetch_kline(symbol, interval, limit=100):
url = f"https://api.exchange.com/v1/klines"
params = {
"symbol": symbol,
"interval": interval,
"limit": limit
}
resp = requests.get(url, params=params, timeout=5)
if resp.status_code == 200:
return resp.json()
else:
# 我曾经遇到过交易所返回 429（限流），所以这里一定要加重试
time.sleep(1)
return fetch_kline(symbol, interval, limit)
3.1.2 WebSocket：实时行情的首选
WebSocket 是长连接，服务端主动推送数据。延迟低，适合 tick 级行情。
核心要点：
- 连接建立后，需要定时发送心跳包（ping/pong）
- 断线重连机制必须实现
- 数据解压：很多交易所会压缩数据（如 gzip）
代码示例：
import websocket
import json
def on_message(ws, message):
data = json.loads(message)
# 这里直接推送到消息队列，不要做复杂处理
print(f"收到行情: {data['symbol']} {data['price']}")
def on_error(ws, error):
print(f"连接异常: {error}")
# 自动重连逻辑
def on_close(ws, close_status_code, close_msg):
print("连接关闭，准备重连...")
# 重连
ws = websocket.WebSocketApp(
"wss://stream.exchange.com/ws",
on_message=on_message,
on_error=on_error,
on_close=on_close
)
ws.run_forever()
3.2 数据清洗与标准化
行情数据到了本地，第一件事不是存，而是洗。你想想看，不同交易所的数据格式千奇百怪，字段名都不一样。不标准化，后面策略根本没法用。
3.2.1 清洗规则
- 去重：同一时间戳的重复数据只保留一条
- 去噪：价格异常值（比如突然跳变 100%）要标记或剔除
- 补全：缺失的时间戳用前值填充或插值
- 对齐：不同交易所的时间戳要统一到同一时区（建议 UTC）
3.2.2 标准化字段
我建议所有行情数据统一成以下结构：
| 字段 | 类型 | 说明 |
|---|---|---|
| symbol | string | 交易对，如 BTCUSDT |
| exchange | string | 交易所名称 |
| timestamp | int64 | Unix 毫秒时间戳 |
| open | float64 | 开盘价 |
| high | float64 | 最高价 |
| low | float64 | 最低价 |
| close | float64 | 收盘价 |
| volume | float64 | 成交量 |
3.3 数据存储选型
存行情数据，选型很关键。我见过不少团队用 MySQL 存 tick 数据，结果一个月后查询就慢成狗。选对存储，事半功倍。
3.3.1 InfluxDB：时序数据库的首选
InfluxDB 是专门为时序数据设计的。写入快，查询快，压缩率高。
优点：
- 写入吞吐极高，单机可支撑百万点/秒
- 自带降采样和保留策略
- 查询语法简单，适合做聚合
缺点：
- 集群版收费，社区版单机有瓶颈
- 不支持复杂 JOIN
3.3.2 ClickHouse：适合大规模分析
ClickHouse 是列式存储，适合做海量数据的聚合查询。如果你需要做回测或统计，它很合适。
优点：
- 压缩比极高，存储成本低
- 查询速度极快，特别是聚合查询
- 支持 SQL，学习成本低
缺点：
- 不适合高频写入（每秒超过 10 万行会有压力）
- 单行更新/删除性能差
3.3.3 PostgreSQL：适合元数据与配置
PostgreSQL 不是用来存行情数据的，但适合存交易对信息、策略配置、用户数据等。
适用场景：
- 交易品种列表
- 策略参数配置
- 交易日志
3.4 数据分片策略
数据量大了，单机扛不住。分片是必须的。分片策略直接影响查询性能和扩展性。
3.4.1 按时间分片
最常用的方式。比如按天、按小时分片。
- 优点：查询时间范围明确时效率高，过期数据容易清理
- 缺点：热点问题——最近的数据写入压力集中
3.4.2 按交易对分片
每个交易对独立一个分片。
- 优点：查询单个交易对时性能好，扩展容易
- 缺点：跨交易对的查询需要合并结果
3.4.3 混合分片
我个人最推荐的方式。先按交易对分片，再按时间分片。
示例：
# 分片键设计
shard_key = f"{symbol}_{date}"
# 例如
BTCUSDT_20240101
ETHUSDT_20240101
BTCUSDT_20240102
- 数据均匀分布，避免热点
- 查询尽量落在单个或少部分分片上
- 分片粒度不要太小（比如按小时分片会导致分片过多）
3.5 整体架构图
下面这张图展示了行情数据从接入到存储的完整流程：
嗯，数据层设计的上半部分就到这里。核心就是三件事：接入要稳、清洗要净、存储要准。下一章我们继续聊数据层的下半部分——缓存、消息队列和数据治理。
公众号:蓝海资料掘金营，微信deep3321