3. 行情数据接入：交易所API对接与数据清洗

行情数据接入，说白了就是量化做市系统的「眼睛」。眼睛要是瞎了，再牛逼的策略也白搭。我见过太多团队，策略模型跑得飞起，结果行情延迟高了50毫秒，直接亏到姥姥家。

今天咱们就聊聊，怎么把这双「眼睛」练得又亮又稳。

3.1 交易所API对接：REST vs WebSocket

交易所给咱们提供了两种接口：REST和WebSocket。这俩有啥区别？我直接说人话：

• REST：你问一句，它答一句。像发微信消息，一问一答。
• WebSocket：它一直跟你连着，有数据就主动推给你。像打电话，实时通话。

你想想看，做市系统需要实时盯盘，肯定得用WebSocket啊。但REST也不是完全没用，我后面会讲。

3.1.1 REST接口：适合做什么？

REST接口虽然慢，但胜在简单可靠。我个人习惯用它来做三件事：

• 获取历史数据：比如拉取过去24小时的K线数据，用于策略回测
• 查询账户信息：余额、持仓这些不常变的数据
• 下单与撤单：虽然WebSocket也能下单，但REST更稳定，不容易丢单

这里有个坑，我曾经踩过：REST接口有频率限制。大部分交易所限制每秒1-10次请求。你要是写个死循环去轮询，分分钟被封IP。

3.1.2 WebSocket：实时行情的命脉

WebSocket是做市系统的核心。它建立一条长连接，交易所一旦有新的行情数据，就立刻推给你。延迟通常在1-10毫秒级别。

但WebSocket也有麻烦事：

• 断线重连：网络波动导致连接断开，你得自动重连
• 心跳检测：长时间没收到数据，要主动发ping包确认连接是否存活
• 数据乱序：多个数据包同时到达，顺序可能错乱

我建议的架构是这样的：

// WebSocket连接管理器伪代码
class WebSocketManager {
    constructor(url) {
        this.url = url;
        this.ws = null;
        this.reconnectInterval = 1000; // 初始重连间隔1秒
        this.maxReconnectInterval = 30000; // 最大重连间隔30秒
    }

    connect() {
        this.ws = new WebSocket(this.url);
        this.ws.onopen = () => {
            console.log('连接成功');
            this.reconnectInterval = 1000; // 重置重连间隔
            this.startHeartbeat();
        };
        this.ws.onmessage = (event) => {
            this.handleMessage(event.data);
        };
        this.ws.onclose = () => {
            console.log('连接断开，准备重连');
            this.reconnect();
        };
        this.ws.onerror = (error) => {
            console.error('连接错误', error);
            this.ws.close();
        };
    }

    reconnect() {
        setTimeout(() => {
            console.log(`尝试重连，间隔${this.reconnectInterval}ms`);
            this.connect();
            this.reconnectInterval = Math.min(
                this.reconnectInterval * 2,
                this.maxReconnectInterval
            );
        }, this.reconnectInterval);
    }

    startHeartbeat() {
        setInterval(() => {
            if (this.ws.readyState === WebSocket.OPEN) {
                this.ws.send('ping');
            }
        }, 30000); // 每30秒发一次心跳
    }
}

嗯，这里要注意：重连间隔要指数退避。第一次断开等1秒，第二次等2秒，第三次等4秒...最多等30秒。这样既不会把交易所服务器打爆，也能尽快恢复连接。

3.2 行情数据格式标准化

不同交易所的行情数据格式五花八门。有的叫「bid」，有的叫「buy」，有的叫「bids」。你总不能为每个交易所写一套解析逻辑吧？

所以，我们需要一个统一的数据模型。

3.2.1 Level1 vs Level2 行情

先搞清楚这两个概念：

类型	包含内容	更新频率	用途
Level1	最新价、最高价、最低价、成交量、涨跌幅	每秒几次	简单监控、普通交易
Level2	完整买卖盘口（通常前50档）、逐笔成交	毫秒级	做市策略、高频交易

做市系统必须用Level2数据。为什么？因为你要看到盘口的深度和厚度，才能判断流动性够不够。Level1数据太粗糙，根本不够用。

3.2.2 统一数据模型设计

我个人习惯这样设计：

// 统一行情数据模型
class UnifiedMarketData {
    constructor() {
        this.exchange = '';      // 交易所名称
        this.symbol = '';       // 交易对，如 BTC/USDT
        this.timestamp = 0;     // 时间戳（毫秒）
        this.bids = [];         // 买单队列 [[价格, 数量], ...]
        this.asks = [];         // 卖单队列 [[价格, 数量], ...]
        this.lastPrice = 0;     // 最新成交价
        this.lastVolume = 0;    // 最新成交量
        this.volume24h = 0;     // 24小时成交量
    }
}

