4、交易所API对接：REST API与WebSocket API的区别、API密钥管理与安全、限频与重试机制

做市商系统跟交易所打交道，说白了就两条路：REST API 和 WebSocket API。这两兄弟性格完全不同，用错了地方，轻则延迟爆炸，重则直接被交易所封号。我刚开始做量化那会儿，就吃过这个亏——全用 REST 去抢行情，结果还没成交，价格已经变了两轮。

4.1 REST API vs WebSocket API：到底该用谁？

先看一张对比图，帮你快速建立直觉：

4.2 什么时候用 REST？什么时候用 WebSocket？

我个人习惯这样分：「查一次」用 REST，「一直查」用 WebSocket。

• 下单、撤单、查余额——这些操作频率低，用 REST 完全够。你想想看，一秒下个几单，REST 的延迟完全能接受。
• 深度行情、成交推送——这些必须用 WebSocket。我在项目中遇到过，用 REST 轮询深度，每 200ms 拉一次，结果交易所直接给我限频了，而且拿到的数据永远慢半拍。

4.3 API 密钥管理与安全

密钥泄露，等于把交易所账户的钥匙交给别人。我见过不止一个团队，因为密钥硬编码在代码里，被实习生 push 到 GitHub 上，几分钟内账户就被洗劫一空。

4.3.1 密钥存储：别偷懒

• 环境变量：最基础的做法。用 os.environ 读取，不要写死在代码里。
• 配置文件加密：用 cryptography 库加密配置文件，运行时解密。
• 密钥管理服务：生产环境建议用 HashiCorp Vault 或 AWS Secrets Manager。

# 错误示范：硬编码
api_key = "abc123def456"  # 千万别这么干！

# 正确做法：环境变量
import os
api_key = os.environ.get("EXCHANGE_API_KEY")
api_secret = os.environ.get("EXCHANGE_API_SECRET")

4.3.2 签名机制：每个请求都要「验明正身」

交易所通常要求对请求进行 HMAC-SHA256 签名。嗯，这里要注意：签名参数必须包含时间戳和随机数，防止重放攻击。

import time
import hmac
import hashlib
import requests

def sign_request(api_secret, method, path, params):
    timestamp = str(int(time.time() * 1000))
    # 拼接签名内容
    message = timestamp + method + path + params
    signature = hmac.new(
        api_secret.encode(),
        message.encode(),
        hashlib.sha256
    ).hexdigest()
    return timestamp, signature

# 示例：发送签名后的请求
timestamp, sig = sign_request(api_secret, "POST", "/api/v1/order", "symbol=BTCUSDT&side=BUY")
headers = {
    "X-API-KEY": api_key,
    "X-TIMESTAMP": timestamp,
    "X-SIGNATURE": sig
}
response = requests.post("https://api.exchange.com/api/v1/order", headers=headers)

4.4 限频与重试机制

交易所不是你家开的，每个 API 都有频率限制。做市商策略每秒可能发几十个请求，一不小心就被封了。

4.4.1 限频策略：别硬闯

限频类型	说明	应对策略
IP 限频	每个 IP 每秒/每分钟允许的请求数	使用多个 IP 或代理
API Key 限频	每个密钥的请求配额	控制请求速率，使用令牌桶算法
权重限频	不同接口消耗不同权重	优先使用低权重接口

我建议用 令牌桶算法 来控制请求速率。说白了就是：桶里有一定数量的令牌，每发一个请求消耗一个令牌，令牌按固定速率补充。

import time
import threading

class TokenBucket:
    def __init__(self, rate, capacity):
        self.rate = rate          # 每秒补充的令牌数
        self.capacity = capacity  # 桶容量
        self.tokens = capacity
        self.last_refill = time.time()
        self.lock = threading.Lock()

    def consume(self, tokens=1):
        with self.lock:
            now = time.time()
            # 补充令牌
            elapsed = now - self.last_refill
            self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
            self.last_refill = now

            if self.tokens >= tokens:
                self.tokens -= tokens
                return True
            return False

# 使用示例：每秒最多 10 个请求
bucket = TokenBucket(rate=10, capacity=10)

def safe_request(url, headers):
    while not bucket.consume():
        time.sleep(0.05)  # 等待 50ms 再试
    return requests.get(url, headers=headers)

4.4.2 重试机制：优雅地处理失败

网络波动、交易所临时限频，这些都会导致请求失败。直接重试？不行，会加重问题。你需要一个 指数退避 + 随机抖动 的重试策略。

import time
import random
import requests

def retry_request(url, headers, max_retries=5):
    for attempt in range(max_retries):
        try:
            response = requests.get(url, headers=headers, timeout=5)
            if response.status_code == 429:  # 限频
                wait = (2 ** attempt) + random.uniform(0, 1)
                print(f"限频了，等待 {wait:.2f} 秒后重试")
                time.sleep(wait)
                continue
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            if attempt == max_retries - 1:
                raise e
            wait = (2 ** attempt) + random.uniform(0, 1)
            print(f"请求失败，{wait:.2f} 秒后重试: {e}")
            time.sleep(wait)
    return None

4.5 WebSocket 连接管理：心跳与重连

WebSocket 是长连接，但网络不稳定时会断开。你需要自己维护心跳和重连逻辑。

import websocket
import json
import threading
import time

class WebSocketClient:
    def __init__(self, url, ping_interval=30):
        self.url = url
        self.ping_interval = ping_interval
        self.ws = None
        self.running = False

    def on_message(self, ws, message):
        data = json.loads(message)
        # 处理行情数据
        print(f"收到行情: {data}")

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

    def on_close(self, ws, close_status_code, close_msg):
        print("连接关闭，准备重连...")
        self.reconnect()

    def on_open(self, ws):
        print("连接成功")
        # 订阅行情
        subscribe_msg = {
            "op": "subscribe",
            "channel": "depth",
            "symbol": "BTCUSDT"
        }
        ws.send(json.dumps(subscribe_msg))

    def connect(self):
        self.ws = websocket.WebSocketApp(
            self.url,
            on_message=self.on_message,
            on_error=self.on_error,
            on_close=self.on_close,
            on_open=self.on_open
        )
        self.running = True
        # 启动心跳线程
        threading.Thread(target=self.heartbeat, daemon=True).start()
        self.ws.run_forever()

    def reconnect(self):
        if not self.running:
            return
        time.sleep(5)  # 等待 5 秒后重连
        print("正在重连...")
        self.connect()

    def heartbeat(self):
        while self.running:
            time.sleep(self.ping_interval)
            if self.ws and self.ws.sock:
                try:
                    self.ws.send(json.dumps({"op": "ping"}))
                except:
                    pass

    def close(self):
        self.running = False
        if self.ws:
            self.ws.close()

4.6 实战建议：做市商系统的 API 架构

最后，给你一个我实际在用的架构参考：

• 行情层：2~3 条 WebSocket 连接，分别订阅深度、成交、K线。每条连接独立心跳。
• 交易层：REST API + 令牌桶限频。下单、撤单、查余额走 REST。
• 监控层：WebSocket 连接状态监控，断线自动报警。
• 重试层：所有 REST 请求统一走重试装饰器，指数退避 + 抖动。

嗯，API 对接这块，说白了就是「稳」字当头。宁可慢一点，也别让交易所把你封了。我见过太多团队，策略写得漂亮，结果 API 层一塌糊涂，上线第一天就被限频，血亏。

---