幣安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 教程。