第1章：数据采集基础

做量化交易，第一步就是拿数据。这事儿听起来简单，但坑特别多。

我刚开始做量化那会儿，觉得数据采集嘛，不就是发个HTTP请求吗？结果第一个月就被交易所封了三次IP。后来才明白，这里面的门道比想象中深得多。

1.1 requests库：你的数据抓手

Python里发HTTP请求，requests库是标配。它简单、直观，几行代码就能拿到数据。

import requests

url = "https://api.binance.com/api/v3/ticker/price"
params = {"symbol": "BTCUSDT"}

response = requests.get(url, params=params)
data = response.json()
print(data)

这段代码干了什么？它向币安的公开接口请求了BTC/USDT的当前价格。返回的是JSON格式，直接解析就能用。

response = requests.get(url, timeout=5)  # 5秒超时

1.2 API鉴权：你的身份证明

公开数据不需要鉴权，但你要拿账户信息、下订单，就必须证明你是谁。

最常见的鉴权方式就是API Key + Secret Key。交易所给你一对密钥，你发请求时带上它们。

鉴权的流程一般是这样的：

1. 把请求参数按字典序排序
2. 拼接成字符串
3. 用Secret Key做HMAC-SHA256签名
4. 把签名放在请求头里

import hmac
import hashlib
import time

api_key = "你的API_KEY"
secret_key = "你的SECRET_KEY"

timestamp = int(time.time() * 1000)
params = {
    "symbol": "BTCUSDT",
    "timestamp": timestamp
}

# 排序并拼接
query_string = "&".join([f"{k}={v}" for k, v in sorted(params.items())])

# 生成签名
signature = hmac.new(
    secret_key.encode(),
    query_string.encode(),
    hashlib.sha256
).hexdigest()

params["signature"] = signature

headers = {"X-MBX-APIKEY": api_key}
response = requests.get("https://api.binance.com/api/v3/account", 
                       headers=headers, params=params)

嗯，这里要注意：时间戳必须和服务器时间同步。偏差太大，交易所会直接拒绝你的请求。

1.3 请求频率控制：别把自己玩死

交易所不是你家开的。你发请求太快，它会限流，甚至封你。

每个交易所的限流规则不一样。币安是每分钟1200次权重请求，火币是100次/秒。你得看文档。

交易所	限流规则	我的建议
币安	1200权重/分钟	控制在1000以内，留点余量
火币	100次/秒	每秒不超过80次
OKX	20次/秒（WebSocket）	用WebSocket代替REST

怎么控制频率？最简单的办法就是加延时。

import time

def rate_limited_request(url, params, delay=0.1):
    """每次请求后等delay秒"""
    response = requests.get(url, params=params)
    time.sleep(delay)
    return response

但这样太粗糙了。我一般用令牌桶算法，更精确。

import time
from collections import deque

class TokenBucket:
    def __init__(self, rate, capacity):
        self.rate = rate  # 每秒生成令牌数
        self.capacity = capacity  # 桶容量
        self.tokens = capacity
        self.last_refill = time.time()
    
    def consume(self, tokens=1):
        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

# 使用示例
bucket = TokenBucket(rate=10, capacity=20)  # 每秒10个，突发20个

def safe_request(url, params):
    while not bucket.consume():
        time.sleep(0.01)  # 等10毫秒再试
    return requests.get(url, params=params)

1.4 错误重试机制：别轻易放弃

网络请求总会出错。超时、连接重置、服务器500...你想想看，这些情况太常见了。

我的做法是：遇到错误别马上放弃，重试几次。但重试不能太频繁，否则就是给交易所添乱。

指数退避（Exponential Backoff）是标准做法。第一次等1秒，第二次等2秒，第三次等4秒，以此类推。

import time
import random

def retry_with_backoff(func, max_retries=5, base_delay=1):
    """带指数退避的重试机制"""
    for attempt in range(max_retries):
        try:
            return func()
        except Exception as e:
            if attempt == max_retries - 1:
                raise  # 最后一次还失败，直接抛异常
            
            # 计算等待时间：指数增长 + 随机抖动
            delay = base_delay * (2 ** attempt) + random.uniform(0, 0.5)
            print(f"请求失败，{delay:.2f}秒后重试... (第{attempt+1}次)")
            time.sleep(delay)
    
    return None

# 使用示例
def fetch_price():
    url = "https://api.binance.com/api/v3/ticker/price"
    params = {"symbol": "BTCUSDT"}
    response = requests.get(url, params=params, timeout=5)
    response.raise_for_status()  # 非200状态码会抛异常
    return response.json()

price = retry_with_backoff(fetch_price)

为什么要加随机抖动？因为如果多个客户端同时重试，它们可能会在同一时间发起请求，造成"惊群效应"。加一点随机性，能分散请求压力。

1.5 完整的数据采集流程

把上面这些整合起来，就是一个健壮的数据采集器了。

class DataCollector:
    def __init__(self, api_key, secret_key, rate_limit=10):
        self.api_key = api_key
        self.secret_key = secret_key
        self.bucket = TokenBucket(rate=rate_limit, capacity=rate_limit*2)
    
    def _sign_request(self, params):
        """生成签名"""
        timestamp = int(time.time() * 1000)
        params["timestamp"] = timestamp
        
        query_string = "&".join([f"{k}={v}" for k, v in sorted(params.items())])
        signature = hmac.new(
            self.secret_key.encode(),
            query_string.encode(),
            hashlib.sha256
        ).hexdigest()
        params["signature"] = signature
        return params
    
    def fetch(self, endpoint, params=None):
        """带限流和重试的数据采集"""
        params = params or {}
        params = self._sign_request(params)
        
        def _request():
            # 限流
            while not self.bucket.consume():
                time.sleep(0.01)
            
            headers = {"X-MBX-APIKEY": self.api_key}
            url = f"https://api.binance.com{endpoint}"
            response = requests.get(url, headers=headers, 
                                  params=params, timeout=5)
            response.raise_for_status()
            return response.json()
        
        return retry_with_backoff(_request)

# 使用
collector = DataCollector("你的KEY", "你的SECRET", rate_limit=10)
data = collector.fetch("/api/v3/ticker/price", {"symbol": "BTCUSDT"})
print(data)

这张图展示了整个流程：准备请求 → 限流检查 → 发送请求 → 成功/失败处理。失败时自动重试，重试间隔指数增长。

好了，数据采集的基础就这些。下一章我们会聊数据清洗和存储，那又是另一番天地了。

---

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