Python yfinance API数据获取：异常处理与稳健性实践

import yfinance as yf
from requests.exceptions import ConnectionError, HTTPError, Timeout

def fetch_stock_data(ticker_symbol, period="max"):
    try:
        # 尝试获取数据并赋值给变量
        data = yf.Ticker(ticker_symbol).history(period=period)
        return data
    except (ConnectionError, HTTPError, Timeout) as e:
        print(f"获取 {ticker_symbol} 数据时发生网络或HTTP错误: {e}")
        return None
    except Exception as e:
        print(f"获取 {ticker_symbol} 数据时发生未知错误: {e}")
        return None

import pandas as pd

def process_data(ticker_symbol, df):
    if df is None:
        print(f"无法获取 {ticker_symbol} 的数据。")
        return
    if df.empty:
        print(f"{ticker_symbol} 返回空数据帧，可能无历史数据或已退市。")
    else:
        print(f"{ticker_symbol} 数据获取成功，共 {len(df)} 条记录。")
        print(df.head())

# 示例使用
# data_0250 = fetch_stock_data("0250.HK")
# process_data("0250.HK", data_0250)

import yfinance as yf
import pandas as pd
import time
from requests.exceptions import ConnectionError, HTTPError, Timeout

def get_robust_stock_history(ticker_symbol, period="max", retries=3, delay=5):
    """
    健壮地获取单个股票的历史数据。
    :param ticker_symbol: 股票代码。
    :param period: 数据周期（例如 "max", "1y", "3mo"）。
    :param retries: 重试次数。
    :param delay: 每次重试前的等待秒数。
    :return: 包含历史数据的DataFrame，如果失败则返回None。
    """
    for attempt in range(retries):
        try:
            print(f"尝试获取 {ticker_symbol} 数据 (尝试 {attempt + 1}/{retries})...")

            # 关键：始终将结果赋值给一个变量
            stock_ticker = yf.Ticker(ticker_symbol)
            data = stock_ticker.history(period=period)

            if data.empty:
                print(f"警告: {ticker_symbol} 返回空数据帧。可能无历史数据或已退市。")
                # 对于空数据帧，我们可能仍希望返回它，或根据需求返回None
                return data 
            else:
                print(f"成功获取 {ticker_symbol} 数据。")
                return data
        except (ConnectionError, HTTPError, Timeout) as e:
            print(f"获取 {ticker_symbol} 数据时发生网络错误: {e}")
            if attempt < retries - 1:
                print(f"等待 {delay} 秒后重试...")
                time.sleep(delay)
            else:
                print(f"已达最大重试次数，放弃获取 {ticker_symbol}。")
                return None
        except Exception as e:
            print(f"获取 {ticker_symbol} 数据时发生未知错误: {e}")
            return None
    return None # 如果所有重试都失败

# 待查询的股票列表，包含有效和可能无效的示例
tickers = ["0001.HK", "0250.HK", "AAPL", "GOOG", "INVALID_TICKER"]

all_stock_data = {}

for ticker in tickers:
    df = get_robust_stock_history(ticker)
    if df is not None:
        all_stock_data[ticker] = df
    print("-" * 30)

# 打印已成功获取的数据
print("\n--- 成功获取的数据概览 ---")
for ticker, df in all_stock_data.items():
    if not df.empty:
        print(f"\n{ticker} (前5行):")
        print(df.head())
    else:
        print(f"\n{ticker} (数据为空或无效).")