📚 课程目录第3章实时数据源接入

3. 实时数据源接入：WebSocket、Kafka、REST API 实时数据获取方式对比与代码实现

实时风控特征工程的第一步是稳定、低延迟地接入市场数据源。不同的数据源协议在延迟、吞吐量、连接方式和资源消耗上差异显著。本章从工程实践角度对比三种主流实时数据接入方式：WebSocket（全双工推送）、Kafka（分布式消息队列）和 REST API（轮询拉取），并给出 Python 核心代码实现。

  核心原则： 风控特征计算对数据新鲜度要求极高（毫秒～秒级），接入层必须优先选择 推送模式（WebSocket / Kafka），避免高频轮询带来的延迟和资源浪费。

3.1 三种接入方式对比总览

对比维度	WebSocket	Kafka	REST API
通信模式	全双工推送	发布/订阅（推送）	请求/响应（轮询拉取）
延迟	毫秒级（极低）	毫秒～亚秒级	秒级（取决于轮询间隔）
吞吐量	中等（单连接）	极高（分区并行）	低（受限于 HTTP 开销）
连接开销	一次握手，长连接	长连接 + 消费者组	每次请求新建连接
数据持久化	无（内存流）	有（磁盘持久化）	无（依赖上游）
适用场景	行情快照、订单簿增量	多消费者、历史回放、高吞吐	低频查询、非实时校验

选型建议： 实时风控特征管道推荐 Kafka 作为主干（高吞吐、可回放），WebSocket 用于 边缘节点直连交易所，REST API 仅用于 启动时补数据或非关键校验。

3.2 实时数据接入架构总览

下图展示了三种接入方式在实时风控特征工程中的典型位置与数据流向：

3.3 WebSocket 实时行情接入

WebSocket 是实时风控中最常用的推送协议。交易所通常提供 深度增量（depth update）和 成交明细（trade）两个主要频道。

3.3.1 核心代码实现

import asyncio
import json
import websockets
from typing import Callable, Dict, Any

class WebSocketMarketClient:
    """异步 WebSocket 行情客户端，支持自动重连与心跳"""

    def __init__(self, url: str, channels: list, callback: Callable):
        self.url = url
        self.channels = channels
        self.callback = callback
        self.ws = None
        self._running = False

    async def _subscribe(self):
        """订阅频道"""
        for ch in self.channels:
            msg = {"op": "subscribe", "channel": ch}
            await self.ws.send(json.dumps(msg))

    async def _heartbeat(self):
        """每 15 秒发送心跳"""
        while self._running:
            await asyncio.sleep(15)
            try:
                await self.ws.send(json.dumps({"op": "ping"}))
            except:
                break

    async def connect(self):
        """建立连接并持续接收消息"""
        self._running = True
        while self._running:
            try:
                async with websockets.connect(self.url) as ws:
                    self.ws = ws
                    await self._subscribe()
                    # 启动心跳任务
                    asyncio.create_task(self._heartbeat())
                    async for message in ws:
                        data = json.loads(message)
                        await self.callback(data)
            except Exception as e:
                print(f"连接断开，5秒后重连: {e}")
                await asyncio.sleep(5)

    def stop(self):
        self._running = False

# 使用示例
async def handle_tick(data: Dict[str, Any]):
    """回调函数：将行情数据送入特征管道"""
    if data.get("type") == "trade":
        price = float(data["price"])
        volume = float(data["volume"])
        # 此处可触发实时特征计算
        print(f"最新成交: {price} @ {volume}")

client = WebSocketMarketClient(
    url="wss://stream.example.com/market",
    channels=["trade.BTC-USDT", "depth.BTC-USDT"],
    callback=handle_tick
)

# 启动
asyncio.run(client.connect())

避坑提示： WebSocket 连接必须实现 心跳保活 和 自动重连 逻辑。交易所通常 30～60 秒无消息即断开连接。此外，回调函数内不要做阻塞操作，应使用异步队列将数据交给下游处理。

3.4 Kafka 高吞吐数据管道

当系统需要同时服务多个风控策略、多个特征计算节点时，Kafka 是首选。Kafka 提供 分区并行消费、消息持久化 和 消费者组负载均衡。

3.4.1 生产者：将行情写入 Kafka

