7个实战案例掌握yfinance金融数据接口：从入门到专业

2026-04-15 08:10:33作者：平淮齐Percy

在量化投资与金融科技领域，高效获取准确的市场数据是构建分析模型的基础。yfinance作为雅虎财经API的非官方客户端，为Python开发者提供了便捷的金融数据获取途径。本文将通过"认知建立→实战突破→问题诊断→效能优化"四阶段框架，结合股票量化分析与另类数据获取双主线，帮助你全面掌握这一强大工具的核心功能与高级应用，实现从数据获取到策略实现的全流程优化。

一、认知建立：yfinance核心架构与数据模型

构建金融数据获取认知框架：从接口到应用

金融数据接口是连接市场数据与量化策略的桥梁，yfinance通过模拟浏览器请求实现对雅虎财经数据的抓取与解析。其核心优势在于无需API密钥、支持多资产类型且数据字段丰富，是个人量化研究者与机构分析师的理想工具。

概念图解	操作流程图
核心组件：Ticker对象（单资产接口）、download方法（批量获取）、共享缓存机制	1. 创建Ticker实例→2. 调用数据方法→3. 缓存处理→4. 数据格式化→5. 异常处理
数据类型：历史价格、实时行情、财务报表、ESG指标、期权数据	请求→解析→验证→存储→应用

import yfinance as yf
import pandas as pd

def initialize_yfinance():
    """初始化yfinance环境并验证配置"""
    try:
        # 检查库版本
        print(f"yfinance版本: {yf.__version__}")
        
        # 创建测试Ticker对象
        test_ticker = yf.Ticker("AAPL")
        
        # 获取基础信息验证连接
        info = test_ticker.info
        if not info:
            raise Exception("无法获取基础信息")
            
        # 验证历史数据获取
        hist = test_ticker.history(period="1d")
        if hist.empty:
            raise Exception("历史数据获取失败")
            
        print("✅ yfinance环境初始化成功")
        return True
        
    except Exception as e:
        print(f"❌ 初始化失败: {str(e)}")
        return False

# 执行环境初始化
initialize_yfinance()

应用场景：新项目初始化或环境配置验证时使用，确保yfinance库能正常连接数据源并返回数据。 注意事项：首次运行可能需要安装依赖库，使用pip install yfinance pandas命令安装。

解析yfinance数据模型：结构与字段解析

yfinance返回的数据主要以pandas DataFrame和字典两种格式呈现，不同数据类型有着特定的结构与字段定义。理解这些数据模型是进行有效分析的基础。

数据类型对比卡片

📊 历史价格数据

结构：DataFrame，时间索引
核心字段：Open(开盘价)、High(最高价)、Low(最低价)、Close(收盘价)、Volume(成交量)
常用方法：Ticker.history(period, interval)

📈 实时行情数据

结构：字典嵌套结构
核心字段：regularMarketPrice(当前价格)、regularMarketChange(价格变动)、regularMarketChangePercent(涨跌幅)
常用方法：Ticker.info()

📑 财务报表数据

结构：字典包含多个DataFrame
核心报表：资产负债表、利润表、现金流量表
常用方法：Ticker.balance_sheet、Ticker.financials

def analyze_data_structure(symbol):
    """解析yfinance返回的各类数据结构"""
    ticker = yf.Ticker(symbol)
    
    # 分析历史价格数据
    hist = ticker.history(period="5d", interval="1d")
    print("📅 历史价格数据结构:")
    print(f"索引类型: {type(hist.index)}")
    print(f"列名: {hist.columns.tolist()}")
    print(f"数据形状: {hist.shape}")
    
    # 分析实时行情数据
    info = ticker.info
    print("\n📊 实时行情数据结构:")
    print(f"数据类型: {type(info)}")
    print(f"关键财务指标: {[k for k in info.keys() if k in ['marketCap', 'trailingPE', 'forwardPE', 'dividendYield']]}")
    
    # 分析财务报表数据
    balance_sheet = ticker.balance_sheet
    print("\n📑 资产负债表结构:")
    print(f"索引类型: {type(balance_sheet.index)}")
    print(f"列名(报告日期): {balance_sheet.columns.tolist()}")

# 分析苹果公司数据结构
analyze_data_structure("AAPL")

应用场景：在开始量化分析前，了解数据组织结构和字段含义，为后续数据清洗和特征工程做准备。 注意事项：不同市场的股票代码格式可能不同，如中国市场需添加".SS"（上交所）或".SZ"（深交所）后缀。

配置高效开发环境：工具链与最佳实践

搭建专业的yfinance开发环境不仅能提高工作效率，还能避免常见的兼容性问题。以下是经过验证的环境配置方案和开发最佳实践。

开发环境配置清单

🔧 核心工具

Python 3.8+：提供稳定的语言特性支持
yfinance 0.2.31+：确保包含最新的数据修复和功能
pandas 1.5.0+：数据处理核心库
Jupyter Notebook：交互式开发与可视化
PyCharm/VS Code：专业Python IDE，支持代码补全和调试

⚙️ 环境优化

设置国内PyPI镜像：加速依赖安装
配置缓存目录：减少重复网络请求
安装TA-Lib：技术指标计算库
配置日志系统：记录数据获取过程

import os
import logging
from yfinance import set_tz_cache_location

def configure_development_environment():
    """配置yfinance开发环境"""
    # 设置缓存目录
    cache_dir = os.path.expanduser("~/.yfinance_cache")
    if not os.path.exists(cache_dir):
        os.makedirs(cache_dir)
    set_tz_cache_location(cache_dir)
    
    # 配置日志
    logging.basicConfig(
        level=logging.INFO,
        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
        handlers=[
            logging.FileHandler("yfinance.log"),
            logging.StreamHandler()
        ]
    )
    
    # 验证环境
    try:
        import talib
        print("✅ TA-Lib已安装，支持技术指标计算")
    except ImportError:
        print("⚠️ TA-Lib未安装，技术指标功能将受限")
        
    print(f"✅ 开发环境配置完成，缓存目录: {cache_dir}")

# 配置开发环境
configure_development_environment()

应用场景：新设备首次配置或项目迁移时使用，建立标准化的开发环境。 注意事项：TA-Lib安装可能需要额外系统依赖，Windows用户建议使用预编译包。

行业应用延伸：在机构量化团队中，通常会搭建共享的yfinance数据服务，通过API网关统一提供数据接口，结合Redis等缓存系统提升多用户并发访问效率。个人开发者可从简单的本地缓存开始，逐步构建自己的数据服务架构。

二、实战突破：量化分析与另类数据获取

构建多线程股票数据采集器：提升效率300%

传统的单线程数据获取方式在处理大量股票代码时效率低下，通过多线程并发技术可以显著提升数据采集速度，特别适合构建股票池数据库或进行行业板块分析。

import concurrent.futures
import yfinance as yf
import pandas as pd
from typing import List, Dict

