3、数据获取：WebSocket实时订阅、REST API快照、数据清洗与对齐

好，咱们进入第三章。这一章讲的是数据获取，说白了就是怎么把订单簿数据从交易所「搬」到你的程序里。我见过太多人在这上面栽跟头——数据没对齐，时间戳搞混，快照和增量对不上，最后策略跑出来全是错的。

嗯，咱们一步步来。先搞清楚三个核心问题：

  • 怎么拿到实时数据？——WebSocket
  • 怎么拿到完整快照？——REST API
  • 怎么把两者拼起来？——数据清洗与对齐

核心原则：WebSocket负责「快」，REST API负责「全」。两者缺一不可。

3.1 WebSocket实时订阅

WebSocket，说白了就是交易所给你开了一条「专线」。你连上去，交易所不断往你这边推数据。我刚开始做量化时，用的是轮询——每秒请求一次REST API，结果被交易所封了IP。后来改用WebSocket，带宽占用从几百KB降到了几十KB，延迟从几百毫秒降到了个位数毫秒。

为什么用WebSocket？你想想看，订单簿每秒可能变化几十次甚至上百次。如果用REST API轮询，你永远拿不到「实时」数据。WebSocket是推送模式，数据一来，你立刻知道。

以币安为例，订阅订单簿增量数据的格式是这样的：

// WebSocket订阅深度数据（币安）
// 订阅深度流：<streamName>@depth@100ms
// 100ms表示每100毫秒推送一次增量

const WebSocket = require('ws');
const ws = new WebSocket('wss://stream.binance.com:9443/ws');

// 订阅BTCUSDT的深度数据
ws.on('open', () => {
  ws.send(JSON.stringify({
    method: 'SUBSCRIBE',
    params: [
      'btcusdt@depth@100ms'
    ],
    id: 1
  }));
});

ws.on('message', (data) => {
  const msg = JSON.parse(data);
  // msg包含：事件类型、时间戳、bids和asks的增量
  // 格式：{ e: 'depthUpdate', E: 123456789, b: [[price, qty], ...], a: [[price, qty], ...] }
  console.log('收到增量数据:', msg);
});

个人经验：我建议订阅时加上100ms或1000ms的推送间隔。太快的推送（比如实时推送）会导致你的程序处理不过来，反而增加延迟。100ms是个不错的平衡点。

这里有个坑——WebSocket推送的是增量数据，不是完整的订单簿。什么意思？就是交易所只告诉你「哪些价格变了」，而不是把整个订单簿重新发一遍。所以，你需要维护一个本地的订单簿副本，每次收到增量就更新它。

3.2 REST API快照

WebSocket增量数据有个前提：你得先有一个「基准」订单簿。这个基准从哪里来？REST API快照。

REST API快照，就是一次性拉取当前时刻的完整订单簿。它包含了所有档位的买盘和卖盘数据。我一般会在程序启动时拉一次快照，然后每隔一段时间（比如5分钟）再拉一次，用来校正本地订单簿。

为什么需要校正？因为WebSocket可能会丢包、断线重连，或者增量数据有延迟。如果不校正，你的本地订单簿会慢慢「漂移」，和真实订单簿产生偏差。

// REST API获取订单簿快照（币安）
// GET /api/v3/depth?symbol=BTCUSDT&limit=1000

const axios = require('axios');

async function getSnapshot(symbol) {
  const url = `https://api.binance.com/api/v3/depth?symbol=${symbol}&limit=1000`;
  const response = await axios.get(url);
  
  // 返回格式：
  // {
  //   lastUpdateId: 123456789,
  //   bids: [[price, qty], ...],
  //   asks: [[price, qty], ...]
  // }
  return response.data;
}

// 使用示例
const snapshot = await getSnapshot('BTCUSDT');
console.log('快照最后更新ID:', snapshot.lastUpdateId);

注意：快照的limit参数很关键。limit=100表示只取前100档，limit=1000表示取前1000档。我个人习惯用1000，因为高频交易中，深度数据越全越好。但要注意，limit越大，数据量越大，网络传输时间也越长。

3.3 数据清洗与对齐

好，现在你有了快照和增量数据。怎么把它们拼起来？这里有个关键点：对齐。

交易所的每个增量数据都有一个「最后更新ID」（lastUpdateId或U）。快照也有一个「最后更新ID」。对齐的规则是：

  1. 先拉取快照，记录它的lastUpdateId
  2. 开始接收WebSocket增量数据
  3. 丢弃所有U <= lastUpdateId的增量（这些是快照之前的数据）
  4. 只处理U > lastUpdateId的增量
  5. 当增量中的U >= lastUpdateId + 1时，开始应用增量到快照上

嗯，这里有点绕。我画个图帮你理解：

数据对齐流程 T0 拉取快照 lastUpdateId=100 T1 增量U=90 丢弃 T2 增量U=101 应用 T3 增量U=105 应用 T4 增量U=108 应用 本地订单簿维护 1. 初始化：用快照数据填充 bids 和 asks 数组 2. 增量处理：对每个增量中的价格档位： - 如果数量为0，删除该价格档位 - 如果数量大于0，更新该价格档位

