3、市场数据接入：实时行情API对接、WebSocket协议、数据清洗与标准化

好，咱们进入第三个章节。说实话，市场数据接入这块，是我在搭建量化系统时踩坑最多的地方。你想想看，策略再牛，模型再准，如果数据源头就出了问题，那后面的分析全是白搭。我见过太多团队，策略回测跑得飞起，一上实盘就崩，十有八九是数据接入环节没处理好。

今天咱们就聊聊，怎么把行情数据稳稳当当地接进来，洗干净，变成系统能用的标准格式。

3.1 实时行情API对接：选对路子，少走弯路

行情API，说白了就是交易所或数据商给你开的一扇门。但门有很多种，你得选对。

常见的行情API类型：

• REST API：一问一答的模式。你发请求，它返回数据。适合拿历史数据，或者低频的快照数据。但实时性差，轮询太频繁还会被限流。
• WebSocket：长连接，服务端主动推数据给你。实时性极好，是实盘交易的首选。
• FIX协议：金融领域的老牌协议，稳定可靠，但配置复杂，一般机构用得多。

我个人习惯，实盘用WebSocket，回测用REST API。这样分工明确，各取所长。

3.2 WebSocket协议：长连接的艺术

WebSocket这东西，本质上就是在TCP之上建立了一个全双工的通道。一旦连上，服务器可以随时把数据推给你，不用你一遍遍去问。

一个典型的WebSocket行情客户端，核心逻辑就三步：

1. 建立连接：用wss://协议连上交易所的网关。
2. 订阅主题：告诉服务器你想听哪些品种的行情（比如BTC/USDT的ticker、深度、成交）。
3. 处理消息：收到数据后，解析、清洗、分发。

这里我贴一段伪代码，帮你理解核心流程：

// 伪代码：WebSocket行情客户端
class MarketDataClient {
    constructor(url, symbols) {
        this.ws = new WebSocket(url);
        this.symbols = symbols;
        this.reconnectAttempts = 0;
    }

    connect() {
        this.ws.onopen = () => {
            // 连接成功后，订阅行情
            this.subscribe(this.symbols);
            this.reconnectAttempts = 0;
        };

        this.ws.onmessage = (event) => {
            // 收到原始数据，交给清洗模块
            const rawData = JSON.parse(event.data);
            this.dataCleaner.process(rawData);
        };

        this.ws.onclose = () => {
            // 断线重连，指数退避
            const delay = Math.min(1000 * 2 ** this.reconnectAttempts, 30000);
            setTimeout(() => this.connect(), delay);
            this.reconnectAttempts++;
        };

        this.ws.onerror = (err) => {
            console.error('WebSocket error:', err);
        };
    }

    subscribe(symbols) {
        const msg = {
            event: 'subscribe',
            channel: 'ticker',
            symbols: symbols
        };
        this.ws.send(JSON.stringify(msg));
    }
}

你看，代码本身不复杂。但真正考验人的，是那些「意外情况」。

3.3 数据清洗：垃圾进，垃圾出

数据接进来了，但你不能直接用。为什么？因为交易所返回的数据，往往带着各种「杂质」。

常见的数据问题：

• 重复数据：网络抖动导致同一笔ticker推送了两次。
• 乱序数据：后发的消息先到了，先发的消息后到了。
• 异常值：价格突然变成0，或者成交量出现天文数字。
• 缺失字段：某些交易所的深度快照，偶尔会漏掉几个档位。

我个人的清洗流程，一般分四步走：

1. 去重：根据交易所的sequence ID或timestamp，丢掉重复包。
2. 排序：用一个环形缓冲区，按时间戳重排乱序数据。
3. 过滤：设定合理的价格和成交量上下限，超出范围的直接丢弃。
4. 补全：对于缺失的字段，用上一个有效值填充，或者标记为NaN。

3.4 数据标准化：统一语言，消除差异

不同交易所，数据格式千差万别。有的用毫秒时间戳，有的用微秒；有的价格用字符串，有的用浮点数；有的深度档位从0开始，有的从1开始。

如果不做标准化，你的策略代码里就会充斥着各种if-else判断，维护起来简直噩梦。

标准化的目标： 把不同来源的数据，统一成系统内部定义的「标准格式」。

举个例子，我常用的标准化结构：

// 标准化的Ticker数据结构
{
    "exchange": "binance",      // 交易所名称
    "symbol": "BTC/USDT",       // 统一交易对格式
    "price": 50000.00,          // 最新价，统一为浮点数
    "volume": 1234.56,          // 24小时成交量
    "timestamp": 1628000000000, // 统一为毫秒时间戳
    "bid": 49999.00,            // 买一价
    "ask": 50001.00,            // 卖一价
    "source": "websocket"       // 数据来源
}

标准化之后，你的策略模块只需要认这一种格式。换交易所？改一下适配器就行，策略代码不用动。

3.5 整体架构：一张图看懂数据流

说了这么多，咱们用一张图把整个流程串起来。下面这张SVG图，展示了我个人比较喜欢的数据接入层架构：

这张图里，从上到下依次是：数据源、连接管理器、数据清洗层、标准化输出。每一层各司其职，上层不用关心下层的细节。这就是事件驱动架构的魅力——解耦。

3.6 实战中的几个关键点

最后，分享几个我在实战中总结出来的关键点，希望能帮你少走弯路：

• 本地缓存：WebSocket断线重连后，交易所通常会推送一次全量快照。这时候要用本地缓存做增量合并，别把之前的深度数据全丢了。
• 数据对齐：不同交易所的ticker更新频率不一样。有的100ms推一次，有的500ms。做套利策略时，一定要把时间戳对齐到同一个精度。
• 日志记录：所有原始数据，建议至少保留一份日志文件。万一策略出了异常，还能回溯排查。我习惯按天滚动日志，压缩后存储。
• 压力测试：行情爆发时（比如重大新闻发布），数据量可能是平时的10倍。你的清洗模块能不能扛住？建议提前做压测。