做加密货币量化交易或套利策略开发,Binance API 是绕不开的基础设施。但在正式接入前,你必须先搞懂它的数据结构——签名算法怎么算、REST 端点返回什么格式、WebSocket 推送的数据结构长什么样。我见过太多开发者卡在 HMAC-SHA256 签名校验上反复折腾,今天用一篇文章把所有核心数据结构讲透,并给出 Python 完整调用示例。

先算一笔账:为什么中转 API 比你想象的更划算

在我们深入数据结构之前,先看一个扎心的价格对比。2026 年主流大模型输出价格如下:

每月 100 万 token 消耗,官方渠道下:

换算人民币(官方汇率 1$=¥7.3):Claude Sonnet 4.5 每月要 ¥1095,DeepSeek V3.2 也要 ¥30.66。但通过 HolySheep 中转站,按 ¥1=$1 结算:Claude Sonnet 4.5 直接降到 ¥150/月,DeepSeek V3.2 只需 ¥4.2/月,节省超过 85%。

对于做高频套利策略、需要实时调用 LLM 判断趋势的开发者,这个差价可能就是你的利润空间。如果你同时在跑 Binance 做市策略,用省下的 API 费用去买币不香吗?

Binance API 认证机制:HMAC-SHA256 签名全解析

Binance 所有需要写操作的 API(下单、撤单、划转等)都需要签名验证。理解签名算法是读懂整个数据结构体系的前提。

签名算法流程

签名公式:HMAC-SHA256(SECRET_KEY, query_string),其中 query_string 是参数字符串按字典序拼接的结果。

import hmac
import hashlib
import time
import requests

