14、实盘交易接口：交易所API对接、签名认证、频率限制处理、错误重试机制

做市系统要真正赚钱，必须连上交易所。这一步，说白了就是让你的程序跟交易所的服务器「对话」。我见过不少团队，策略写得漂亮，结果卡在API对接上，一上线就爆仓。今天咱们就把这块硬骨头啃下来。

14.1 交易所API的基本结构

交易所API通常分两类：REST API和WebSocket API。REST用来下单、查余额，WebSocket用来收实时行情和成交回报。

我个人习惯把API调用封装成三层：

• 传输层：处理HTTP请求、SSL证书、连接池
• 签名层：生成认证签名，保证请求合法
• 业务层：封装具体接口，比如下单、撤单

你想想看，如果这三层混在一起，出问题排查起来有多痛苦？我在项目中遇到过，同事把签名逻辑写在业务代码里，结果换了个交易所，改了一整天。

14.2 签名认证：别让交易所把你拒之门外

交易所的签名认证，说白了就是证明「你是你」。常见的签名方式有HMAC-SHA256和RSA。我建议新手先用HMAC，简单可靠。

签名流程大致如下：

1. 拼接请求参数（时间戳、API Key、请求体等）
2. 用密钥对字符串做HMAC-SHA256哈希
3. 把签名放到请求头里发出去

这里有个坑——时间戳。交易所会校验你的时间戳跟服务器时间差，一般允许5秒内的偏差。我曾经因为服务器时间慢了10秒，所有请求都被拒绝，排查了半小时才发现是NTP服务没开。

代码示例（Python）：

import hmac
import hashlib
import time
import requests

def sign_request(api_key, secret_key, params):
    # 1. 加入时间戳
    params['timestamp'] = int(time.time() * 1000)
    params['api_key'] = api_key
    
    # 2. 按字典序排序参数
    sorted_params = sorted(params.items())
    query_string = '&'.join(f'{k}={v}' for k, v in sorted_params)
    
    # 3. 生成签名
    signature = hmac.new(
        secret_key.encode(),
        query_string.encode(),
        hashlib.sha256
    ).hexdigest()
    
    params['signature'] = signature
    return params

# 使用示例
params = {'symbol': 'BTCUSDT', 'side': 'BUY', 'quantity': 0.01}
signed_params = sign_request('your_api_key', 'your_secret', params)
response = requests.post('https://api.exchange.com/order', json=signed_params)

14.3 频率限制：别把交易所搞崩了

交易所都有频率限制，比如每秒最多10次请求。超过限制，轻则返回429错误，重则封IP。做市系统每秒可能要发几十次请求，所以必须处理限频。

我常用的策略有三种：

策略	适用场景	缺点
令牌桶	请求速率均匀的场景	突发流量可能被截断
滑动窗口	需要精确控制窗口内请求数	实现稍复杂
队列+延迟	对延迟不敏感的操作	可能堆积

我个人偏爱令牌桶。为什么呢？因为它允许短时间内的突发请求，同时保证长期平均速率。做市系统经常需要快速补单，令牌桶正好合适。

令牌桶实现示例：

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个请求，桶容量20
bucket = TokenBucket(rate=10, capacity=20)
if bucket.consume():
    # 发送请求
    pass
else:
    # 等待或丢弃
    time.sleep(0.1)

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

交易所API不可能100%稳定。网络抖动、服务器过载、临时维护，都会导致请求失败。做市系统必须优雅地处理这些错误。

我总结了一套重试策略：

• 可重试的错误：超时、429（限频）、5xx（服务器错误）
• 不可重试的错误：400（参数错误）、401（签名错误）、403（权限不足）

嗯，这里要注意：千万别无脑重试。我曾经见过一个系统，遇到429后疯狂重试，结果被封了IP，整整一天没法交易。

正确的做法是：

1. 第一次失败后，等待1秒重试
2. 第二次失败，等待2秒
3. 第三次失败，等待4秒
4. 最多重试3次，超过就记录日志并报警

这就是指数退避（Exponential Backoff）。加上随机抖动（Jitter），可以避免多个客户端同时重试造成雪崩。

带抖动的重试实现：

import time
import random

def retry_with_backoff(func, max_retries=3, base_delay=1):
    for attempt in range(max_retries):
        try:
            return func()
        except (TimeoutError, ConnectionError) as e:
            if attempt == max_retries - 1:
                raise  # 最后一次失败，直接抛出
            
            # 指数退避 + 随机抖动
            delay = base_delay * (2 ** attempt) + random.uniform(0, 0.5)
            print(f"请求失败，{delay:.2f}秒后重试...")
            time.sleep(delay)
    
    return None

# 使用
def place_order():
    # 实际下单逻辑
    pass

retry_with_backoff(place_order)

14.5 整体架构：把这些串起来

下面这张图展示了API对接的完整流程。你可以看到，请求从业务层出发，经过签名、限频、重试，最后到达交易所。每个环节都有对应的处理逻辑。

你看，整个流程其实不复杂。但每个环节都有细节，忽略任何一个都可能出大问题。我建议你先把签名和限频跑通，再加重试机制。一步一步来，别想一口吃成胖子。

好了，API对接这块就聊到这儿。记住：签名要准，限频要稳，重试要狠。把这三点做好，你的做市系统就成功了一半。

---

交易系统化学习资料 微信Strategy888888