用 Python 接入币安 API：从零开始打造自动交易脚本

关键词：币安 API、Python、加密货币、自动交易、API 密钥、REST 接口、WebSocket、交易机器人

快速入门：为何选择币安 API

币安 API 是全球加密资产交易网络中最成熟的 加密货币 API 之一。无论你是量化交易者、数据分析师，还是想用 Python 做策略回测，它都能提供 市价、账户、实时行情、资金管理 等几十种接口。
想一步到位提升到生产级？👉一文掌握高阶调试技巧，90% 的初学者忽略这一步，让你的脚本上线就稳。

创建与管理 API 密钥

1. 获取密钥

1. 登录币安网页端 →「用户中心」→「API 管理」。
2. 点击「创建新密钥」，系统会同时给出 API Key 与 Secret Key。
3. 务必立即启用「IP 白名单」和「可交易」权限，关闭不必要的提币权限，最小化泄露风险。

2. 三种认证方式概览

下文以最易落地的 HMAC 为例。

HMAC 签名范例代码

import hmac, hashlib, time, requests
from urllib.parse import urlencode

base_url = "https://api.binance.com"
API_KEY = "替换为你自己的"
API_SECRET = "替换为你自己的"

def _sign(query_string):
    return hmac.new(API_SECRET.encode(), query_string.encode(), hashlib.sha256).hexdigest()

def get_account():
    endpoint = "/api/v3/account"
    ts = int(time.time() * 1000)
    params = {"timestamp": ts, "recvWindow": 5000}
    query = urlencode(params)
    sig = _sign(query)
    url = f"{base_url}{endpoint}?{query}&signature={sig}"
    headers = {"X-MBX-APIKEY": API_KEY}
    return requests.get(url, headers=headers).json()

print(get_account())  # 示例输出账户信息

提示：recvWindow 是容错窗口，默认 5000 ms；若服务器时间与本地差异大，可调至 60000 ms。

核心端点速查表

连通测试与服务器时间

def test_ping():
    return requests.get(base_url + "/api/v3/ping").json()

def server_time():
    return requests.get(base_url + "/api/v3/time").json()["serverTime"]

获取行情

def depth(symbol="BTCUSDT", limit=20):
    url = f"{base_url}/api/v3/depth?symbol={symbol}&limit={limit}"
    return requests.get(url).json()

K 线数据

def klines(symbol="BTCUSDT", interval="1h", limit=100):
    url = f"{base_url}/api/v3/klines?symbol={symbol}&interval={interval}&limit={limit}"
    raw = requests.get(url).json()
    return [{
        "open_time": r[0],
        "open": float(r[1]),
        "high": float(r[2]),
        "low": float(r[3]),
        "close": float(r[4]),
        "volume": float(r[5])
    } for r in raw]

实战：下第一笔订单

限价单

def limit_order(symbol, side, qty, price):
    params = {
        "symbol": symbol,
        "side": side,
        "type": "LIMIT",
        "timeInForce": "GTC",
        "quantity": qty,
        "price": price,
        "timestamp": int(time.time() * 1000)
    }
    query = urlencode(params)
    sig = _sign(query)
    url = f"{base_url}/api/v3/order?{query}&signature={sig}"
    headers = {"X-MBX-APIKEY": API_KEY}
    return requests.post(url, headers=headers).json()

市价单简化示例

def market_buy(symbol, qty):
    return limit_order(symbol, "BUY", qty, None)

直接把 type 改成 MARKET 即可。

账户与资金流水

def trades(symbol):
    params = {"symbol": symbol, "timestamp": int(time.time() * 1000)}
    query = urlencode(params)
    sig = _sign(query)
    url = f"{base_url}/api/v3/myTrades?{query}&signature={sig}"
    headers = {"X-MBX-APIKEY": API_KEY}
    return requests.get(url, headers=headers).json()

该接口支持 startTime 与 endTime，便于做月度对账。

WebSocket 实时推送

获取推送 listenKey：

