做加密货币量化交易或套利策略开发,Binance API 是绕不开的基础设施。但在正式接入前,你必须先搞懂它的数据结构——签名算法怎么算、REST 端点返回什么格式、WebSocket 推送的数据结构长什么样。我见过太多开发者卡在 HMAC-SHA256 签名校验上反复折腾,今天用一篇文章把所有核心数据结构讲透,并给出 Python 完整调用示例。
先算一笔账:为什么中转 API 比你想象的更划算
在我们深入数据结构之前,先看一个扎心的价格对比。2026 年主流大模型输出价格如下:
- GPT-4.1 output:$8/MTok
- Claude Sonnet 4.5 output:$15/MTok
- Gemini 2.5 Flash output:$2.50/MTok
- DeepSeek V3.2 output:$0.42/MTok
每月 100 万 token 消耗,官方渠道下:
- Claude Sonnet 4.5 = $150/月
- GPT-4.1 = $80/月
- Gemini 2.5 Flash = $25/月
- DeepSeek V3.2 = $4.2/月
换算人民币(官方汇率 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 的核心优势:
- 汇率优势:¥1=$1,官方汇率 ¥7.3=$1 的基础上节省超过 85%
- 国内直连:延迟 <50ms,无需魔法上网
- 充值便捷:支持微信/支付宝直接充值
- 注册福利:新用户赠送免费试用额度
# 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 结算,汇率稳定不浮动
- 模型覆盖全:GPT-4.1、Claude Sonnet 4.5、Gemini 2.5 Flash、DeepSeek V3.2 全部支持
- 国内直连:延迟 <50ms,不掉线不超时,微信/支付宝秒充值
我自己在跑一个 Binance 做市策略时,用 DeepSeek V3.2 做订单簿异常检测,每月 API 费用从 ¥300+ 降到 ¥40+,这个差价就是纯利润。
结语
Binance API 的数据结构本身并不复杂,关键在于:签名算法必须用毫秒级时间戳且参数按 ASCII 排序、市场深度有精度限制需要严格取整、WebSocket 推送是增量更新而非全量数据。把这几个核心点理解透,再配合上面的代码示例,你应该能快速跑通自己的交易策略。
如果你的策略还需要 LLM 做辅助决策,比如实时分析链上数据判断趋势方向,不妨试试 HolySheep 中转 API,国内直连低价,每月能省下不少成本。
有问题欢迎评论区交流,我会尽量解答。