バイナンスAPIをPythonで使う方法

バイナンスAPIの紹介

バイナンスAPIは、開発者にバイナンスの取引プラットフォームへのプログラム的なアクセスを提供し、取引ボット、アプリケーション、ツールなどの交換と対話するものを構築できるようにします。このチュートリアルでは、認証の設定から様々な種類のリクエストを送信するプロセスをガイドします。

多くの開発者が従来APIテストにはPostmanを利用してきましたが、Apidogは、バイナンスのような暗号通貨APIに特化した、より効率的で専門的な体験を提供します。その直感的なインターフェース、強化された認証処理、および暗号特有の機能により、Apidogは、バイナンスAPIエンドポイントのテストと実装を開始するために必要なセットアップ時間を大幅に短縮します。

このプラットフォームのリアルタイム共同作業ツールと包括的なドキュメント生成機能は、暗号取引アプリケーションに取り組むチームにとって特に価値があります。

バイナンスAPIの統合を効率的に行いたいなら、Apidogへの移行は、開発ワークフローを明らかに改善し、暗号プロジェクトの展開時間を短縮します。

バイナンスAPI認証の設定

バイナンスAPIを使用する前に、バイナンスアカウントからAPIキーを作成する必要があります。このセクションでは、APIキーの設定と保護方法について説明します。

バイナンスAPIキーの作成

1. バイナンスアカウントにログインします
2. アカウント設定で「API管理」に移動します
3. 新しいAPIキーを作成します
4. APIキーとシークレットキーを書き留めます
5. 制限（IPホワイトリスト、取引権限など）を設定します

バイナンスAPIがサポートするAPIキーの種類

バイナンスは、次の3種類のAPIキー認証方法をサポートしています：

1. HMACキー - 最も一般的な認証方法
2. RSAキー - 強化されたセキュリティ機能を提供
3. Ed25519キー - 最高のパフォーマンスとセキュリティを提供

HMAC認証の例

HMACキーを使った認証方法は次のとおりです：

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

# API credentials
api_key = 'your_api_key'
api_secret = 'your_api_secret'

# Base URL for Binance API
base_url = 'https://api.binance.com'

# Function to create a signature for authenticated endpoints
def get_signature(query_string):
    return hmac.new(
        api_secret.encode('utf-8'),
        query_string.encode('utf-8'),
        hashlib.sha256
    ).hexdigest()

# Example request to get account information
def get_account():
    endpoint = '/api/v3/account'
    timestamp = int(time.time() * 1000)

    params = {
        'timestamp': timestamp,
        'recvWindow': 5000  # Optional parameter
    }

    query_string = urlencode(params)
    signature = get_signature(query_string)

    url = f"{base_url}{endpoint}?{query_string}&signature={signature}"

    headers = {
        'X-MBX-APIKEY': api_key
    }

    response = requests.get(url, headers=headers)
    return response.json()

# Call the function
account_info = get_account()
print(account_info)

バイナンスAPIエンドポイントの理解

バイナンスAPIは、その機能に基づいていくつかのエンドポイントカテゴリに整理されています。

一般的なバイナンスAPIエンドポイント

これらのエンドポイントは、取引所の一般情報を提供します：

# Test connectivity
def test_connectivity():
    endpoint = '/api/v3/ping'
    url = f"{base_url}{endpoint}"
    response = requests.get(url)
    return response.json()

# Get server time
def get_server_time():
    endpoint = '/api/v3/time'
    url = f"{base_url}{endpoint}"
    response = requests.get(url)
    return response.json()

# Get exchange information
def get_exchange_info():
    endpoint = '/api/v3/exchangeInfo'
    url = f"{base_url}{endpoint}"
    response = requests.get(url)
    return response.json()

マーケットデータバイナンスAPIエンドポイント

これらのエンドポイントは、市場データにアクセスを提供します：

