数据采集与存储：API接入、WebSocket实时流、CSV/Parquet文件存储、数据库选型

做量化交易，数据就是你的弹药库。弹药质量不行，再好的策略也白搭。这一章我重点聊聊数据怎么来、怎么存、怎么选型。说白了，就是要把行情数据从交易所搬到你的系统里，还要保证它不丢、不乱、能快速查。

API接入：REST vs WebSocket，怎么选？

我个人习惯把API接入分成两类：拉模式（REST）和推模式（WebSocket）。

REST API 适合做历史数据补全和定时快照。比如每天收盘后拉一次日线数据，或者每隔5秒拉一次最新价。优点是简单、稳定，不容易断。缺点也很明显——有延迟，而且频繁请求容易被限流。

WebSocket 则是实时行情的标配。交易所一有成交，服务器就主动推给你。延迟通常在毫秒级别。我在项目中遇到过一个问题：WebSocket连接偶尔会断开，而且断开后不会自动重连。嗯，这里要注意，必须自己实现心跳检测和重连机制。

// 伪代码：WebSocket 重连逻辑
function connectWebSocket() {
    ws = new WebSocket(url);
    ws.onOpen = () => { 
        heartbeatInterval = setInterval(sendPing, 30000); 
    };
    ws.onClose = () => {
        clearInterval(heartbeatInterval);
        setTimeout(connectWebSocket, 1000); // 1秒后重连
    };
    ws.onMessage = (data) => {
        // 校验数据完整性
        if (isValid(data)) {
            processData(data);
        }
    };
}

WebSocket实时流：数据校验是第一道防线

实时流的数据量很大，每秒可能上千条。如果每条都做完整校验，CPU扛不住。我的做法是分层校验：

• 第一层：格式校验——JSON能不能解析？字段齐不齐？
• 第二层：逻辑校验——时间戳是不是递增的？价格是不是正数？
• 第三层：一致性校验——和REST API拉到的快照对比，偏差超过阈值就报警

我曾经踩过一个坑：某交易所的WebSocket推送的成交价偶尔会多一个零。比如实际价格是5000，它推成50000。如果直接入库，策略就全乱了。所以我现在都会加一个「价格突变检测」——如果当前价格和前一笔相差超过5%，就标记为可疑数据，等REST确认后再处理。

CSV/Parquet文件存储：离线分析的基石

实时数据存数据库，历史数据存文件。这是我的铁律。

CSV 的好处是通用，Excel都能打开。但缺点也很明显：没有压缩，没有索引，大文件读写慢。我一般只用CSV做数据交换，比如给同事发一份日线数据。

Parquet 才是正经的列式存储格式。压缩率高（能压到CSV的1/5），读取快（只读需要的列），还支持谓词下推。我建议所有历史行情数据都用Parquet存。

# Python 示例：写入 Parquet
import pandas as pd
import pyarrow.parquet as pq

df = pd.DataFrame({
    'timestamp': [1625097600, 1625097601],
    'symbol': ['BTCUSDT', 'BTCUSDT'],
    'price': [35000.5, 35001.2],
    'volume': [0.1, 0.2]
})

# 按日期分区存储
df.to_parquet('data/2025/01/15/btc_trades.parquet', 
              compression='snappy')

数据库选型：InfluxDB vs ClickHouse

这是大家问得最多的问题。我直接说结论：

特性	InfluxDB	ClickHouse
写入速度	极快（专门为时序优化）	快（批量写入更优）
查询灵活性	一般（SQL-like，但有限制）	强（完整SQL支持）
压缩率	中等	高（列式存储）
运维复杂度	低（单机就能跑）	中等（集群配置稍复杂）
适合场景	实时监控、Tick级数据	历史分析、多表关联查询

我的选择逻辑：

• 如果只存行情数据，不做复杂分析 → InfluxDB。部署简单，写入快，查询也够用。
• 如果既要存行情，又要做因子计算、回测分析 → ClickHouse。SQL灵活，能直接跑聚合查询。

我曾经在一个项目里两个都用过：InfluxDB 存实时Tick，ClickHouse 存分钟级聚合数据。实时看盘用 InfluxDB，回测分析用 ClickHouse。各取所长，互不干扰。

数据校验与错误恢复：实战中的血泪教训

数据采集做得再好，也难免出问题。网络抖动、交易所宕机、磁盘写满……什么情况都可能发生。我的恢复策略分三步：

1. 本地缓存——WebSocket 数据先写本地文件（按分钟滚动），再异步写入数据库。即使数据库挂了，数据也不会丢。
2. 数据对账——每天凌晨跑一次对账任务，用 REST API 拉取交易所的历史数据，和本地数据库做对比。发现缺失就补录。
3. 异常标记——校验不通过的数据不要直接丢弃，而是标记为「可疑」，存入单独的异常表。人工审核后再决定是修正还是删除。

-- ClickHouse 示例：创建异常数据表
CREATE TABLE trade_anomalies (
    timestamp DateTime,
    symbol String,
    price Float64,
    volume Float64,
    anomaly_reason String,
    raw_data String
) ENGINE = MergeTree()
ORDER BY timestamp;

嗯，这套机制帮我避免过好几次灾难。有一次交易所的WebSocket推送了半小时的乱序数据，因为我有本地缓存和对账机制，最终只丢了3秒的数据，而且很快补回来了。