def fetch_single_ticker(symbol: str) -> Dict:
    """获取单个股票的关键数据"""
    try:
        ticker = yf.Ticker(symbol)
        
        # 获取基本信息
        info = ticker.info
        
        # 获取最新100天的价格数据
        hist = ticker.history(period="100d")
        
        # 提取关键指标
        return {
            "symbol": symbol,
            "name": info.get("longName", "N/A"),
            "sector": info.get("sector", "N/A"),
            "market_cap": info.get("marketCap", 0),
            "current_price": info.get("currentPrice", 0),
            "price_change": info.get("regularMarketChangePercent", 0),
            "volatility": hist["Close"].pct_change().std() * (252**0.5) if not hist.empty else 0,
            "data_available": not hist.empty
        }
        
    except Exception as e:
        print(f"⚠️ 获取{symbol}数据失败: {str(e)}")
        return {"symbol": symbol, "data_available": False}

def multi_thread_data_collector(symbols: List[str], max_workers: int = 10) -> pd.DataFrame:
    """多线程股票数据采集器"""
    results = []
    
    # 使用线程池并发获取数据
    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
        # 提交所有任务
        future_to_symbol = {executor.submit(fetch_single_ticker, symbol): symbol for symbol in symbols}
        
        # 处理结果
        for future in concurrent.futures.as_completed(future_to_symbol):
            symbol = future_to_symbol[future]
            try:
                data = future.result()
                results.append(data)
                print(f"✅ 完成{symbol}数据采集")
            except Exception as e:
                print(f"❌ {symbol}处理出错: {str(e)}")
    
    # 转换为DataFrame并返回
    return pd.DataFrame(results)

# 示例：采集道琼斯30成分股数据
dow_jones_symbols = [
    "AAPL", "MSFT", "JPM", "V", "JNJ", "WMT", "PG", "INTC", "IBM", "DIS",
    "HD", "MCD", "MRK", "UNH", "KO", "CAT", "VZ", "CSCO", "AXP", "XOM",
    "BA", "GS", "CVX", "MMM", "PFE", "WBA", "NKE", "MMM", "DD", "T"
]

# 执行多线程采集 (预计耗时: 15-30秒，取决于网络状况)
stock_data = multi_thread_data_collector(dow_jones_symbols, max_workers=15)
print(stock_data[["symbol", "name", "sector", "current_price", "volatility"]].head())

应用场景：指数成分股分析、行业板块比较、股票池构建等需要批量获取数据的场景。 注意事项：线程数量不宜过多（建议10-20个），避免触发服务器反爬机制；添加适当的异常处理，确保单个股票数据获取失败不会影响整体任务。

实现股票量化因子计算引擎：从基础到进阶

量化因子是量化策略的核心组成部分，yfinance提供的价格和成交量数据可以用来计算多种技术指标因子。以下实现了一个灵活的因子计算引擎，支持多种常用技术指标。

技术指标计算过程分步拆解

数据准备：获取历史价格数据，确保数据完整性
基础因子：计算收益率、波动率等基础指标
趋势因子：计算移动平均线、MACD等趋势指标
动量因子：计算RSI、CCI等动量指标
成交量因子：计算成交量变化率、资金流向等指标

import pandas as pd
import numpy as np
import yfinance as yf

class FactorEngine:
    """股票量化因子计算引擎"""
    
    def __init__(self, symbol: str, start_date: str = None, end_date: str = None):
        self.symbol = symbol
        self.start_date = start_date
        self.end_date = end_date
        self.data = self._load_data()
        self.factors = pd.DataFrame(index=self.data.index)
        
    def _load_data(self) -> pd.DataFrame:
        """加载基础价格数据"""
        ticker = yf.Ticker(self.symbol)
        if self.start_date and self.end_date:
            data = ticker.history(start=self.start_date, end=self.end_date)
        else:
            data = ticker.history(period="1y")
            
        if data.empty:
            raise ValueError(f"无法获取{self.symbol}的历史数据")
            
        return data
    
    def calculate_basic_factors(self):
        """计算基础因子"""
        # 日收益率
        self.factors['return'] = self.data['Close'].pct_change()
        
        # 波动率 (20日滚动)
        self.factors['volatility'] = self.factors['return'].rolling(window=20).std() * np.sqrt(252)
        
        # 收盘价与52周高点比率
        high_52week = self.data['High'].rolling(window=252).max()
        self.factors['price_to_high52'] = self.data['Close'] / high_52week
        
        return self
    
    def calculate_trend_factors(self):
        """计算趋势因子"""
        # 移动平均线
        self.factors['ma5'] = self.data['Close'].rolling(window=5).mean()
        self.factors['ma20'] = self.data['Close'].rolling(window=20).mean()
        self.factors['ma50'] = self.data['Close'].rolling(window=50).mean()
        
        # MACD指标
        ema12 = self.data['Close'].ewm(span=12, adjust=False).mean()
        ema26 = self.data['Close'].ewm(span=26, adjust=False).mean()
        self.factors['macd'] = ema12 - ema26
        self.factors['macd_signal'] = self.factors['macd'].ewm(span=9, adjust=False).mean()
        
        return self
    
    def calculate_momentum_factors(self):
        """计算动量因子"""
        # RSI (相对强弱指数)
        delta = self.data['Close'].diff(1)
        gain = delta.where(delta > 0, 0)
        loss = -delta.where(delta < 0, 0)
        
        avg_gain = gain.rolling(window=14).mean()
        avg_loss = loss.rolling(window=14).mean()
        
        rs = avg_gain / avg_loss
        self.factors['rsi'] = 100 - (100 / (1 + rs))
        
        # 动量指标 (60日收益率)
        self.factors['momentum_60d'] = self.data['Close'].pct_change(periods=60)
        
        return self
    
    def calculate_volume_factors(self):
        """计算成交量因子"""
        # 成交量变化率
        self.factors['volume_change'] = self.data['Volume'].pct_change()
        
        # 成交量加权平均价
        self.factors['vwap'] = (self.data['Volume'] * (self.data['High'] + self.data['Low'] + self.data['Close']) / 3).cumsum() / self.data['Volume'].cumsum()
        
        return self
    
    def get_factors(self) -> pd.DataFrame:
        """获取所有计算的因子"""
        return self.factors.dropna()

# 使用因子引擎计算苹果公司股票因子
try:
    factor_engine = FactorEngine("AAPL", start_date="2023-01-01", end_date="2023-12-31")
    factors = factor_engine.calculate_basic_factors() \
                          .calculate_trend_factors() \
                          .calculate_momentum_factors() \
                          .calculate_volume_factors() \
                          .get_factors()
                          
    print(f"✅ 成功计算{len(factors)}个交易日的因子数据")
    print(factors[['return', 'volatility', 'rsi', 'macd', 'vwap']].tail())
except Exception as e:
    print(f"❌ 因子计算失败: {str(e)}")

