第1章：数据获取——用Python搞定多交易所实时行情

做跨市场套利，第一道坎就是数据。没有数据，什么策略都是空谈。

我个人习惯把数据获取分成两类：一类是实时数据，用WebSocket推过来；另一类是历史数据，用REST API去拉。这两者各有各的用处，也各有各的坑。今天我们就来聊聊，怎么用Python把这套东西搭起来。

1.1 为什么需要多交易所数据？

跨市场套利的本质，就是同一个资产在不同交易所之间存在价差。你想想看，比特币在币安卖10000 USDT，在OKX卖10005 USDT，这5块钱的差价就是机会。

但问题来了——你得同时看到两个交易所的报价。如果只看一个，你根本不知道另一边发生了什么。所以，多交易所数据获取是套利策略的起点。

核心要点： 套利机会转瞬即逝，数据延迟直接决定策略生死。

1.2 WebSocket vs REST API：怎么选？

先说结论：实时行情用WebSocket，历史数据用REST API。

为什么？

  • WebSocket：建立一次连接，服务器主动推送数据。延迟低，适合盯盘。
  • REST API：每次请求都要建立HTTP连接。适合拉取K线、订单簿快照等历史数据。

我在项目中遇到过一个问题：刚开始做套利时，我用REST API轮询行情，每秒钟请求一次。结果呢？交易所把我IP封了。后来换成WebSocket，不仅没被封，数据还更及时。

小技巧： 如果交易所同时支持WebSocket和REST，优先用WebSocket获取实时数据。REST API只用来补数据或做校验。

1.3 实战：用WebSocket获取币安实时行情

我们以币安为例，写一个简单的WebSocket客户端。币安的WebSocket地址是：wss://stream.binance.com:9443/ws。

import websocket
import json

def on_message(ws, message):
    data = json.loads(message)
    print(f"BTC/USDT 最新价: {data['c']}")

def on_error(ws, error):
    print(f"连接出错: {error}")

def on_close(ws, close_status_code, close_msg):
    print("连接已关闭")

def on_open(ws):
    # 订阅BTC/USDT的实时行情
    subscribe_msg = {
        "method": "SUBSCRIBE",
        "params": ["btcusdt@ticker"],
        "id": 1
    }
    ws.send(json.dumps(subscribe_msg))
    print("订阅成功")

if __name__ == "__main__":
    ws = websocket.WebSocketApp(
        "wss://stream.binance.com:9443/ws",
        on_open=on_open,
        on_message=on_message,
        on_error=on_error,
        on_close=on_close
    )
    ws.run_forever()

这段代码做了几件事：

  1. 建立WebSocket连接
  2. 订阅BTC/USDT的ticker数据
  3. 收到数据后打印最新价

嗯，这里要注意：币安的WebSocket数据是JSON格式，字段很多。比如c是当前价，v是成交量。具体字段可以查官方文档。

1.4 实战：用REST API获取OKX历史K线

有时候我们需要拉取历史数据做回测。这时候REST API就派上用场了。以OKX为例，获取BTC/USDT的1小时K线：

import requests
import pandas as pd

def get_okx_klines(symbol, bar='1H', limit=100):
    url = "https://www.okx.com/api/v5/market/candles"
    params = {
        'instId': symbol,
        'bar': bar,
        'limit': limit
    }
    response = requests.get(url, params=params)
    data = response.json()['data']
    
    # 转换成DataFrame
    df = pd.DataFrame(data, columns=[
        'timestamp', 'open', 'high', 'low', 'close', 'vol', 'vol_ccy'
    ])
    df['timestamp'] = pd.to_datetime(df['timestamp'].astype(int), unit='ms')
    df[['open', 'high', 'low', 'close']] = df[['open', 'high', 'low', 'close']].astype(float)
    return df

# 获取最近100根1小时K线
df = get_okx_klines('BTC-USDT', bar='1H', limit=100)
print(df.head())

这段代码返回一个DataFrame，包含时间戳、开盘价、最高价、最低价、收盘价和成交量。说白了，这就是我们做回测的基础数据。

避坑指南： 我曾经因为没处理REST API的限频，导致程序频繁报错。每个交易所都有请求频率限制，比如币安是1200次/分钟。建议在代码里加个sleep或者用限流库。

1.5 多交易所数据聚合架构

单交易所的数据获取不难，难的是同时管理多个交易所的连接。我画了一张图，展示我常用的架构：

多交易所数据聚合架构 币安 WebSocket OKX WebSocket Bybit WebSocket 数据聚合层（统一格式、去重、对齐时间戳） 套利策略引擎（价差计算、信号生成）

