3. 数据采集基础：使用 requests 库抓取公开价格 API

数据采集是定价稳定性监控的起点。本章将带你掌握使用 Python 的 requests 库从 CoinGecko、股票行情等公开 API 抓取价格数据，并完成 JSON 解析与数据清洗。你将学会构建一个健壮的数据采集层，为后续的监控与告警打下基础。

3.1 公开价格 API 概览

选择合适的 API 是数据采集的第一步。下表对比了三种常见的公开价格 API 及其特点：

3.2 使用 requests 库发送 HTTP 请求

requests 是 Python 中最流行的 HTTP 库。安装后，只需几行代码即可获取 API 数据。

3.2.1 安装与导入

# 安装 requests 库（如未安装）
# pip install requests

import requests
import json
from datetime import datetime

3.2.2 发送 GET 请求获取比特币价格

CoinGecko 的 /simple/price 端点可以快速获取指定币种的当前价格。

# 定义 API 端点
url = "https://api.coingecko.com/api/v3/simple/price"

# 定义请求参数
params = {
    "ids": "bitcoin",
    "vs_currencies": "usd"
}

# 发送 GET 请求
response = requests.get(url, params=params)

# 检查响应状态码
if response.status_code == 200:
    data = response.json()
    print(data)  # 输出: {'bitcoin': {'usd': 67000.0}}
else:
    print(f"请求失败，状态码: {response.status_code}")

3.3 JSON 解析与数据提取

API 返回的数据通常是 JSON 格式。Python 的 json 模块和 requests 内置的 .json() 方法可以轻松解析。

3.3.1 解析嵌套 JSON

CoinGecko 的返回结构是嵌套字典。以下示例展示了如何提取价格和时间戳：

# 获取更详细的数据（包含市场数据）
url_detail = "https://api.coingecko.com/api/v3/coins/bitcoin"
params_detail = {
    "localization": "false",
    "tickers": "false",
    "market_data": "true"
}

response_detail = requests.get(url_detail, params=params_detail)
data_detail = response_detail.json()

# 提取当前价格
current_price = data_detail["market_data"]["current_price"]["usd"]
# 提取 24 小时最高价
high_24h = data_detail["market_data"]["high_24h"]["usd"]
# 提取 24 小时最低价
low_24h = data_detail["market_data"]["low_24h"]["usd"]

print(f"当前价格: ${current_price}")
print(f"24h 最高: ${high_24h}")
print(f"24h 最低: ${low_24h}")

3.3.2 处理缺失字段

API 数据有时会缺少某些字段。使用 .get() 方法可以安全地提取数据，避免 KeyError。

# 安全提取数据
price = data_detail.get("market_data", {}).get("current_price", {}).get("usd", None)
if price is None:
    print("价格数据不可用")
else:
    print(f"价格: ${price}")

3.4 数据清洗与格式化

原始 API 数据通常需要清洗才能用于监控。常见清洗步骤包括：

• 类型转换： 将字符串数字转换为 float。
• 时间戳格式化： 将 Unix 时间戳转换为可读的日期时间。
• 缺失值处理： 填充或删除空值。
• 单位统一： 确保所有价格使用同一货币单位。

# 数据清洗示例
def clean_price_data(raw_data):
    """清洗价格数据，返回结构化字典"""
    cleaned = {}
    
    # 提取并转换价格
    market_data = raw_data.get("market_data", {})
    cleaned["price_usd"] = float(market_data.get("current_price", {}).get("usd", 0))
    cleaned["high_24h"] = float(market_data.get("high_24h", {}).get("usd", 0))
    cleaned["low_24h"] = float(market_data.get("low_24h", {}).get("usd", 0))
    
    # 格式化时间戳
    timestamp = raw_data.get("last_updated")
    if timestamp:
        # CoinGecko 返回 ISO 8601 格式时间
        cleaned["timestamp"] = datetime.fromisoformat(timestamp.replace("Z", "+00:00"))
    else:
        cleaned["timestamp"] = datetime.utcnow()
    
    # 计算价格变动百分比
    cleaned["price_change_24h"] = round(
        (cleaned["price_usd"] - cleaned["high_24h"]) / cleaned["high_24h"] * 100, 2
    )
    
    return cleaned

# 使用示例
cleaned_data = clean_price_data(data_detail)
print(cleaned_data)

3.5 数据采集流程架构

下图展示了从 API 请求到清洗后数据的完整流程：

3.6 错误处理与重试机制

网络请求不可靠，必须加入错误处理。以下是一个带重试机制的采集函数：

import time
from requests.exceptions import RequestException

def fetch_with_retry(url, params, max_retries=3, delay=2):
    """带重试机制的 API 请求函数"""
    for attempt in range(max_retries):
        try:
            response = requests.get(url, params=params, timeout=10)
            response.raise_for_status()  # 触发 HTTPError
            return response.json()
        except RequestException as e:
            print(f"第 {attempt+1} 次请求失败: {e}")
            if attempt < max_retries - 1:
                time.sleep(delay * (attempt + 1))  # 指数退避
            else:
                print("达到最大重试次数，请求失败")
                return None

# 使用示例
data = fetch_with_retry(
    url="https://api.coingecko.com/api/v3/simple/price",
    params={"ids": "bitcoin", "vs_currencies": "usd"}
)
if data:
    print(f"获取成功: {data}")

3.7 多资产批量采集

实际监控中通常需要同时采集多个资产。CoinGecko 支持批量查询：

# 批量查询多个币种
assets = ["bitcoin", "ethereum", "ripple", "cardano", "solana"]
params_batch = {
    "ids": ",".join(assets),
    "vs_currencies": "usd",
    "include_24hr_change": "true"
}

data_batch = fetch_with_retry(
    url="https://api.coingecko.com/api/v3/simple/price",
    params=params_batch
)

if data_batch:
    for asset, info in data_batch.items():
        price = info.get("usd", "N/A")
        change = info.get("usd_24h_change", "N/A")
        print(f"{asset}: ${price} (24h 变化: {change:.2f}%)")

3.8 数据持久化：保存为 CSV

清洗后的数据可以保存到本地文件，便于后续分析和历史回溯。

import csv

def save_to_csv(data, filename="price_data.csv"):
    """将清洗后的数据保存为 CSV 文件"""
    with open(filename, mode="a", newline="", encoding="utf-8") as file:
        writer = csv.writer(file)
        # 如果文件为空，写入表头
        if file.tell() == 0:
            writer.writerow(["timestamp", "asset", "price_usd", "high_24h", "low_24h", "change_24h"])
        
        for asset, info in data.items():
            writer.writerow([
                datetime.utcnow().isoformat(),
                asset,
                info.get("usd", ""),
                info.get("usd_24h_high", ""),
                info.get("usd_24h_low", ""),
                info.get("usd_24h_change", "")
            ])

# 使用示例
save_to_csv(data_batch)

3.9 本章小结

通过本章学习，你已经掌握了：

• 使用 requests 库发送 GET 请求并处理响应。
• 解析嵌套 JSON 数据并安全提取字段。
• 执行数据清洗（类型转换、时间格式化、缺失值处理）。
• 实现带重试机制的健壮采集函数。
• 批量采集多资产数据并持久化到 CSV 文件。