应用场景：量化策略开发、因子选股、风险模型构建等需要技术指标的场景。 注意事项：不同指标有不同的参数设置（如RSI的14天窗口），需要根据具体策略需求调整；因子计算前应确保数据质量，处理缺失值和异常值。

设计另类数据获取方案：新闻情绪与产业链数据

除了传统的价格和财务数据，另类数据（如新闻情绪、社交媒体热度、产业链数据）正成为量化分析的新维度。以下实现了一个结合新闻情绪分析的多源数据获取方案。

import yfinance as yf
import pandas as pd
import requests
from datetime import datetime, timedelta
import nltk
from nltk.sentiment import SentimentIntensityAnalyzer

# 下载NLTK情感分析模型（首次运行时需要）
try:
    sia = SentimentIntensityAnalyzer()
except LookupError:
    nltk.download('vader_lexicon')
    sia = SentimentIntensityAnalyzer()

class AlternativeDataCollector:
    """另类数据收集器"""
    
    def __init__(self, symbol: str):
        self.symbol = symbol
        self.ticker = yf.Ticker(symbol)
        self.company_name = self._get_company_name()
    
    def _get_company_name(self) -> str:
        """获取公司全称"""
        info = self.ticker.info
        return info.get("longName", self.symbol)
    
    def get_news_sentiment(self, days: int = 7) -> pd.DataFrame:
        """获取新闻情绪数据"""
        try:
            # 获取公司新闻
            news = self.ticker.news
            
            if not news:
                return pd.DataFrame(columns=["datetime", "title", "sentiment_score"])
            
            # 处理新闻数据
            news_data = []
            cutoff_date = datetime.now() - timedelta(days=days)
            
            for item in news:
                # 解析日期
                publish_time = datetime.fromtimestamp(item["providerPublishTime"])
                
                # 只保留指定天数内的新闻
                if publish_time < cutoff_date:
                    continue
                
                # 分析标题情感
                title = item["title"]
                sentiment = sia.polarity_scores(title)
                
                news_data.append({
                    "datetime": publish_time,
                    "title": title,
                    "sentiment_score": sentiment["compound"],
                    "source": item["publisher"]
                })
            
            # 转换为DataFrame并排序
            df = pd.DataFrame(news_data)
            return df.sort_values("datetime", ascending=False)
            
        except Exception as e:
            print(f"⚠️ 获取新闻情绪数据失败: {str(e)}")
            return pd.DataFrame()
    
    def get_peer_data(self) -> pd.DataFrame:
        """获取同行公司数据"""
        try:
            # 获取公司同行列表
            info = self.ticker.info
            peers = info.get("sectorPeers", [])
            
            if not peers:
                return pd.DataFrame()
                
            # 获取同行公司的关键指标
            peer_data = []
            for peer in peers[:5]:  # 限制获取前5个同行
                try:
                    peer_ticker = yf.Ticker(peer)
                    peer_info = peer_ticker.info
                    
                    peer_data.append({
                        "symbol": peer,
                        "name": peer_info.get("longName", peer),
                        "market_cap": peer_info.get("marketCap", 0),
                        "pe_ratio": peer_info.get("trailingPE", 0),
                        "sector": peer_info.get("sector", "N/A"),
                        "industry": peer_info.get("industry", "N/A")
                    })
                except Exception:
                    continue
            
            return pd.DataFrame(peer_data)
            
        except Exception as e:
            print(f"⚠️ 获取同行数据失败: {str(e)}")
            return pd.DataFrame()

# 收集特斯拉的另类数据
adc = AlternativeDataCollector("TSLA")

# 获取新闻情绪数据
news_sentiment = adc.get_news_sentiment(days=14)
print("📰 新闻情绪数据:")
print(news_sentiment[["datetime", "title", "sentiment_score"]].head())

# 获取同行数据
peer_data = adc.get_peer_data()
print("\n🏢 同行公司数据:")
print(peer_data[["symbol", "name", "market_cap", "pe_ratio"]])

应用场景：事件驱动策略、市场情绪分析、行业比较研究等需要非结构化数据的场景。 注意事项：新闻数据可能存在延迟或重复，情绪分析结果仅作参考，需结合其他指标使用；同行数据获取可能受到API限制，建议控制请求频率。

行业应用延伸：对冲基金正越来越多地将另类数据整合到投资决策中。例如，通过卫星图像分析零售停车场车流量预测销售数据，或通过社交媒体情绪预测短期股价波动。个人投资者可以从简单的新闻情绪分析入手，逐步构建多源数据融合的分析框架。

三、问题诊断：数据质量与异常处理

构建数据质量检测体系：完整性与一致性校验

金融数据质量直接影响量化策略的可靠性，建立完善的数据质量检测体系是量化分析的关键环节。以下实现了一个全面的数据质量检测工具，涵盖完整性、一致性和合理性校验。

数据质量检测流程

完整性检测：检查数据是否存在缺失值、时间序列是否连续
一致性检测：验证价格逻辑关系（如最高价≥收盘价≥最低价）
合理性检测：识别异常值和不合理数据（如价格异常波动）
时间戳检测：确保时间序列规范且无重复

import pandas as pd
import numpy as np
import yfinance as yf
from datetime import datetime

