3. 实时数据源接入：WebSocket、Kafka、REST API，数据解析与缓存策略

做量化交易，最怕什么？

怕行情来了，你的数据还没到。或者更惨——数据到了，你解析错了。

今天咱们聊聊实时数据源接入。这个话题我做了快十年，踩过的坑能写一本小册子。WebSocket、Kafka、REST API，这三种方式我都在生产环境里用过。各有各的脾气，也各有各的坑。

3.1 三种数据源接入方式

先说说这三种方式。我个人习惯把它们分成两类：推模式和拉模式。

方式	模式	延迟	适用场景
WebSocket	推	毫秒级	实时行情、订单簿
Kafka	推/拉	亚毫秒级	内部数据总线、多消费者
REST API	拉	秒级	历史数据、非实时查询

3.1.1 WebSocket：实时行情的首选

WebSocket 说白了就是一条长连接。服务器有数据了，直接推给你。不用你一遍遍去问「有数据了吗？」。

我在项目中遇到过一个问题：某交易所的 WebSocket 连接，每 30 秒会发一个 ping 帧。如果你没及时回复 pong，连接就被断开了。嗯，这个坑我踩过，当时策略跑得好好的，突然断连了，行情停了 3 分钟。

import asyncio
import websockets
import json

async def subscribe_market_data():
    uri = "wss://api.exchange.com/ws"
    async with websockets.connect(uri) as ws:
        # 订阅 BTC/USDT 的 ticker
        subscribe_msg = {
            "type": "subscribe",
            "channel": "ticker",
            "symbol": "BTC/USDT"
        }
        await ws.send(json.dumps(subscribe_msg))
        
        while True:
            try:
                data = await asyncio.wait_for(ws.recv(), timeout=10)
                # 解析数据
                tick = json.loads(data)
                process_tick(tick)
            except asyncio.TimeoutError:
                # 发送心跳
                await ws.send(json.dumps({"type": "ping"}))

3.1.2 Kafka：内部数据总线的王者

Kafka 适合做内部的数据分发。你想想看，一个行情数据进来，可能要同时喂给风控模块、策略模块、日志模块。用 WebSocket 一个个推？太累了。

Kafka 的玩法是：生产者把数据扔到 topic 里，消费者自己去拉。我建议把行情数据按品种分区，比如 BTC 一个分区，ETH 一个分区。这样消费的时候可以并行处理。

from kafka import KafkaConsumer, KafkaProducer
import json

# 生产者：把行情数据发到 Kafka
producer = KafkaProducer(
    bootstrap_servers=['localhost:9092'],
    value_serializer=lambda v: json.dumps(v).encode('utf-8')
)

def publish_tick(tick_data):
    producer.send('market_data', 
                  key=tick_data['symbol'].encode(),
                  value=tick_data)

# 消费者：策略模块消费行情
consumer = KafkaConsumer(
    'market_data',
    bootstrap_servers=['localhost:9092'],
    group_id='strategy_group',
    auto_offset_reset='latest'
)

for msg in consumer:
    tick = json.loads(msg.value)
    # 处理行情
    update_strategy(tick)

3.1.3 REST API：历史数据的救星

REST API 适合拉取历史数据。比如你要回测，需要过去一年的分钟线。用 WebSocket 等？等到猴年马月。

但 REST API 有个问题：限流。交易所通常限制每秒多少次请求。我见过有人写了个 for 循环去拉数据，结果被交易所封了 IP。

import requests
import time
from ratelimit import limits, sleep_and_retry

@sleep_and_retry
@limits(calls=10, period=1)  # 每秒最多10次
def fetch_klines(symbol, interval, limit=500):
    url = f"https://api.exchange.com/v1/klines"
    params = {
        "symbol": symbol,
        "interval": interval,
        "limit": limit
    }
    resp = requests.get(url, params=params)
    if resp.status_code == 429:
        # 被限流了，等一会再试
        time.sleep(1)
        return fetch_klines(symbol, interval, limit)
    return resp.json()

3.2 数据解析策略

数据拿到了，怎么解析？

我个人习惯用 Protocol Buffers 或者 FlatBuffers。为什么？因为 JSON 解析太慢了。你想想看，每秒几千笔行情，每笔都做 json.loads，CPU 很快就满了。

格式	解析速度	数据大小	适用场景
JSON	慢	大	调试、小规模
Protobuf	快	小	生产环境
FlatBuffers	极快	小	高频交易

# 使用 Protobuf 定义行情数据结构
syntax = "proto3";

message Tick {
    string symbol = 1;
    double price = 2;
    double volume = 3;
    int64 timestamp = 4;
    string exchange = 5;
}

# Python 中解析
import market_pb2

def parse_tick(raw_bytes):
    tick = market_pb2.Tick()
    tick.ParseFromString(raw_bytes)
    return tick

3.3 缓存策略

数据解析完了，放哪？

我建议用 Redis 做缓存。为什么？因为快。而且 Redis 支持多种数据结构，非常适合存行情数据。

3.3.1 缓存什么？

• 最新行情：用 Redis 的 String 类型，key 是 "tick:BTC/USDT"，value 是序列化的行情数据
• 订单簿：用 Redis 的 Sorted Set，score 是价格，member 是数量
• K 线数据：用 Redis 的 List，左进右出，保持固定长度

import redis
import json

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

# 缓存最新行情
def cache_tick(tick):
    key = f"tick:{tick['symbol']}"
    r.setex(key, 60, json.dumps(tick))  # 60秒过期

# 缓存订单簿
def cache_orderbook(symbol, bids, asks):
    key = f"orderbook:{symbol}"
    # 清空旧数据
    r.delete(key)
    # 添加买单
    for price, qty in bids:
        r.zadd(f"{key}:bids", {qty: price})
    # 添加卖单
    for price, qty in asks:
        r.zadd(f"{key}:asks", {qty: price})

3.3.2 缓存淘汰策略

缓存不能无限大。我建议用 LRU（最近最少使用）策略。Redis 默认就支持。

# 设置 Redis 最大内存和淘汰策略
# 在 redis.conf 中配置：
# maxmemory 1gb
# maxmemory-policy allkeys-lru

3.4 整体架构图

说了这么多，咱们画个图看看整体架构。

这张图展示了从数据源到应用的完整链路。数据从 WebSocket、Kafka、REST API 进来，经过解析层处理，缓存到 Redis，最后被策略引擎、风控模块等消费。

嗯，今天就聊到这。这些经验都是我用真金白银换来的，希望能帮你少走弯路。