# Get order book for a symbol
def get_order_book(symbol, limit=100):
    endpoint = '/api/v3/depth'
    params = {
        'symbol': symbol,
        'limit': limit
    }
    url = f"{base_url}{endpoint}?{urlencode(params)}"
    response = requests.get(url)
    return response.json()

# Get recent trades
def get_recent_trades(symbol, limit=500):
    endpoint = '/api/v3/trades'
    params = {
        'symbol': symbol,
        'limit': limit
    }
    url = f"{base_url}{endpoint}?{urlencode(params)}"
    response = requests.get(url)
    return response.json()

# Get candlestick data
def get_klines(symbol, interval, start_time=None, end_time=None, limit=500):
    endpoint = '/api/v3/klines'
    params = {
        'symbol': symbol,
        'interval': interval,
        'limit': limit
    }

    if start_time:
        params['startTime'] = start_time
    if end_time:
        params['endTime'] = end_time

    url = f"{base_url}{endpoint}?{urlencode(params)}"
    response = requests.get(url)

    # Format the response
    klines = response.json()
    formatted_klines = []
    for k in klines:
        formatted_klines.append({
            'open_time': k[0],
            'open': float(k[1]),
            'high': float(k[2]),
            'low': float(k[3]),
            'close': float(k[4]),
            'volume': float(k[5]),
            'close_time': k[6],
            'quote_volume': float(k[7]),
            'trades': k[8],
            'taker_buy_base_volume': float(k[9]),
            'taker_buy_quote_volume': float(k[10])
        })

    return formatted_klines

バイナンスAPIを利用した取引

取引エンドポイントを使用すると、注文を出し、管理することができます。

バイナンスAPIを使用した注文の出し方

異なる種類の注文を出す方法は次のとおりです：

# Place a LIMIT order
def place_limit_order(symbol, side, quantity, price):
    endpoint = '/api/v3/order'
    timestamp = int(time.time() * 1000)

    params = {
        'symbol': symbol,
        'side': side,  # 'BUY' or 'SELL'
        'type': 'LIMIT',
        'timeInForce': 'GTC',  # Good Till Canceled
        'quantity': quantity,
        'price': price,
        'timestamp': timestamp,
        'recvWindow': 5000
    }

    query_string = urlencode(params)
    signature = get_signature(query_string)

    url = f"{base_url}{endpoint}?{query_string}&signature={signature}"

    headers = {
        'X-MBX-APIKEY': api_key
    }

    response = requests.post(url, headers=headers)
    return response.json()

# Place a MARKET order
def place_market_order(symbol, side, quantity):
    endpoint = '/api/v3/order'
    timestamp = int(time.time() * 1000)

    params = {
        'symbol': symbol,
        'side': side,  # 'BUY' or 'SELL'
        'type': 'MARKET',
        'quantity': quantity,
        'timestamp': timestamp,
        'recvWindow': 5000
    }

    query_string = urlencode(params)
    signature = get_signature(query_string)

    url = f"{base_url}{endpoint}?{query_string}&signature={signature}"

    headers = {
        'X-MBX-APIKEY': api_key
    }

    response = requests.post(url, headers=headers)
    return response.json()

# Place a STOP_LOSS order
def place_stop_loss_order(symbol, side, quantity, stop_price):
    endpoint = '/api/v3/order'
    timestamp = int(time.time() * 1000)

    params = {
        'symbol': symbol,
        'side': side,
        'type': 'STOP_LOSS',
        'quantity': quantity,
        'stopPrice': stop_price,
        'timestamp': timestamp,
        'recvWindow': 5000
    }

    query_string = urlencode(params)
    signature = get_signature(query_string)

    url = f"{base_url}{endpoint}?{query_string}&signature={signature}"

    headers = {
        'X-MBX-APIKEY': api_key
    }

    response = requests.post(url, headers=headers)
    return response.json()

バイナンスAPIを使用した注文の管理

注文の照会、キャンセル、追跡を行う方法は次のとおりです：

