4、交易所API对接：REST API与WebSocket API、API密钥管理、限频处理、错误重试机制

做高频做市，第一关就是跟交易所打交道。说白了，你的策略再牛，连不上交易所、拿不到数据、下不了单，一切都是白搭。我见过太多新手，策略逻辑写得漂漂亮亮，结果一上线就被交易所封了API，或者因为限频没处理好，订单全卡在队列里。

这一章，咱们就把API对接这件事彻底讲透。我会把我在实战中踩过的坑、总结的经验，全都掏出来。

REST API vs WebSocket API：什么时候用哪个？

交易所通常提供两种API：REST和WebSocket。很多人搞不清它们的区别，我简单说下。

REST API，就是请求-响应模式。你发一个HTTP请求，交易所给你返回结果。适合做查询、下单、撤单这类操作。但有个问题——它慢。一次请求怎么也得几十毫秒，高频场景下根本扛不住。

WebSocket API，是长连接模式。你跟交易所建立一条持久通道，数据实时推送过来。适合接收行情、订单状态更新。延迟低，几毫秒就能收到数据。

我个人习惯是：行情用WebSocket，交易用REST。为什么？行情需要实时性，WebSocket天然适合。交易操作（下单、撤单）用REST更稳妥，因为WebSocket的连接状态不稳定，万一断线了，订单状态可能丢失。

核心原则：

  • 行情订阅 → WebSocket
  • 下单/撤单 → REST
  • 账户查询 → REST（偶尔用WebSocket也可以）
  • 订单状态更新 → WebSocket（实时推送）

我在项目中遇到过一个问题：用WebSocket下单，结果网络波动导致连接断开，订单到底成交了没有？完全不知道。后来我改成REST下单，WebSocket只用来收成交回报，问题就解决了。

API密钥管理：别让你的密钥裸奔

密钥泄露，等于你把交易所账户的钥匙交给了别人。我见过有人把API密钥硬编码在代码里，还上传到GitHub……嗯，后果可想而知。

我的做法是这样的：

  1. 环境变量存储：密钥放在环境变量里，代码里用 os.getenv() 读取。绝对不要写死在代码里。
  2. 权限最小化：只给API密钥必要的权限。比如做市策略只需要交易权限，就别开提币权限。
  3. IP白名单：限制API密钥只能从你的服务器IP访问。这样就算密钥泄露，别人也用不了。
  4. 定期轮换：每隔一段时间换一次密钥。我一般一个月换一次。
# 错误示范：密钥硬编码
api_key = "your-api-key-here"
api_secret = "your-api-secret-here"

# 正确做法：从环境变量读取
import os
api_key = os.getenv("EXCHANGE_API_KEY")
api_secret = os.getenv("EXCHANGE_API_SECRET")

警告：千万不要把密钥提交到Git仓库。建议在 .gitignore 里加上密钥文件，或者用 python-dotenv 管理本地环境变量。

限频处理：别被交易所封了

每个交易所都有频率限制。你发请求太快，交易所会直接拒绝，甚至封你的IP。高频做市策略对频率要求很高，所以限频处理是必修课。

常见的限频策略：

交易所 限频规则 我的建议
币安 每IP 1200次/分钟 控制在1000次/分钟以内
OKX 每IP 600次/分钟 控制在500次/分钟以内
Bybit 每IP 600次/分钟 控制在500次/分钟以内

你想想看，如果策略同时跑多个币对，一不小心就超限了。我一般用令牌桶算法来控制请求频率。简单说就是：维护一个桶，桶里有令牌，每次请求消耗一个令牌，令牌按固定速率补充。如果桶空了，就等一会儿再发。

import time
import threading

class TokenBucket:
    def __init__(self, rate, capacity):
        self.rate = rate          # 令牌补充速率（个/秒）
        self.capacity = capacity  # 桶容量
        self.tokens = capacity    # 当前令牌数
        self.last_time = time.time()
        self.lock = threading.Lock()
    
    def consume(self, tokens=1):
        with self.lock:
            now = time.time()
            # 补充令牌
            elapsed = now - self.last_time
            self.tokens = min(self.capacity, 
                             self.tokens + elapsed * self.rate)
            self.last_time = now
            
            if self.tokens >= tokens:
                self.tokens -= tokens
                return True
            else:
                return False

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

def send_request():
    while not bucket.consume():
        time.sleep(0.01)  # 等待10毫秒
    # 发送实际请求
    print("请求已发送")

小技巧：我习惯在每次请求前检查令牌桶，如果桶空了，就sleep一小段时间再重试。这样既不会超限，也不会浪费CPU。

错误重试机制：别让一次失败毁了整个策略

网络波动、交易所维护、请求超时……这些在高频交易中太常见了。如果一次请求失败就放弃，策略的稳定性会非常差。

我的重试策略：

  1. 指数退避：第一次失败等1秒，第二次等2秒，第三次等4秒……最多重试3-5次。
  2. 区分错误类型：网络超时可以重试，但认证失败（比如密钥错误）就不要重试了，重试也没用。
  3. 记录日志：每次重试都记录日志，方便排查问题。
import time
import requests
from requests.exceptions import RequestException

def api_call_with_retry(url, params, max_retries=3):
    for attempt in range(max_retries):
        try:
            response = requests.get(url, params=params, timeout=5)
            response.raise_for_status()
            return response.json()
        except RequestException as e:
            if attempt == max_retries - 1:
                # 最后一次重试失败，抛出异常
                raise
            # 指数退避：等待 1, 2, 4 秒
            wait_time = 2 ** attempt
            print(f"请求失败，{wait_time}秒后重试... 错误：{e}")
            time.sleep(wait_time)

注意：下单操作的重试要特别小心。如果请求超时了，但订单实际上已经成交了，重试可能会导致重复下单。我的做法是：下单前生成一个唯一的client_order_id，重试时带上这个ID，交易所会识别为重复订单并拒绝。

知识体系总览

下面这张图，把本章的核心逻辑串起来了。你可以看到，API对接不是简单的发请求，而是一个完整的体系：从连接方式选择，到密钥安全，再到限频和重试，环环相扣。

交易所API对接知识体系 交易所API对接 API类型选择 REST API（交易） WebSocket（行情） API密钥管理 环境变量存储 权限最小化 IP白名单 + 定期轮换 限频处理 令牌桶算法 控制请求速率 避免被封IP 错误重试机制 指数退避重试 区分错误类型 唯一订单ID防重复

嗯，这一章的内容就这些。API对接看起来琐碎，但每一步都关系到策略的生死。密钥管理不好，钱没了；限频没处理好，策略停了；重试机制没设计好，订单乱成一锅粥。把这些基础打牢，后面的策略开发才能稳扎稳打。

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

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