バイナンスAPIをPythonで動かす方法｜python-binanceとccxtの実装ガイド

バイナンスの Python SDK において、最も成熟した3つの選択肢は、python-binance（公式推奨）、ccxt（複数取引所の統一インターフェース）、aiohttp ネイティブ実装（高性能な自作）です。これらはそれぞれ「迅速な開発」「複数取引所の互換性」「究極のパフォーマンス」という異なるニーズに対応しています。本記事では、これら3つのインストール、署名付き呼び出し、注文、WebSocket 購読の完全なコードを提示し、パフォーマンスのベンチマークテスト結果を付記します。バイナンスのアカウントをお持ちでない方は、まず バイナンス公式サイト で本人確認（KYC）を完了させてください。新規ユーザーの方は 無料登録 から口座を開設できます。

1. 3つの SDK 比較表

2. python-binance の完全な実装例

インストールと初期化

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}

残高と相場の確認

# 現物アカウントの残高
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']}")

# ローソク足
klines = client.get_klines(symbol="BTCUSDT", interval=Client.KLINE_INTERVAL_1HOUR, limit=100)
for k in klines[-5:]:
    print(f"時間 {k[0]}, 始値:{k[1]} 高値:{k[2]} 安値:{k[3]} 終値:{k[4]} 出来高:{k[5]}")

# 板情報（オーダーブック）
depth = client.get_order_book(symbol="BTCUSDT", limit=20)
print(f"最良買気配 {depth['bids'][0]}, 最良売気配 {depth['asks'][0]}")

現物注文の実行

# 指値注文
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"])

先物注文（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
)

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()

非同期バージョン 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())

3. ccxt 統一インターフェースの実装例

インストールと初期化

pip install ccxt

import ccxt

exchange = ccxt.binance({
    "apiKey": "YOUR_KEY",
    "secret": "YOUR_SECRET",
    "enableRateLimit": True,  # 自動レート制限
    "options": {
        "defaultType": "spot"  # または "future"
    }
})

# マーケット情報のロード
markets = exchange.load_markets()
print(f"{len(markets)} 個の通貨ペアをサポートしています")

統一スタイルの呼び出し

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")

先物への切り替え

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

# 先物注文：シンボルは引き続き BTC/USDT（内部で BTCUSDT 永続にマッピング）
order = exchange.create_order(
    symbol="BTC/USDT",
    type="limit",
    side="buy",
    amount=0.01,
    price=60000,
    params={"positionSide": "BOTH", "reduceOnly": False}
)

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())

4. 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())

5. パフォーマンス・ベンチマークテスト

上海の阿里云 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++ 実装を推奨します。

6. よくある質問 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: 以下の3つの可能性があります。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 チュートリアルをご覧ください。