第4章 HTTP客户端封装

4、HTTP客户端封装：基于aiohttp/httpx构建可复用的异步HTTP客户端

做量化交易，说白了就是跟交易所的API打交道。你想想看，一个策略同时跑在币安、OKX、Bybit上，每个交易所的请求频率限制、超时设置、错误码都不一样。如果每个交易所都单独写一套HTTP请求逻辑，那代码维护起来简直是一场噩梦。

我个人习惯是，先封装一个通用的异步HTTP客户端，把超时、重试、限流这些基础设施一次性搞定。这样后面对接任何交易所，只需要配置不同的参数就行。今天我们就来聊聊怎么用aiohttp和httpx这两个库，构建一个真正可复用的客户端。

4.1 为什么选择异步HTTP？

很多新手会问：「同步请求不行吗？」嗯，如果你只对接一个交易所，同步确实够用。但量化交易系统通常要同时监控多个市场，发送多个订单。同步请求会阻塞整个事件循环，你想想看，一个请求卡住5秒，其他所有交易都得等着。

我在项目中遇到过最典型的情况：同时监控10个交易对，每个交易对需要订阅深度数据。如果用同步方式，光数据采集就得开10个线程，线程切换的开销非常大。换成异步后，一个事件循环就能搞定，CPU占用直接降了70%。

核心原则：异步HTTP是量化交易系统的标配。aiohttp和httpx都支持异步，但底层实现略有不同。aiohttp更轻量，httpx功能更全（支持HTTP/2）。我个人建议：如果主要对接REST API，用aiohttp；如果需要HTTP/2或更丰富的特性，选httpx。

4.2 基础客户端封装

我们先从最基础的开始。一个可复用的HTTP客户端，至少需要处理三件事：超时控制、请求重试、限流保护。下面是我在实际项目中用的模板，你可以直接拿来改。

import asyncio
import aiohttp
from typing import Optional, Dict, Any
from tenacity import retry, stop_after_attempt, wait_exponential

class AsyncHttpClient:
    """可复用的异步HTTP客户端"""
    
    def __init__(self, 
                 base_url: str,
                 timeout: int = 10,
                 max_retries: int = 3,
                 rate_limit: int = 10):
        self.base_url = base_url
        self.timeout = aiohttp.ClientTimeout(total=timeout)
        self.max_retries = max_retries
        self.rate_limit = rate_limit
        self._session: Optional[aiohttp.ClientSession] = None
        self._rate_limiter = asyncio.Semaphore(rate_limit)
        
    async def __aenter__(self):
        self._session = aiohttp.ClientSession(
            timeout=self.timeout,
            headers={'Content-Type': 'application/json'}
        )
        return self
        
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if self._session:
            await self._session.close()
            
    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=1, max=10)
    )
    async def request(self, 
                      method: str, 
                      endpoint: str,
                      params: Optional[Dict] = None,
                      data: Optional[Dict] = None) -> Dict[str, Any]:
        """带重试和限流的请求方法"""
        async with self._rate_limiter:
            url = f"{self.base_url}{endpoint}"
            async with self._session.request(
                method, url, params=params, json=data
            ) as response:
                response.raise_for_status()
                return await response.json()

小技巧：用tenacity库做重试，比手写while循环优雅得多。我一般设置指数退避，第一次重试等1秒，第二次等2秒，第三次等4秒。这样既不会把交易所打爆，也不会因为短暂网络波动就放弃。

4.3 超时处理的坑

超时设置看起来简单，但坑不少。我记得有一次，某个交易所的深度接口偶尔会卡住30秒，我设置的超时是5秒。结果呢？每次超时后重试，连续3次都超时，订单直接错过了最佳成交时机。

为什么会这样？因为我把超时设得太死了。交易所的深度接口在行情剧烈波动时，响应时间会显著增加。后来我改成动态超时：正常情况用5秒，如果连续两次超时，自动放宽到15秒。

class AdaptiveTimeout:
    """自适应超时管理器"""
    
    def __init__(self, base_timeout: int = 5, max_timeout: int = 30):
        self.base_timeout = base_timeout
        self.max_timeout = max_timeout
        self._consecutive_timeouts = 0
        
    def get_timeout(self) -> int:
        if self._consecutive_timeouts >= 2:
            return min(self.base_timeout * 2, self.max_timeout)
        return self.base_timeout
        
    def record_timeout(self):
        self._consecutive_timeouts += 1
        
    def record_success(self):
        self._consecutive_timeouts = max(0, self._consecutive_timeouts - 1)

注意：超时不是越大越好。如果交易所真的挂了，你等30秒也是白等。我一般设一个上限，比如30秒，超过这个时间直接放弃，切换到备用交易所。