class DataQualityChecker:
    """金融数据质量检测工具"""
    
    def __init__(self, symbol: str, data: pd.DataFrame = None):
        self.symbol = symbol
        self.data = data if data is not None else self._load_default_data()
        self.quality_report = {
            "symbol": symbol,
            "check_time": datetime.now(),
            "total_records": len(self.data),
            "start_date": self.data.index.min(),
            "end_date": self.data.index.max(),
            "issues": []
        }
    
    def _load_default_data(self) -> pd.DataFrame:
        """加载默认数据（1年日线数据）"""
        ticker = yf.Ticker(self.symbol)
        return ticker.history(period="1y", interval="1d")
    
    def check_completeness(self) -> "DataQualityChecker":
        """检查数据完整性"""
        # 检查缺失值
        missing_values = self.data.isnull().sum()
        for column, count in missing_values.items():
            if count > 0:
                self.quality_report["issues"].append({
                    "type": "missing_value",
                    "severity": "medium",
                    "message": f"列 {column} 存在 {count} 个缺失值",
                    "count": count
                })
        
        # 检查时间序列连续性
        expected_dates = pd.date_range(
            start=self.data.index.min(), 
            end=self.data.index.max(),
            freq='B'  # 工作日频率
        )
        
        missing_dates = set(expected_dates) - set(self.data.index)
        if missing_dates:
            self.quality_report["issues"].append({
                "type": "missing_dates",
                "severity": "low",
                "message": f"存在 {len(missing_dates)} 个缺失的交易日",
                "count": len(missing_dates)
            })
        
        return self
    
    def check_consistency(self) -> "DataQualityChecker":
        """检查数据一致性"""
        # 检查价格逻辑关系
        invalid_high_low = self.data[(self.data['High'] < self.data['Low'])].index
        if not invalid_high_low.empty:
            self.quality_report["issues"].append({
                "type": "price_logic",
                "severity": "high",
                "message": f"存在 {len(invalid_high_low)} 条记录最高价低于最低价",
                "count": len(invalid_high_low)
            })
        
        # 检查收盘价是否在高低价之间
        invalid_close = self.data[
            (self.data['Close'] < self.data['Low']) | 
            (self.data['Close'] > self.data['High'])
        ].index
        
        if not invalid_close.empty:
            self.quality_report["issues"].append({
                "type": "close_price",
                "severity": "high",
                "message": f"存在 {len(invalid_close)} 条记录收盘价超出高低价范围",
                "count": len(invalid_close)
            })
        
        return self
    
    def check_plausibility(self, z_threshold: float = 3.0) -> "DataQualityChecker":
        """检查数据合理性"""
        # 使用Z-score检测价格异常波动
        price_changes = self.data['Close'].pct_change().dropna()
        z_scores = np.abs((price_changes - price_changes.mean()) / price_changes.std())
        abnormal_changes = z_scores[z_scores > z_threshold].index
        
        if not abnormal_changes.empty:
            self.quality_report["issues"].append({
                "type": "price_volatility",
                "severity": "medium",
                "message": f"存在 {len(abnormal_changes)} 条记录价格异常波动 (Z-score > {z_threshold})",
                "count": len(abnormal_changes)
            })
        
        # 检查成交量异常
        volume_z_scores = np.abs((self.data['Volume'] - self.data['Volume'].mean()) / self.data['Volume'].std())
        abnormal_volume = volume_z_scores[volume_z_scores > z_threshold].index
        
        if not abnormal_volume.empty:
            self.quality_report["issues"].append({
                "type": "volume_anomaly",
                "severity": "medium",
                "message": f"存在 {len(abnormal_volume)} 条记录成交量异常 (Z-score > {z_threshold})",
                "count": len(abnormal_volume)
            })
        
        return self
    
    def generate_report(self) -> dict:
        """生成质量报告"""
        # 计算总体质量评分 (0-100)
        issue_count = len(self.quality_report["issues"])
        severity_score = sum(
            {"low": 1, "medium": 3, "high": 5}[issue["severity"]] 
            for issue in self.quality_report["issues"]
        )
        
        quality_score = max(0, 100 - (issue_count * 2 + severity_score * 3))
        self.quality_report["quality_score"] = round(quality_score, 2)
        self.quality_report["status"] = "PASS" if quality_score >= 80 else "WARNING" if quality_score >= 60 else "FAIL"
        
        return self.quality_report

# 检测苹果公司股票数据质量
dq_checker = DataQualityChecker("AAPL")
report = dq_checker.check_completeness() \
                   .check_consistency() \
                   .check_plausibility() \
                   .generate_report()

print(f"📊 {report['symbol']} 数据质量报告")
print(f"总体评分: {report['quality_score']} ({report['status']})")
print(f"数据范围: {report['start_date'].strftime('%Y-%m-%d')} 至 {report['end_date'].strftime('%Y-%m-%d')}")
print(f"记录总数: {report['total_records']}")

if report["issues"]:
    print("\n⚠️ 发现问题:")
    for i, issue in enumerate(report["issues"], 1):
        print(f"{i}. [{issue['severity'].upper()}] {issue['message']}")
else:
    print("\n✅ 未发现数据质量问题")

应用场景：量化策略回测前的数据验证、数据 pipeline 质量监控、历史数据清洗等场景。 注意事项：不同市场和资产类型的数据质量标准可能不同，需根据具体情况调整检测参数（如Z-score阈值）；并非所有异常都是错误，有些可能是真实的市场事件（如分红、拆股），需要人工确认。

实现智能数据修复方案：从异常检测到自动修复

识别数据质量问题后，需要进行有效的数据修复。以下实现了一个智能数据修复系统，能够根据不同类型的异常自动选择合适的修复策略，并保留修复记录以便追溯。

常见数据问题与修复策略对照表

数据问题类型	检测方法	修复策略	适用场景
缺失值	空值检查	前向填充/线性插值	短期数据缺失
价格逻辑错误	高低价关系验证	基于邻近数据修正	数据记录错误
异常波动	Z-score/箱线图	移动平均替换	非事件驱动的异常
时间戳不规范	时间序列连续性检查	重新索引对齐	数据合并后时间不一致
成交量为零	零值检查	中位数填充	非停牌日的成交量缺失

import pandas as pd
import numpy as np
from scipy import stats
from datetime import timedelta