代码实现大概是这样的：

// 数据对齐核心逻辑
class OrderBook {
  constructor() {
    this.bids = new Map();  // 买盘：价格 -> 数量
    this.asks = new Map();  // 卖盘：价格 -> 数量
    this.lastUpdateId = 0;
    this.snapshotApplied = false;
  }

  // 应用快照
  applySnapshot(snapshot) {
    this.bids.clear();
    this.asks.clear();
    
    snapshot.bids.forEach(([price, qty]) => {
      this.bids.set(price, parseFloat(qty));
    });
    snapshot.asks.forEach(([price, qty]) => {
      this.asks.set(price, parseFloat(qty));
    });
    
    this.lastUpdateId = snapshot.lastUpdateId;
    this.snapshotApplied = true;
    console.log(`快照已应用，lastUpdateId: ${this.lastUpdateId}`);
  }

  // 应用增量
  applyUpdate(update) {
    // 对齐检查：丢弃快照之前的增量
    if (update.U <= this.lastUpdateId) {
      return; // 丢弃
    }
    
    // 处理买盘增量
    update.b.forEach(([price, qty]) => {
      const qtyNum = parseFloat(qty);
      if (qtyNum === 0) {
        this.bids.delete(price);  // 删除档位
      } else {
        this.bids.set(price, qtyNum);  // 更新档位
      }
    });
    
    // 处理卖盘增量
    update.a.forEach(([price, qty]) => {
      const qtyNum = parseFloat(qty);
      if (qtyNum === 0) {
        this.asks.delete(price);
      } else {
        this.asks.set(price, qtyNum);
      }
    });
    
    // 更新lastUpdateId
    this.lastUpdateId = update.u;
  }

  // 获取当前最优买卖价
  getTopOfBook() {
    const bestBid = Math.max(...this.bids.keys());
    const bestAsk = Math.min(...this.asks.keys());
    return { bestBid, bestAsk, spread: bestAsk - bestBid };
  }
}

// 使用示例
const orderBook = new OrderBook();

// 1. 拉取快照
const snapshot = await getSnapshot('BTCUSDT');
orderBook.applySnapshot(snapshot);

// 2. 订阅WebSocket增量
ws.on('message', (data) => {
  const update = JSON.parse(data);
  if (update.e === 'depthUpdate') {
    orderBook.applyUpdate(update);
    const top = orderBook.getTopOfBook();
    console.log(`最优买价: ${top.bestBid}, 最优卖价: ${top.bestAsk}, 价差: ${top.spread}`);
  }
});

避坑指南：我曾经遇到过一个问题——WebSocket增量数据中的价格是字符串，数量也是字符串。如果不做parseFloat转换，直接比较字符串会导致排序错误。嗯，这个坑我踩过，所以代码里我特意加了parseFloat。

3.4 数据清洗的细节

数据清洗，说白了就是去掉「脏数据」。我在项目中遇到过几种常见情况：

  • 重复数据：WebSocket偶尔会推送重复的增量，需要根据updateId去重
  • 乱序数据：网络延迟可能导致增量数据到达顺序错乱，需要按updateId排序
  • 空数据：某些交易所会在市场波动剧烈时推送空增量，直接忽略
  • 价格精度问题：不同交易所的价格精度不同，需要统一处理

举个例子，价格精度问题：币安的BTCUSDT价格精度是2位小数（比如50000.00），而某些小交易所可能是8位小数。如果你不做归一化，策略计算价差时会出现精度错误。

// 价格精度归一化
function normalizePrice(price, decimals = 2) {
  return parseFloat(price.toFixed(decimals));
}

// 数量精度归一化
function normalizeQty(qty, decimals = 6) {
  return parseFloat(qty.toFixed(decimals));
}

// 使用示例
const rawPrice = 50000.12345678;
const normalizedPrice = normalizePrice(rawPrice, 2);
console.log(normalizedPrice); // 50000.12

重要提醒：数据对齐是订单簿分析的基础。如果对齐做不好，后面的所有分析都是错的。我建议你在开发阶段，每隔一段时间打印一次本地订单簿的快照，和交易所的REST API快照做对比，确保两者一致。

3.5 实战中的注意事项

最后，分享几个实战中的小技巧：

  • 断线重连：WebSocket一定会断线。我建议实现自动重连机制，重连后重新拉取快照
  • 数据缓存：把原始数据缓存到本地文件或数据库，方便复盘分析
  • 性能监控：监控数据处理的延迟，如果延迟超过100ms，说明你的程序处理不过来
  • 多交易所对齐：如果你同时订阅多个交易所，每个交易所的订单簿要独立维护

嗯，这一章的内容就到这里。数据获取是订单簿分析的「地基」，地基打牢了，后面的分析才能站得住脚。

本章核心要点：

  • WebSocket负责实时增量数据，REST API负责完整快照
  • 数据对齐的关键是lastUpdateId，丢弃快照之前的增量
  • 数据清洗包括去重、排序、精度归一化
  • 断线重连后必须重新拉取快照

蓝海数据掘金营，专注资料整理

← 上一章 📖 返回目录 下一章 →