Binance Futures Historical Data: Python Automation

안녕하세요! 오늘은 **Binance Futures的历史数据批量下载**를 Python으로 자동화하는 방법에 대해 알아보겠습니다. Binance Futures는全球最大の暗号通貨先物取引所の1つであり、トレーダーや研究者にとって 풍부な歷史データへのアクセスが重要です。

目次

1. [Binance Futures APIとは？](#binance-futures-apiとは) 2. [事前準備](#事前準備) 3. [Python実装](#python実装) 4. [大量データダウンロードの最適化](#大量データダウンロードの最適化) 5. [エラー処理とベストプラクティス](#エラー処理とベストプラクティス) 6. [よくある問題と解決策](#よくある問題と解決策) ---

Binance Futures APIとは？

Binance Futuresは、世界最大級の暗号通貨先物取引所で、 풍부한歷史市場データを提供します。APIを活用することで、以下のようなデータが取得可能です： - **K线数据**（ローソク足）：1分足から1ヶ月足まで - **取引データ**：約定履歴 - **注文簿データ**：板情報 - **ポジション情報**：保有ポジション - **資金調達率**：Funding Rate

APIエンドポイント一覧

| エンドポイント | 説明 | 制限 | |--------------|------|------| | /fapi/v1/klines | K线データ取得 | 1200 requests/min | | /fapi/v1/aggTrades | 集約取引取得 | 1200 requests/min | | /fapi/v1/fundingRate | 資金調達率 | 1200 requests/min | | /fapi/v1/openInterest | オープンインタレスト | 1200 requests/min | ---

事前準備

必要なライブラリのインストール

pip install requests pandas python-dotenv asyncio aiohttp

プロジェクト構造

binance_futures_data/
├── config.py          # API設定
├── client.py          # APIクライアント
├── downloader.py      # データダウンロード
├── utils.py           # ユーティリティ関数
├── requirements.txt   # 依存関係
└── data/              # データ保存先

---

Python実装

1. 基本設定（config.py）

import os
from dotenv import load_dotenv

load_dotenv()

Binance Futures API設定
BINANCE_BASE_URL = "https://fapi.binance.com"
BINANCE_WS_URL = "wss://fstream.binance.com/ws"

APIキー（オプション：パブリックエンドポイントには不要）
API_KEY = os.getenv("BINANCE_API_KEY", "")
API_SECRET = os.getenv("BINANCE_API_SECRET", "")

レートリミット設定
REQUESTS_PER_MINUTE = 1100  # 安全マージンあり
REQUEST_DELAY = 60 / REQUESTS_PER_MINUTE

データ保存設定
DATA_DIR = "./data"
SAVE_FORMAT = "parquet"  # parquet, csv, json

2. APIクライアント（client.py）

import requests
import time
from typing import Optional, List, Dict, Any
from datetime import datetime
import pandas as pd

class BinanceFuturesClient:
    """Binance Futures APIクライアント"""
    
    def __init__(self, base_url: str = "https://fapi.binance.com"):
        self.base_url = base_url
        self.session = requests.Session()
        self.last_request_time = 0
        
    def _rate_limit(self):
        """レートリミット制御"""
        elapsed = time.time() - self.last_request_time
        if elapsed < REQUEST_DELAY:
            time.sleep(REQUEST_DELAY - elapsed)
        self.last_request_time = time.time()
    
    def _request(self, method: str, endpoint: str, **kwargs) -> Dict[str, Any]:
        """HTTPリクエスト実行"""
        url = f"{self.base_url}{endpoint}"
        self._rate_limit()
        
        response = self.session.request(method, url, **kwargs)
        response.raise_for_status()
        
        return response.json()
    
    def get_klines(
        self,
        symbol: str,
        interval: str,
        start_time: Optional[int] = None,
        end_time: Optional[int] = None,
        limit: int = 1000
    ) -> pd.DataFrame:
        """
        K线データ（ローソク足）を取得
        
        Args:
            symbol: 取引ペア（例：BTCUSDT）
            interval: 間隔（1m, 5m, 1h, 1d, etc.）
            start_time: 開始時間（ミリ秒）
            end_time: 終了時間（ミリ秒）
            limit: 取得件数（最大1000）
        """
        params = {
            "symbol": symbol.upper(),
            "interval": interval,
            "limit": limit
        }
        
        if start_time:
            params["startTime"] = start_time
        if end_time:
            params["endTime"] = end_time
        
        data = self._request("GET", "/fapi/v1/klines", params=params)
        
        # DataFrameに変換
        df = pd.DataFrame(data, columns=[
            "open_time", "open", "high", "low", "close", "volume",
            "close_time", "quote_volume", "trades", "taker_buy_volume",
            "taker_buy_quote_volume", "ignore"
        ])
        
        # 数値型に変換
        for col in ["open", "high", "low", "close", "volume", "quote_volume"]:
            df[col] = pd.to_numeric(df[col])
        
        # 日時変換
        df["open_time"] = pd.to_datetime(df["open_time"], unit="ms")
        df["close_time"] = pd.to_datetime(df["close_time"], unit="ms")
        
        return df
    
    def get_funding_rate(
        self,
        symbol: str,
        start_time: Optional[int] = None,
        end_time: Optional[int] = None,
        limit: int = 1000
    ) -> pd.DataFrame:
        """資金調達率を取得"""
        params = {
            "symbol": symbol.upper(),
            "limit": limit
        }
        
        if start_time:
            params["startTime"] = start_time
        if end_time:
            params["endTime"] = end_time
        
        data = self._request("GET", "/fapi/v1/fundingRate", params=params)
        
        df = pd.DataFrame(data)
        df["fundingTime"] = pd.to_datetime(df["fundingTime"], unit="ms")
        df["fundingRate"] = pd.to_numeric(df["fundingRate"])
        
        return df

3. 的大量データダウンロード（downloader.py）

```python import asyncio import aiohttp from datetime import datetime, timedelta from typing import List, Tuple, Optional import pandas as pd from pathlib import Path import json class FuturesDataDownloader: """Binance Futures 大量データ一括ダウンロード""" def __init__( self, base_url: str = "https://fapi.binance.com", save_dir: str = "./data" ): self.base_url = base_url self.save_dir = Path(save_dir) self.save_dir.mkdir(parents=True, exist_ok=True) async def _fetch_klines_async( self, session: aiohttp.ClientSession, symbol: str, interval: str, start_time: int, end_time: int, limit: int = 1000 ) -> List: """非同期でK线データを取得""" url = f"{self.base_url}/fapi/v1/klines" params = { "symbol": symbol.upper(), "interval": interval, "startTime": start_time, "endTime": end_time, "limit": limit } async with session.get(url, params=params) as response: response.raise_for_status() return await response.json() async def download_historical_klines( self, symbol: str, interval: str, start_date: datetime, end_date: datetime, delay: float = 0.1 ) -> pd.DataFrame: """ 歴史的K线データを一括ダウンロード Args: symbol: 取引ペア interval: 時間間隔 start_date: 開始日時 end_date