数据获取：从交易所API获取Level-2订单簿快照数据

做量化交易，数据就是命根子。尤其是订单簿不平衡度这种信号，你拿不到高质量的Level-2数据，后面所有分析都是空中楼阁。

今天咱们就聊聊怎么从Binance把订单簿快照数据拉下来。我自己的经验是，这一步看似简单，但坑不少。嗯，咱们一个一个说。

为什么选Binance？

说实话，市面上交易所很多，但我个人习惯用Binance。原因有三：

• API文档清晰：接口设计规范，参数说明到位
• Level-2数据深度够：支持5000档订单簿，做不平衡度分析绰绰有余
• 限流策略友好：每分钟1200次请求，对个人策略来说够用了

当然，如果你做高频交易，可能得考虑更专业的方案。但咱们课程讲的是信号挖掘，Binance完全够用。

Level-2订单簿长什么样？

先看个实际例子。我从Binance拉了一组BTC/USDT的订单簿快照，结构是这样的：

字段	说明	示例值
lastUpdateId	快照更新ID	123456789
bids	买单列表 [价格, 数量]	[["45000.00", "1.5"], ...]
asks	卖单列表 [价格, 数量]	[["45001.00", "2.3"], ...]

说白了，就是当前市场上所有挂单的「价格-数量」快照。买单按价格从高到低排，卖单从低到高排。

代码实现：拉取订单簿快照

直接上代码。我习惯用 requests 库，轻量又好用。

import requests
import json

def fetch_order_book_snapshot(symbol="BTCUSDT", limit=100):
    """
    从Binance获取订单簿快照数据
    
    参数:
        symbol: 交易对，默认BTCUSDT
        limit: 返回的档位数，最大5000
    
    返回:
        dict: 包含bids和asks的订单簿数据
    """
    url = "https://api.binance.com/api/v3/depth"
    params = {
        "symbol": symbol,
        "limit": limit
    }
    
    try:
        response = requests.get(url, params=params, timeout=5)
        response.raise_for_status()
        data = response.json()
        
        print(f"成功获取 {symbol} 订单簿快照")
        print(f"更新ID: {data['lastUpdateId']}")
        print(f"买单数量: {len(data['bids'])} 档")
        print(f"卖单数量: {len(data['asks'])} 档")
        
        return data
    
    except requests.exceptions.RequestException as e:
        print(f"请求失败: {e}")
        return None

# 测试一下
snapshot = fetch_order_book_snapshot("BTCUSDT", 100)

数据清洗：别让脏数据坑了你

数据拉下来之后，不能直接用。我在项目中遇到过好几次，Binance返回的价格和数量都是字符串，不是浮点数。你想想看，如果你直接拿字符串做加减乘除，结果会怎样？

所以，第一步就是类型转换：

def clean_order_book(data):
    """
    清洗订单簿数据：将字符串转为浮点数
    """
    cleaned = {
        "lastUpdateId": data["lastUpdateId"],
        "bids": [],
        "asks": []
    }
    
    for price, qty in data["bids"]:
        cleaned["bids"].append([float(price), float(qty)])
    
    for price, qty in data["asks"]:
        cleaned["asks"].append([float(price), float(qty)])
    
    return cleaned

# 使用
snapshot = fetch_order_book_snapshot("BTCUSDT", 100)
if snapshot:
    cleaned_data = clean_order_book(snapshot)
    print(f"清洗后买单第一档: 价格={cleaned_data['bids'][0][0]}, 数量={cleaned_data['bids'][0][1]}")

数据存储：别每次都重新拉

做策略回测时，你不可能每次都去交易所拉数据。一来慢，二来容易被限流。我的做法是：把快照数据存到本地文件，按日期归档。

import os
import time
from datetime import datetime

def save_snapshot_to_file(data, symbol, save_dir="orderbook_data"):
    """
    将订单簿快照保存到本地文件
    """
    if not os.path.exists(save_dir):
        os.makedirs(save_dir)
    
    timestamp = int(time.time() * 1000)
    filename = f"{symbol}_{timestamp}.json"
    filepath = os.path.join(save_dir, filename)
    
    with open(filepath, "w") as f:
        json.dump(data, f)
    
    print(f"数据已保存至: {filepath}")

# 定时拉取并保存
def collect_order_book_loop(symbol="BTCUSDT", interval=1, duration=60):
    """
    持续采集订单簿数据
    
    参数:
        interval: 采集间隔（秒）
        duration: 采集时长（秒）
    """
    start_time = time.time()
    count = 0
    
    while time.time() - start_time < duration:
        snapshot = fetch_order_book_snapshot(symbol, 100)
        if snapshot:
            cleaned = clean_order_book(snapshot)
            save_snapshot_to_file(cleaned, symbol)
            count += 1
        
        time.sleep(interval)
    
    print(f"采集完成，共保存 {count} 个快照文件")

# 运行1分钟，每5秒采集一次
collect_order_book_loop("BTCUSDT", interval=5, duration=60)

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

Binance对API请求有限制。个人账户每分钟1200次权重，每次请求消耗的权重不一样。订单簿快照请求消耗的权重是 limit / 100。

举个例子：你拉100档数据，消耗1点权重；拉5000档，消耗50点权重。所以，别贪心，够用就行。

知识体系总览

下面这张图，是我自己总结的订单簿数据获取流程。你看一眼，心里就有数了：

这张图把整个流程串起来了。你从API拉数据，清洗后存本地，同时注意限流。然后定时循环，就能拿到连续的时间序列数据了。

常见问题与解决方案

最后，分享几个我踩过的坑：

• 网络超时：加个重试机制，失败后等1秒再试，最多重试3次
• 数据不完整：检查 lastUpdateId 是否连续，不连续说明中间有丢包
• 时间戳对齐：本地时间和服务器时间可能有偏差，用 time.time() 记录本地时间戳

好了，数据获取这块就聊到这儿。你拿到数据后，下一件事就是计算订单簿不平衡度了。那个咱们后面再细说。

---