第2章数据源接入实战

数据源接入实战：WebSocket实时行情接入（以Binance为例）、REST API历史数据获取（以Coinbase为例）

做量化交易，第一关就是搞定数据。没有数据，策略就是空中楼阁。

我个人习惯把数据源分成两类：实时数据和历史数据。实时数据用来感知市场脉搏，历史数据用来回测策略。今天咱们就拿两个最主流的交易所开刀——Binance的WebSocket实时行情，和Coinbase的REST API历史数据。

为什么选这两个交易所？

Binance的WebSocket接口，文档清晰，延迟低，币种全。我在项目中遇到过好几次，用它的深度流做高频信号，效果不错。Coinbase呢？它的REST API对历史数据支持很友好，分页稳定，数据质量高。说白了，一个适合吃实时，一个适合挖历史。

核心思路：实时数据用WebSocket长连接，历史数据用REST轮询。两者互补，构成完整的数据管道。

一、Binance WebSocket 实时行情接入

WebSocket 说白了就是一条双向通道。客户端连上服务器后，服务器会主动推数据给你。你不需要反复问「有新数据吗？」，它自己会告诉你。

1.1 连接与订阅

Binance的WebSocket地址是固定的：wss://stream.binance.com:9443/ws。你只需要发一个订阅消息，告诉它你想要什么。

import websocket
import json

def on_message(ws, message):
    data = json.loads(message)
    print(f"收到行情: {data}")

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

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

def on_open(ws):
    # 订阅 BTCUSDT 的实时成交数据
    subscribe_msg = {
        "method": "SUBSCRIBE",
        "params": ["btcusdt@trade"],
        "id": 1
    }
    ws.send(json.dumps(subscribe_msg))
    print("订阅成功")

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()

嗯，这里要注意。订阅的流名称有固定格式：小写交易对@事件类型。比如 btcusdt@trade 是逐笔成交，btcusdt@depth20 是20档深度。我曾经因为大小写问题折腾了半小时，后来才发现Binance要求全小写。

1.2 数据解析与缓存

收到数据后，不能直接扔给策略。你需要解析、校验、缓存。我一般用 deque 做环形缓冲区，控制内存。

from collections import deque
import threading

class BinanceStream:
    def __init__(self, maxlen=1000):
        self.buffer = deque(maxlen=maxlen)
        self.lock = threading.Lock()
    
    def on_message(self, ws, message):
        data = json.loads(message)
        # 只保留成交数据
        if 'e' in data and data['e'] == 'trade':
            trade = {
                'price': float(data['p']),
                'qty': float(data['q']),
                'time': data['T'],
                'side': 'buy' if data['m'] else 'sell'
            }
            with self.lock:
                self.buffer.append(trade)
    
    def get_latest(self, n=10):
        with self.lock:
            return list(self.buffer)[-n:]

避坑指南：WebSocket 连接会断。我曾经在凌晨三点被报警吵醒，发现连接断了两个小时。一定要加心跳检测和自动重连机制。Binance 每3分钟会发一个 ping，你必须在10分钟内回复 pong，否则它会踢你。

二、Coinbase REST API 历史数据获取

实时数据有了，但回测需要历史数据。Coinbase 的 REST API 在这方面做得不错。它支持按时间范围查询，粒度到分钟级。

2.1 获取历史K线

Coinbase 的接口地址是 https://api.exchange.coinbase.com。获取历史K线用 /products/{product_id}/candles。

import requests
import time
from datetime import datetime, timedelta

def get_coinbase_klines(symbol, granularity=3600, start=None, end=None):
    """
    获取Coinbase历史K线
    :param symbol: 交易对，如 'BTC-USD'
    :param granularity: 粒度，秒为单位。3600=1小时
    :param start: 开始时间，datetime对象
    :param end: 结束时间，datetime对象
    """
    url = f"https://api.exchange.coinbase.com/products/{symbol}/candles"
    
    params = {
        'granularity': granularity
    }
    
    if start:
        params['start'] = start.isoformat()
    if end:
        params['end'] = end.isoformat()
    
    resp = requests.get(url, params=params)
    resp.raise_for_status()
    
    data = resp.json()
    # Coinbase返回格式: [时间, 低, 高, 开, 关, 成交量]
    klines = []
    for item in data:
        kline = {
            'time': datetime.fromtimestamp(item[0]),
            'low': float(item[1]),
            'high': float(item[2]),
            'open': float(item[3]),
            'close': float(item[4]),
            'volume': float(item[5])
        }
        klines.append(kline)
    
    return sorted(klines, key=lambda x: x['time'])