Binance API 凭证(从 https://www.binance.com/userCenter/content/apiManagement 获取)

API_KEY = "YOUR_BINANCE_API_KEY" SECRET_KEY = "YOUR_BINANCE_SECRET_KEY" def create_signed_params(params: dict) -> dict: """ 生成带签名的请求参数 """ # 1. 添加时间戳 params['timestamp'] = int(time.time() * 1000) # 2. 按字典序拼接 query_string query_string = '&'.join([f"{k}={v}" for k, v in sorted(params.items())]) # 3. HMAC-SHA256 签名 signature = hmac.new( SECRET_KEY.encode('utf-8'), query_string.encode('utf-8'), hashlib.sha256 ).hexdigest() # 4. 添加签名到参数 params['signature'] = signature return params

示例:查询账户余额

def get_account_balance(): url = "https://api.binance.com/api/v3/account" params = create_signed_params({}) headers = {"X-MBX-APIKEY": API_KEY} response = requests.get(url, params=params, headers=headers) return response.json()

测试

result = get_account_balance() print(result)

签名后的请求头结构

Headers:
  X-MBX-APIKEY: YOUR_API_KEY
  Content-Type: application/x-www-form-urlencoded

Query Parameters (signed):
  symbol=BNBUSDT
  side=BUY
  type=LIMIT
  timeInForce=GTC
  quantity=1
  price=300
  timestamp=1709654321000
  signature=3d4f2b1a8c9e7f6d5a4b3c2d1e0f9a8b7c6d5e4f3a2b1c0d9e8f7a6b5c4d3e2f

核心 REST API 数据结构

1. 账户信息结构

# GET /api/v3/account 返回结构
{
  "makerCommission": 10,          # 挂单手续费率(基点,10 = 0.1%)
  "takerCommission": 10,           # 吃单手续费率
  "buyerCommission": 0,            # 买币手续费率(已废弃)
  "sellerCommission": 0,           # 卖币手续费率(已废弃)
  "canTrade": true,                # 是否可以交易
  "canWithdraw": true,             # 是否可以提币
  "canDeposit": true,              # 是否可以充值
  "updateTime": 1672515781000,     # 最后更新时间戳(毫秒)
  "accountType": "SPOT",           # 账户类型:SPOT/MARGIN/UST
  "balances": [
    {
      "asset": "BTC",              # 资产名称
      "free": "0.00000000",        # 可用数量
      "locked": "0.00000000"       # 锁定数量(下单冻结)
    },
    {
      "asset": "BNB",
      "free": "100.00000000",
      "locked": "10.00000000"
    }
  ],
  "permissions": ["SPOT", "MARGIN"]  # 权限列表
}

2. 订单响应结构

# POST /api/v3/order 返回结构(成功下单)
{
  "symbol": "BTCUSDT",             # 交易对
  "orderId": 123456789,            # 订单ID(唯一标识)
  "orderListId": -1,               # OCO订单ID,单个订单为-1
  "clientOrderId": "my_order_001", # 客户端自定义订单ID
  "transactTime": 1672515782000,   # 成交时间戳
  "price": "45000.00000000",      # 订单价格
  "origQty": "0.10000000",        # 原始数量
  "executedQty": "0.00000000",     # 已成交数量
  "cummulativeQuoteQty": "0.00000000",  # 累计成交金额
  "status": "NEW",                # 订单状态
  "type": "LIMIT",                # 订单类型
  "side": "BUY",                  # 买卖方向
  "fills": []                     # 成交记录(部分成交时非空)
}

订单状态枚举:

NEW(新订单) / PARTIALLY_FILLED(部分成交) / FILLED(完全成交)

CANCELED(已取消) / PENDING_CANCEL(取消中) / REJECTED(被拒绝)

EXPIRED(已过期)

3. 市场深度数据结构

# GET /api/v3/depth?symbol=BTCUSDT&limit=100 返回结构
{
  "lastUpdateId": 160,            # 推送的最后一个update ID
  "bids": [                        # 买方深度(价格从高到低)
    ["4023.00000000", "0.10000000"],  # [价格, 数量]
    ["4022.50000000", "0.20000000"]
  ],
  "asks": [                        # 卖方深度(价格从低到高)
    ["4024.00000000", "0.15000000"],
    ["4024.50000000", "0.30000000"]
  ]
}

深度限制可选值:5, 10, 20, 50, 100, 500, 1000, 5000

4. K线(K线图)数据结构

# GET /api/v3/klines?symbol=BTCUSDT&interval=1h&limit=100 返回结构
[
  [
    1672515780000,    // 开盘时间(毫秒)
    "38547.00000000", // 开盘价
    "38820.00000000", // 最高价
    "38456.00000000", // 最低价
    "38745.00000000", // 收盘价
    "123.45678",      // 成交量
    1672519340000,    // 收盘时间
    "4781234.56789000", // 成交额
    9523,            // 成交笔数
    "98.76543",      // 主动买入成交量
    "3823456.78900000", // 主动买入成交额
    "0"              // 忽略
  ]
]

周期可选值:1m, 3m, 5m, 15m, 30m, 1h, 2h, 4h, 6h, 8h, 12h, 1d, 3d, 1w, 1M

5. 近期成交记录

# GET /api/v3/myTrades?symbol=BTCUSDT 返回结构
[
  {
    "symbol": "BTCUSDT",
    "id": 2845732,
    "orderId": 1002342,           # 关联订单ID
    "orderListId": -1,
    "price": "38650.00000000",    // 成交价格
    "qty": "0.10000000",          // 成交数量
    "quoteQty": "3865.00000000",  // 成交金额
    "commission": "0.00010000",   // 手续费
    "commissionAsset": "BNB",    // 手续费资产类型
    "time": 1672515782000,       // 成交时间
    "isBuyer": true,             // 是否买方
    "isMaker": false,            // 是否挂单方(False=吃单)
    "isBestMatch": true
  }
]

WebSocket 实时数据结构

Binance WebSocket 提供低延迟的实时数据推送,适合做市策略和价格监控。

# WebSocket 连接示例(Python)
import websocket
import json
import threading

class BinanceWebSocket:
    def __init__(self, symbols: list, streams: list):
        self.symbols = symbols
        self.streams = streams
        self.ws = None
        self.thread = None
        
    def connect(self):
        # streams 参数格式:symbol@stream1/symbol@stream2
        streams = '/'.join([f"{s.lower()}@{t}" for s in self.symbols for t in self.streams])
        url = f"wss://stream.binance.com:9443/stream?streams={streams}"
        
        self.ws = websocket.WebSocketApp(
            url,
            on_message=self.on_message,
            on_error=self.on_error,
            on_close=self.on_close
        )
        self.thread = threading.Thread(target=self.ws.run_forever)
        self.thread.daemon = True
        self.thread.start()
        
    def on_message(self, ws, message):
        data = json.loads(message)
        event = data.get('e')  # 事件类型
        symbol = data.get('s')  # 交易对
        
        if event == "trade":
            # 实时成交事件
            print(f"成交事件: {symbol} 价格:{data['p']} 数量:{data['q']} 时间:{data['T']}")
            # data 结构:
            # {
            #   "e": "trade",           // 事件类型
            #   "E": 1672515783000,    // 事件时间
            #   "s": "BTCUSDT",        // 交易对
            #   "t": 12345,            // 成交ID
            #   "p": "38650.00000000", // 成交价格
            #   "q": "0.10000000",     // 成交数量
            #   "b": 100234,           // 买方订单ID
            #   "a": 100235,           // 卖方订单ID
            #   "T": 1672515782999,    // 成交时间
            #   "m": true              // 是否做市商成交
            # }
            
        elif event == "depthUpdate":
            # 深度更新事件(@depth 订阅)
            print(f"深度更新: {symbol} 买一:{data['b'][0]} 卖一:{data['a'][0]}")
            # data 结构:
            # {
            #   "e": "depthUpdate",
            #   "E": 1672515783000,
            #   "s": "BTCUSDT",
            #   "U": 100,              // 推送更新起始ID
            #   "u": 200,              // 推送更新结束ID
            #   "b": [["38650", "1.5"]], // 买方深度变化
            #   "a": [["38651", "2.0"]]  // 卖方深度变化
            # }
            
        elif event == "kline":
            # K线更新事件
            kline = data['k']
            print(f"K线更新: {symbol} {kline['t']}-{kline['T']} OHLC:{kline['o']}/{kline['h']}/{kline['l']}/{kline['c']}")
            
    def on_error(self, ws, error):
        print(f"WebSocket错误: {error}")
        
    def on_close(self, ws):
        print("WebSocket连接关闭")

使用示例

ws = BinanceWebSocket(['BTCUSDT', 'ETHUSDT'], ['trade', 'depth20@100ms']) ws.connect()

常见报错排查

错误码对照表

错误码含义解决方案
-1013 下单数量小于最小值/精度不符 检查 LOT_SIZE 过滤规则,用 precision 取整
-1021 时间戳偏差过大(>5分钟) 同步服务器时间:time.time() * 1000 校准
-1022 签名不匹配 检查 SECRET_KEY 是否正确,query_string 排序方式
-2015 无效的 API-KEY 确认 API Key 是否过期或权限不足
-3023 余额不足 检查 GET /api/v3/account 的 free 字段
-2010 订单不存在 确认 orderId 和 origClientOrderId 是否正确

高频报错场景与修复代码

# 场景1:签名错误(最常见)

错误:{"code":-1022,"msg":"Signature for this request was not found."}

def fix_signature_issue(): """ 签名问题的三个排查方向: 1. SECRET_KEY 包含额外空格或换行符 2. query_string 拼接顺序错误 3. timestamp 使用秒而非毫秒 """ import binance.helpers as helpers # 正确做法:直接用 Binance 官方签名函数(如果有SDK) # 或者手动实现时确保: params = { 'symbol': 'BTCUSDT', 'side': 'BUY', 'type': 'LIMIT', 'quantity': '0.1', 'price': '40000', 'timeInForce': 'GTC', 'timestamp': int(time.time() * 1000) # 必须是毫秒! } # 排序必须按 ASCII 顺序 sorted_params = dict(sorted(params.items())) query_string = '&'.join([f"{k}={v}" for k, v in sorted_params.items()]) # 确保密钥没有隐藏字符 secret_key = SECRET_KEY.strip() signature = hmac.new( secret_key.encode('utf-8'), query_string.encode('utf-8'), hashlib.sha256 ).hexdigest() return signature

场景2:数量精度错误

错误:{"code":-1013,"msg":"Filter failure: LOT_SIZE"}

def fix_lot_size(): """ 从交易所获取交易对精度信息 """ def get_symbol_info(symbol): url = "https://api.binance.com/api/v3/exchangeInfo" resp = requests.get(url) data = resp.json() for s in data['symbols']: if s['symbol'] == symbol: # 提取数量精度 for f in s['filters']: if f['filterType'] == 'LOT_SIZE': return { 'minQty': float(f['minQty']), 'maxQty': float(f['maxQty']), 'stepSize': float(f['stepSize']) } return None info = get_symbol_info('BTCUSDT') # 例如: {'minQty': 0.00001, 'maxQty': 9000, 'stepSize': 0.00001} # 数量取整函数 def round_step_size(quantity, step_size): precision = int(round(-math.log(step_size, 10), 10)) return round(float(quantity) // step_size * step_size, precision) original_qty = 0.12345678 fixed_qty = round_step_size(original_qty, info['stepSize']) # 输出: 0.12345 return fixed_qty

场景3:时间戳偏差过大

错误:{"code":-1021,"msg":"Timestamp for this request was a minute ahead of the server's time."}

def fix_timestamp_sync(): """ 同步 Binance 服务器时间的正确方法 """ import datetime # 方法1:使用 HTTP Header 中的服务器时间 def get_server_time(): url = "https://api.binance.com/api/v3/time" resp = requests.get(url) server_time = resp.json()['serverTime'] local_time = int(time.time() * 1000) time_diff = server_time - local_time print(f"本地时间与服务器时间差: {time_diff}ms") return time_diff time_offset = get_server_time() def adjusted_timestamp(): return int(time.time() * 1000) + time_offset # 之后所有请求都用 adjusted_timestamp() 替代 int(time.time() * 1000)

Python 完整下单流程实战

"""
Binance 现货市价下单完整流程
包含:签名生成 -> 订单提交 -> 响应解析 -> 异常处理
"""
import hmac
import hashlib
import time
import requests
from typing import Optional, Dict, Any

class BinanceSpotClient:
    BASE_URL = "https://api.binance.com"
    
    def __init__(self, api_key: str, api_secret: str):
        self.api_key = api_key
        self.api_secret = api_secret
        self.time_offset = self._sync_time()
        
    def _sync_time(self) -> int:
        """同步服务器时间偏移"""
        resp = requests.get(f"{self.BASE_URL}/api/v3/time")
        server_time = resp.json()['serverTime']
        local_time = int(time.time() * 1000)
        return server_time - local_time
    
    def _sign(self, params: Dict) -> str:
        """生成签名"""
        params['timestamp'] = int(time.time() * 1000) + self.time_offset
        query_string = '&'.join([f"{k}={v}" for k, v in sorted(params.items())])
        return hmac.new(
            self.api_secret.encode('utf-8'),
            query_string.encode('utf-8'),
            hashlib.sha256
        ).hexdigest()
    
    def place_market_buy(self, symbol: str, quantity: float) -> Dict[str, Any]:
        """
        市价买入
        
        Args:
            symbol: 交易对,如 'BTCUSDT'
            quantity: 购买数量
            
        Returns:
            订单响应字典
        """
        params = {
            'symbol': symbol.upper(),
            'side': 'BUY',
            'type': 'MARKET',
            'quantity': str(quantity)  # 必须转字符串
        }
        params['signature'] = self._sign(params)
        
        headers = {'X-MBX-APIKEY': self.api_key}
        url = f"{self.BASE_URL}/api/v3/order"
        
        try:
            resp = requests.post(url, data=params, headers=headers, timeout=10)
            data = resp.json()
            
            if 'code' in data:  # 错误响应
                raise Exception(f"Binance API错误 [{data['code']}]: {data['msg']}")
            
            return {
                'success': True,
                'orderId': data['orderId'],
                'symbol': data['symbol'],
                'executedQty': data['executedQty'],
                'status': data['status'],
                'cummulativeQuoteQty': data['cummulativeQuoteQty']
            }
            
        except requests.exceptions.Timeout:
            # 超时不代表下单失败,需要查询订单状态
            print("请求超时,正在查询订单状态...")
            return {'success': False, 'error': 'timeout'}
            
        except requests.exceptions.RequestException as e:
            return {'success': False, 'error': str(e)}

    def get_order_status(self, symbol: str, order_id: int) -> Dict:
        """查询订单状态(用于处理超时情况)"""
        params = {
            'symbol': symbol.upper(),
            'orderId': order_id
        }
        params['signature'] = self._sign(params)
        
        headers = {'X-MBX-APIKEY': self.api_key}
        url = f"{self.BASE_URL}/api/v3/order"
        
        resp = requests.get(url, params=params, headers=headers)
        return resp.json()

使用示例

if __name__ == "__main__": client = BinanceSpotClient( api_key="YOUR_API_KEY", api_secret="YOUR_SECRET_KEY" ) # 尝试市价买入 0.01 BTC result = client.place_market_buy('BTCUSDT', 0.01) if result.get('success'): print(f"下单成功!订单ID: {result['orderId']}") print(f"成交数量: {result['executedQty']}") print(f"成交金额: {result['cummulativeQuoteQty']} USDT") else: print(f"下单失败: {result.get('error')}")

推荐:使用 HolySheep 中转优化你的策略开发

在做 Binance 量化策略时,你可能还需要调用 LLM 做情绪分析、新闻解读或自定义技术指标计算。通过 HolySheep 中转 API,你可以用国内直连 <50ms 的延迟访问 GPT-4.1、Claude Sonnet 4.5、Gemini 2.5 Flash 和 DeepSeek V3.2 等模型。

HolySheep 的核心优势:

# HolySheep API 调用示例(与 OpenAI 兼容)
import openai

client = openai.OpenAI(
    api_key="YOUR_HOLYSHEEP_API_KEY",  # 从 HolySheep 获取
    base_url="https://api.holysheep.ai/v1"  # 固定中转地址
)

调用 Claude Sonnet 4.5 分析加密市场情绪

response = client.chat.completions.create( model="claude-sonnet-4-20250514", messages=[ {"role": "system", "content": "你是一个加密货币分析师"}, {"role": "user", "content": "分析 BTC 当前市场情绪,给出短期交易建议"} ], temperature=0.7, max_tokens=500 ) print(response.choices[0].message.content)

使用 DeepSeek V3.2 做批量数据处理(超低价)

response = client.chat.completions.create( model="deepseek-chat", messages=[ {"role": "user", "content": "帮我分析这100条交易记录中的异常模式"} ] )

适合谁与不适合谁

场景推荐使用 Binance API 直连推荐使用 HolySheep 中转
高频交易 ✓ 对延迟敏感,必须直连 ✗ 不适合
套利策略 ✓ 需要实时深度数据 ✓ 配合 LLM 做价差分析
新闻/情绪分析 ✗ Binance 不提供 ✓ 调用 LLM 做 NLP 处理
技术指标计算 ✓ 自计算更灵活 ✓ LLM 辅助策略优化
Dashboard/报告 ✓ 聚合多交易所数据 ✓ LLM 生成自然语言总结

价格与回本测算

假设你每月在 Claude Sonnet 4.5 上花费 $100(GPT-4.1 的 API 调用):

方案月消耗汇率月花费(¥)年花费(¥)
官方 API $100 ¥7.3/$1 ¥730 ¥8760
HolySheep 中转 $100 ¥1/$1 ¥100 ¥1200
节省 - - ¥630/月 ¥7560/年

回本测算:HolySheep 注册即送免费额度,对于轻度使用用户可能完全免费使用。对于月消耗 $50+ 的开发者,半年内省下的费用就超过一次策略开发的人工成本。

为什么选 HolySheep

在对比了国内主流 AI 中转服务后,我选择 HolySheep 的原因有三:

  1. 价格透明:没有隐藏费用,按 ¥1=$1 结算,汇率稳定不浮动
  2. 模型覆盖全:GPT-4.1、Claude Sonnet 4.5、Gemini 2.5 Flash、DeepSeek V3.2 全部支持
  3. 国内直连:延迟 <50ms,不掉线不超时,微信/支付宝秒充值

我自己在跑一个 Binance 做市策略时,用 DeepSeek V3.2 做订单簿异常检测,每月 API 费用从 ¥300+ 降到 ¥40+,这个差价就是纯利润。

结语

Binance API 的数据结构本身并不复杂,关键在于:签名算法必须用毫秒级时间戳且参数按 ASCII 排序、市场深度有精度限制需要严格取整、WebSocket 推送是增量更新而非全量数据。把这几个核心点理解透,再配合上面的代码示例,你应该能快速跑通自己的交易策略。

如果你的策略还需要 LLM 做辅助决策,比如实时分析链上数据判断趋势方向,不妨试试 HolySheep 中转 API,国内直连低价,每月能省下不少成本。

有问题欢迎评论区交流,我会尽量解答。

👉 免费注册 HolySheep AI,获取首月赠额度