# Query order status
def query_order(symbol, order_id=None, orig_client_order_id=None):
    endpoint = '/api/v3/order'
    timestamp = int(time.time() * 1000)

    params = {
        'symbol': symbol,
        'timestamp': timestamp,
        'recvWindow': 5000
    }

    if order_id:
        params['orderId'] = order_id
    elif orig_client_order_id:
        params['origClientOrderId'] = orig_client_order_id
    else:
        raise ValueError("Either order_id or orig_client_order_id must be provided")

    query_string = urlencode(params)
    signature = get_signature(query_string)

    url = f"{base_url}{endpoint}?{query_string}&signature={signature}"

    headers = {
        'X-MBX-APIKEY': api_key
    }

    response = requests.get(url, headers=headers)
    return response.json()

# Cancel an order
def cancel_order(symbol, order_id=None, orig_client_order_id=None):
    endpoint = '/api/v3/order'
    timestamp = int(time.time() * 1000)

    params = {
        'symbol': symbol,
        'timestamp': timestamp,
        'recvWindow': 5000
    }

    if order_id:
        params['orderId'] = order_id
    elif orig_client_order_id:
        params['origClientOrderId'] = orig_client_order_id
    else:
        raise ValueError("Either order_id or orig_client_order_id must be provided")

    query_string = urlencode(params)
    signature = get_signature(query_string)

    url = f"{base_url}{endpoint}?{query_string}&signature={signature}"

    headers = {
        'X-MBX-APIKEY': api_key
    }

    response = requests.delete(url, headers=headers)
    return response.json()

# Get all open orders
def get_open_orders(symbol=None):
    endpoint = '/api/v3/openOrders'
    timestamp = int(time.time() * 1000)

    params = {
        'timestamp': timestamp,
        'recvWindow': 5000
    }

    if symbol:
        params['symbol'] = symbol

    query_string = urlencode(params)
    signature = get_signature(query_string)

    url = f"{base_url}{endpoint}?{query_string}&signature={signature}"

    headers = {
        'X-MBX-APIKEY': api_key
    }

    response = requests.get(url, headers=headers)
    return response.json()

バイナンスAPIを使用したアカウントおよびユーザーデータ

これらのエンドポイントを使用すると、アカウント情報にアクセスし、ユーザーデータをストリーミングできます。

バイナンスAPIを使用したアカウント情報

# Get account information
def get_account_information():
    endpoint = '/api/v3/account'
    timestamp = int(time.time() * 1000)

    params = {
        'timestamp': timestamp,
        'recvWindow': 5000
    }

    query_string = urlencode(params)
    signature = get_signature(query_string)

    url = f"{base_url}{endpoint}?{query_string}&signature={signature}"

    headers = {
        'X-MBX-APIKEY': api_key
    }

    response = requests.get(url, headers=headers)
    return response.json()

# Get account trade list
def get_account_trades(symbol, start_time=None, end_time=None, limit=500):
    endpoint = '/api/v3/myTrades'
    timestamp = int(time.time() * 1000)

    params = {
        'symbol': symbol,
        'timestamp': timestamp,
        'limit': limit,
        'recvWindow': 5000
    }

    if start_time:
        params['startTime'] = start_time
    if end_time:
        params['endTime'] = end_time

    query_string = urlencode(params)
    signature = get_signature(query_string)

    url = f"{base_url}{endpoint}?{query_string}&signature={signature}"

    headers = {
        'X-MBX-APIKEY': api_key
    }

    response = requests.get(url, headers=headers)
    return response.json()

バイナンスAPIを使用したユーザーデータストリーム

# Start a user data stream
def start_user_data_stream():
    endpoint = '/api/v3/userDataStream'

    headers = {
        'X-MBX-APIKEY': api_key
    }

    response = requests.post(f"{base_url}{endpoint}", headers=headers)
    return response.json()

# Keep alive a user data stream
def keep_alive_user_data_stream(listen_key):
    endpoint = '/api/v3/userDataStream'

    params = {
        'listenKey': listen_key
    }

    headers = {
        'X-MBX-APIKEY': api_key
    }

    response = requests.put(f"{base_url}{endpoint}?{urlencode(params)}", headers=headers)
    return response.json()

