币安API Python怎么调？python-binance与ccxt实战

币安 Python SDK 最成熟的三个选择是 python-binance（官方推荐）、ccxt（多交易所统一接口）、aiohttp 原生封装（高性能自研），分别对应「快速开发」「多所兼容」「极致性能」三种场景。本文给出三者的安装、签名调用、下单、WebSocket 订阅完整代码，并附性能基准测试。没有币安账号请先到 币安官网 完成 KYC，新用户可通过 免费注册 开通。

一、三种 SDK 对比表

二、python-binance 完整示例

1. 安装与初始化

pip install python-binance

from binance.client import Client
from binance.enums import *

API_KEY = "YOUR_API_KEY"
SECRET_KEY = "YOUR_SECRET_KEY"

client = Client(API_KEY, SECRET_KEY)

# 测试网
# client = Client(API_KEY, SECRET_KEY, testnet=True)

# 验证连接
print(client.get_server_time())  # {'serverTime': 1713027384562}

2. 查询余额和行情

# Spot 账户余额
account = client.get_account()
non_zero = [b for b in account["balances"] if float(b["free"]) > 0]
for b in non_zero:
    print(f"{b['asset']}: 可用 {b['free']}, 冻结 {b['locked']}")

# 单品种价格
price = client.get_symbol_ticker(symbol="BTCUSDT")
print(f"BTC 最新价: {price['price']}")

# K 线
klines = client.get_klines(symbol="BTCUSDT", interval=Client.KLINE_INTERVAL_1HOUR, limit=100)
for k in klines[-5:]:
    print(f"时间 {k[0]}, O:{k[1]} H:{k[2]} L:{k[3]} C:{k[4]} V:{k[5]}")

# 深度
depth = client.get_order_book(symbol="BTCUSDT", limit=20)
print(f"最优买 {depth['bids'][0]}, 最优卖 {depth['asks'][0]}")

3. 现货下单

# 限价单
order = client.order_limit_buy(
    symbol="BTCUSDT",
    quantity=0.001,
    price="60000.00"
)
print(f"订单 ID: {order['orderId']}, 状态: {order['status']}")

# 市价买（按数量）
client.order_market_buy(symbol="BTCUSDT", quantity=0.001)

# 市价买（按 USDT 金额）
client.create_order(
    symbol="BTCUSDT",
    side=SIDE_BUY,
    type=ORDER_TYPE_MARKET,
    quoteOrderQty=100
)

# 查订单
status = client.get_order(symbol="BTCUSDT", orderId=order["orderId"])
print(f"已成交: {status['executedQty']} / {status['origQty']}")

# 撤单
client.cancel_order(symbol="BTCUSDT", orderId=order["orderId"])

4. 合约下单（Futures）

# 查询合约持仓
positions = client.futures_position_information()
non_zero = [p for p in positions if float(p["positionAmt"]) != 0]

# 设置杠杆
client.futures_change_leverage(symbol="BTCUSDT", leverage=10)

# 合约限价做多
futures_order = client.futures_create_order(
    symbol="BTCUSDT",
    side="BUY",
    type="LIMIT",
    timeInForce="GTC",
    quantity=0.01,
    price="60000.00"
)
print(futures_order)

# 合约市价平仓
client.futures_create_order(
    symbol="BTCUSDT",
    side="SELL",
    type="MARKET",
    quantity=0.01,
    reduceOnly=True
)

5. WebSocket 订阅

from binance import ThreadedWebsocketManager

twm = ThreadedWebsocketManager(api_key=API_KEY, api_secret=SECRET_KEY)
twm.start()

def handle_ticker(msg):
    print(f"{msg['s']} 最新价 {msg['c']}, 24h量 {msg['v']}")

def handle_user(msg):
    if msg["e"] == "executionReport":
        print(f"订单 {msg['i']} 状态 {msg['X']}, 成交量 {msg['z']}")

twm.start_symbol_ticker_socket(callback=handle_ticker, symbol="BTCUSDT")
twm.start_user_socket(callback=handle_user)  # 订单流自动监听

twm.join()

6. 异步版本 AsyncClient

import asyncio
from binance import AsyncClient, BinanceSocketManager

async def main():
    client = await AsyncClient.create(API_KEY, SECRET_KEY)

    # 并行请求
    price, account = await asyncio.gather(
        client.get_symbol_ticker(symbol="BTCUSDT"),
        client.get_account()
    )
    print(price, len(account["balances"]))

    # WebSocket
    bsm = BinanceSocketManager(client)
    async with bsm.symbol_ticker_socket("BTCUSDT") as stream:
        for _ in range(10):
            msg = await stream.recv()
            print(msg)

    await client.close_connection()

asyncio.run(main())

三、ccxt 统一接口示例

1. 安装与初始化

pip install ccxt

import ccxt

exchange = ccxt.binance({
    "apiKey": "YOUR_KEY",
    "secret": "YOUR_SECRET",
    "enableRateLimit": True,  # 自动限频
    "options": {
        "defaultType": "spot"  # 或 "future"
    }
})

# 自动加载 markets
markets = exchange.load_markets()
print(f"支持 {len(markets)} 个交易对")

2. 统一风格的调用

ccxt 最大优势是 API 风格对所有交易所一致：