class SmartDataRepairer:
    """智能数据修复系统"""
    
    def __init__(self, symbol: str, data: pd.DataFrame = None):
        self.symbol = symbol
        self.original_data = data.copy() if data is not None else self._load_default_data()
        self.repaired_data = self.original_data.copy()
        self.repair_log = []
    
    def _load_default_data(self) -> pd.DataFrame:
        """加载默认数据"""
        ticker = yf.Ticker(self.symbol)
        return ticker.history(period="1y", interval="1d")
    
    def repair_missing_values(self, method: str = "forward_fill") -> "SmartDataRepairer":
        """修复缺失值"""
        # 记录原始缺失情况
        original_missing = self.repaired_data.isnull().sum().sum()
        
        if original_missing == 0:
            self.repair_log.append({
                "action": "repair_missing_values",
                "method": method,
                "affected_records": 0,
                "message": "未发现缺失值"
            })
            return self
        
        # 根据选择的方法修复缺失值
        if method == "forward_fill":
            self.repaired_data = self.repaired_data.ffill()
        elif method == "backward_fill":
            self.repaired_data = self.repaired_data.bfill()
        elif method == "interpolate":
            self.repaired_data = self.repaired_data.interpolate(method="time")
        else:
            raise ValueError(f"不支持的修复方法: {method}")
        
        # 记录修复结果
        remaining_missing = self.repaired_data.isnull().sum().sum()
        self.repair_log.append({
            "action": "repair_missing_values",
            "method": method,
            "affected_records": original_missing - remaining_missing,
            "message": f"修复了 {original_missing - remaining_missing} 个缺失值，仍有 {remaining_missing} 个未修复"
        })
        
        return self
    
    def repair_price_logic(self) -> "SmartDataRepairer":
        """修复价格逻辑错误"""
        # 找出最高价 < 最低价的记录
        invalid_high_low = self.repaired_data[self.repaired_data['High'] < self.repaired_data['Low']].index
        count_high_low = len(invalid_high_low)
        
        # 找出收盘价超出高低价范围的记录
        invalid_close = self.repaired_data[
            (self.repaired_data['Close'] < self.repaired_data['Low']) | 
            (self.repaired_data['Close'] > self.repaired_data['High'])
        ].index
        count_close = len(invalid_close)
        
        # 修复最高价 < 最低价的问题
        if count_high_low > 0:
            # 交换高低价
            self.repaired_data.loc[invalid_high_low, ['High', 'Low']] = \
                self.repaired_data.loc[invalid_high_low, ['Low', 'High']].values
            
            self.repair_log.append({
                "action": "repair_price_logic",
                "issue": "high_low_inversion",
                "affected_records": count_high_low,
                "message": f"修复了 {count_high_low} 条最高价低于最低价的记录"
            })
        
        # 修复收盘价超出范围的问题
        if count_close > 0:
            # 将收盘价限制在高低价范围内
            self.repaired_data.loc[invalid_close, 'Close'] = self.repaired_data.loc[invalid_close, ['Close', 'High', 'Low']].apply(
                lambda x: min(max(x['Close'], x['Low']), x['High']), axis=1
            )
            
            self.repair_log.append({
                "action": "repair_price_logic",
                "issue": "close_out_of_range",
                "affected_records": count_close,
                "message": f"修复了 {count_close} 条收盘价超出高低价范围的记录"
            })
        
        return self
    
    def repair_abnormal_volatility(self, z_threshold: float = 3.0, window: int = 20) -> "SmartDataRepairer":
        """修复异常价格波动"""
        # 计算收益率
        returns = self.repaired_data['Close'].pct_change().dropna()
        
        # 计算Z-score
        z_scores = np.abs(stats.zscore(returns))
        abnormal_indices = returns[z_scores > z_threshold].index
        
        if len(abnormal_indices) == 0:
            self.repair_log.append({
                "action": "repair_abnormal_volatility",
                "affected_records": 0,
                "message": "未发现异常价格波动"
            })
            return self
        
        # 使用移动平均修复异常值
        for idx in abnormal_indices:
            # 获取窗口数据（排除当前异常值）
            start_idx = max(0, self.repaired_data.index.get_loc(idx) - window)
            window_data = self.repaired_data.iloc[start_idx:self.repaired_data.index.get_loc(idx)]
            
            if len(window_data) >= window // 2:  # 确保有足够的数据点
                # 保存原始值用于日志
                original_value = self.repaired_data.loc[idx, 'Close']
                
                # 使用窗口数据的移动平均替换
                self.repaired_data.loc[idx, 'Close'] = window_data['Close'].mean()
                
                # 相应调整开盘价、最高价和最低价
                price_ratio = self.repaired_data.loc[idx, 'Close'] / original_value
                self.repaired_data.loc[idx, ['Open', 'High', 'Low']] *= price_ratio
        
        self.repair_log.append({
            "action": "repair_abnormal_volatility",
            "affected_records": len(abnormal_indices),
            "message": f"使用 {window} 天移动平均修复了 {len(abnormal_indices)} 条异常价格波动记录"
        })
        
        return self
    
    def get_repaired_data(self) -> pd.DataFrame:
        """获取修复后的数据"""
        return self.repaired_data
    
    def get_repair_summary(self) -> dict:
        """获取修复摘要"""
        total_affected = sum(log["affected_records"] for log in self.repair_log)
        
        return {
            "symbol": self.symbol,
            "original_records": len(self.original_data),
            "repaired_records": total_affected,
            "repair_actions": len(self.repair_log),
            "repair_details": self.repair_log
        }

# 修复特斯拉股票数据
try:
    # 首先获取数据并故意引入一些问题用于演示
    ticker = yf.Ticker("TSLA")
    data = ticker.history(period="1y")
    
    # 模拟一些数据问题
    data.loc[data.sample(frac=0.05).index, 'Volume'] = 0  # 5%的成交量为零
    data.iloc[10:15, data.columns.get_loc('High')] = data.iloc[10:15, data.columns.get_loc('Low')] - 1  # 制造高低价颠倒
    
    # 创建修复器并执行修复
    repairer = SmartDataRepairer("TSLA", data)
    repaired_data = repairer.repair_missing_values(method="interpolate") \
                            .repair_price_logic() \
                            .repair_abnormal_volatility(z_threshold=3.5) \
                            .get_repaired_data()
    
    # 获取修复摘要
    summary = repairer.get_repair_summary()
    print(f"🔧 {summary['symbol']} 数据修复摘要")
    print(f"原始记录数: {summary['original_records']}")
    print(f"修复记录数: {summary['repaired_records']}")
    print(f"执行修复操作: {summary['repair_actions']}")
    
    print("\n修复详情:")
    for i, detail in enumerate(summary['repair_details'], 1):
        print(f"{i}. {detail['message']}")
        
except Exception as e:
    print(f"❌ 数据修复过程出错: {str(e)}")

应用场景：历史数据清洗、回测数据预处理、实时数据质量监控等场景。 注意事项：数据修复可能引入偏差，修复后的数据应标注并在策略回测中评估修复对结果的影响；对于重大异常（如长时间停牌后的复牌），自动修复可能不如人工处理准确。

常见误区解析

误区	正确认知
所有缺失值都应该被修复	应先分析缺失原因，停牌导致的缺失不应修复
数据修复越彻底越好	过度修复可能平滑真实的市场波动，影响策略有效性
异常值一定是错误	某些异常值可能是真实的市场事件（如 earnings 发布）
修复方法可以通用	不同类型的数据问题需要匹配不同的修复策略

设计鲁棒的API请求策略：重试、限流与故障转移

yfinance作为非官方API客户端，面临着请求限制和连接不稳定等问题。设计鲁棒的API请求策略可以显著提高数据获取的可靠性，减少因网络问题或服务器限制导致的失败。

API请求优化策略

指数退避重试：失败后逐渐增加重试间隔，避免加重服务器负担
请求限流：控制并发请求数量和频率，避免触发反爬机制
故障转移：当主数据源不可用时，切换到备用数据源
缓存策略：合理使用缓存减少重复请求
超时控制：设置适当的请求超时时间，避免无限等待

import yfinance as yf
import time
import random
import pandas as pd
from requests.exceptions import RequestException
from functools import wraps

