第二章：数据源接入——交易所API对接、WebSocket实时行情、RESTful历史数据、数据缓存策略

做市系统里，数据源就是你的眼睛和耳朵。没有数据，策略就是瞎猜。我见过不少团队，策略模型写得漂漂亮亮，结果一上线，数据延迟了200毫秒，直接亏穿。所以这一章，咱们把数据接入这件事彻底聊透。

2.1 交易所API对接——先搞定“门”在哪

每个交易所都像一栋楼，API就是它的门。有的门是旋转门（RESTful），有的门是自动滑门（WebSocket）。你得先搞清楚门朝哪开，才能进去拿数据。

我个人习惯，对接一个新交易所时，先看三样东西：

• API文档的版本号——别用v1接口，很多交易所v1已经废弃了，我踩过这个坑。
• 频率限制（Rate Limit）——比如每秒最多10次请求，超了直接封IP。
• 签名方式——HMAC-SHA256还是RSA？这个搞错了，请求永远过不去。

举个例子，对接币安的现货行情API，代码大概长这样：

import requests
import hashlib
import hmac
import time

API_KEY = 'your_api_key'
SECRET_KEY = 'your_secret_key'

def get_klines(symbol='BTCUSDT', interval='1m'):
    url = 'https://api.binance.com/api/v3/klines'
    params = {
        'symbol': symbol,
        'interval': interval,
        'limit': 100
    }
    resp = requests.get(url, params=params)
    return resp.json()

嗯，这里要注意：RESTful接口适合拿历史数据，但别用它做实时行情。为什么？因为每次请求都有网络延迟，你请求一次，行情已经变了三次。

2.2 WebSocket实时行情——真正的“实时”

WebSocket就像一根水管，交易所那边一有数据，就顺着管子流到你这边。你不用去“拉”，它自己“推”过来。

做市系统里，实时行情通常包括：

• 深度数据（Order Book）——买一卖一、买五卖五，甚至全量深度
• 成交数据（Trade）——每一笔成交的价格和数量
• Ticker数据——24小时涨跌幅、成交量等

我个人建议，订阅深度数据时，别订阅全量深度。全量深度每次更新都推送几千条数据，带宽扛不住。我见过一个团队，同时订阅了5个交易所的全量深度，结果服务器带宽被打满，连策略都跑不动了。

正确的做法是订阅增量深度（Incremental Depth），只推送变化的部分。代码示例：

import websocket
import json

def on_message(ws, message):
    data = json.loads(message)
    # 处理增量深度数据
    if data['e'] == 'depthUpdate':
        # 更新本地order book
        update_local_book(data['b'], data['a'])

def on_error(ws, error):
    print(f"WebSocket error: {error}")

def on_close(ws, close_status_code, close_msg):
    print("WebSocket closed, reconnecting...")
    # 自动重连逻辑
    connect_websocket()

def on_open(ws):
    # 订阅BTCUSDT的增量深度
    sub_msg = {
        "method": "SUBSCRIBE",
        "params": ["btcusdt@depth@100ms"],
        "id": 1
    }
    ws.send(json.dumps(sub_msg))

def connect_websocket():
    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()

2.3 RESTful历史数据——补数据用的“时光机”

做回测或者策略初始化时，你需要历史数据。WebSocket只能拿实时数据，拿不到昨天的K线。这时候就得靠RESTful接口。

RESTful历史数据有几个坑：

• 数据粒度——1分钟K线、5分钟K线、日线，粒度不同，请求参数不同
• 数据完整性——有些交易所的历史数据有缺失，比如某一天突然少了几根K线
• 请求限制——一次最多拿1000根K线，想拿一年的数据，得循环请求

我遇到过最头疼的问题：回测时发现某天的数据对不上，查了半天，原来是交易所那天做了系统升级，数据丢了两个小时。所以，我建议从至少两个数据源交叉验证历史数据。

批量拉取历史数据的代码模式：

def fetch_historical_klines(symbol, interval, start_time, end_time):
    all_klines = []
    current_start = start_time
    
    while current_start < end_time:
        url = f'https://api.binance.com/api/v3/klines'
        params = {
            'symbol': symbol,
            'interval': interval,
            'startTime': current_start,
            'limit': 1000
        }
        resp = requests.get(url, params=params)
        data = resp.json()
        
        if not data:
            break
            
        all_klines.extend(data)
        # 更新起始时间为最后一根K线的结束时间
        current_start = data[-1][0] + 1
        
        # 控制请求频率，别被封
        time.sleep(0.1)
    
    return all_klines

2.4 数据缓存策略——别让CPU闲着等网络

数据到了本地，不能直接用完就扔。做市系统里，数据缓存是性能的关键。你想想看，每次策略需要深度数据时都去请求交易所，那延迟得多高？

我常用的缓存策略分三层：

说白了，L1缓存是为了让策略能“零延迟”拿到最新数据。L2缓存是为了防止程序重启后重新拉取大量数据。L3缓存则是为了回测和分析。

我习惯用Redis做L1缓存，因为它快，而且支持发布订阅模式。举个例子：

import redis

r = redis.Redis(host='localhost', port=6379, db=0)

# 存储最新深度数据
def update_order_book(symbol, bids, asks):
    key = f'orderbook:{symbol}'
    data = {
        'bids': bids[:5],  # 只存前5档
        'asks': asks[:5],
        'timestamp': int(time.time() * 1000)
    }
    r.setex(key, 10, json.dumps(data))  # 10秒过期

# 策略读取深度数据
def get_order_book(symbol):
    key = f'orderbook:{symbol}'
    data = r.get(key)
    if data:
        return json.loads(data)
    else:
        # 缓存失效，从WebSocket重新获取
        return fetch_from_websocket(symbol)

2.5 本章知识体系总览

下面这张图，是我自己梳理的数据源接入整体架构。你看一眼，心里就有数了：

这张图把整个流程串起来了：交易所数据通过WebSocket和RESTful两种方式接入，经过API认证后，进入三层缓存体系，最终供策略层使用。说白了，这就是做市系统的数据高速公路。

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