第四章：数据获取：使用Python获取多交易所实时行情数据

做跨市场套利，最核心的命脉是什么？

我个人觉得，不是策略多精妙，也不是资金多雄厚。而是——你能不能拿到干净、及时、对齐的数据。

你想想看，两个市场的行情如果差了一秒，套利窗口可能就关上了。我在2019年做BTC期现套利时，就因为数据延迟了200毫秒，眼睁睁看着价差从0.5%缩到0.05%。嗯，那次教训挺深刻的。

4.1 数据获取的核心挑战

跨市场套利的数据获取，说白了就三个难点：

1. 多源异构：每个交易所的API格式、字段命名、推送方式都不一样
2. 时间对齐：不同服务器的时钟偏差，可能导致价差计算失真
3. 稳定性：任何一个交易所断连，你的套利模型就瘸了一条腿

4.2 架构设计：统一数据管道

我习惯用一套「发布-订阅」模式来组织数据流。每个交易所的行情是一个独立的生产者，策略模块是消费者。中间用消息队列解耦。

4.3 实战代码：Binance + OKX 实时行情获取

我拿两个最常用的交易所举例。Binance和OKX，一个偏现货，一个偏合约。套利机会经常出现在它们之间。

4.3.1 统一Tick数据结构

先定义一个标准化的Tick对象。不管哪个交易所来的数据，都转成这个格式：

from dataclasses import dataclass
from datetime import datetime
from typing import Optional

@dataclass
class UnifiedTick:
    """统一行情Tick结构"""
    exchange: str          # 交易所名称
    symbol: str            # 交易对，如 'BTC/USDT'
    bid_price: float       # 买一价
    ask_price: float       # 卖一价
    bid_volume: float      # 买一量
    ask_volume: float      # 卖一量
    last_price: float      # 最新成交价
    timestamp: int         # 毫秒级时间戳
    local_time: datetime   # 本地接收时间
    
    @property
    def spread(self) -> float:
        """买卖价差"""
        return self.ask_price - self.bid_price
    
    @property
    def mid_price(self) -> float:
        """中间价"""
        return (self.bid_price + self.ask_price) / 2

4.3.2 Binance WebSocket 行情订阅

import asyncio
import json
import websockets
from typing import Callable

class BinanceWebSocket:
    """Binance 实时行情订阅"""
    
    def __init__(self, symbols: list, callback: Callable):
        self.symbols = symbols
        self.callback = callback
        self.base_url = "wss://stream.binance.com:9443/ws"
        
    async def connect(self):
        """建立WebSocket连接"""
        streams = [f"{s.lower()}@bookTicker" for s in self.symbols]
        url = f"{self.base_url}/{'/'.join(streams)}"
        
        async with websockets.connect(url) as ws:
            while True:
                try:
                    data = await ws.recv()
                    tick = self._parse_tick(json.loads(data))
                    await self.callback(tick)
                except Exception as e:
                    print(f"Binance WS error: {e}")
                    # 自动重连
                    await asyncio.sleep(1)
                    break
    
    def _parse_tick(self, raw: dict) -> UnifiedTick:
        """解析Binance原始数据"""
        return UnifiedTick(
            exchange='binance',
            symbol=raw['s'],
            bid_price=float(raw['b']),
            ask_price=float(raw['a']),
            bid_volume=float(raw['B']),
            ask_volume=float(raw['A']),
            last_price=0.0,  # bookTicker不含成交价
            timestamp=raw['E'],
            local_time=datetime.now()
        )

4.3.3 OKX WebSocket 行情订阅

class OKXWebSocket:
    """OKX 实时行情订阅"""
    
    def __init__(self, symbols: list, callback: Callable):
        self.symbols = symbols
        self.callback = callback
        self.base_url = "wss://ws.okx.com:8443/ws/v5/public"
        
    async def connect(self):
        """建立WebSocket连接并订阅"""
        async with websockets.connect(self.base_url) as ws:
            # 发送订阅请求
            subscribe_msg = {
                "op": "subscribe",
                "args": [{
                    "channel": "tickers",
                    "instId": s
                } for s in self.symbols]
            }
            await ws.send(json.dumps(subscribe_msg))
            
            while True:
                try:
                    data = await ws.recv()
                    tick = self._parse_tick(json.loads(data))
                    if tick:
                        await self.callback(tick)
                except Exception as e:
                    print(f"OKX WS error: {e}")
                    await asyncio.sleep(1)
                    break
    
    def _parse_tick(self, raw: dict) -> Optional[UnifiedTick]:
        """解析OKX原始数据"""
        if 'data' not in raw:
            return None
            
        d = raw['data'][0]
        return UnifiedTick(
            exchange='okx',
            symbol=d['instId'],
            bid_price=float(d['bidPx']),
            ask_price=float(d['askPx']),
            bid_volume=float(d['bidSz']),
            ask_volume=float(d['askSz']),
            last_price=float(d['last']),
            timestamp=int(d['ts']),
            local_time=datetime.now()
        )