# Close a user data stream
def close_user_data_stream(listen_key):
    endpoint = '/api/v3/userDataStream'

    params = {
        'listenKey': listen_key
    }

    headers = {
        'X-MBX-APIKEY': api_key
    }

    response = requests.delete(f"{base_url}{endpoint}?{urlencode(params)}", headers=headers)
    return response.json()

バイナンスAPIの高度な機能

このセクションでは、バイナンスAPIのより高度な機能について説明します。

バイナンスAPIにおける注文リスト（OCO）の操作

一方がトリガーされると他方が自動的にキャンセルされるペアの注文を配置することを許可するOCO（One-Cancels-the-Other）注文：

# Place an OCO order
def place_oco_order(symbol, side, quantity, price, stop_price):
    endpoint = '/api/v3/orderList/oco'
    timestamp = int(time.time() * 1000)

    params = {
        'symbol': symbol,
        'side': side,
        'quantity': quantity,
        'price': price,
        'stopPrice': stop_price,
        'timestamp': timestamp,
        'recvWindow': 5000
    }

    # Define the types for the OCO order based on side
    if side == 'SELL':
        params['aboveType'] = 'STOP_LOSS'
        params['belowType'] = 'LIMIT_MAKER'
    else:
        params['aboveType'] = 'LIMIT_MAKER'
        params['belowType'] = 'STOP_LOSS'

    params['aboveTimeInForce'] = 'GTC'
    params['belowTimeInForce'] = 'GTC'

    query_string = urlencode(params)
    signature = get_signature(query_string)

    url = f"{base_url}{endpoint}?{query_string}&signature={signature}"

    headers = {
        'X-MBX-APIKEY': api_key
    }

    response = requests.post(url, headers=headers)
    return response.json()

バイナンスAPIにおけるレート制限の処理

バイナンスはAPIリクエストに対してレート制限を適用します。これを処理する方法は次のとおりです：

import time
from functools import wraps

# Decorator to handle rate limits
def handle_rate_limit(max_retries=3, retry_delay=30):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            retries = 0
            while retries <= max_retries:
                try:
                    response = func(*args, **kwargs)
                    if 'code' in response and response['code'] == -1429:  # Rate limit exceeded
                        retries += 1
                        if retries > max_retries:
                            raise Exception(f"Rate limit exceeded after {max_retries} retries")
                        print(f"レート制限を超えました。{retry_delay}秒待機して再試行します...")
                        time.sleep(retry_delay)
                    else:
                        return response
                except Exception as e:
                    if "429" in str(e) or "418" in str(e):
                        retries += 1
                        if retries > max_retries:
                            raise
                        print(f"レート制限を超えました。{retry_delay}秒待機して再試行します...")
                        time.sleep(retry_delay)
                    else:
                        raise
        return wrapper
    return decorator

# Example usage
@handle_rate_limit()
def get_exchange_info_with_retry():
    endpoint = '/api/v3/exchangeInfo'
    url = f"{base_url}{endpoint}"
    response = requests.get(url)
    return response.json()

完全なバイナンスAPI取引ボットの構築

すべてをまとめてシンプルな取引ボットを構築しましょう：

import hmac
import hashlib
import time
import requests
import json
import websocket
import threading
from urllib.parse import urlencode

