重构金融数据获取：yfinance高级应用与最佳实践指南

2026-04-12 09:54:03作者：管翌锬

一、认知重构：深入理解yfinance数据获取机制

1.1 揭开yfinance工作原理的黑箱

yfinance作为雅虎财经数据的非官方接口，通过模拟浏览器请求获取金融数据，其核心工作流程包括：请求构造→数据解析→格式转换→缓存管理。与传统API不同，yfinance无需API密钥，但需要理解其请求限制和数据结构。

认知锚点：

请求模拟：yfinance通过构造特定URL和请求头模拟浏览器行为，绕过简单的反爬机制
数据缓存：内置缓存系统减少重复请求，默认缓存位置在用户主目录下
数据标准化：将不同金融产品的原始数据统一转换为pandas DataFrame格式

场景化任务：实现一个基础的yfinance数据请求分析器，打印出实际发送的URL和响应状态码。

import yfinance as yf
from yfinance import shared

def analyze_yfinance_request(symbol):
    """分析yfinance数据请求过程"""
    # 保存原始请求函数
    original_request = shared.get

    # 自定义请求处理函数
    def log_request(url, *args, **kwargs):
        print(f"请求URL: {url}")
        response = original_request(url, *args, **kwargs)
        print(f"响应状态码: {response.status_code}")
        return response
    
    # 替换请求函数
    shared.get = log_request
    
    try:
        # 请求数据以触发请求
        ticker = yf.Ticker(symbol)
        ticker.history(period="1d")
        print("请求分析完成")
    finally:
        # 恢复原始请求函数
        shared.get = original_request

# 执行请求分析 (执行耗时: ~2秒)
analyze_yfinance_request("AAPL")

1.2 yfinance架构与核心组件解析

yfinance库采用模块化设计，主要包含以下核心组件：

组件	功能描述	关键类/函数
数据获取层	处理网络请求与响应	Ticker, download
数据解析层	解析原始数据为结构化格式	_parse_price_history
缓存管理层	处理数据缓存与失效策略	set_tz_cache_location
异常处理层	处理网络错误与数据异常	YFinanceException
工具函数层	提供辅助功能与数据转换	utils模块

场景化任务：创建一个自定义Ticker类，继承并扩展yfinance的Ticker类，添加自定义数据处理方法。

import yfinance as yf
from yfinance import Ticker

class EnhancedTicker(Ticker):
    """增强版Ticker类，添加自定义分析功能"""
    
    def __init__(self, ticker, session=None):
        super().__init__(ticker, session)
        
    def get_volatility(self, period="1mo"):
        """计算指定周期内的波动率"""
        hist = self.history(period=period)
        if hist.empty:
            return None
        return hist['Close'].pct_change().std() * (252 ** 0.5)  # 年化波动率
    
    def get_beta(self, market_symbol="^GSPC", period="1y"):
        """计算股票相对于市场的beta值"""
        stock_data = self.history(period=period)['Close'].pct_change().dropna()
        market_data = yf.Ticker(market_symbol).history(period=period)['Close'].pct_change().dropna()
        
        # 对齐日期
        combined = stock_data.to_frame('stock').join(
            market_data.to_frame('market'), how='inner'
        )
        
        if len(combined) < 2:
            return None
            
        # 计算协方差和市场方差
        covariance = combined.cov().iloc[0, 1]
        market_variance = combined['market'].var()
        
        return covariance / market_variance

# 使用增强版Ticker类 (执行耗时: ~4秒)
aapl = EnhancedTicker("AAPL")
print(f"波动率: {aapl.get_volatility():.4f}")
print(f"Beta值: {aapl.get_beta():.4f}")

二、应用突破：yfinance在量化分析中的创新应用

2.1 构建多资产类别投资组合分析系统

传统金融数据获取工具往往局限于单一资产类别，而yfinance支持股票、指数、加密货币、ETF等多种资产，为构建多元化投资组合分析提供了可能。

认知锚点：

多资产支持：通过不同的代码后缀区分资产类别（如-USD表示加密货币）
投资组合构建：利用pandas的多索引功能组织不同资产数据
相关性分析：计算不同资产间的相关性，优化投资组合分散度

场景化任务：创建一个投资组合分析工具，支持股票、债券和加密货币的混合分析。

import yfinance as yf
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt

class PortfolioAnalyzer:
    """投资组合分析工具"""
    
    def __init__(self, assets, weights=None):
        """
        初始化投资组合分析器
        
        参数:
            assets: 字典，键为资产名称，值为yfinance代码
            weights: 字典，键为资产名称，值为权重
        """
        self.assets = assets
        self.weights = weights or {name: 1/len(assets) for name in assets}
        self.data = None
        
    def fetch_data(self, start_date=None, end_date=None, period="1y"):
        """获取资产历史数据"""
        # 下载所有资产数据
        data = {}
        for name, symbol in self.assets.items():
            ticker = yf.Ticker(symbol)
            hist = ticker.history(period=period, start=start_date, end=end_date)['Close']
            data[name] = hist
        
        self.data = pd.DataFrame(data)
        return self.data
    
    def calculate_returns(self):
        """计算资产收益率"""
        if self.data is None:
            raise ValueError("请先调用fetch_data获取数据")
        return self.data.pct_change().dropna()
    
    def portfolio_metrics(self):
        """计算投资组合关键指标"""
        returns = self.calculate_returns()
        
        # 组合收益率
        portfolio_returns = (returns * pd.Series(self.weights)).sum(axis=1)
        
        # 计算指标
        total_return = (1 + portfolio_returns).prod() - 1
        annualized_return = (1 + total_return) ** (252/len(portfolio_returns)) - 1
        volatility = portfolio_returns.std() * np.sqrt(252)
        sharpe_ratio = annualized_return / volatility  # 假设无风险利率为0
        
        return {
            "总收益率": total_return,
            "年化收益率": annualized_return,
            "波动率": volatility,
            "夏普比率": sharpe_ratio
        }
    
    def plot_correlation_matrix(self):
        """绘制资产相关性矩阵热力图"""
        returns = self.calculate_returns()
        corr_matrix = returns.corr()
        
        plt.figure(figsize=(10, 8))
        plt.imshow(corr_matrix, cmap='coolwarm', interpolation='nearest')
        plt.colorbar(label='相关系数')
        plt.xticks(range(len(corr_matrix.columns)), corr_matrix.columns, rotation=45)
        plt.yticks(range(len(corr_matrix.columns)), corr_matrix.columns)
        
        # 添加相关系数值
        for i in range(len(corr_matrix.columns)):
            for j in range(len(corr_matrix.columns)):
                plt.text(j, i, f"{corr_matrix.iloc[i, j]:.2f}", 
                         ha='center', va='center', color='white')
        
        plt.title('资产相关性矩阵')
        plt.tight_layout()
        plt.show()

# 创建投资组合分析器 (执行耗时: ~8秒，取决于资产数量)
portfolio = PortfolioAnalyzer({
    "美国股票": "SPY",
    "黄金": "GLD",
    "比特币": "BTC-USD",
    "国债": "TLT"
}, {
    "美国股票": 0.4,
    "黄金": 0.2,
    "比特币": 0.1,
    "国债": 0.3
})

portfolio.fetch_data(period="2y")
print("投资组合指标:")
metrics = portfolio.portfolio_metrics()
for name, value in metrics.items():
    print(f"{name}: {value:.4f}")

portfolio.plot_correlation_matrix()

2.2 实现实时市场监控与异常检测系统

yfinance不仅能获取历史数据，还可以通过定期轮询实现接近实时的市场监控，结合异常检测算法及时发现市场突变。

认知锚点：

实时数据获取：通过定期调用Ticker.info()方法获取最新价格
异常检测：使用统计方法识别价格异常波动
事件驱动：当检测到异常时触发自定义事件处理

场景化任务：开发一个实时市场监控系统，当股票价格出现异常波动时发送通知。

import yfinance as yf
import time
import numpy as np
from datetime import datetime
from collections import deque