def retry_with_backoff(max_retries=3, backoff_factor=0.3, exceptions=(RequestException,)):
    """带指数退避的重试装饰器"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            retry_count = 0
            while retry_count < max_retries:
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    retry_count += 1
                    if retry_count >= max_retries:
                        raise
                    
                    # 指数退避策略
                    sleep_time = backoff_factor * (2 ** (retry_count - 1)) + random.uniform(0, 0.1)
                    print(f"⚠️ 请求失败 (尝试 {retry_count}/{max_retries})，{sleep_time:.2f}秒后重试: {str(e)}")
                    time.sleep(sleep_time)
            return func(*args, **kwargs)
        return wrapper
    return decorator

class RobustDataFetcher:
    """鲁棒的数据获取器"""
    
    def __init__(self, cache_dir=None, max_concurrent=5, request_delay=1.0):
        self.cache = {}
        self.cache_dir = cache_dir
        self.max_concurrent = max_concurrent  # 最大并发数
        self.request_delay = request_delay  # 请求间隔（秒）
        self.last_request_time = 0
    
    def _rate_limit(self):
        """速率限制控制"""
        current_time = time.time()
        elapsed = current_time - self.last_request_time
        
        if elapsed < self.request_delay:
            sleep_time = self.request_delay - elapsed
            time.sleep(sleep_time)
        
        self.last_request_time = time.time()
    
    @retry_with_backoff(max_retries=5, backoff_factor=0.5)
    def fetch_ticker_data(self, symbol, period="1y", interval="1d", use_cache=True):
        """获取单只股票数据"""
        # 检查缓存
        cache_key = f"{symbol}_{period}_{interval}"
        if use_cache and cache_key in self.cache:
            print(f"📦 使用缓存数据: {symbol}")
            return self.cache[cache_key]
        
        # 速率限制
        self._rate_limit()
        
        # 获取数据
        print(f"📥 获取数据: {symbol}")
        ticker = yf.Ticker(symbol)
        data = ticker.history(period=period, interval=interval)
        
        if data.empty:
            raise ValueError(f"无法获取 {symbol} 的数据")
        
        # 存入缓存
        self.cache[cache_key] = data
        
        return data
    
    def fetch_multiple_tickers(self, symbols, period="1y", interval="1d", batch_size=5):
        """获取多只股票数据"""
        all_data = {}
        
        # 分批处理
        for i in range(0, len(symbols), batch_size):
            batch = symbols[i:i+batch_size]
            print(f"\n处理批次 {i//batch_size + 1}/{(len(symbols)+batch_size-1)//batch_size}")
            
            # 逐个获取（避免并发过高）
            for symbol in batch:
                try:
                    data = self.fetch_ticker_data(symbol, period, interval)
                    all_data[symbol] = data
                except Exception as e:
                    print(f"❌ 无法获取 {symbol} 数据: {str(e)}")
                    continue
        
        return all_data

# 使用鲁棒数据获取器
try:
    fetcher = RobustDataFetcher(request_delay=1.5)  # 设置1.5秒请求间隔
    
    # 获取FAANG股票数据
    faang_symbols = ["META", "AAPL", "AMZN", "NFLX", "GOOGL"]
    faang_data = fetcher.fetch_multiple_tickers(faang_symbols, period="6mo", batch_size=2)
    
    # 打印结果摘要
    print("\n📊 数据获取结果:")
    for symbol, data in faang_data.items():
        print(f"{symbol}: {len(data)} 条记录, 日期范围: {data.index.min().strftime('%Y-%m-%d')} 至 {data.index.max().strftime('%Y-%m-%d')}")
        
except Exception as e:
    print(f"❌ 数据获取过程出错: {str(e)}")

应用场景：批量数据获取、定时数据更新任务、对可靠性要求高的生产环境等场景。 注意事项：请求间隔和并发数应根据实际情况调整，过于频繁的请求可能导致IP被临时封禁；缓存策略应考虑数据时效性，对实时性要求高的数据不应长时间缓存。

四、效能优化：从代码到架构的全方位提升

构建分布式数据采集系统：基于任务队列的架构设计

对于大规模数据采集需求，单进程单线程的方式效率低下，构建基于任务队列的分布式系统可以显著提升数据获取能力，支持大规模股票池和高频数据采集。

系统架构组件

任务队列：存储待采集的股票代码和参数
工作节点：多个并行工作的采集进程/线程
结果存储：数据库或文件系统存储采集结果
监控系统：跟踪任务进度和系统状态
调度器：分配任务并处理失败重试

import yfinance as yf
import pandas as pd
from queue import Queue
from threading import Thread, Lock
from dataclasses import dataclass
from typing import List, Dict, Callable
import time
import logging
from pathlib import Path

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(threadName)s - %(levelname)s - %(message)s'
)

@dataclass
class DataTask:
    """数据采集任务"""
    symbol: str
    period: str
    interval: str
    priority: int = 1  # 1-5，5为最高优先级
    retries: int = 0
    max_retries: int = 3

class TaskQueue:
    """任务队列"""
    def __init__(self):
        self.queue = Queue()
        self.lock = Lock()
        self.completed_tasks = 0
        self.failed_tasks = 0
        self.total_tasks = 0
    
    def add_task(self, task: DataTask):
        """添加任务"""
        with self.lock:
            self.queue.put((task.priority, task))
            self.total_tasks += 1
    
    def get_task(self) -> DataTask:
        """获取任务（按优先级）"""
        priority, task = self.queue.get()
        return task
    
    def task_complete(self, success: bool):
        """标记任务完成"""
        with self.lock:
            if success:
                self.completed_tasks += 1
            else:
                self.failed_tasks += 1
        self.queue.task_done()
    
    def get_status(self) -> Dict:
        """获取队列状态"""
        with self.lock:
            return {
                "total": self.total_tasks,
                "completed": self.completed_tasks,
                "failed": self.failed_tasks,
                "remaining": self.queue.qsize()
            }

class Worker(Thread):
    """工作节点"""
    def __init__(self, queue: TaskQueue, result_handler: Callable, name: str = None):
        super().__init__(name=name)
        self.queue = queue
        self.result_handler = result_handler
        self.running = True
        self.delay = 1.5  # 基础延迟时间（秒）
    
    def run(self):
        """运行工作节点"""
        logging.info(f"工作节点 {self.name} 启动")
        
        while self.running:
            try:
                # 获取任务
                task = self.queue.get_task()
                logging.info(f"处理任务: {task.symbol} (重试: {task.retries}/{task.max_retries})")
                
                try:
                    # 获取数据
                    ticker = yf.Ticker(task.symbol)
                    data = ticker.history(period=task.period, interval=task.interval)
                    
                    if data.empty:
                        raise ValueError("返回空数据")
                    
                    # 处理结果
                    self.result_handler(task.symbol, data)
                    
                    # 标记成功
                    self.queue.task_complete(success=True)
                    logging.info(f"完成任务: {task.symbol}")
                    
                    # 随机延迟，避免请求过于规律
                    time.sleep(self.delay + random.uniform(-0.3, 0.3))
                    
                except Exception as e:
                    logging.error(f"任务 {task.symbol} 失败: {str(e)}")
                    
                    # 检查是否需要重试
                    if task.retries < task.max_retries:
                        task.retries += 1
                        logging.info(f"重新排队任务: {task.symbol} (重试 {task.retries}/{task.max_retries})")
                        self.queue.add_task(task)
                    
                    # 标记失败
                    self.queue.task_complete(success=False)
                
            except Exception as e:
                logging.error(f"工作节点错误: {str(e)}")
                time.sleep(5)
        
        logging.info(f"工作节点 {self.name} 停止")
    
    def stop(self):
        """停止工作节点"""
        self.running = False

class DistributedDataCollector:
    """分布式数据采集系统"""
    def __init__(self, output_dir: str = "data_output", num_workers: int = 3):
        self.task_queue = TaskQueue()
        self.output_dir = Path(output_dir)
        self.output_dir.mkdir(exist_ok=True)
        self.workers = []
        self.num_workers = num_workers
        
        # 初始化工作节点
        for i in range(num_workers):
            worker = Worker(
                queue=self.task_queue,
                result_handler=self.save_data,
                name=f"worker-{i+1}"
            )
            self.workers.append(worker)
    
    def save_data(self, symbol: str, data: pd.DataFrame):
        """保存数据到文件"""
        # 创建符号目录
        symbol_dir = self.output_dir / symbol
        symbol_dir.mkdir(exist_ok=True)
        
        # 保存为CSV
        filename = f"{symbol}_data.csv"
        data.to_csv(symbol_dir / filename)
    
    def add_tasks(self, symbols: List[str], period: str = "1y", interval: str = "1d", priority: int = 1):
        """添加任务"""
        for symbol in symbols:
            task = DataTask(
                symbol=symbol,
                period=period,
                interval=interval,
                priority=priority
            )
            self.task_queue.add_task(task)
    
    def start(self):
        """启动采集系统"""
        # 启动所有工作节点
        for worker in self.workers:
            worker.start()
        
        logging.info(f"分布式采集系统启动，{self.num_workers}个工作节点，{self.task_queue.get_status()['total']}个任务")
    
    def wait_complete(self, check_interval: int = 5):
        """等待所有任务完成"""
        while True:
            status = self.task_queue.get_status()
            
            if status["remaining"] == 0:
                logging.info("所有任务处理完成")
                break
                
            logging.info(f"任务状态: 总任务 {status['total']}, 已完成 {status['completed']}, 失败 {status['failed']}, 剩余 {status['remaining']}")
            time.sleep(check_interval)
    
    def stop(self):
        """停止采集系统"""
        # 停止所有工作节点
        for worker in self.workers:
            worker.stop()
        
        # 等待所有工作节点退出
        for worker in self.workers:
            worker.join()
        
        logging.info("分布式采集系统已停止")

# 使用分布式数据采集系统
if __name__ == "__main__":
    # 从文件读取股票列表（示例使用FAANG+一些额外股票）
    stock_symbols = ["META", "AAPL", "AMZN", "NFLX", "GOOGL", "TSLA", "MSFT", "NVDA", "BABA", "PDD"]
    
    # 创建采集系统（3个工作节点）
    collector = DistributedDataCollector(output_dir="distributed_data", num_workers=3)
    
    # 添加任务
    collector.add_tasks(stock_symbols, period="1y", interval="1d")
    
    try:
        # 启动系统
        collector.start()
        
        # 等待完成
        collector.wait_complete()
        
    except KeyboardInterrupt:
        print("\n收到中断信号，正在停止系统...")
    finally:
        # 停止系统
        collector.stop()
        
        # 打印最终状态
        final_status = collector.task_queue.get_status()
        print(f"\n最终状态: 总任务 {final_status['total']}, 成功 {final_status['completed']}, 失败 {final_status['failed']}")

应用场景：大规模股票池数据采集、历史数据批量下载、定期数据更新等需要处理大量数据的场景。 注意事项：工作节点数量不宜过多，避免触发API限制；分布式系统需考虑错误恢复和任务追踪机制；应实现数据完整性校验，确保所有任务都被正确处理。

实现智能缓存系统：多级缓存与过期策略

缓存是提升数据获取效率的关键技术，实现智能缓存系统可以显著减少重复请求，降低API调用频率，同时保证数据的新鲜度和准确性。

多级缓存架构

内存缓存：最快的缓存层，存储最近访问的热点数据
磁盘缓存：持久化存储，保存历史数据和不常访问数据
分布式缓存：多节点共享缓存，适用于分布式系统

缓存过期策略

时间过期：根据数据类型设置不同的过期时间（如实时数据5分钟，历史数据1天）
访问频率：不常访问的数据优先淘汰
数据重要性：核心数据设置更长的过期时间

import yfinance as yf
import pandas as pd
import os
import time
import hashlib
import json
from pathlib import Path
from functools import lru_cache
from datetime import datetime, timedelta
from typing import Dict, Optional, Any

class SmartCache:
    """智能缓存系统"""
    
    def __init__(self, cache_dir: str = "smart_cache", memory_cache_size: int = 100):
        self.cache_dir = Path(cache_dir)
        self.cache_dir.mkdir(exist_ok=True)
        
        # 内存缓存 - 使用LRU策略
        self.memory_cache = {}
        
        # 缓存配置 - 根据数据类型定义过期时间（秒）
        self.cache_config = {
            "history_daily": 86400,  # 日线数据：1天
            "history_intraday": 300,  # 日内数据：5分钟
            "info": 3600,  # 公司信息：1小时
            "financials": 43200,  # 财务数据：12小时
            "news": 600,  # 新闻数据：10分钟
            "other": 1800  # 其他数据：30分钟
        }
    
    def _generate_key(self, symbol: str, data_type: str, **params) -> str:
        """生成缓存键"""
        # 将参数排序并转换为字符串
        params_str = json.dumps(sorted(params.items()), sort_keys=True)
        # 创建唯一哈希键
        key_str = f"{symbol}_{data_type}_{params_str}"
        return hashlib.md5(key_str.encode()).hexdigest()
    
    def _get_cache_path(self, key: str) -> Path:
        """获取缓存文件路径"""
        # 使用哈希的前两位作为子目录，分散文件
        subdir = key[:2]
        subdir_path = self.cache_dir / subdir
        subdir_path.mkdir(exist_ok=True)
        return subdir_path / f"{key}.pkl"
    
    def _is_valid(self, cache_path: Path, data_type: str) -> bool:
        """检查缓存是否有效"""
        if not cache_path.exists():
            return False
            
        # 检查文件修改时间
        modified_time = os.path.getmtime(cache_path)
        current_time = time.time()
        ttl = self.cache_config.get(data_type, self.cache_config["other"])
        
        return (current_time - modified_time) < ttl
    
    def get(self, symbol: str, data_type: str, **params) -> Optional[Any]:
        """获取缓存数据"""
        key = self._generate_key(symbol, data_type, **params)
        
        # 1. 先检查内存缓存
        if key in self.memory_cache:
            mem_data, timestamp = self.memory_cache[key]
            ttl = self.cache_config.get(data_type, self.cache_config["other"])
            
            if (time.time() - timestamp) < ttl:
                # 内存缓存有效
                return mem_data
        
        # 2. 检查磁盘缓存
        cache_path = self._get_cache_path(key)
        if self._is_valid(cache_path, data_type):
            try:
                # 从磁盘加载
                data = pd.read_pickle(cache_path)
                
                # 更新内存缓存
                self.memory_cache[key] = (data, time.time())
                
                # 限制内存缓存大小，简单LRU淘汰
                if len(self.memory_cache) > 1000:
                    # 按时间戳排序并删除最旧的100个
                    sorted_items = sorted(self.memory_cache.items(), key=lambda x: x[1][1])
                    for k, _ in sorted_items[:100]:
                        del self.memory_cache[k]
                
                return data
            except Exception as e:
                print(f"⚠️ 读取缓存失败: {str(e)}")
                # 删除损坏的缓存文件
                if cache_path.exists():
                    cache_path.unlink()
        
        # 缓存无效或不存在
        return None
    
    def set(self, symbol: str, data_type: str, data: Any, **params) -> None:
        """设置缓存数据"""
        key = self._generate_key(symbol, data_type, **params)
        cache_path = self._get_cache_path(key)
        
        try:
            # 保存到磁盘
            pd.to_pickle(data, cache_path)
            
            # 更新内存缓存
            self.memory_cache[key] = (data, time.time())
            
        except Exception as e:
            print(f"⚠️ 写入缓存失败: {str(e)}")
    
    def clear_expired(self) -> int:
        """清理过期缓存"""
        count = 0
        
        # 遍历所有缓存文件
        for subdir in self.cache_dir.iterdir():
            if not subdir.is_dir():
                continue
                
            for cache_file in subdir.iterdir():
                if not cache_file.suffix == ".pkl":
                    continue
                    
                # 从文件名提取key
                key = cache_file.stem
                
                # 解析数据类型（简单方式，实际应用中可能需要更复杂的解析）
                # 这里简化处理，假设所有缓存都使用默认TTL
                data_type = "other"
                
                if not self._is_valid(cache_file, data_type):
                    cache_file.unlink()
                    count += 1
        
        # 清理内存中过期的缓存
        current_time = time.time()
        to_remove = []
        
        for key, (_, timestamp) in self.memory_cache.items():
            # 简化处理，实际应用需要解析data_type
            ttl = self.cache_config["other"]
            if (current_time - timestamp) >= ttl:
                to_remove.append(key)
                
        for key in to_remove:
            del self.memory_cache[key]
            
        return count

class CachedDataFetcher:
    """带缓存的数据获取器"""
    
    def __init__(self, cache: SmartCache = None):
        self.cache = cache if cache is not None else SmartCache()
    
    def get_history(self, symbol: str, period: str = "1y", interval: str = "1d") -> pd.DataFrame:
        """获取历史价格数据（带缓存）"""
        # 确定数据类型
        if interval in ["1m", "2m", "5m", "15m", "30m", "60m", "90m"]:
            data_type = "history_intraday"
        else:
            data_type = "history_daily"
        
        # 尝试从缓存获取
        cached_data = self.cache.get(
            symbol=symbol,
            data_type=data_type,
            period=period,
            interval=interval
        )
        
        if cached_data is not None:
            print(f"📦 从缓存加载 {symbol} 历史数据")
            return cached_data
        
        # 缓存未命中，从API获取
        print(f"📥 从API获取 {symbol} 历史数据")
        ticker = yf.Ticker(symbol)
        data = ticker.history(period=period, interval=interval)
        
        if not data.empty:
            # 存入缓存
            self.cache.set(
                symbol=symbol,
                data_type=data_type,
                data=data,
                period=period,
                interval=interval
            )
        
        return data
    
    def get_company_info(self, symbol: str) -> Dict:
        """获取公司信息（带缓存）"""
        # 尝试从缓存获取
        cached_data = self.cache.get(
            symbol=symbol,
            data_type="info"
        )
        
        if cached_data is not None:
            print(f"📦 从缓存加载 {symbol} 公司信息")
            return cached_data
        
        # 缓存未命中，从API获取
        print(f"📥 从API获取 {symbol} 公司信息")
        ticker = yf.Ticker(symbol)
        info = ticker.info
        
        if info:
            # 存入缓存
            self.cache.set(
                symbol=symbol,
                data_type="info",
                data=info
            )
        
        return info

# 使用智能缓存系统
if __name__ == "__main__":
    # 创建智能缓存
    smart_cache = SmartCache(cache_dir="yfinance_smart_cache")
    
    # 创建带缓存的数据获取器
    fetcher = CachedDataFetcher(smart_cache)
    
    # 第一次获取（无缓存）
    start_time = time.time()
    aapl_history = fetcher.get_history("AAPL", period="1y", interval="1d")
    print(f"首次获取耗时: {time.time() - start_time:.2f}秒")
    
    # 第二次获取（有缓存）
    start_time = time.time()
    aapl_history_cached = fetcher.get_history("AAPL", period="1y", interval="1d")
    print(f"缓存获取耗时: {time.time() - start_time:.2f}秒")
    
    # 获取公司信息
    msft_info = fetcher.get_company_info("MSFT")
    print(f"微软公司名称: {msft_info.get('longName')}")
    print(f"微软市值: {msft_info.get('marketCap'):,} USD")
    
    # 清理过期缓存
    expired_count = smart_cache.clear_expired()
    print(f"清理了 {expired_count} 个过期缓存文件")

应用场景：高频数据查询、交互式分析工具、多用户共享数据服务等场景。 注意事项：缓存策略需根据数据更新频率和重要性调整；内存缓存大小应根据可用内存合理设置；定期清理过期缓存可以释放磁盘空间。

行业应用延伸：在专业量化交易系统中，缓存系统通常与数据预处理流水线结合，将原始数据转换为策略所需的特征数据后再进行缓存，进一步提高策略回测和实盘运行效率。对于机构用户，分布式缓存系统（如Redis）常被用于多节点共享数据，降低整体API调用成本。

附录：实用资源与高频问题

高频问题速查表

问题	解决方案
API请求频繁失败	1. 增加请求间隔 2. 实现指数退避重试 3. 检查网络代理
数据不完整或缺失	1. 使用repair_price_repair模块 2. 结合其他数据源验证 3. 手动检查异常日期
内存占用过高	1. 分块处理大数据 2. 使用高效数据类型 3. 及时释放不再使用的变量
与pandas版本不兼容	1. 固定pandas版本到1.5.x 2. 修改过时的语法 3. 更新yfinance到最新版本
中文显示乱码	1. 设置matplotlib字体 2. 使用seaborn替代matplotlib 3. 保存为SVG格式

资源链接汇总

官方文档：项目内doc目录
源代码：项目根目录
测试数据：tests/data目录
示例代码：doc/source/reference/examples目录
开发指南：doc/source/development目录

通过本教程的学习，你已经掌握了yfinance库的核心功能、实战应用、问题诊断和效能优化的全方位知识。从基础的数据获取到构建分布式采集系统，从简单的技术指标计算到智能缓存策略，这些技能将帮助你在量化分析领域构建可靠、高效的数据基础架构。随着实践的深入，你可以进一步探索更复杂的应用场景，如实时数据处理、多源数据融合和机器学习预测模型构建，不断提升你的量化分析能力。

yfinance

Download market data from Yahoo! Finance's API

项目地址：https://gitcode.com/GitHub_Trending/yf/yfinance

登录后查看全文