class BinanceTrader:
    def __init__(self, api_key, api_secret, symbol='BTCUSDT'):
        self.api_key = api_key
        self.api_secret = api_secret
        self.base_url = 'https://api.binance.com'
        self.symbol = symbol
        self.ws = None
        self.listen_key = None
        self.keep_alive_thread = None

    def get_signature(self, query_string):
        return hmac.new(
            self.api_secret.encode('utf-8'),
            query_string.encode('utf-8'),
            hashlib.sha256
        ).hexdigest()

    def get_server_time(self):
        endpoint = '/api/v3/time'
        response = requests.get(f"{self.base_url}{endpoint}")
        return response.json()['serverTime']

    def get_account_info(self):
        endpoint = '/api/v3/account'
        timestamp = int(time.time() * 1000)

        params = {
            'timestamp': timestamp,
            'recvWindow': 5000
        }

        query_string = urlencode(params)
        signature = self.get_signature(query_string)

        url = f"{self.base_url}{endpoint}?{query_string}&signature={signature}"

        headers = {
            'X-MBX-APIKEY': self.api_key
        }

        response = requests.get(url, headers=headers)
        return response.json()

    def place_market_order(self, side, quantity):
        endpoint = '/api/v3/order'
        timestamp = int(time.time() * 1000)

        params = {
            'symbol': self.symbol,
            'side': side,
            'type': 'MARKET',
            'quantity': quantity,
            'timestamp': timestamp,
            'recvWindow': 5000
        }

        query_string = urlencode(params)
        signature = self.get_signature(query_string)

        url = f"{self.base_url}{endpoint}?{query_string}&signature={signature}"

        headers = {
            'X-MBX-APIKEY': self.api_key
        }

        response = requests.post(url, headers=headers)
        return response.json()

    def start_user_data_stream(self):
        endpoint = '/api/v3/userDataStream'

        headers = {
            'X-MBX-APIKEY': self.api_key
        }

        response = requests.post(f"{self.base_url}{endpoint}", headers=headers)
        self.listen_key = response.json()['listenKey']
        return self.listen_key

    def keep_alive_user_data_stream(self):
        while True:
            if self.listen_key:
                endpoint = '/api/v3/userDataStream'

                params = {
                    'listenKey': self.listen_key
                }

                headers = {
                    'X-MBX-APIKEY': self.api_key
                }

                requests.put(f"{self.base_url}{endpoint}?{urlencode(params)}", headers=headers)
            time.sleep(30 * 60)  # 30分ごとに維持

    def on_message(self, ws, message):
        data = json.loads(message)

        # Handle account updates
        if data.get('e') == 'outboundAccountPosition':
            print("アカウント更新を受信しました:")
            print(f"更新時間: {data['u']}")
            print("残高:")
            for balance in data['B']:
                print(f"資産: {balance['a']}, 自由: {balance['f']}, ロック: {balance['l']}")

        # Handle order updates
        elif data.get('e') == 'executionReport':
            print("注文更新を受信しました:")
            print(f"シンボル: {data['s']}")
            print(f"サイド: {data['S']}")
            print(f"注文タイプ: {data['o']}")
            print(f"注文ステータス: {data['X']}")
            print(f"価格: {data['p']}")
            print(f"数量: {data['q']}")
            print(f"実行された数量: {data['z']}")

            # Implement trading logic here based on order updates
            if data['X'] == 'FILLED' and data['S'] == 'BUY':
                # 例: 買いが成立したら売り注文を出す
                sell_price = float(data['p']) * 1.05  # 5%の利益目標
                self.place_limit_order('SELL', data['q'], sell_price)

    def on_error(self, ws, error):
        print(f"エラー: {error}")

    def on_close(self, ws, close_status_code, close_msg):
        print("WebSocket接続が閉じられました")

    def on_open(self, ws):
        print("WebSocket接続がオープンしました")

    def start_websocket(self):
        self.start_user_data_stream()
        socket_url = f"wss://stream.binance.com:9443/ws/{self.listen_key}"

        self.ws = websocket.WebSocketApp(
            socket_url,
            on_message=self.on_message,
            on_error=self.on_error,
            on_close=self.on_close,
            on_open=self.on_open
        )

        # Start keep-alive thread
        self.keep_alive_thread = threading.Thread(target=self.keep_alive_user_data_stream)
        self.keep_alive_thread.daemon = True
        self.keep_alive_thread.start()

        # Run the WebSocket in a separate thread
        ws_thread = threading.Thread(target=self.ws.run_forever)
        ws_thread.daemon = True
        ws_thread.start()

        return ws_thread