4.4 限流策略实战

限流是量化交易中最容易被忽视的环节。每个交易所的限流规则都不一样：币安是每分钟1200次请求，OKX是每秒100次，Bybit是每分钟600次。如果你不控制请求频率，轻则被限流，重则被封号。

我常用的限流方案有两种：令牌桶和滑动窗口。令牌桶适合突发流量，滑动窗口适合平滑请求。对于交易所API，我建议用滑动窗口，因为交易所的限流策略本身就是基于滑动窗口的。

import time
from collections import deque

class SlidingWindowRateLimiter:
    """滑动窗口限流器"""
    
    def __init__(self, max_requests: int, window_size: int = 60):
        self.max_requests = max_requests
        self.window_size = window_size
        self._requests = deque()
        
    async def acquire(self):
        now = time.time()
        # 移除窗口外的请求记录
        while self._requests and self._requests[0] < now - self.window_size:
            self._requests.popleft()
            
        if len(self._requests) >= self.max_requests:
            # 计算需要等待的时间
            wait_time = self._requests[0] + self.window_size - now
            if wait_time > 0:
                await asyncio.sleep(wait_time)
                
        self._requests.append(now)

避坑指南：我曾经因为限流没处理好，被币安封了IP。原因是我的策略在开仓时同时发送了多个订单，瞬间超过了限流阈值。后来我加了一个「预热」机制：在真正交易前，先发几个小请求测试限流状态。如果发现接近阈值，就主动降低频率。

4.5 多交易所路由适配

有了通用的HTTP客户端，接下来就是路由适配了。每个交易所的API路径、签名方式、错误码都不一样。我的做法是定义一个抽象基类，然后每个交易所实现自己的适配器。

from abc import ABC, abstractmethod

class BaseExchangeAdapter(ABC):
    """交易所适配器基类"""
    
    def __init__(self, config: Dict):
        self.config = config
        self.client = AsyncHttpClient(
            base_url=config['base_url'],
            timeout=config.get('timeout', 10),
            rate_limit=config.get('rate_limit', 10)
        )
        
    @abstractmethod
    async def get_ticker(self, symbol: str) -> Dict:
        """获取最新行情"""
        pass
        
    @abstractmethod
    async def place_order(self, order: Dict) -> Dict:
        """下单"""
        pass
        
    @abstractmethod
    def sign_request(self, params: Dict) -> Dict:
        """签名请求"""
        pass

class BinanceAdapter(BaseExchangeAdapter):
    """币安适配器"""
    
    async def get_ticker(self, symbol: str) -> Dict:
        return await self.client.request(
            'GET', f'/api/v3/ticker/price',
            params={'symbol': symbol}
        )
        
    def sign_request(self, params: Dict) -> Dict:
        # 币安签名逻辑
        params['timestamp'] = int(time.time() * 1000)
        query_string = '&'.join([f"{k}={v}" for k, v in params.items()])
        params['signature'] = hmac.new(
            self.config['secret_key'].encode(),
            query_string.encode(),
            hashlib.sha256
        ).hexdigest()
        return params

4.6 知识体系总览

下面这张图总结了本章的核心逻辑。你可以看到，从底层HTTP客户端到上层交易所适配器，每一层都有明确的职责。

4.7 实战建议

最后，给你几个实战中的建议：

不要复用session：每个交易所应该有自己的session实例，避免请求互相干扰。我见过有人用一个session对接所有交易所，结果一个交易所的限流影响了其他交易所的请求。
日志要详细：记录每次请求的URL、耗时、状态码、错误信息。这样排查问题的时候，一眼就能看出是网络问题还是交易所问题。
测试要全面：模拟超时、限流、错误码等各种异常情况。我曾经因为没测试限流场景，上线第一天就被交易所封了IP。
监控要到位：对每个交易所的请求成功率、平均耗时、限流次数做监控。一旦发现异常，及时告警。

个人经验：我习惯在客户端里加一个「健康检查」方法。每隔30秒，向交易所发送一个轻量级请求（比如查询服务器时间）。如果连续3次失败，就自动切换到备用交易所。这个机制帮我避免了好几次因为交易所宕机导致的损失。

好了，关于HTTP客户端的封装就聊到这里。记住，好的基础设施能让你在对接新交易所时事半功倍。下一节我们会深入具体的交易所API对接细节，到时候这些封装就能派上大用场了。

📚 课程目录

4、HTTP客户端封装：基于aiohttp/httpx构建可复用的异步HTTP客户端

4.1 为什么选择异步HTTP？

4.2 基础客户端封装

4.3 超时处理的坑

4.4 限流策略实战

4.5 多交易所路由适配

4.6 知识体系总览

4.7 实战建议