class MarketMonitor:
    """实时市场监控系统"""
    
    def __init__(self, symbols, window_size=20, z_threshold=3.0, check_interval=60):
        """
        初始化市场监控器
        
        参数:
            symbols: 要监控的股票代码列表
            window_size: 计算移动统计量的窗口大小
            z_threshold: Z分数异常检测阈值
            check_interval: 检查间隔(秒)
        """
        self.symbols = symbols
        self.window_size = window_size
        self.z_threshold = z_threshold
        self.check_interval = check_interval
        self.price_history = {symbol: deque(maxlen=window_size) for symbol in symbols}
        self.running = False
        
    def start_monitoring(self, duration=None):
        """开始监控市场"""
        self.running = True
        start_time = datetime.now()
        
        print(f"开始市场监控 - {start_time.strftime('%Y-%m-%d %H:%M:%S')}")
        print(f"监控股票: {', '.join(self.symbols)}")
        print(f"检查间隔: {self.check_interval}秒")
        print("-" * 50)
        
        try:
            while self.running:
                # 检查是否达到持续时间
                if duration and (datetime.now() - start_time).total_seconds() >= duration:
                    break
                    
                for symbol in self.symbols:
                    self._check_symbol(symbol)
                    
                time.sleep(self.check_interval)
                
        except KeyboardInterrupt:
            print("\n监控被用户中断")
        finally:
            self.running = False
            print(f"\n监控结束 - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    
    def stop_monitoring(self):
        """停止监控"""
        self.running = False
    
    def _check_symbol(self, symbol):
        """检查单个股票的价格异常"""
        try:
            ticker = yf.Ticker(symbol)
            info = ticker.info
            current_price = info.get('currentPrice') or info.get('regularMarketPrice')
            
            if current_price is None:
                print(f"{symbol}: 无法获取当前价格")
                return
                
            timestamp = datetime.now().strftime('%H:%M:%S')
            self.price_history[symbol].append(current_price)
            
            # 只有在收集到足够数据后才进行异常检测
            if len(self.price_history[symbol]) >= self.window_size:
                prices = np.array(self.price_history[symbol])
                mean = np.mean(prices)
                std = np.std(prices)
                
                if std > 0:  # 避免除以零
                    z_score = (current_price - mean) / std
                    
                    if abs(z_score) > self.z_threshold:
                        self._handle_anomaly(symbol, current_price, z_score, mean, std)
                    else:
                        print(f"{timestamp} {symbol}: {current_price:.2f} (Z-score: {z_score:.2f})")
                else:
                    print(f"{timestamp} {symbol}: {current_price:.2f} (价格无波动)")
            else:
                print(f"{timestamp} {symbol}: {current_price:.2f} (收集数据中: {len(self.price_history[symbol])}/{self.window_size})")
                
        except Exception as e:
            print(f"{symbol} 检查出错: {str(e)}")
    
    def _handle_anomaly(self, symbol, price, z_score, mean, std):
        """处理检测到的价格异常"""
        timestamp = datetime.now().strftime('%H:%M:%S')
        anomaly_type = "上涨异常" if z_score > 0 else "下跌异常"
        
        print(f"\n⚠️ {timestamp} {symbol} 检测到{anomaly_type}!")
        print(f"当前价格: {price:.2f}, 平均价格: {mean:.2f}, Z分数: {z_score:.2f}")
        print(f"价格波动: {(price - mean)/mean*100:.2f}%")
        print("-" * 50)
        
        # 这里可以添加通知逻辑，如发送邮件、短信等

# 启动市场监控 (执行耗时: 持续运行，按Ctrl+C停止)
monitor = MarketMonitor(
    symbols=["AAPL", "MSFT", "GOOGL", "TSLA"],
    window_size=30,
    z_threshold=3.0,
    check_interval=30  # 每30秒检查一次
)

# 运行监控10分钟 (600秒) 或直到用户中断
monitor.start_monitoring(duration=600)

三、问题攻坚：解决yfinance数据获取中的关键挑战

3.1 突破API请求限制的高级策略

yfinance作为非官方API客户端，面临着雅虎财经的请求限制和反爬机制。理解并规避这些限制对于构建稳定的数据获取系统至关重要。

认知锚点：

请求限流：控制请求频率，避免触发临时封禁
会话复用：使用持久会话减少连接开销和识别风险
用户代理轮换：模拟不同浏览器和设备，降低被识别为爬虫的概率

场景化任务：实现一个智能请求管理器，自动处理请求限制和临时错误。

import yfinance as yf
import time
import random
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry
import requests

class SmartSessionManager:
    """智能会话管理器，处理请求限制和错误恢复"""
    
    def __init__(self, max_retries=3, backoff_factor=0.3, timeout=10):
        """
        初始化智能会话管理器
        
        参数:
            max_retries: 最大重试次数
            backoff_factor: 退避因子，用于计算重试间隔
            timeout: 请求超时时间(秒)
        """
        self.max_retries = max_retries
        self.backoff_factor = backoff_factor
        self.timeout = timeout
        self.user_agents = [
            "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36",
            "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/14.1.1 Safari/605.1.15",
            "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/92.0.4515.107 Safari/537.36",
            "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:89.0) Gecko/20100101 Firefox/89.0"
        ]
        self.session = self._create_session()
        
    def _create_session(self):
        """创建带有重试机制的会话"""
        session = requests.Session()
        
        # 设置重试策略
        retry_strategy = Retry(
            total=self.max_retries,
            backoff_factor=self.backoff_factor,
            status_forcelist=[429, 500, 502, 503, 504]
        )
        
        adapter = HTTPAdapter(max_retries=retry_strategy)
        session.mount("https://", adapter)
        session.mount("http://", adapter)
        
        return session
    
    def _random_user_agent(self):
        """随机选择一个用户代理"""
        return random.choice(self.user_agents)
    
    def fetch_ticker_data(self, symbol, period="1d", interval="1m"):
        """获取股票数据，带有智能重试和限流"""
        ticker = yf.Ticker(symbol, session=self.session)
        
        # 设置随机用户代理
        self.session.headers.update({"User-Agent": self._random_user_agent()})
        
        for attempt in range(self.max_retries + 1):
            try:
                data = ticker.history(period=period, interval=interval)
                
                if not data.empty:
                    return data
                    
                print(f"警告: {symbol} 返回空数据，尝试重试 ({attempt+1}/{self.max_retries+1})")
                
            except Exception as e:
                print(f"获取{symbol}数据失败: {str(e)}")
                
                if attempt < self.max_retries:
                    sleep_time = self.backoff_factor * (2 ** attempt) + random.uniform(0, 1)
                    print(f"将在{sleep_time:.2f}秒后重试...")
                    time.sleep(sleep_time)
        
        print(f"错误: 无法获取{symbol}数据，已达到最大重试次数")
        return None

# 使用智能会话管理器获取数据 (执行耗时: ~5秒)
session_manager = SmartSessionManager(max_retries=3, backoff_factor=0.5)
data = session_manager.fetch_ticker_data("AAPL", period="5d", interval="5m")

if data is not None:
    print(f"成功获取数据: {len(data)}行")
    print(data[['Open', 'High', 'Low', 'Close', 'Volume']].head())

3.2 处理金融数据中的特殊情况与异常值

金融数据往往包含各种特殊情况，如股票拆分、分红、停牌等，这些都可能导致数据异常。有效的异常处理策略是确保分析准确性的关键。

认知锚点：

除权除息调整：理解并处理分红和拆股对价格的影响
数据完整性检查：验证时间序列的连续性和完整性
异常值处理：区分真实价格波动和数据错误

场景化任务：构建一个金融数据清洗器，自动识别并处理常见的数据异常。

import yfinance as yf
import pandas as pd
import numpy as np
from scipy import stats

class FinancialDataCleaner:
    """金融数据清洗器，处理常见的数据异常"""
    
    def __init__(self, symbol):
        """初始化数据清洗器"""
        self.symbol = symbol
        self.data = None
        self.cleaned_data = None
        
    def fetch_data(self, period="1y", interval="1d"):
        """获取原始数据"""
        ticker = yf.Ticker(self.symbol)
        self.data = ticker.history(period=period, interval=interval)
        print(f"获取原始数据: {len(self.data)}行")
        return self.data
    
    def detect_and_handle_anomalies(self, z_threshold=3.0, volume_multiplier=5):
        """检测并处理数据异常"""
        if self.data is None:
            raise ValueError("请先调用fetch_data获取数据")
            
        self.cleaned_data = self.data.copy()
        
        # 1. 检查并处理时间序列连续性
        self._check_time_continuity()
        
        # 2. 检测并处理价格异常值
        self._detect_price_anomalies(z_threshold)
        
        # 3. 检测并处理成交量异常值
        self._detect_volume_anomalies(volume_multiplier)
        
        # 4. 处理缺失值
        self._handle_missing_values()
        
        print(f"清洗后数据: {len(self.cleaned_data)}行")
        return self.cleaned_data
    
    def _check_time_continuity(self):
        """检查时间序列连续性"""
        expected_index = pd.date_range(
            start=self.data.index.min(), 
            end=self.data.index.max(),
            freq=self.data.index.freq
        )
        
        missing_dates = expected_index[~expected_index.isin(self.data.index)]
        
        if len(missing_dates) > 0:
            print(f"发现{len(missing_dates)}个缺失日期，将添加并填充")
            # 创建包含所有日期的DataFrame
            self.cleaned_data = self.cleaned_data.reindex(expected_index)
    
    def _detect_price_anomalies(self, z_threshold):
        """使用Z分数检测价格异常值"""
        for column in ['Open', 'High', 'Low', 'Close']:
            if column not in self.cleaned_data.columns:
                continue
                
            # 计算Z分数
            z_scores = stats.zscore(self.cleaned_data[column].dropna())
            outliers = np.abs(z_scores) > z_threshold
            
            # 记录异常值数量
            outlier_count = outliers.sum()
            if outlier_count > 0:
                print(f"在{column}中发现{outlier_count}个价格异常值")
                
                # 使用移动平均替换异常值
                window = min(5, len(self.cleaned_data) // 10)  # 自适应窗口大小
                self.cleaned_data[column] = self.cleaned_data[column].mask(
                    outliers, 
                    self.cleaned_data[column].rolling(window=window, min_periods=1).mean()
                )
    
    def _detect_volume_anomalies(self, volume_multiplier):
        """检测成交量异常值"""
        if 'Volume' not in self.cleaned_data.columns:
            return
            
        # 使用移动平均和标准差检测异常成交量
        window = min(20, len(self.cleaned_data) // 5)
        volume_mean = self.cleaned_data['Volume'].rolling(window=window).mean()
        volume_std = self.cleaned_data['Volume'].rolling(window=window).std()
        
        # 定义异常阈值：超过平均成交量N倍标准差
        upper_bound = volume_mean + volume_multiplier * volume_std
        lower_bound = volume_mean - volume_multiplier * volume_std
        
        # 标记异常值
        volume_outliers = (self.cleaned_data['Volume'] > upper_bound) | (self.cleaned_data['Volume'] < lower_bound)
        outlier_count = volume_outliers.sum()
        
        if outlier_count > 0:
            print(f"发现{outlier_count}个成交量异常值")
            # 使用中位数替换异常成交量
            self.cleaned_data['Volume'] = self.cleaned_data['Volume'].mask(
                volume_outliers, 
                self.cleaned_data['Volume'].median()
            )
    
    def _handle_missing_values(self):
        """处理缺失值"""
        # 前向填充然后后向填充，处理大多数缺失情况
        self.cleaned_data = self.cleaned_data.ffill().bfill()
        
        # 检查是否还有缺失值
        missing_values = self.cleaned_data.isnull().sum().sum()
        if missing_values > 0:
            print(f"仍有{missing_values}个缺失值，使用均值填充")
            self.cleaned_data = self.cleaned_data.fillna(self.cleaned_data.mean())

# 使用数据清洗器处理数据 (执行耗时: ~3秒)
cleaner = FinancialDataCleaner("AAPL")
cleaner.fetch_data(period="2y")
cleaned_data = cleaner.detect_and_handle_anomalies(z_threshold=3.0, volume_multiplier=4)

# 比较清洗前后的数据
print("\n清洗前后收盘价对比:")
comparison = pd.DataFrame({
    '原始收盘价': cleaner.data['Close'],
    '清洗后收盘价': cleaned_data['Close']
}).dropna()

print(comparison.tail(10))

四、效能进化：yfinance性能优化与大规模数据处理

4.1 构建分布式金融数据获取系统

当需要获取大规模金融数据时，单线程请求效率低下。通过多线程和任务队列可以显著提高数据获取效率。

认知锚点：

并发请求：使用线程池同时获取多个股票数据
任务队列：合理分配请求任务，避免过载
结果合并：高效合并分散获取的数据

场景化任务：实现一个多线程金融数据获取器，支持大规模股票数据并行下载。

import yfinance as yf
import pandas as pd
from concurrent.futures import ThreadPoolExecutor, as_completed
import time
from tqdm import tqdm

class ParallelDataDownloader:
    """并行金融数据下载器"""
    
    def __init__(self, max_workers=8, request_timeout=10):
        """
        初始化并行下载器
        
        参数:
            max_workers: 最大工作线程数
            request_timeout: 请求超时时间(秒)
        """
        self.max_workers = max_workers
        self.request_timeout = request_timeout
        
    def download_batch(self, symbols, period="1d", interval="1m", batch_size=20):
        """
        批量下载多个股票数据
        
        参数:
            symbols: 股票代码列表
            period: 数据周期
            interval: 数据间隔
            batch_size: 每批处理的股票数量
            
        返回:
            包含所有股票数据的字典，键为股票代码，值为DataFrame
        """
        results = {}
        start_time = time.time()
        
        # 将股票列表分成批次
        batches = [symbols[i:i+batch_size] for i in range(0, len(symbols), batch_size)]
        
        print(f"开始下载 {len(symbols)} 个股票数据，分为 {len(batches)} 批")
        
        for batch_num, batch in enumerate(batches, 1):
            batch_start = time.time()
            
            # 使用线程池并行下载批次中的股票
            with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
                # 创建未来任务字典
                futures = {
                    executor.submit(
                        self._download_single_symbol, 
                        symbol, period, interval
                    ): symbol for symbol in batch
                }
                
                # 处理完成的任务
                for future in tqdm(as_completed(futures), total=len(futures), 
                                  desc=f"批次 {batch_num}/{len(batches)}"):
                    symbol = futures[future]
                    try:
                        data = future.result()
                        if data is not None and not data.empty:
                            results[symbol] = data
                    except Exception as e:
                        print(f"下载 {symbol} 时出错: {str(e)}")
            
            batch_time = time.time() - batch_start
            print(f"批次 {batch_num} 完成，耗时 {batch_time:.2f} 秒")
            
            # 批次之间添加延迟，避免请求过于集中
            if batch_num < len(batches):
                time.sleep(2)  # 等待2秒再开始下一批
        
        total_time = time.time() - start_time
        print(f"全部下载完成，耗时 {total_time:.2f} 秒")
        print(f"成功获取 {len(results)}/{len(symbols)} 个股票数据")
        
        return results
    
    def _download_single_symbol(self, symbol, period, interval):
        """下载单个股票数据"""
        try:
            ticker = yf.Ticker(symbol)
            data = ticker.history(period=period, interval=interval)
            return data
        except Exception as e:
            print(f"{symbol}: {str(e)}")
            return None
    
    def combine_results(self, results, output_format='multi_index'):
        """
        合并下载结果
        
        参数:
            results: download_batch返回的结果字典
            output_format: 输出格式，'multi_index'或'panel'
            
        返回:
            合并后的DataFrame
        """
        if not results:
            return None
            
        if output_format == 'multi_index':
            # 创建多级索引DataFrame
            return pd.concat(results, names=['Symbol', 'Date'])
        elif output_format == 'panel':
            # 创建宽格式DataFrame，列为股票，行为日期
            panels = {}
            for col in results[next(iter(results))].columns:
                panels[col] = pd.DataFrame({
                    symbol: df[col] for symbol, df in results.items()
                })
            return panels
        else:
            raise ValueError("不支持的输出格式")

# 使用并行下载器获取股票数据 (执行耗时: ~60秒，取决于股票数量)
if __name__ == "__main__":
    # 纳斯达克100成分股示例列表
    nasdaq_100_symbols = [
        "AAPL", "MSFT", "GOOGL", "AMZN", "META", "TSLA", "NVDA", "PEP", "COST", "ADBE",
        "AVGO", "CMCSA", "CSCO", "INTC", "AMD", "TXN", "QCOM", "AMAT", "BKNG", "GILD"
    ]
    
    downloader = ParallelDataDownloader(max_workers=4)
    results = downloader.download_batch(
        nasdaq_100_symbols,
        period="1mo",
        interval="1d",
        batch_size=10
    )
    
    # 合并结果
    combined_data = downloader.combine_results(results)
    if combined_data is not None:
        print(f"合并后数据形状: {combined_data.shape}")
        print(combined_data.head())
        
        # 保存到CSV文件
        combined_data.to_csv("nasdaq_100_data.csv")
        print("数据已保存到nasdaq_100_data.csv")

4.2 yfinance项目开发与贡献指南

参与开源项目是提升技能和影响力的重要途径。yfinance作为活跃的开源项目，欢迎开发者贡献代码和改进。

认知锚点：

分支策略：理解项目的分支管理模型
代码规范：遵循项目的编码标准和最佳实践
贡献流程：掌握从提交Issue到PR合并的完整流程

yfinance项目采用结构化的版本控制策略，通过主分支(main)、开发分支(dev)和功能分支(feature)的分离，确保代码质量和项目稳定性。

如图所示yfinance的开发流程包括：

从dev分支创建功能分支(feature)
在功能分支上开发新功能
完成后合并回dev分支进行测试
测试稳定后合并到main分支发布新版本
紧急修复通过urgent bugfixes直接合并到main和dev分支

场景化任务：为yfinance项目贡献一个新功能，实现自定义数据缓存过期策略。

# 贡献示例：自定义缓存管理器
import os
import json
import time
from datetime import datetime, timedelta

class CustomCacheManager:
    """自定义缓存管理器，支持设置缓存过期时间"""
    
    def __init__(self, cache_dir="./yfinance_cache", default_ttl=86400):
        """
        初始化缓存管理器
        
        参数:
            cache_dir: 缓存目录路径
            default_ttl: 默认缓存过期时间(秒)，默认为1天
        """
        self.cache_dir = cache_dir
        self.default_ttl = default_ttl
        
        # 创建缓存目录
        os.makedirs(self.cache_dir, exist_ok=True)
        
    def _get_cache_path(self, key):
        """获取缓存文件路径"""
        # 对键进行哈希以创建文件名
        import hashlib
        key_hash = hashlib.md5(key.encode()).hexdigest()
        return os.path.join(self.cache_dir, f"{key_hash}.json")
    
    def set(self, key, data, ttl=None):
        """
        存储数据到缓存
        
        参数:
            key: 缓存键
            data: 要缓存的数据
            ttl: 缓存过期时间(秒)，为None则使用默认值
        """
        ttl = ttl or self.default_ttl
        expiry_time = time.time() + ttl
        
        # 准备缓存数据
        cache_data = {
            "data": data,
            "expiry": expiry_time,
            "timestamp": time.time()
        }
        
        # 保存到文件
        cache_path = self._get_cache_path(key)
        with open(cache_path, 'w') as f:
            json.dump(cache_data, f)
    
    def get(self, key):
        """
        从缓存获取数据
        
        参数:
            key: 缓存键
            
        返回:
            如果缓存有效则返回数据，否则返回None
        """
        cache_path = self._get_cache_path(key)
        
        if not os.path.exists(cache_path):
            return None
            
        try:
            with open(cache_path, 'r') as f:
                cache_data = json.load(f)
                
            # 检查是否过期
            if time.time() > cache_data["expiry"]:
                # 删除过期缓存
                os.remove(cache_path)
                return None
                
            return cache_data["data"]
            
        except Exception as e:
            print(f"缓存读取错误: {str(e)}")
            return None
    
    def clear_expired(self):
        """清理所有过期缓存"""
        count = 0
        for filename in os.listdir(self.cache_dir):
            if filename.endswith(".json"):
                cache_path = os.path.join(self.cache_dir, filename)
                try:
                    with open(cache_path, 'r') as f:
                        cache_data = json.load(f)
                        
                    if time.time() > cache_data["expiry"]:
                        os.remove(cache_path)
                        count += 1
                except Exception:
                    # 删除损坏的缓存文件
                    os.remove(cache_path)
                    count += 1
                    
        print(f"清理了{count}个过期或损坏的缓存文件")
    
    def clear_all(self):
        """清除所有缓存"""
        count = 0
        for filename in os.listdir(self.cache_dir):
            if filename.endswith(".json"):
                os.remove(os.path.join(self.cache_dir, filename))
                count += 1
        print(f"清除了{count}个缓存文件")

# 使用自定义缓存管理器 (执行耗时: ~1秒)
cache_manager = CustomCacheManager(default_ttl=3600)  # 1小时缓存

# 存储数据
cache_manager.set("AAPL_1d", {"price": 150.25, "volume": 50000000})

# 获取数据
data = cache_manager.get("AAPL_1d")
print("从缓存获取数据:", data)

# 清理过期缓存
cache_manager.clear_expired()