3. 数据获取策略：API接口调用、文件下载、Web爬虫、数据订阅服务

在量化回测实验中，数据获取是决定实验质量与效率的首要环节。不同的数据源特性、更新频率与合规要求，决定了必须采用差异化的获取策略。本章系统梳理四种主流数据获取方式，帮助你在不同场景下做出最优选择。

3.1 四种获取策略总览

3.2 API接口调用

API（应用程序编程接口）是量化回测中最标准化的数据获取方式。主流金融数据提供商（如Wind、Tushare、Quandl、Alpha Vantage）均提供RESTful或WebSocket接口。

3.2.1 核心调用流程

1. 注册与认证：获取API Key或Token，配置访问权限。
2. 构建请求：确定端点URL、请求参数（证券代码、时间范围、字段列表）。
3. 发送请求：使用HTTP GET/POST方法，设置超时与重试机制。
4. 解析响应：将JSON/CSV/XML格式数据转换为DataFrame。
5. 错误处理：处理限流（Rate Limit）、认证失败、数据缺失等异常。

3.2.2 代码示例：Python调用行情API

import requests
import pandas as pd
import time

def fetch_kline(api_key, symbol, start_date, end_date, retries=3):
    url = "https://api.example.com/v1/kline"
    params = {
        "symbol": symbol,
        "start": start_date,
        "end": end_date,
        "interval": "1d",
        "apikey": api_key
    }
    for attempt in range(retries):
        try:
            resp = requests.get(url, params=params, timeout=10)
            resp.raise_for_status()
            data = resp.json()
            df = pd.DataFrame(data["records"])
            df["date"] = pd.to_datetime(df["date"])
            return df.set_index("date")
        except requests.exceptions.RequestException as e:
            print(f"Attempt {attempt+1} failed: {e}")
            time.sleep(2 ** attempt)
    raise Exception("Max retries exceeded")

# 使用示例
df = fetch_kline("your_api_key", "AAPL", "2024-01-01", "2024-12-31")
print(df.head())

3.3 文件下载

文件下载适用于获取批量历史数据，常见格式包括CSV、Parquet、HDF5。许多数据提供商（如CSI、Yahoo Finance历史数据、交易所公开数据）提供每日更新的压缩包。

3.3.1 下载策略对比

3.3.2 增量更新策略

对于每日更新的文件数据，建议采用增量下载而非全量覆盖：

import os
import pandas as pd
from datetime import datetime, timedelta

def incremental_download(base_dir, symbol, last_date):
    """仅下载缺失日期数据并追加到本地文件"""
    local_file = f"{base_dir}/{symbol}.parquet"
    today = datetime.now().date()
    
    # 读取已有数据
    if os.path.exists(local_file):
        existing = pd.read_parquet(local_file)
        last_local = existing.index.max().date()
    else:
        existing = pd.DataFrame()
        last_local = last_date - timedelta(days=1)
    
    # 仅下载缺失区间
    if last_local < today:
        new_data = fetch_kline("your_key", symbol, 
                               last_local + timedelta(days=1), 
                               today)
        if not new_data.empty:
            combined = pd.concat([existing, new_data])
            combined = combined[~combined.index.duplicated(keep='last')]
            combined.sort_index().to_parquet(local_file)
            print(f"Updated {symbol}: {len(new_data)} new rows")
    else:
        print(f"{symbol} is up to date")

3.4 Web爬虫

Web爬虫用于从财经网站、新闻门户、社交媒体等非结构化来源提取数据。在量化回测中，爬虫常用于获取另类数据（如舆情情绪、供应链信息、专利数据）。

3.4.1 爬虫技术栈选择

3.4.2 爬虫数据清洗要点

import re
from bs4 import BeautifulSoup

def clean_financial_text(html_content):
    """从财经新闻HTML中提取并清洗正文"""
    soup = BeautifulSoup(html_content, 'html.parser')
    
    # 移除脚本、样式、广告标签
    for tag in soup(['script', 'style', 'nav', 'footer', 'aside']):
        tag.decompose()
    
    # 提取正文（根据实际页面结构调整选择器）
    article = soup.find('article') or soup.find('div', class_='content')
    if not article:
        return ""
    
    text = article.get_text(separator='\n')
    # 清洗多余空白与特殊字符
    text = re.sub(r'\s+', ' ', text)
    text = re.sub(r'[^\w\s\.\,\!\?\-\%\$]', '', text)
    return text.strip()

3.5 数据订阅服务

数据订阅服务面向专业量化机构，提供实时、低延迟、高可靠的数据流。典型服务包括：Bloomberg Terminal、Reuters、IQFeed、Polygon.io、以及交易所直连（DMA）数据。

3.5.1 订阅服务架构

3.5.2 订阅服务关键指标

3.6 策略选择决策树

3.7 数据获取最佳实践

3.7.1 数据获取清单