4.4 数据对齐：时钟偏差处理

这里有个坑，我曾经踩过。

两个交易所的服务器时间可能差几百毫秒。如果你直接用它们的时间戳计算价差，会发现价差忽大忽小，其实不是市场变了，是时钟没对齐。

我的解决方案：

1. 启动时先获取两个交易所的服务器时间，计算偏差
2. 每次收到Tick，用本地时间 + 偏差修正
3. 如果偏差超过1秒，发出告警

class TimeSync:
    """时间同步管理器"""
    
    def __init__(self):
        self.binance_offset = 0.0
        self.okx_offset = 0.0
        
    async def sync(self):
        """获取各交易所时间偏差"""
        import aiohttp
        
        async with aiohttp.ClientSession() as session:
            # Binance
            async with session.get('https://api.binance.com/api/v3/time') as resp:
                data = await resp.json()
                server_time = data['serverTime'] / 1000
                local_time = datetime.now().timestamp()
                self.binance_offset = server_time - local_time
                
            # OKX
            async with session.get('https://www.okx.com/api/v5/public/time') as resp:
                data = await resp.json()
                server_time = int(data['data'][0]['ts']) / 1000
                local_time = datetime.now().timestamp()
                self.okx_offset = server_time - local_time
                
    def adjust_timestamp(self, exchange: str, local_ts: float) -> float:
        """修正时间戳"""
        offset = getattr(self, f"{exchange}_offset", 0.0)
        return local_ts + offset

4.5 数据质量监控

数据拿到手了，不代表就能直接用。我一般会加三道检查：

检查项	阈值	处理方式
价格异常跳动	单次变动 > 5%	丢弃该Tick，记录日志
买卖价差为负	bid > ask	丢弃，触发告警
数据延迟	本地时间 - 服务器时间 > 2秒	标记为延迟数据，不参与策略计算
重复数据	相同时间戳出现2次以上	去重，保留第一条

class DataQualityChecker:
    """数据质量检查器"""
    
    def __init__(self):
        self.last_prices = {}  # symbol -> price
        self.seen_timestamps = set()
        
    def check(self, tick: UnifiedTick) -> bool:
        """检查Tick数据质量，返回True表示通过"""
        key = f"{tick.exchange}_{tick.symbol}"
        
        # 1. 检查买卖价差
        if tick.bid_price >= tick.ask_price:
            print(f"Invalid spread: {tick}")
            return False
            
        # 2. 检查价格异常跳动
        if key in self.last_prices:
            last = self.last_prices[key]
            change = abs(tick.last_price - last) / last
            if change > 0.05:  # 5%阈值
                print(f"Price jump detected: {tick}")
                return False
                
        # 3. 检查重复数据
        ts_key = f"{key}_{tick.timestamp}"
        if ts_key in self.seen_timestamps:
            return False
            
        # 更新状态
        self.last_prices[key] = tick.last_price
        self.seen_timestamps.add(ts_key)
        return True

4.6 启动与运行

最后，把整个系统串起来：

async def main():
    """主函数：启动多交易所数据获取"""
    
    symbols = ['BTC-USDT', 'ETH-USDT']
    
    # 数据回调
    async def on_tick(tick: UnifiedTick):
        checker = DataQualityChecker()
        if checker.check(tick):
            # 推送到消息队列或策略模块
            print(f"{tick.exchange} {tick.symbol}: "
                  f"Bid={tick.bid_price:.2f} Ask={tick.ask_price:.2f}")
    
    # 启动时间同步
    time_sync = TimeSync()
    await time_sync.sync()
    print(f"Time offsets: Binance={time_sync.binance_offset:.3f}s, "
          f"OKX={time_sync.okx_offset:.3f}s")
    
    # 启动WebSocket连接
    binance_ws = BinanceWebSocket(symbols, on_tick)
    okx_ws = OKXWebSocket(symbols, on_tick)
    
    # 并发运行
    await asyncio.gather(
        binance_ws.connect(),
        okx_ws.connect()
    )

if __name__ == "__main__":
    asyncio.run(main())

数据获取这块，说白了就是「稳」字当头。代码写得再花哨，不如一个稳定的连接来得实在。我见过太多人把精力花在优化策略上，结果数据源一断，全白搭。

嗯，先把数据管道搭稳了，后面的套利策略才能跑得安心。

---

公众号:蓝海数据掘金营，微信deep3321