from kafka import KafkaProducer
import json
import time

producer = KafkaProducer(
    bootstrap_servers=['kafka1:9092', 'kafka2:9092'],
    value_serializer=lambda v: json.dumps(v).encode('utf-8'),
    acks='all',          # 等待所有副本确认
    retries=3,
    linger_ms=5,         # 批量发送延迟，平衡延迟与吞吐
    batch_size=32768     # 32KB 批量
)

def publish_tick(symbol: str, price: float, volume: float, ts: int):
    """将行情 tick 发布到 Kafka topic"""
    msg = {
        "symbol": symbol,
        "price": price,
        "volume": volume,
        "timestamp": ts,
        "source": "websocket"
    }
    future = producer.send('market-ticks', key=symbol.encode(), value=msg)
    # 可选：阻塞等待发送结果
    record_metadata = future.get(timeout=5)
    return record_metadata

3.4.2 消费者：实时特征计算

from kafka import KafkaConsumer
import json

consumer = KafkaConsumer(
    'market-ticks',
    bootstrap_servers=['kafka1:9092'],
    group_id='risk-feature-group',
    auto_offset_reset='latest',
    enable_auto_commit=True,
    auto_commit_interval_ms=1000,
    value_deserializer=lambda m: json.loads(m.decode('utf-8'))
)

def consume_ticks():
    """持续消费行情消息，计算实时特征"""
    for message in consumer:
        tick = message.value
        symbol = tick['symbol']
        price = tick['price']
        # 示例：计算 1 分钟价格波动率特征
        # 实际应使用滑动窗口数据结构
        print(f"[{symbol}] 价格: {price} | 分区: {message.partition} | 偏移: {message.offset}")

工程要点： Kafka 消费者应设置 enable_auto_commit=False 并在特征计算成功后再手动提交偏移量，避免数据丢失。风控场景下，至少一次语义（at-least-once）比精确一次更实用。

3.5 REST API 轮询接入

REST API 适用于 非关键数据 或 启动时补全历史数据。轮询间隔通常 ≥ 1 秒，不适合高频特征。

3.5.1 带限速的异步轮询

import aiohttp
import asyncio
from typing import Dict, Any

class RestPollingClient:
    """异步 REST 轮询客户端，内置限速与指数退避"""

    def __init__(self, base_url: str, interval: float = 1.0):
        self.base_url = base_url
        self.interval = interval
        self.session = None

    async def __aenter__(self):
        self.session = aiohttp.ClientSession()
        return self

    async def __aexit__(self, *args):
        await self.session.close()

    async def poll_once(self, endpoint: str) -> Dict[str, Any]:
        """单次轮询请求"""
        url = f"{self.base_url}/{endpoint}"
        async with self.session.get(url, timeout=5) as resp:
            if resp.status == 429:  # 限速
                retry_after = int(resp.headers.get('Retry-After', 5))
                await asyncio.sleep(retry_after)
                return await self.poll_once(endpoint)
            resp.raise_for_status()
            return await resp.json()

    async def poll_loop(self, endpoint: str, callback):
        """持续轮询循环"""
        while True:
            try:
                data = await self.poll_once(endpoint)
                await callback(data)
            except Exception as e:
                print(f"轮询异常: {e}")
            await asyncio.sleep(self.interval)

# 使用示例
async def main():
    async with RestPollingClient("https://api.example.com/v1", interval=2.0) as client:
        await client.poll_loop("ticker/BTCUSDT", lambda d: print(d))

asyncio.run(main())

避坑提示： REST 轮询必须处理 限速（429）、连接超时 和 数据去重。轮询间隔不要短于交易所允许的最小间隔，否则会被封 IP。建议使用 指数退避 策略应对限速。

3.6 三种接入方式选型决策树

以下决策树帮助你在实际项目中快速选择合适的数据接入方式：

3.7 本章小结

WebSocket：延迟最低，适合直连交易所获取行情快照和增量数据。必须实现心跳与重连。
Kafka：吞吐量最高，支持多消费者并行消费和消息回放，是实时风控特征管道的中枢。
REST API：实现简单，但延迟高、资源浪费大，仅用于非关键场景或启动补数据。
实际生产环境通常采用 WebSocket → Kafka → 特征计算 的混合架构，兼顾低延迟与高可用。