3. 多币种数据批量下载

任务目标

编写一个Python脚本，从币安公开API批量下载多个加密货币的日线K线数据，统一为DataFrame格式并存储为CSV文件，便于后续量化策略回测使用。

环境依赖

需要 requests、pandas 两个库，安装命令：

pip install requests pandas

核心实现

脚本通过币安REST API的 /api/v3/klines 接口获取K线数据。每次请求最多返回1000根K线，因此采用循环分页拉取全部历史数据。下载完成后统一列名并保存为CSV。

import time
import requests
import pandas as pd
from datetime import datetime, timezone

# 配置区
SYMBOLS = ["BTCUSDT", "ETHUSDT", "BNBUSDT"]  # 要下载的交易对
INTERVAL = "1d"                                # K线周期
START_STR = "2020-01-01"                       # 起始日期
END_STR = "2024-12-31"                         # 结束日期
OUTPUT_DIR = "./history_data"                  # 保存目录

BASE_URL = "https://api.binance.com/api/v3/klines"
COLUMNS = [
    "open_time", "open", "high", "low", "close", "volume",
    "close_time", "quote_volume", "count",
    "taker_buy_volume", "taker_buy_quote_volume", "ignore"
]

def fetch_klines(symbol: str, interval: str, start_ms: int, end_ms: int) -> list:
    """分页拉取全部K线数据"""
    all_rows = []
    current_start = start_ms
    while current_start < end_ms:
        params = {
            "symbol": symbol,
            "interval": interval,
            "startTime": current_start,
            "endTime": end_ms,
            "limit": 1000
        }
        resp = requests.get(BASE_URL, params=params, timeout=10)
        resp.raise_for_status()
        rows = resp.json()
        if not rows:
            break
        all_rows.extend(rows)
        # 下一次请求从最后一根K线的close_time + 1开始
        current_start = rows[-1][6] + 1
        time.sleep(0.2)  # 避免触发频率限制
    return all_rows

def download_symbol(symbol: str) -> pd.DataFrame:
    """下载单个交易对并转为DataFrame"""
    start_dt = datetime.strptime(START_STR, "%Y-%m-%d").replace(tzinfo=timezone.utc)
    end_dt = datetime.strptime(END_STR, "%Y-%m-%d").replace(tzinfo=timezone.utc)
    start_ms = int(start_dt.timestamp() * 1000)
    end_ms = int(end_dt.timestamp() * 1000)

    raw = fetch_klines(symbol, INTERVAL, start_ms, end_ms)
    df = pd.DataFrame(raw, columns=COLUMNS)

    # 统一格式：保留数值列并转换类型
    numeric_cols = ["open", "high", "low", "close", "volume"]
    for col in numeric_cols:
        df[col] = df[col].astype(float)
    df["open_time"] = pd.to_datetime(df["open_time"], unit="ms")
    df = df[["open_time", "open", "high", "low", "close", "volume"]]
    return df

def main():
    import os
    os.makedirs(OUTPUT_DIR, exist_ok=True)
    for symbol in SYMBOLS:
        print(f"正在下载 {symbol} ...")
        df = download_symbol(symbol)
        file_path = f"{OUTPUT_DIR}/{symbol}_{INTERVAL}.csv"
        df.to_csv(file_path, index=False)
        print(f"已保存 {len(df)} 条记录 -> {file_path}")

if __name__ == "__main__":
    main()

运行方式

将上述代码保存为 download_history.py，在命令行执行：

python download_history.py

验证方法

运行结束后检查 history_data 目录，应生成3个CSV文件。用以下命令快速验证数据完整性：

import pandas as pd
df = pd.read_csv("history_data/BTCUSDT_1d.csv")
print(df.shape)        # 应约为 (1827, 6)，即2020-2024年约5年日线数据
print(df.head(3))      # 查看前3行，open_time应从2020-01-01开始
print(df["close"].isna().sum())  # 应为0，无缺失值

通过条件：每个CSV文件均包含6列（时间、开高低收、成交量），时间范围覆盖设定的起止日期，且无空值。若网络不稳定导致请求失败，脚本会抛出异常，可适当增大 time.sleep 间隔后重试。