# 行情（任意交易所都是 fetch_ticker）
ticker = exchange.fetch_ticker("BTC/USDT")
print(f"最新价 {ticker['last']}, 涨幅 {ticker['percentage']}%")

# 余额
balance = exchange.fetch_balance()
print(balance["total"])  # {'BTC': 0.001, 'USDT': 100, ...}

# 下单
order = exchange.create_order(
    symbol="BTC/USDT",
    type="limit",
    side="buy",
    amount=0.001,
    price=60000
)

# 撤单
exchange.cancel_order(order["id"], symbol="BTC/USDT")

# 查询订单
my_orders = exchange.fetch_open_orders("BTC/USDT")

3. 切换到合约

exchange.options["defaultType"] = "future"

# 合约下单：symbol 仍是 BTC/USDT（内部映射到 BTCUSDT 永续）
order = exchange.create_order(
    symbol="BTC/USDT",
    type="limit",
    side="buy",
    amount=0.01,
    price=60000,
    params={"positionSide": "BOTH", "reduceOnly": False}
)

4. WebSocket（ccxt.pro，商业版）

# pip install ccxt --upgrade
import ccxt.pro as ccxtpro
import asyncio

async def watch_ticker():
    exchange = ccxtpro.binance({"apiKey": "KEY", "secret": "SECRET"})
    while True:
        ticker = await exchange.watch_ticker("BTC/USDT")
        print(ticker["last"])

asyncio.run(watch_ticker())

四、aiohttp 原生封装（高性能场景）

需要榨干最后 10ms 延迟时，自己封装最合适：

import aiohttp, asyncio, hmac, hashlib, time
from urllib.parse import urlencode

class BinanceAsync:
    def __init__(self, key, secret):
        self.key = key
        self.secret = secret
        self.base = "https://api.binance.com"
        self.session = None

    async def __aenter__(self):
        self.session = aiohttp.ClientSession(
            connector=aiohttp.TCPConnector(limit=100, ttl_dns_cache=300),
            timeout=aiohttp.ClientTimeout(total=5)
        )
        return self

    async def __aexit__(self, *args):
        await self.session.close()

    def _sign(self, params):
        params["timestamp"] = int(time.time() * 1000)
        query = urlencode(params)
        sig = hmac.new(self.secret.encode(), query.encode(), hashlib.sha256).hexdigest()
        return f"{query}&signature={sig}"

    async def get_account(self):
        query = self._sign({})
        async with self.session.get(
            f"{self.base}/api/v3/account?{query}",
            headers={"X-MBX-APIKEY": self.key}
        ) as r:
            return await r.json()

    async def place_order(self, symbol, side, qty, price):
        query = self._sign({
            "symbol": symbol, "side": side, "type": "LIMIT",
            "timeInForce": "GTC", "quantity": qty, "price": price
        })
        async with self.session.post(
            f"{self.base}/api/v3/order?{query}",
            headers={"X-MBX-APIKEY": self.key}
        ) as r:
            return await r.json()

async def demo():
    async with BinanceAsync("KEY", "SECRET") as api:
        account = await api.get_account()
        print(account["canTrade"])

asyncio.run(demo())

五、性能基准测试

在上海阿里云 ECS 到东京币安节点的实测：

import asyncio, time

async def benchmark(client, n=100):
    start = time.time()
    tasks = [client.get_symbol_ticker(symbol="BTCUSDT") for _ in range(n)]
    await asyncio.gather(*tasks)
    elapsed = time.time() - start
    return elapsed, n/elapsed

# 结果（n=100 并发）：
# python-binance AsyncClient: 8.2s, 12.2 req/s
# ccxt (同步包装异步):        12.5s, 8.0 req/s
# aiohttp 原生:              6.4s, 15.6 req/s

结论：中低频策略首选 python-binance；跨所套利选 ccxt；高频做市机（HFT）选自研 aiohttp 或 C++ 封装。

六、常见问题 FAQ

Q1: python-binance 和 binance-connector-python 有什么区别？

A: python-binance（sammchardy）是社区老牌库，功能最全、文档最多；binance-connector-python 是币安官方出的薄封装，功能少但紧跟 API 更新。生产环境推荐 python-binance。

Q2: ccxt 能不能订阅 WebSocket？

A: 免费版 ccxt 只有 REST；WebSocket 在 ccxt.pro（付费）。开源项目可用社区衍生的 ccxt-ws 替代，但稳定性不如原生 python-binance。

Q3: 调用报 APIError: Invalid API-key, IP, or permissions 怎么办？

A: 三种可能：1) API Key 或 Secret 复制错；2) 开了 IP 白名单但当前服务器 IP 不在列表；3) 调用的接口权限未勾选（例如合约接口没勾 Enable Futures）。

Q4: 如何测试 SDK 而不影响真实账户？

A: python-binance 和 ccxt 都支持 testnet，只需初始化时传 testnet=True。测试币在 https://testnet.binance.vision 免费领取 10000 USDT。

Q5: 生产环境需要连接池吗？

A: 需要。默认 requests 每次请求新建 TCP 连接，延迟增加 20-30ms。使用 requests.Session() 或 aiohttp 的 TCPConnector(limit=100) 可保持长连接，延迟降到 5ms 以内。

看完 Python 方案，回到 分类导航 查看「API接入」分类其它语言 SDK 教程。