def start_stream():
    url = f"{base_url}/api/v3/userDataStream"
    headers = {"X-MBX-APIKEY": API_KEY}
    return requests.post(url, headers=headers).json()["listenKey"]

用 wss://stream.binance.com:9443/ws/{listenKey} 建立持久连接，即可监听 订单回报与资金变化。

进阶玩法

OCO（一单取消另一单）

def oco_sell(symbol, qty, profit_price, stop_price):
    params = {
        "symbol": symbol,
        "side": "SELL",
        "quantity": qty,
        "price": profit_price,
        "stopPrice": stop_price,
        "stopLimitPrice": stop_price * 0.99,
        "stopLimitTimeInForce": "GTC",
        "timestamp": int(time.time() * 1000)
    }
    query = urlencode(params)
    sig = _sign(query)
    url = f"{base_url}/api/v3/orderList/oco?{query}&signature={sig}"
    headers = {"X-MBX-APIKEY": API_KEY}
    return requests.post(url, headers=headers).json()

优雅的限流处理

使用指数退避策略，自测脚本时别忘了打开调试日志。👉三步实现智能重连，减少无效请求，尤其在高并发场景下能省下大量 429 错误。

构建完整交易机器人骨架

以下是一个最小可运行类框架，可根据策略自行扩展：

class Trader:
    def __init__(self, key, secret, symbol):
        self.key, self.secret, self.symbol = key, secret, symbol
        self.base_url = "https://api.binance.com"

    def _signed(self, method, endpoint, params):
        params["timestamp"] = int(time.time() * 1000)
        query = urlencode(params)
        sig = _sign(query)
        url = f"{self.base_url}{endpoint}?{query}&signature={sig}"
        headers = {"X-MBX-APIKEY": self.key}
        return requests.request(method, url, headers=headers).json()

    def buy_market(self, qty):
        return self._signed("POST","/api/v3/order",
                {"symbol": self.symbol, "side": "BUY", "type": "MARKET", "quantity": qty})

# 初始化
bot = Trader(API_KEY, API_SECRET, "BTCUSDT")
print(bot.buy_market(0.001))

最佳实践速览

1. 环境变量保存密钥，永不上传代码仓库。

   export BINANCE_API_KEY=xxx
   export BINANCE_SECRET=xxx

2. IP 白名单必须启用，防止密钥泄露后异地登录。
3. 定期轮转密钥，至少 90 天更换一次。
4. 单元测试务必在 币安现货测试网 完成，避免真仓意外成交。

常见问题 FAQ

Q1：总是提示「-1021 INVALID_TIMESTAMP」怎么办？
A：本地时间与币安服务器时间不同步。先抓取 /api/v3/time，再对比本地，误差超过 1000 ms 就会被拒绝。

Q2：如何批量查询订单？
A：用 /api/v3/allOrders 可一次返回某交易对全部订单；在参数里添加 limit=1000 即可拉取最近 1000 条记录。

Q3：机器人运行一段时间直接 429 报错，如何重试？
A：建议用指数退避 + Retry-After 响应头。可用 tenacity 包装饰所有请求函数。

Q4：需要订阅千档深度，但 REST 延迟高怎么办？
A：切到 WebSocket 深度流 wss://stream.binance.com:9443/ws/btcusdt@depth@100ms，即可毫秒级推送。

Q5：Python 老报 SSL 错误？
A：升级到最新 urllib3 或 requests，或在脚本顶部加 export SSL_CERT_FILE=/etc/ssl/cert.pem（Linux 环境）。

Q6：担心风险，有没有无风险的测试币？
A：测试网 faucet 会发放免费的 tBNB、BTC、USDT，登录 testnet.binance.vision 领取即可安全演练。

小结

本文系统梳理了用 Python 接入币安 API 的完整链路：从密钥管理、签名、REST 与 WebSocket 数据，到下单、止损、OCOS、限流和异常处理。
先用少量仓位跑通流程，再逐渐叠加高阶功能，切忌一次性全仓操作。祝你在加密世界做出稳健的 量化交易策略，自动收益，风控无忧！