4. 数据采集入门：使用 Python requests 库获取实时行情数据

实时行情数据是坏账率哨兵系统的第一道防线。只有稳定、准确地获取交易所的报价数据，后续的坏账计算、阈值预警才有意义。本章从零开始，带你掌握 requests 库的核心用法，并完成一次真实的行情数据采集。

4.1 为什么选择 requests 库？

Python 标准库中的 urllib 虽然也能发送 HTTP 请求，但接口繁琐、错误处理不直观。requests 库以“为人类设计”为理念，提供了简洁的 API，是金融数据采集领域的事实标准。

4.2 环境准备与安装

在终端或 Jupyter Notebook 中执行以下命令安装：

pip install requests

验证安装是否成功：

import requests
print(requests.__version__)
# 输出示例：2.31.0

4.3 第一次行情请求：获取 BTC/USDT 最新价

我们以币安（Binance）公开的 REST API 为例，获取比特币兑 USDT 的实时价格。币安 API 无需认证即可查询行情数据，非常适合入门练习。

import requests

# 币安现货行情接口
url = "https://api.binance.com/api/v3/ticker/price"
params = {"symbol": "BTCUSDT"}

try:
    response = requests.get(url, params=params, timeout=5)
    response.raise_for_status()  # 检查 HTTP 状态码
    data = response.json()
    print(f"最新价格: {data['price']} USDT")
except requests.exceptions.RequestException as e:
    print(f"请求失败: {e}")

代码解析：

• params 字典自动将参数拼接到 URL 末尾，生成 https://api.binance.com/api/v3/ticker/price?symbol=BTCUSDT。
• timeout=5 防止请求卡死，超过 5 秒未响应则抛出异常。
• response.json() 直接将返回的 JSON 字符串解析为 Python 字典。

4.4 理解 HTTP 响应结构

一次完整的 HTTP 响应包含状态行、响应头和响应体三部分。在金融数据采集调试中，响应头中的 Content-Type 和 X-MBX-* 自定义头常包含重要信息。

response = requests.get("https://api.binance.com/api/v3/ping")
print(f"状态码: {response.status_code}")
print(f"响应头: {dict(response.headers)}")
print(f"响应体: {response.text}")

4.5 批量获取多个交易对行情

坏账率哨兵通常需要监控多个交易对。币安提供了批量查询接口，一次请求即可获取所有交易对的最新价格。

url = "https://api.binance.com/api/v3/ticker/price"
# 不传 symbol 参数，返回全部交易对
response = requests.get(url, timeout=10)
all_prices = response.json()

# 只提取我们关注的几个交易对
watch_list = ["BTCUSDT", "ETHUSDT", "BNBUSDT", "ADAUSDT"]
for item in all_prices:
    if item["symbol"] in watch_list:
        print(f"{item['symbol']}: {item['price']}")

4.6 处理常见异常与错误

网络波动、交易所维护、参数错误都会导致请求失败。一个健壮的采集程序必须覆盖以下异常场景：

import requests
from requests.exceptions import Timeout, ConnectionError, HTTPError

url = "https://api.binance.com/api/v3/ticker/price"
params = {"symbol": "BTCUSDT"}

try:
    resp = requests.get(url, params=params, timeout=3)
    resp.raise_for_status()
except Timeout:
    print("请求超时，交易所可能响应缓慢")
except ConnectionError:
    print("网络连接失败，请检查网络或代理设置")
except HTTPError as e:
    print(f"HTTP 错误 {resp.status_code}: {resp.text}")
except Exception as e:
    print(f"未知错误: {e}")
else:
    data = resp.json()
    print(f"成功获取价格: {data['price']}")

4.7 数据采集流程架构

下图展示了从发送请求到数据落地的完整链路，这也是坏账率哨兵系统数据层的核心骨架。

4.8 实战练习：构建你的第一个行情采集函数

将本节知识封装为一个可复用的函数，为后续章节的定时采集任务打下基础。

import requests
from typing import Optional, Dict, Any

def fetch_ticker_price(symbol: str, base_url: str = "https://api.binance.com") -> Optional[Dict[str, Any]]:
    """
    获取指定交易对的实时行情价格

    参数:
        symbol: 交易对名称，如 "BTCUSDT"
        base_url: API 基础地址，默认币安

    返回:
        成功返回包含价格的字典，失败返回 None
    """
    url = f"{base_url}/api/v3/ticker/price"
    params = {"symbol": symbol.upper()}

    try:
        resp = requests.get(url, params=params, timeout=5)
        resp.raise_for_status()
        return resp.json()
    except requests.exceptions.RequestException as e:
        print(f"[ERROR] 获取 {symbol} 行情失败: {e}")
        return None

# 使用示例
result = fetch_ticker_price("ethusdt")
if result:
    print(f"ETH/USDT 最新价: {result['price']}")