# Example usage
if __name__ == "__main__":
    api_key = "your_api_key"
    api_secret = "your_api_secret"

    trader = BinanceTrader(api_key, api_secret, 'BTCUSDT')

    # Get account information
    account_info = trader.get_account_info()
    print("アカウント情報:")
    print(f"取引可能: {account_info['canTrade']}")
    print(f"引き出し可能: {account_info['canWithdraw']}")
    print(f"入金可能: {account_info['canDeposit']}")

    # Print balances
    print("\n残高:")
    for balance in account_info['balances']:
        if float(balance['free']) > 0 or float(balance['locked']) > 0:
            print(f"資産: {balance['asset']}, 自由: {balance['free']}, ロック: {balance['locked']}")

    # Start the WebSocket for real-time updates
    ws_thread = trader.start_websocket()

    try:
        # Keep the main thread running
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        print("停止中...")
        trader.ws.close()

バイナンスAPIを使用するためのベストプラクティス

バイナンスAPIを使用する際に従うべきベストプラクティスは以下の通りです：

バイナンスAPIのセキュリティプラクティス

1. APIキーの権限を制限する: 必要な権限のみを有効にします（例：読み取り専用、取引、引き出し）
2. IPアドレスをホワイトリストに登録する: APIキーを特定のIPアドレスのみで動作するように制限します
3. APIキーの安全な保管: コード内にAPIキーをハードコーディングせず、環境変数やセキュアボルトを使用します
4. 定期的なキーのローテーション: リスクを軽減するためにAPIキーを定期的に変更します

import os

# Load API keys from environment variables
api_key = os.environ.get('BINANCE_API_KEY')
api_secret = os.environ.get('BINANCE_API_SECRET')

if not api_key or not api_secret:
    raise ValueError("環境変数にAPIキーが見つかりませんでした")

バイナンスAPIによるエラーハンドリング

APIエラーや接続問題に対処するために、堅牢なエラーハンドリングを実装します：

def safe_request(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        try:
            response = func(*args, **kwargs)
            # Check for API error codes
            if isinstance(response, dict) and 'code' in response:
                print(f"APIエラー: コード {response['code']}, メッセージ: {response.get('msg', 'メッセージなし')}")
                # Handle specific error codes
                if response['code'] == -1021:  # INVALID_TIMESTAMP
                    print("サーバーとローカル時間が同期していません。タイムスタンプを調整しています...")
                    # Implement timestamp adjustment logic here
                elif response['code'] == -2010:  # INSUFFICIENT_BALANCE
                    print("注文に対する残高が不足しています。")
                # Add more specific error handling as needed
            return response
        except requests.exceptions.RequestException as e:
            print(f"リクエストエラー: {e}")
            # Implement retry logic or fallback behavior
            return None
        except json.JSONDecodeError:
            print("JSONレスポンスのデコードに失敗しました")
            return None
        except Exception as e:
            print(f"予期しないエラー: {e}")
            return None
    return wrapper

@safe_request
def get_account_info_safe():
    # Implementation of get_account_info with proper error handling
    # ...

結論

バイナンスAPIは、バイナンス取引所とプログラム的に対話する強力な方法を提供します。このチュートリアルでは、認証から注文の発行、アカウント管理に至るまで、APIの使用に関する重要な側面を取り上げました。ベストプラクティスを遵守し、適切なエラーハンドリングを実装することで、バイナンスの豊富な機能セットを活用する堅牢なアプリケーションや取引ボットを構築できます。

APIキーを安全に保つこと、レート制限を尊重すること、実際の資金で使用する前にテスト環境でコードを十分にテストすることを忘れないでください。バイナンスがAPIを更新する際には、エンドポイントやパラメータの変更についての情報を常に把握し、アプリケーションが正常に機能し続けるようにしてください。

このチュートリアルから得た知識を基に、自分自身の取引ツールやアプリケーションをバイナンスAPIを使用して構築し始める準備が整ったでしょう。