寻求替代get_data_yahoo的快速可靠历史股票数据获取方案（1500个ticker、95天数据）

针对批量拉取大量股票历史数据的需求，以下是几个比get_data_yahoo更稳定高效的方案，附带具体实现思路：

1. 优化yfinance调用（低成本改进） #

yfinance本身支持批量请求，比循环单个调用效率高很多，搭配重试机制可解决大部分临时报错问题：

• 核心思路：用yfinance.download批量拉取，通过重试装饰器处理网络/服务端错误，同时拆分大列表为小批次避免限流
• 代码示例：

import yfinance as yf
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import requests
import time

# 定义重试规则：最多重试3次，间隔指数增长
@retry(stop=stop_after_attempt(3),
       wait=wait_exponential(multiplier=1, min=2, max=10),
       retry=retry_if_exception_type((requests.exceptions.RequestException, Exception)))
def fetch_batch_data(tickers, start_date, end_date):
    return yf.download(tickers, start=start_date, end=end_date, threads=True)

# 拆分大列表为小批次
def split_tickers(tickers, batch_size=50):
    for i in range(0, len(tickers), batch_size):
        yield tickers[i:i+batch_size]

# 使用示例
tickers_list = ["AAPL", "MSFT"] + [...]  # 你的1500个ticker
start = "2024-01-01"
end = "2024-04-05"  # 对应95天左右的时间范围

all_data = {}
for batch in split_tickers(tickers_list):
    batch_data = fetch_batch_data(batch, start, end)
    all_data.update(batch_data)
    time.sleep(0.5)  # 增加间隔避免触发限流

• 优势：无需额外API密钥，批量请求效率提升明显，重试机制覆盖大部分临时错误
• 注意：若仍频繁报错，可适当减小batch_size或延长请求间隔

2. 使用Alpha Vantage API（免费/付费可选） #

Alpha Vantage提供稳定的股票数据接口，免费版有每分钟5次、每天500次的调用限制，可通过多个API密钥轮询突破限制，付费版无严格限制：

• 核心思路：循环调用API，用密钥池轮询，搭配重试机制控制请求频率
• 代码示例：

import requests
from tenacity import retry, stop_after_attempt, wait_fixed
import time
import pandas as pd

API_KEYS = ["KEY1", "KEY2"] + [...]  # 多个API密钥轮换
key_index = 0

@retry(stop=stop_after_attempt(3), wait=wait_fixed(2))
def fetch_ticker_data(ticker, start_date, end_date):
    global key_index
    key = API_KEYS[key_index % len(API_KEYS)]
    key_index += 1
    
    url = f"https://www.alphavantage.co/query?function=TIME_SERIES_DAILY_ADJUSTED&symbol={ticker}&outputsize=full&apikey={key}"
    response = requests.get(url)
    data = response.json()
    
    # 解析数据并筛选目标日期范围
    if "Time Series (Daily)" not in data:
        return None
    df = pd.DataFrame.from_dict(data["Time Series (Daily)"], orient="index")
    df.index = pd.to_datetime(df.index)
    df = df.sort_index()
    return df.loc[start_date:end_date]

# 批量处理
all_data = {}
for ticker in tickers_list:
    data = fetch_ticker_data(ticker, start, end)
    if data is not None:
        all_data[ticker] = data
    time.sleep(0.2)  # 控制请求频率，避免触发限流

• 优势：数据稳定可靠，免费版可满足小批量需求，付费版支持高并发
• 注意：免费版需严格控制调用频率，需自行解析返回的JSON格式

3. 使用Polygon.io API（高可靠性批量方案） #

Polygon.io是专为金融数据打造的API服务，付费版支持批量请求、高并发，数据延迟低，适合大规模ticker场景：

• 核心思路：使用其aggs端点批量获取多个ticker的历史数据，支持时间范围筛选
• 代码示例：

import requests
import pandas as pd

API_KEY = "YOUR_POLYGON_KEY"

def fetch_batch_polygon(tickers, start_date, end_date):
    tickers_str = ",".join(tickers)
    # 按日期分组批量请求
    url = f"https://api.polygon.io/v2/aggs/grouped/locale/us/market/stocks/{start_date}?adjusted=true&tickers={tickers_str}&apiKey={API_KEY}"
    response = requests.get(url)
    data = response.json()
    
    # 解析返回数据（示例逻辑）
    result = {}
    for item in data.get("results", []):
        ticker = item["T"]
        if ticker not in result:
            result[ticker] = []
        result[ticker].append({
            "date": pd.to_datetime(item["t"], unit="ms"),
            "open": item["o"],
            "high": item["h"],
            "low": item["l"],
            "close": item["c"],
            "volume": item["v"]
        })
    # 转换为DataFrame
    for ticker in result:
        result[ticker] = pd.DataFrame(result[ticker]).set_index("date").sort_index()
    return result

# 按批次处理ticker列表
all_data = {}
for batch in split_tickers(tickers_list, batch_size=100):
    batch_result = fetch_batch_polygon(batch, start, end)
    all_data.update(batch_result)

• 优势：批量请求效率极高，数据质量高，支持实时和历史数据，适合大规模ticker列表
• 注意：需要付费订阅（免费版有严格的调用限制）

4. 本地缓存优化（通用辅助方案） #

不管用哪个数据源，都建议添加本地缓存机制，避免重复请求已获取的数据，减少报错概率：

• 实现思路：将每个ticker的数据保存到本地CSV/SQLite数据库，每次请求前先检查本地是否已有符合时间范围的数据，仅请求缺失部分
• 示例逻辑：

import pandas as pd
import os

def load_cached_data(ticker, start_date, end_date):
    cache_path = f"./stock_cache/{ticker}.csv"
    if os.path.exists(cache_path):
        df = pd.read_csv(cache_path, index_col=0, parse_dates=True)
        # 检查是否覆盖所需时间范围
        if df.index.min() <= pd.to_datetime(start_date) and df.index.max() >= pd.to_datetime(end_date):
            return df.loc[start_date:end_date]
    return None

def save_cached_data(ticker, data):
    os.makedirs("./stock_cache", exist_ok=True)
    data.to_csv(f"./stock_cache/{ticker}.csv")

# 使用时先查缓存，再请求
all_data = {}
for ticker in tickers_list:
    cached_data = load_cached_data(ticker, start, end)
    if cached_data is not None:
        all_data[ticker] = cached_data
    else:
        fresh_data = fetch_ticker_data(ticker, start, end)
        if fresh_data is not None:
            save_cached_data(ticker, fresh_data)
            all_data[ticker] = fresh_data

内容的提问来源于stack exchange，提问作者brodsky_star