这个架构的核心思想是：每个交易所独立维护一个WebSocket连接，收到数据后统一送到聚合层。聚合层负责三件事：

  • 统一格式：不同交易所的数据字段名不一样，比如币安用c表示最新价，OKX用last。聚合层把它们转成统一格式。
  • 去重：WebSocket偶尔会推送重复数据，需要过滤掉。
  • 对齐时间戳：不同交易所的服务器时间可能有偏差，需要统一对齐到同一个时间基准。

1.6 代码实战：多交易所数据聚合

下面是一个简化版的聚合代码，演示如何同时监听币安和OKX的行情：

import websocket
import json
import threading

class MultiExchangeAggregator:
    def __init__(self):
        self.data = {}
        self.lock = threading.Lock()
    
    def on_message_binance(self, ws, message):
        data = json.loads(message)
        with self.lock:
            self.data['binance'] = {
                'price': float(data['c']),
                'timestamp': data['E']
            }
    
    def on_message_okx(self, ws, message):
        data = json.loads(message)
        with self.lock:
            self.data['okx'] = {
                'price': float(data['data'][0]['last']),
                'timestamp': data['data'][0]['ts']
            }
    
    def start(self):
        # 币安连接
        binance_ws = websocket.WebSocketApp(
            "wss://stream.binance.com:9443/ws",
            on_message=self.on_message_binance
        )
        binance_ws.on_open = lambda ws: ws.send(
            json.dumps({"method": "SUBSCRIBE", "params": ["btcusdt@ticker"], "id": 1})
        )
        
        # OKX连接
        okx_ws = websocket.WebSocketApp(
            "wss://ws.okx.com:8443/ws/v5/public",
            on_message=self.on_message_okx
        )
        okx_ws.on_open = lambda ws: ws.send(
            json.dumps({"op": "subscribe", "args": [{"channel": "tickers", "instId": "BTC-USDT"}]})
        )
        
        # 启动两个线程
        t1 = threading.Thread(target=binance_ws.run_forever)
        t2 = threading.Thread(target=okx_ws.run_forever)
        t1.start()
        t2.start()
        
        # 主线程打印价差
        while True:
            with self.lock:
                if 'binance' in self.data and 'okx' in self.data:
                    spread = self.data['okx']['price'] - self.data['binance']['price']
                    print(f"价差: {spread:.2f}")

if __name__ == "__main__":
    aggregator = MultiExchangeAggregator()
    aggregator.start()

这段代码用两个线程分别维护币安和OKX的连接，主线程不断计算价差。实际生产中，你还需要处理断线重连、心跳检测等问题。

个人经验： 我建议用websocket-client库而不是websockets库。前者更稳定，对多线程支持更好。我在生产环境跑了半年，没出过问题。

1.7 数据存储：别让数据白跑

实时数据拿到手，不存下来就浪费了。我一般用两种方式：

  • 内存队列：用queue.Queue暂存，策略直接从队列取数据。
  • 磁盘存储：用InfluxDB或CSV文件保存历史数据，方便回测。

举个例子，把数据写入CSV：

import csv
from datetime import datetime

def save_to_csv(data, filename='market_data.csv'):
    with open(filename, 'a', newline='') as f:
        writer = csv.writer(f)
        writer.writerow([
            datetime.now().isoformat(),
            data['exchange'],
            data['symbol'],
            data['price'],
            data['volume']
        ])

嗯，这里要注意：CSV文件会越来越大，建议每天或每小时轮转一次文件。

1.8 常见问题与避坑

做数据获取这几年，我踩过不少坑。挑几个典型的说说：

问题 原因 解决方案
WebSocket频繁断线 网络不稳定或交易所主动断开 实现自动重连机制，设置心跳包
数据延迟高 服务器距离远或代码处理慢 使用就近服务器，优化数据处理逻辑
数据格式不一致 不同交易所字段名不同 在聚合层做统一映射
API被限频 请求频率超过限制 加sleep或使用限流库

避坑指南： 我曾经因为没处理WebSocket的断线重连，导致策略在凌晨3点断线，第二天早上才发现。那晚的套利机会全错过了。所以，一定要加自动重连和告警机制。

1.9 小结

数据获取是套利策略的基石。WebSocket负责实时推送，REST API负责历史数据拉取。两者结合，才能构建一个完整的数据体系。

我个人建议：先从单个交易所的WebSocket开始，跑通了再扩展到多个。别一上来就想搞大而全的架构，容易翻车。

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

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