数据采集基础：用requests库抓取金融API数据

金融数据采集，说白了就是跟各种API打交道。我做了这么多年数据工程，发现很多新手一上来就想着搞什么分布式爬虫，其实连最基础的requests库都没用好。今天咱们就把这块地基打扎实。

3.1 为什么选requests库？

Python里做HTTP请求的库不少，urllib、httpx、aiohttp都有。但我个人习惯用requests，原因很简单——它把复杂的东西藏起来了。你想想看，金融数据接口返回的要么是JSON，要么是CSV，requests处理这两种格式就像切豆腐一样顺手。

我记得刚入行那会儿，有个同事用urllib写了个数据采集脚本，光处理请求头就写了三十行。我用requests重构后，代码量直接砍掉三分之二。嗯，这就是工具选对的好处。

3.2 基础用法：GET请求与参数

先看一个最典型的场景——从股票API获取实时行情。大多数金融API都支持通过GET请求传递参数，比如股票代码、时间范围这些。

import requests

# 设置请求参数
params = {
    'symbol': '000001.SZ',
    'fields': 'open,high,low,close,volume',
    'start_date': '2024-01-01',
    'end_date': '2024-01-31'
}

# 发送GET请求
response = requests.get(
    'https://api.example.com/stock/daily',
    params=params,
    timeout=10  # 设置超时，这个很重要
)

# 检查状态码
if response.status_code == 200:
    data = response.json()
    print(f"获取到 {len(data['records'])} 条记录")
else:
    print(f"请求失败，状态码: {response.status_code}")

小技巧：我个人习惯把API的base_url和参数分开定义，这样后期维护时改起来方便。你想想看，如果哪天接口地址变了，你只需要改一个变量就行。

3.3 处理JSON格式数据

金融API返回的数据，90%以上都是JSON格式。处理JSON其实就三步：解析、提取、转换。但这里有个坑——很多新手拿到数据就直接用，完全不检查结构。

我曾经在项目中遇到过这么个事：某家券商的API文档写的是返回字段叫"close_price"，结果实际返回的是"closing_price"。如果代码里写死了字段名，那数据就全丢了。所以我的做法是——先打印出来看看。

# 安全解析JSON
try:
    json_data = response.json()
    
    # 先打印结构，确认字段名
    print("返回数据结构:", json_data.keys())
    
    # 提取核心数据
    records = json_data.get('data', {}).get('records', [])
    
    # 转换为结构化数据
    stock_data = []
    for record in records:
        stock_data.append({
            'date': record['trade_date'],
            'open': float(record['open_price']),
            'high': float(record['high_price']),
            'low': float(record['low_price']),
            'close': float(record['close_price']),
            'volume': int(record['volume'])
        })
        
except ValueError as e:
    print(f"JSON解析失败: {e}")
except KeyError as e:
    print(f"字段缺失: {e}")

注意：千万别直接用response.json()而不做异常处理。我见过太多线上事故就是因为API返回了错误信息，结果json()直接抛异常，整个采集流程就断了。

3.4 处理CSV格式数据

有些金融数据接口会返回CSV格式，比如一些历史行情数据。处理CSV我推荐用pandas，但如果你不想引入那么重的库，用csv模块也够用。

import csv
from io import StringIO

# 假设response.text是CSV格式的字符串
csv_content = response.text

# 用StringIO包装成文件对象
csv_file = StringIO(csv_content)
reader = csv.DictReader(csv_file)

# 逐行读取
rows = []
for row in reader:
    rows.append({
        'date': row['日期'],
        'open': float(row['开盘价']),
        'close': float(row['收盘价']),
        'volume': int(row['成交量'])
    })

print(f"共读取 {len(rows)} 行CSV数据")

避坑指南：我曾经遇到过CSV文件里混入了BOM头（就是文件开头那几个看不见的字符），导致第一列字段名前面多了个"\ufeff"。解决办法很简单——读取前用.lstrip('\ufeff')处理一下。

3.5 异常处理与重试机制

做金融数据采集，最怕的就是网络波动。API偶尔超时、返回500错误、甚至直接断连，这些都是家常便饭。我的原则是——能自动恢复的绝不手动干预。

这里分享一个我用了好几年的重试模板：

import time
from requests.exceptions import RequestException

