4、金融数据获取：REST API基础、使用CCXT库获取交易所行情、WebSocket实时数据流、数据缓存策略

做市系统说白了就是跟数据打交道。你策略再牛，拿不到行情数据也是白搭。这一章我们聊聊金融数据怎么拿，怎么存，怎么让它又快又稳。

我个人习惯把数据获取分成三层：REST API 做基础查询，WebSocket 做实时推送，缓存层 做性能兜底。这三层缺一不可。

核心观点：数据获取不是简单的"发请求-收响应"，而是一个需要精心设计的管道系统。管道任何一个环节出问题，你的做市策略都会跟着遭殃。

4.1 REST API 基础

REST API 是啥？说白了就是 HTTP 请求。你给交易所服务器发个 GET 请求，它给你返回 JSON 数据。简单、直接、容易理解。

但这里有个坑——频率限制。我在项目中遇到过，刚开始写代码时疯狂请求，结果 IP 直接被交易所封了半小时。嗯，那半小时我的策略完全瞎了。

常见的 REST API 端点包括：

  • GET /api/v3/ticker/price — 获取最新价格
  • GET /api/v3/depth — 获取订单簿深度
  • GET /api/v3/klines — 获取K线数据
  • GET /api/v3/trades — 获取近期成交记录

你想想看，这些接口返回的数据结构其实大同小异。但不同交易所的字段名、限频规则、认证方式都不一样。这就是为什么我们需要 CCXT 这样的统一接口库。

小技巧：写 REST 客户端时，一定要加重试机制。网络抖动是常态，不是异常。我一般用指数退避策略，第一次失败等1秒，第二次等2秒，第三次等4秒，最多重试3次。

4.2 使用 CCXT 库获取交易所行情

CCXT 是我最常用的加密货币交易所统一接口库。它支持 100+ 交易所，API 风格一致。你换交易所只需要改一行代码，不用重写整个数据获取模块。

安装很简单：

pip install ccxt

来看一个获取 Binance 行情的小例子：

import ccxt

# 初始化交易所对象
exchange = ccxt.binance({
    'rateLimit': 1200,  # 限频控制，毫秒
    'enableRateLimit': True,  # 自动限频
})

# 获取 BTC/USDT 的最新 ticker
ticker = exchange.fetch_ticker('BTC/USDT')
print(f"最新价: {ticker['last']}")
print(f"24h涨跌: {ticker['percentage']}%")

# 获取订单簿深度
orderbook = exchange.fetch_order_book('BTC/USDT', limit=10)
print(f"买一价: {orderbook['bids'][0][0]}, 量: {orderbook['bids'][0][1]}")
print(f"卖一价: {orderbook['asks'][0][0]}, 量: {orderbook['asks'][0][1]}")

为什么推荐 CCXT？因为它帮你处理了：

  • 认证签名 — 私有接口的 HMAC 签名自动完成
  • 限频控制 — 内置请求队列，不会触发封禁
  • 数据标准化 — 不同交易所返回统一格式的数据
  • 错误处理 — 网络异常、交易所维护等场景都有对应异常类

注意：CCXT 的 fetch_ticker 返回的是字典，但不同交易所的字段可能略有差异。比如有些交易所返回 'last'，有些返回 'close'。建议在代码里做一层适配，统一成自己系统的内部格式。

4.3 WebSocket 实时数据流

REST API 有个致命问题——延迟。你轮询一次需要几百毫秒，这期间价格可能已经变了好几次。做市交易对延迟极其敏感，所以必须上 WebSocket。

WebSocket 是长连接，交易所主动推送数据给你。延迟通常在 10-50ms 级别，比 REST 快一个数量级。

我曾经踩过一个坑：WebSocket 连接断了，程序没检测到，结果策略一直用旧数据做市，亏了不少。后来我加了个心跳检测，每 5 秒检查一次连接状态，断了就自动重连。

来看一个用 websockets 库连接 Binance 的例子：

import asyncio
import websockets
import json