# 获取最近7天的1小时K线
end_time = datetime.utcnow()
start_time = end_time - timedelta(days=7)
klines = get_coinbase_klines('BTC-USD', granularity=3600, 
                              start=start_time, end=end_time)
print(f"获取到 {len(klines)} 根K线")

2.2 分页与限频

Coinbase 对历史数据有300条的限制。如果你要拉一年的数据，必须分页。你想想看，一次300条，一天86400秒，粒度3600秒的话，一天才24条。但如果你要秒级数据，那就得循环很多次。

粒度	每次最多返回	覆盖时间范围
60秒（1分钟）	300条	5小时
300秒（5分钟）	300条	25小时
3600秒（1小时）	300条	12.5天
86400秒（1天）	300条	300天

注意：Coinbase 的限频是每秒10次请求。我刚开始做数据采集时，没控制好频率，直接被封了IP半小时。建议加个 time.sleep(0.1) 做限流。

三、数据融合：实时+历史的统一接口

有了两个数据源，怎么融合？我习惯做一个统一的数据接口层。上层策略不用关心数据是从WebSocket来的还是REST来的，它只管取数据。

class UnifiedDataFeed:
    def __init__(self):
        self.real_time = BinanceStream()
        self.historical_cache = {}
    
    def get_klines(self, symbol, start, end, granularity):
        """统一获取K线，优先从缓存取，没有则从Coinbase拉"""
        cache_key = f"{symbol}_{granularity}"
        if cache_key not in self.historical_cache:
            self.historical_cache[cache_key] = []
        
        # 检查缓存是否覆盖了请求范围
        cached = self.historical_cache[cache_key]
        if cached and cached[0]['time'] <= start and cached[-1]['time'] >= end:
            return [k for k in cached if start <= k['time'] <= end]
        
        # 从Coinbase拉取
        new_data = get_coinbase_klines(symbol, granularity, start, end)
        self.historical_cache[cache_key].extend(new_data)
        return new_data
    
    def get_latest_trade(self):
        """获取最新一笔成交"""
        trades = self.real_time.get_latest(1)
        return trades[0] if trades else None

这样做的好处很明显。策略回测时走历史数据，实盘时走实时数据。切换只需要改一个配置项。我在项目中就是这么干的，省了不少事。

四、整体架构图

下面这张图展示了整个数据管道的流转逻辑。从两个交易所分别接入，经过解析、缓存、融合，最终提供给策略层使用。

你看，左边是实时流，右边是历史流。两条线在中间汇合，统一对外提供服务。这样做的好处是，策略层完全不需要关心数据是从哪来的。你换交易所也好，换数据源也好，只需要改底层接入代码，上层纹丝不动。

个人经验：刚开始做数据融合时，我犯过一个错——把实时数据和历史数据混在一起存。结果回测时用了实盘数据，策略表现异常好，一上实盘就崩。后来我强制区分数据来源，给每条数据打上标签：source: 'realtime' 或 source: 'historical'。这个习惯一直保留到现在。

好了，数据源接入这块就聊到这儿。记住一句话：实时数据要稳，历史数据要全。Binance的WebSocket帮你搞定稳，Coinbase的REST帮你搞定全。两者结合，你的数据管道就立起来了。

📚 课程目录

数据源接入实战：WebSocket实时行情接入（以Binance为例）、REST API历史数据获取（以Coinbase为例）

为什么选这两个交易所？

一、Binance WebSocket 实时行情接入

1.1 连接与订阅

1.2 数据解析与缓存

二、Coinbase REST API 历史数据获取

2.1 获取历史K线

2.2 分页与限频

三、数据融合：实时+历史的统一接口

四、整体架构图