3. 数据获取基础：从 Yahoo Finance 获取历史行情、从 CSV 文件读取数据、API 接口调用规范

数据获取是量化回测的起点。原始数据的质量直接决定回测结果的可靠性。本章聚焦三种最常用的数据获取方式：Yahoo Finance 在线接口、本地 CSV 文件读取以及通用 API 调用规范。掌握这些基础，你就能为后续清洗与回测建立稳定的数据管道。

3.1 从 Yahoo Finance 获取历史行情

Yahoo Finance 提供免费、稳定的历史日线数据，是回测入门最常用的数据源。Python 中通过 yfinance 库即可快速获取。

3.1.1 安装与基本调用

# 安装（如未安装）
# pip install yfinance

import yfinance as yf

# 获取单只股票历史行情
ticker = "AAPL"
data = yf.download(ticker, start="2023-01-01", end="2023-12-31")
print(data.head())

3.1.2 返回数据结构

yf.download() 返回 pandas.DataFrame，包含以下标准列：

3.1.3 批量获取与异常处理

tickers = ["AAPL", "MSFT", "GOOGL"]
data = yf.download(tickers, start="2023-01-01", end="2023-12-31")

# 处理缺失数据（如某只股票某日停牌）
data = data.dropna(how='any')  # 删除任意列为 NaN 的行
print(data.shape)

3.2 从 CSV 文件读取数据

本地 CSV 文件是离线回测、历史数据备份、机构数据交换的通用格式。读取时需注意编码、分隔符、日期解析等细节。

3.2.1 标准读取方式

import pandas as pd

# 标准读取
df = pd.read_csv("historical_data.csv", 
                 parse_dates=["Date"], 
                 index_col="Date",
                 encoding="utf-8")

# 查看基本信息
print(df.info())
print(df.head())

3.2.2 常见 CSV 格式问题与解决方案

3.2.3 读取后字段对齐

# 统一列名为小写 + 下划线
df.columns = [col.lower().replace(' ', '_') for col in df.columns]

# 确保包含必要字段
required_cols = ['open', 'high', 'low', 'close', 'volume']
for col in required_cols:
    if col not in df.columns:
        raise ValueError(f"缺少必要字段: {col}")

# 类型强制转换
df['close'] = pd.to_numeric(df['close'], errors='coerce')

3.3 API 接口调用规范

除 Yahoo Finance 外，量化交易常对接专业数据服务商（如 Wind、Bloomberg、Tushare、AKShare）。统一的 API 调用规范能降低维护成本，提高代码可读性。

3.3.1 通用 API 调用模板

import requests
import time
from typing import Dict, Optional

def fetch_market_data(symbol: str, 
                      start_date: str, 
                      end_date: str,
                      api_key: str,
                      base_url: str = "https://api.example.com") -> Optional[Dict]:
    """
    通用市场数据 API 调用模板
    
    参数:
        symbol: 股票代码
        start_date: 起始日期 (YYYY-MM-DD)
        end_date: 结束日期 (YYYY-MM-DD)
        api_key: API 密钥
        base_url: API 基础地址
    
    返回:
        解析后的 JSON 数据，失败返回 None
    """
    headers = {
        "Authorization": f"Bearer {api_key}",
        "User-Agent": "QuantBacktest/1.0"
    }
    params = {
        "symbol": symbol,
        "start": start_date,
        "end": end_date,
        "format": "json"
    }
    
    try:
        response = requests.get(base_url + "/history", 
                                headers=headers, 
                                params=params,
                                timeout=10)
        response.raise_for_status()  # 触发 HTTP 错误
        return response.json()
    except requests.exceptions.RequestException as e:
        print(f"API 请求失败: {e}")
        return None
    finally:
        time.sleep(0.5)  # 限速，避免被封

3.3.2 API 调用规范清单

• 认证方式：优先使用 API Key 放在 Header 中，避免 URL 明文传输。
• 限速与重试：每次请求后至少等待 0.5 秒；遇到 429 状态码时指数退避重试。
• 错误处理：区分网络错误（超时、断连）与业务错误（参数非法、无权限），分别记录日志。
• 数据缓存：对同一参数组合的请求结果缓存到本地 CSV/Parquet，减少重复调用。
• 字段映射：将 API 返回的字段名（如 trade_date）统一映射为内部标准名（如 date）。

3.4 三种数据获取方式对比