async def subscribe_depth(symbol='btcusdt'):
    uri = "wss://stream.binance.com:9443/ws"
    async with websockets.connect(uri) as ws:
        # 订阅订单簿深度数据
        subscribe_msg = {
            "method": "SUBSCRIBE",
            "params": [f"{symbol}@depth20@100ms"],
            "id": 1
        }
        await ws.send(json.dumps(subscribe_msg))
        
        while True:
            data = await ws.recv()
            depth = json.loads(data)
            # 处理深度数据...
            print(f"收到深度更新: {depth['bids'][:3]}")

# 运行
asyncio.run(subscribe_depth())

WebSocket 数据流的核心要点：

  • 订阅管理 — 连接建立后先发订阅消息，告诉交易所你要哪些数据
  • 心跳保活 — 定期发送 ping 帧，防止连接被断开
  • 断线重连 — 必须实现自动重连逻辑，网络不可能永远稳定
  • 数据去重 — 重连后可能收到重复数据，需要做去重处理

经验之谈：WebSocket 收到的数据是增量更新的。比如订单簿，交易所只推送变化的部分，你需要自己维护一个本地订单簿快照。我习惯用红黑树或者跳表来维护，查询和更新都是 O(log n)。

4.4 数据缓存策略

数据缓存是容易被忽视但极其重要的一环。你想想看，如果 WebSocket 断了，REST API 又限频，你的策略拿什么做决策？

缓存策略分三层：

层级 存储位置 适用数据 过期时间
L1 内存缓存 Python 字典 / Redis 最新价格、订单簿快照 秒级
L2 本地缓存 SQLite / 文件 K线数据、历史成交 分钟级
L3 持久化存储 PostgreSQL / InfluxDB 全量历史数据 永久

我个人习惯用 Redis 做 L1 缓存。为什么？因为它快，而且支持数据结构丰富。比如订单簿可以用 Redis 的 Sorted Set 来存，价格作为 score，数量作为 value，查询和更新都很快。

来看一个简单的缓存实现思路：

import redis
import json

class MarketDataCache:
    def __init__(self):
        self.redis = redis.Redis(host='localhost', port=6379, db=0)
        
    def update_ticker(self, symbol, ticker):
        # 缓存最新 ticker，过期时间 5 秒
        key = f"ticker:{symbol}"
        self.redis.setex(key, 5, json.dumps(ticker))
        
    def get_ticker(self, symbol):
        key = f"ticker:{symbol}"
        data = self.redis.get(key)
        if data:
            return json.loads(data)
        return None
        
    def update_depth(self, symbol, depth):
        # 用 Sorted Set 存储订单簿
        bid_key = f"depth:{symbol}:bids"
        ask_key = f"depth:{symbol}:asks"
        
        # 清空旧数据
        self.redis.delete(bid_key, ask_key)
        
        # 写入新数据
        for price, qty in depth['bids']:
            self.redis.zadd(bid_key, {price: qty})
        for price, qty in depth['asks']:
            self.redis.zadd(ask_key, {price: qty})
            
        # 设置过期时间 10 秒
        self.redis.expire(bid_key, 10)
        self.redis.expire(ask_key, 10)

避坑指南：缓存一定要设置过期时间！我曾经犯过一个错误——缓存永不过期，结果交易所数据变了，我的策略还在用旧数据做市。那天的回撤我现在还记得。过期时间要根据数据更新频率来定，ticker 可以短一点（1-5秒），K线可以长一点（几分钟）。

本章小结

数据获取是做市系统的地基。REST API 负责基础查询，WebSocket 负责实时推送，缓存层负责性能和容错。这三者配合好了，你的策略才能稳定运行。

最后分享一个我自己的数据获取架构图，你可以参考这个思路来设计：

做市系统数据获取架构 交易所数据源 Binance | OKX | Bybit | Coinbase | ... 数据获取层 REST API (轮询) WebSocket (实时) CCXT 统一接口 数据缓存层 L1: Redis 内存缓存 L2: SQLite 本地缓存 L3: 持久化存储 数据流向：交易所 → 获取层 → 缓存层 → 策略引擎

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

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