// 交易所适配器接口
class ExchangeAdapter {
    constructor(exchangeName) {
        this.exchangeName = exchangeName;
    }

    // 将交易所原始数据转换为统一格式
    normalize(rawData) {
        throw new Error('子类必须实现normalize方法');
    }
}

// 币安适配器示例
class BinanceAdapter extends ExchangeAdapter {
    normalize(rawData) {
        const data = new UnifiedMarketData();
        data.exchange = 'binance';
        data.symbol = rawData.s;
        data.timestamp = rawData.E;
        data.bids = rawData.b.map(item => [parseFloat(item[0]), parseFloat(item[1])]);
        data.asks = rawData.a.map(item => [parseFloat(item[0]), parseFloat(item[1])]);
        data.lastPrice = parseFloat(rawData.c);
        data.lastVolume = parseFloat(rawData.v);
        data.volume24h = parseFloat(rawData.V);
        return data;
    }
}

这样设计的好处是：新增一个交易所，只需要写一个适配器。核心策略完全不用改，直接消费统一格式的数据就行。

3.3 数据降噪与清洗

行情数据不是拿来就能用的。交易所推送的数据里，有很多「噪音」。比如：

• 重复数据：同一个tick推了两次
• 异常数据：价格突然跳变100倍（交易所bug或网络问题）
• 过期数据：时间戳比当前时间还早

这些噪音如果不处理，策略会做出错误决策。我曾经遇到过：某交易所的WebSocket偶尔会推送一条历史数据，导致策略以为价格暴跌，直接清仓了...嗯，亏了不少钱。

3.3.1 数据去重

最简单的去重方法：维护一个序列号或时间戳缓存。

class DataDeduplicator {
    constructor() {
        this.lastTimestamp = 0;
        this.lastSequence = 0;
    }

    isDuplicate(data) {
        // 如果时间戳和序列号都相同，说明是重复数据
        if (data.timestamp === this.lastTimestamp && 
            data.sequence === this.lastSequence) {
            return true;
        }
        this.lastTimestamp = data.timestamp;
        this.lastSequence = data.sequence;
        return false;
    }
}

3.3.2 异常值过滤

对于价格数据，我建议用滑动窗口标准差来检测异常：

class OutlierFilter {
    constructor(windowSize = 100, threshold = 3) {
        this.windowSize = windowSize;
        this.threshold = threshold;
        this.priceHistory = [];
    }

    isOutlier(price) {
        if (this.priceHistory.length < this.windowSize) {
            this.priceHistory.push(price);
            return false; // 数据不够，先不判断
        }

        // 计算均值和标准差
        const mean = this.priceHistory.reduce((a, b) => a + b, 0) / this.windowSize;
        const variance = this.priceHistory.reduce((sum, p) => sum + Math.pow(p - mean, 2), 0) / this.windowSize;
        const std = Math.sqrt(variance);

        // 如果价格偏离均值超过3个标准差，视为异常
        if (Math.abs(price - mean) > this.threshold * std) {
            return true;
        }

        // 更新滑动窗口
        this.priceHistory.shift();
        this.priceHistory.push(price);
        return false;
    }
}

3.3.3 数据对齐

不同交易所的数据到达时间不一样。比如币安的数据比OKX早到10毫秒。如果你直接比较两个交易所的价格，可能会得出错误的价差。

我的做法是：按时间戳对齐。把数据放进一个时间桶里，比如每100毫秒一个桶，同一个桶里的数据才进行比较。

class TimeAligner {
    constructor(bucketSizeMs = 100) {
        this.bucketSizeMs = bucketSizeMs;
        this.buckets = new Map();
    }

    addData(data) {
        const bucketKey = Math.floor(data.timestamp / this.bucketSizeMs);
        if (!this.buckets.has(bucketKey)) {
            this.buckets.set(bucketKey, []);
        }
        this.buckets.get(bucketKey).push(data);
    }

    getAlignedData() {
        // 返回上一个完整时间桶的数据
        const now = Date.now();
        const currentBucket = Math.floor(now / this.bucketSizeMs);
        const previousBucket = currentBucket - 1;
        
        const data = this.buckets.get(previousBucket);
        this.buckets.delete(previousBucket); // 清理旧数据
        return data || [];
    }
}

3.4 整体架构图

说了这么多，咱们画张图总结一下行情数据接入的整体流程：

这张图展示了从交易所原始数据到统一行情数据的完整链路。每一层都有明确的职责，出了问题也好排查。

行情数据接入这块，说白了就是「接进来、洗干净、统一化」。做好了，你的做市系统就成功了一半。做不好，后面策略写得再漂亮也是白搭。

我个人建议，在开发初期就把数据接入模块做得足够健壮。别急着跑策略，先把数据质量搞上去。你想想看，如果连数据都是错的，策略怎么可能赚钱？

---

公众号:蓝海资料掘金营，微信deep3321