def fetch_with_retry(url, params=None, max_retries=3, backoff_factor=2):
    """
    带重试机制的数据采集函数
    
    参数:
        url: API地址
        params: 请求参数
        max_retries: 最大重试次数
        backoff_factor: 退避因子（每次重试等待时间翻倍）
    """
    for attempt in range(max_retries):
        try:
            response = requests.get(
                url, 
                params=params,
                timeout=10,
                headers={'User-Agent': 'Mozilla/5.0'}
            )
            
            # 处理HTTP错误状态码
            if response.status_code == 429:
                # 请求太频繁，等一会儿再试
                wait_time = 2 ** attempt
                print(f"触发限流，等待 {wait_time} 秒...")
                time.sleep(wait_time)
                continue
                
            elif response.status_code == 503:
                # 服务暂时不可用
                wait_time = backoff_factor ** attempt
                print(f"服务不可用，{wait_time}秒后重试...")
                time.sleep(wait_time)
                continue
                
            elif response.status_code != 200:
                print(f"HTTP错误: {response.status_code}")
                continue
                
            # 成功返回数据
            return response.json()
            
        except RequestException as e:
            print(f"第 {attempt + 1} 次请求失败: {e}")
            
            if attempt < max_retries - 1:
                wait_time = backoff_factor ** attempt
                print(f"等待 {wait_time} 秒后重试...")
                time.sleep(wait_time)
            else:
                print("重试次数用尽，放弃请求")
                return None
    
    return None

# 使用示例
data = fetch_with_retry(
    'https://api.example.com/stock/daily',
    params={'symbol': '000001.SZ'}
)

if data:
    print("数据采集成功")
else:
    print("数据采集失败，请检查网络或API状态")

核心要点：

  • 重试次数一般设3次就够了，太多反而会加重服务器负担
  • 退避策略用指数退避，第一次等2秒，第二次等4秒，第三次等8秒
  • 一定要区分不同类型的错误——限流(429)和服务不可用(503)的处理方式不同

3.6 实战：完整的金融数据采集流程

把上面这些知识点串起来，就是一个完整的采集流程。我一般会把它封装成一个类，方便复用：

class FinancialDataCollector:
    """金融数据采集器"""
    
    def __init__(self, base_url, api_key=None):
        self.base_url = base_url
        self.session = requests.Session()
        
        if api_key:
            self.session.headers.update({'X-API-Key': api_key})
    
    def collect_stock_data(self, symbol, start_date, end_date):
        """采集股票日线数据"""
        params = {
            'symbol': symbol,
            'start_date': start_date,
            'end_date': end_date,
            'format': 'json'
        }
        
        data = fetch_with_retry(
            f"{self.base_url}/stock/daily",
            params=params
        )
        
        if data:
            return self._parse_stock_data(data)
        return []
    
    def _parse_stock_data(self, raw_data):
        """解析原始数据"""
        records = raw_data.get('data', [])
        parsed = []
        
        for record in records:
            try:
                parsed.append({
                    'symbol': record['symbol'],
                    'date': record['trade_date'],
                    'open': float(record['open']),
                    'high': float(record['high']),
                    'low': float(record['low']),
                    'close': float(record['close']),
                    'volume': int(record['volume'])
                })
            except (KeyError, ValueError, TypeError) as e:
                print(f"解析记录失败: {e}, 原始数据: {record}")
                continue
        
        return parsed

# 使用示例
collector = FinancialDataCollector(
    base_url='https://api.example.com',
    api_key='your_api_key_here'
)

result = collector.collect_stock_data(
    symbol='000001.SZ',
    start_date='2024-01-01',
    end_date='2024-01-31'
)

print(f"成功采集 {len(result)} 条数据")

3.7 本章知识体系

下面这张图把本章的核心内容串起来了，你可以对照着回顾一下：

数据采集基础：知识体系 requests库 GET请求与参数 params、timeout、headers JSON格式处理 解析、提取、异常处理 CSV格式处理 csv模块、pandas、BOM处理 异常处理与重试 指数退避、状态码判断 实战：完整采集流程封装

我的建议：刚开始做数据采集时，别想着一步到位。先把GET请求跑通，再慢慢加上JSON解析、异常处理、重试机制。每加一个功能就测试一次，这样出了问题能快速定位。

好了，这一章的内容就到这里。记住一句话——数据采集是ETL流程的入口，入口不稳，后面全是白搭。把requests库用熟练了，后面的路就好走了。

公众号:蓝海数据掘金营，微信deep3321

← 上一章 📖 返回目录 下一章 →