在很多量化分析和投资研究项目中,yfinance 几乎是一个绕不开的名字。它不需要申请 API Key,安装简单,几行代码就能拿到股票、指数、ETF 的历史行情和基础财务数据,对个人开发者和研究者来说非常友好。因此,不少人第一次用 Python 做金融数据分析,接触的第一个数据源就是 yfinance。
你可能已经了解过,yfinance的数据来源是大名鼎鼎的雅虎财经,不过,yfinance 并不是雅虎的官方产品,它的诞生本身就带着一点应急方案的色彩。
时间回到 2017 年前后,Yahoo Finance 逐步关闭了原有的历史数据 API,大量依赖该接口的程序和研究代码突然失效。为了延续这些工作,社区中出现了一个名为 fix-yahoo-finance 的项目,尝试修复旧接口的调用方式。随着使用者越来越多,这个项目被重新设计并更名为 yfinance,也逐渐发展成今天大家熟悉的样子。
从本质上看,yfinance 是一个对 Yahoo Finance 公共数据进行整理和封装的开源工具。它最大的优势在于开箱即用,数据直接以表格形式返回,和 Pandas 配合得非常自然,适合做快速分析、回测和原型验证。正因为门槛低、上手快,yfinance 被广泛用于教学、个人研究,以及不少早期的量化项目中。
但也正因为它依赖的是非官方的数据接口,yfinance 在实际使用中并不总是稳定,有时数据突然取不到,有时字段含义发生变化,有时同一个品种在不同时间返回的结果并不一致。这些问题并不是代码写错了,而是 yfinance 本身出身背景所决定的。
本文接下来将围绕 yfinance 使用过程中常见的一些疑难杂症展开,结合实际场景分析问题出现的原因,并给出相对稳妥的应对方案,帮助你在继续使用 yfinance 的同时,尽量降低踩坑的概率。
yfinance 用户常见问题详细清单与解决方案
1. 网络连接与限流问题
1.1 HTTP 429 错误 (Too Many Requests)
这是 yfinance 最常见的问题之一。用户在短时间内发送大量请求后,会收到 HTTP 429 错误。这个问题在 2024 年后变得更加严重,Yahoo Finance 加强了反爬虫措施。
这种情况一般出现在:
- 短时间内下载大量股票数据时(如一次下载数百或数千个股票)
- 频繁调用
ticker.info或其他需要单独请求的方法时 - 使用循环批量处理股票时没有添加延迟
- 曾经工作正常的代码突然开始失败(Yahoo 收紧了限制)
- 从 2024 年 11 月开始,用户报告即使下载 950 个股票也会触发限流
# 代码示例
tickers = ['AAPL', 'GOOGL', 'MSFT', ...] # 100+ 股票
data = yf.download(tickers, period='1mo')
# 错误信息
# 429 Client Error: Too Many Requests
# 或者部分数据成功,部分失败
根本原因:
- yfinance 不是官方 API,而是通过爬取 Yahoo Finance 网站获取数据
- Yahoo Finance 有速率限制:大约每分钟 60 次价格请求,每分钟 10 次 info 请求
- Yahoo 将多次请求视为潜在的 DDoS 攻击
- 2024 年后 Yahoo 显著收紧了限制策略
解决方案:
方案 1: 添加请求延迟
import yfinance as yf
import time
tickers = ['AAPL', 'GOOGL', 'MSFT', 'TSLA', 'AMZN']
data = {}
for ticker in tickers:
try:
data[ticker] = yf.download(ticker, period='1mo', progress=False)
time.sleep(2) # 每次请求间隔 2 秒
except Exception as e:
print(f"{ticker} 下载失败: {e}")
time.sleep(5) # 遇到错误时等待更长时间
方案 2: 使用批量下载(推荐)
# 一次性下载多个股票(yfinance 会优化请求)
tickers = ['AAPL', 'GOOGL', 'MSFT', 'TSLA', 'AMZN']
data = yf.download(tickers, period='1mo', group_by='ticker', threads=False)
# 注意: threads=True 可能会加剧限流问题
方案 3: 实现指数退避重试机制
import time
import random
def download_with_retry(ticker, max_retries=5):
for attempt in range(max_retries):
try:
data = yf.download(ticker, period='1mo', progress=False)
if not data.empty:
return data
except Exception as e:
if attempt < max_retries - 1:
# 指数退避: 2^attempt + 随机值
wait_time = (2 ** attempt) + random.uniform(0, 1)
print(f"{ticker} 失败,等待 {wait_time:.2f} 秒后重试...")
time.sleep(wait_time)
else:
print(f"{ticker} 达到最大重试次数")
raise e
return None
方案 4: 分批处理大量股票
def download_in_batches(tickers, batch_size=50, delay=10):
all_data = {}
for i in range(0, len(tickers), batch_size):
batch = tickers[i:i+batch_size]
print(f"处理批次 {i//batch_size + 1}: {len(batch)} 个股票")
try:
batch_data = yf.download(batch, period='1mo',
group_by='ticker',
progress=False)
all_data.update(batch_data)
except Exception as e:
print(f"批次失败: {e}")
if i + batch_size < len(tickers):
print(f"等待 {delay} 秒...")
time.sleep(delay)
return all_data
1.2 连接重置错误 (Connection Reset)
有时会收到 ConnectionResetError 或 Connection aborted 错误,表明远程服务器强制关闭了连接。
一般出现在:
- 使用某些云平台上运行(如 PythonAnywhere、Heroku)
- IP 地址被 Yahoo 标记为可疑
- 使用 VPN 或代理时
- 网络不稳定时
报错信息:
ConnectionResetError(10054, 'An existing connection was forcibly closed
by the remote host', None, 10054, None)
解决方案:
方案 1: 增加超时和重试
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
# 配置重试策略
session = requests.Session()
retry = Retry(
total=5,
backoff_factor=1,
status_forcelist=[429, 500, 502, 503, 504]
)
adapter = HTTPAdapter(max_retries=retry)
session.mount('http://', adapter)
session.mount('https://', adapter)
# yfinance 会使用这个 session
# 但需要修改 yfinance 源码或等待库更新
方案 2: 更换网络环境
如果在云平台失败,尝试:
1. 在本地机器上运行
2. 更换不同的云服务商
3. 使用 Google Colab(通常不会被封)
方案 3: 检查防火墙设置
确保允许 HTTPS 连接到:
- query1.finance.yahoo.com
- query2.finance.yahoo.com
- finance.yahoo.com
2. 数据获取失败问题
2.1 返回空 DataFrame
data = yf.download('INVALID_TICKER', start='2023-01-01')
# 返回:
# Empty DataFrame
# Columns: [Open, High, Low, Close, Adj Close, Volume]
# Index: []
# 或者看到警告:
# [*********************100%***********************] 1 of 1 completed
# 1 Failed download:
# - INVALID_TICKER: possibly delisted; no price data found
yf.download() 或 ticker.history() 返回空的 DataFrame,没有任何数据,也没有明确的错误信息。
何时出现:
- 股票代码错误或拼写错误
- 股票已退市
- 日期范围不正确(如开始日期晚于结束日期)
- Yahoo Finance 暂时没有该股票的数据
- 2024 年 2 月,Yahoo API 大规模变更导致大量股票返回空数据
根本原因:
- yfinance 依赖 Yahoo Finance 的数据,如果 Yahoo 没有数据,yfinance 也无法获取
- Yahoo Finance 经常更改其内部 API 结构
- 某些国际市场或小型股票的数据可能不完整
解决方案:
方案 1: 验证股票代码
import yfinance as yf
def validate_ticker(symbol):
"""验证股票代码是否有效"""
ticker = yf.Ticker(symbol)
# 方法 1: 检查 info 是否有基本信息
try:
info = ticker.info
if 'regularMarketPrice' in info or 'previousClose' in info:
print(f"{symbol} 是有效的股票代码")
return True
except:
pass
# 方法 2: 尝试获取少量历史数据
try:
hist = ticker.history(period='5d')
if not hist.empty:
print(f"{symbol} 有可用的历史数据")
return True
except:
pass
print(f"{symbol} 无效或无数据")
return False
# 使用前先验证
if validate_ticker('AAPL'):
data = yf.download('AAPL', period='1mo')
方案 2: 检查 Yahoo Finance 网站
def check_data_availability(symbol, start_date='2023-01-01'):
"""
先检查数据可用性,再下载
"""
import webbrowser
ticker = yf.Ticker(symbol)
data = ticker.history(period='5d')
if data.empty:
print(f"{symbol} 在 yfinance 中无数据")
print(f"请访问 Yahoo Finance 网站确认:")
url = f"https://finance.yahoo.com/quote/{symbol}/history"
print(url)
# webbrowser.open(url) # 自动打开浏览器
return None
# 如果有数据,继续下载完整数据
full_data = yf.download(symbol, start=start_date, progress=False)
return full_data
方案 3: 使用 try-except 处理多个股票
def safe_download(tickers, **kwargs):
"""
安全地下载多个股票,记录失败的股票
"""
successful = {}
failed = []
if isinstance(tickers, str):
tickers = [tickers]
for ticker in tickers:
try:
data = yf.download(ticker, progress=False, **kwargs)
if data.empty:
print(f"警告: {ticker} 返回空数据")
failed.append(ticker)
else:
successful[ticker] = data
print(f"✓ {ticker} 下载成功")
except Exception as e:
print(f"✗ {ticker} 下载失败: {str(e)}")
failed.append(ticker)
print(f"\n成功: {len(successful)}, 失败: {len(failed)}")
if failed:
print(f"失败的股票: {', '.join(failed)}")
return successful, failed
# 使用示例
successful, failed = safe_download(['AAPL', 'INVALID', 'GOOGL'],
period='1mo')
方案 4: 2024 年 2 月 API 变更后的修复
如果遇到大规模数据获取失败,尝试:
1. 升级到最新版本
2. 如果最新版本仍有问题,可能需要降级pip install yfinance==0.2.54 (或其他稳定版本)
3. 检查版本
import yfinance as yf
print(f"yfinance 版本: {yf.__version__}")2.2 “No data found” 错误
明确的 “No data found” 错误信息,通常伴随具体的股票代码和日期范围。
何时出现:
- 请求的日期范围超出了可用数据范围
- 股票在请求的时间段内尚未上市
- 周末或节假日请求实时数据
- 加密货币或外汇市场在非交易时段
具体症状:
data = yf.download('AAPL', start='1900-01-01', end='1950-12-31')
# 错误: No data found for this date range, symbol may be delisted
解决方案:
方案 1: 使用 period=’max’ 获取所有可用数据
# 不指定具体日期,获取最大可用范围
data = yf.download('AAPL', period='max')
print(f"数据范围: {data.index[0]} 到 {data.index[-1]}")
方案 2: 动态调整日期范围
from datetime import datetime, timedelta
def smart_download(symbol, days_back=365):
"""
智能下载:如果指定范围失败,自动缩短范围
"""
end_date = datetime.now()
for days in [days_back, 180, 90, 30, 7]:
start_date = end_date - timedelta(days=days)
try:
data = yf.download(symbol,
start=start_date.strftime('%Y-%m-%d'),
end=end_date.strftime('%Y-%m-%d'),
progress=False)
if not data.empty:
print(f"成功获取 {len(data)} 天的数据 ({days} 天范围)")
return data
except:
continue
print(f"{symbol} 无法获取任何数据")
return None
3. ticker.info 相关问题
3.1 KeyError – 缺少预期的字段
尝试访问 ticker.info 中的特定字段时出现 KeyError,这是因为 Yahoo Finance 不保证所有股票都有相同的字段,且字段可能随时变化。
何时出现:
- 访问不常见的指标(如某些 ETF 或国际股票缺少 P/E ratio)
- Yahoo Finance 更改了数据结构(经常发生)
- 股票类型特殊(如债券、期货)
- 2023 年 8 月后,很多字段突然消失(如
fullTimeEmployees,city,phone等)
具体症状:
ticker = yf.Ticker('AAPL')
pe_ratio = ticker.info['trailingPE'] # 可能成功
# 但这些可能失败:
employees = ticker.info['fullTimeEmployees'] # KeyError!
ceo = ticker.info['companyOfficers'][0]['name'] # KeyError!
# 常见失败的字段:
# - exchangeTimezoneName
# - regularMarketOpen
# - longName
# - fullTimeEmployees
# - city, phone, fax
根本原因:
- Yahoo Finance 的数据结构不一致
- 不同类型的证券有不同的可用字段
- Yahoo 经常修改其网页结构,导致某些字段无法爬取
- 2023 年 8 月,Yahoo 移除了许多基本信息字段
解决方案:
方案 1: 使用 .get() 方法安全访问
ticker = yf.Ticker('AAPL')
info = ticker.info
# 不要这样:
# pe_ratio = info['trailingPE'] # 可能抛出 KeyError
# 应该这样:
pe_ratio = info.get('trailingPE', None)
market_cap = info.get('marketCap', 'N/A')
employees = info.get('fullTimeEmployees', 0)
# 或提供默认值并记录
def safe_get_info(ticker_obj, key, default=None):
value = ticker_obj.info.get(key, default)
if value == default:
print(f"警告: {ticker_obj.ticker} 缺少字段 '{key}'")
return value
方案 2: 检查可用字段
ticker = yf.Ticker('AAPL')
info = ticker.info
# 打印所有可用的键
print("可用字段:")
print(sorted(info.keys()))
# 检查特定字段是否存在
required_fields = ['marketCap', 'trailingPE', 'forwardPE', 'dividendYield']
available_fields = [f for f in required_fields if f in info]
missing_fields = [f for f in required_fields if f not in info]
print(f"可用: {available_fields}")
print(f"缺失: {missing_fields}")
方案 3: 使用 fast_info 获取基本数据
# fast_info 提供有限但更可靠的字段
ticker = yf.Ticker('AAPL')
try:
fast_info = ticker.fast_info
# fast_info 可用字段:
price = fast_info.get('last_price') # 或 fast_info['last_price']
currency = fast_info.get('currency')
market_cap = fast_info.get('market_cap')
shares = fast_info.get('shares')
print(f"价格: {price}, 市值: {market_cap}")
except Exception as e:
print(f"fast_info 也失败了: {e}")
方案 4: 创建健壮的信息提取函数
def extract_stock_info(symbol):
"""
安全地提取股票信息,处理缺失字段
"""
ticker = yf.Ticker(symbol)
info = ticker.info
# 定义期望的字段和默认值
fields = {
'symbol': symbol,
'shortName': info.get('shortName', symbol),
'longName': info.get('longName', info.get('shortName', symbol)),
'sector': info.get('sector', 'Unknown'),
'industry': info.get('industry', 'Unknown'),
'marketCap': info.get('marketCap', None),
'trailingPE': info.get('trailingPE', None),
'forwardPE': info.get('forwardPE', None),
'dividendYield': info.get('dividendYield', None),
'fiftyTwoWeekHigh': info.get('fiftyTwoWeekHigh', None),
'fiftyTwoWeekLow': info.get('fiftyTwoWeekLow', None),
}
# 记录缺失的重要字段
important_fields = ['marketCap', 'sector']
missing_important = [f for f in important_fields if fields[f] is None]
if missing_important:
print(f"{symbol} 缺少重要字段: {missing_important}")
return fields
# 使用
stock_data = extract_stock_info('AAPL')
print(stock_data)
3.2 info 返回不完整或全为 None
ticker.info 返回一个字典,但大部分字段的值都是 None,或者字典本身几乎为空。
何时出现:
- Yahoo Finance 临时服务问题
- 被限流后返回的不完整数据
- 某些 ETF、共同基金或国际股票数据不完整
- 刚上市的股票信息不全
具体症状:
ticker = yf.Ticker('SOME_ETF')
info = ticker.info
print(info)
# 输出:
# {'regularMarketPrice': None, 'preMarketPrice': None,
# 'logo_url': '', 'trailingPegRatio': None, ...}
解决方案:
方案 1: 等待并重试
import time
def get_info_with_retry(symbol, max_attempts=3, delay=5):
"""
多次尝试获取信息
"""
ticker = yf.Ticker(symbol)
for attempt in range(max_attempts):
info = ticker.info
# 检查是否有足够的非 None 字段
non_none_count = sum(1 for v in info.values() if v is not None)
if non_none_count > 5: # 至少有 5 个字段有值
print(f"成功获取 {symbol} 的信息 ({non_none_count} 个字段)")
return info
print(f"尝试 {attempt + 1}/{max_attempts}: 只有 {non_none_count} 个字段有值")
if attempt < max_attempts - 1:
time.sleep(delay)
print(f"警告: {symbol} 的信息不完整")
return info
方案 2: 使用历史数据作为备选
def get_basic_info_from_history(symbol):
"""
当 info 不可用时,从历史数据获取基本信息
"""
ticker = yf.Ticker(symbol)
# 先尝试 info
info = ticker.info
if info.get('regularMarketPrice'):
return info
# 如果 info 失败,从历史数据获取
print(f"{symbol} info 不可用,尝试从历史数据获取...")
hist = ticker.history(period='5d')
if hist.empty:
return None
# 构建基本信息
basic_info = {
'symbol': symbol,
'regularMarketPrice': hist['Close'].iloc[-1],
'volume': hist['Volume'].iloc[-1],
'previousClose': hist['Close'].iloc[-2] if len(hist) > 1 else None,
}
return basic_info
4. 时区和时间问题
4.1 Timezone-aware 时间戳导致比较困难
yfinance 返回的 DataFrame 的索引是 timezone-aware 的 DatetimeIndex(通常是 UTC 或交易所时区),这使得与 timezone-naive 的日期进行比较或合并变得复杂。
何时出现:
- 尝试使用字符串日期过滤数据时:
df[df.index > '2023-01-01'] - 合并多个 DataFrame 时时区不匹配
- 与 pandas 的其他操作混合使用时
- 将数据保存到数据库时
具体症状:
data = yf.download('AAPL', start='2023-01-01', end='2023-12-31')
print(data.index)
# 输出:
# DatetimeIndex(['2023-01-03 00:00:00-05:00', '2023-01-04 00:00:00-05:00', ...],
# dtype='datetime64[ns, America/New_York]', name='Date')
# 尝试比较时出错:
# df[df.index > '2023-01-01'] # 可能失败
# 或者:
# TypeError: '>' not supported between instances of 'Timestamp' and 'str'
# ValueError: Tz-aware datetime.datetime cannot be converted to datetime64
# unless utc=True
根本原因:
- yfinance 为了准确性,返回带时区信息的时间戳
- 不同市场有不同的时区(纽约、伦敦、东京等)
- pandas 对 timezone-aware 和 timezone-naive 的处理很严格
解决方案:
方案 1: 移除时区信息(最常用)
import yfinance as yf
data = yf.download('AAPL', period='1mo')
# 方法 1: 使用 tz_localize(None)
data.index = data.index.tz_localize(None)
# 方法 2: 使用 tz_convert(None) - 如果已有时区
# data.index = data.index.tz_convert(None).tz_localize(None)
# 现在可以正常比较了
filtered = data[data.index > '2023-01-15']
方案 2: 转换到特定时区
# 转换到美国东部时区
data.index = data.index.tz_convert('America/New_York')
# 转换到 UTC
data.index = data.index.tz_convert('UTC')
# 转换到本地时区
import datetime
local_tz = datetime.datetime.now().astimezone().tzinfo
data.index = data.index.tz_convert(local_tz)
方案 3: 创建处理函数
def download_and_clean_timezone(symbol, **kwargs):
"""
下载数据并自动移除时区信息
"""
data = yf.download(symbol, progress=False, **kwargs)
if not data.empty and data.index.tz is not None:
data.index = data.index.tz_localize(None)
return data
# 使用
data = download_and_clean_timezone('AAPL', period='1mo')
方案 4: 使用 pd.to_datetime 进行时区aware的比较
import pandas as pd
data = yf.download('AAPL', period='1mo')
# 创建同样时区的日期进行比较
start_date = pd.to_datetime('2023-01-15').tz_localize(data.index.tz)
filtered = data[data.index > start_date]
# 或者
start_date = pd.Timestamp('2023-01-15', tz=data.index.tz)
filtered = data[data.index > start_date]
4.2 不同时区导致数据不一致
在不同时区的机器上运行相同的代码,得到不同的结果,特别是在处理盘前盘后数据或股息时。
何时出现:
- 在不同地理位置的服务器上运行
- 处理跨越日期边界的数据
- 股息发放日期恰好在请求的开始日期前一天
具体症状:
# 在 UTC 时区运行:
data = yf.download('VLB.TO', start='2021-09-01', end='2021-09-03', actions=True)
# 可能包含 2021-08-31 的股息数据(NaN 价格)
# 在 EST 时区运行:
data = yf.download('VLB.TO', start='2021-09-01', end='2021-09-03', actions=True)
# 不包含 2021-08-31 的数据
解决方案:
方案 1: 标准化到 UTC
import pandas as pd
def download_standardized(symbol, start, end):
"""
下载并标准化到 UTC
"""
data = yf.download(symbol, start=start, end=end, progress=False)
if not data.empty:
# 转换到 UTC
if data.index.tz is not None:
data.index = data.index.tz_convert('UTC')
else:
data.index = data.index.tz_localize('UTC')
return data
方案 2: 过滤掉边界日期的部分数据
import pandas as pd
def download_with_strict_dates(symbol, start_date, end_date):
"""
严格按照日期范围过滤数据
"""
data = yf.download(symbol, start=start_date, end=end_date, progress=False)
if not data.empty:
# 移除时区信息以便比较
data.index = data.index.tz_localize(None)
# 严格过滤日期范围
start = pd.to_datetime(start_date)
end = pd.to_datetime(end_date)
data = data[(data.index >= start) & (data.index < end)]
return data5. 分钟级数据限制问题
5.1 1 分钟数据只能获取 7 天
这是 yfinance 用户最常见的困惑之一。Yahoo Finance 对不同时间间隔的数据有不同的历史长度限制,用户无法获取超过限制的历史分钟数据。
何时出现:
- 尝试获取 1 个月以上的 1 分钟数据
- 需要分析长期的盘中交易模式
- 量化交易回测需要大量历史分钟数据
具体限制:
# 不同时间间隔的最大历史长度:
# 1m (1分钟): 最多 7 天
# 2m (2分钟): 最多 60 天
# 5m (5分钟): 最多 60 天
# 15m (15分钟): 最多 60 天
# 30m (30分钟): 最多 60 天
# 60m/1h (1小时): 最多 730 天
# 1d (日线): 无限制
# 1wk (周线): 无限制
# 1mo (月线): 无限制具体症状:
# 这会失败或只返回 7 天数据:
data = yf.download('AAPL', start='2023-01-01', end='2023-12-31', interval='1m')
# 可能返回: Empty DataFrame 或只有最近 7 天
# 这会成功:
data = yf.download('AAPL', period='7d', interval='1m')根本原因:
- Yahoo Finance 服务器限制,不是 yfinance 的限制
- 分钟级数据量巨大,Yahoo 不存储太久的历史
- 这是 Yahoo Finance 免费服务的限制
解决方案:
方案 1: 分批下载并合并(仅适用于 60 天内)
import pandas as pd
from datetime import datetime, timedelta
def get_extended_intraday(symbol, days=30, interval='5m'):
"""
分批获取更长时间的分钟数据(最多 60 天)
"""
if interval == '1m' and days > 7:
print("警告: 1分钟数据最多只能获取 7 天")
days = 7
elif interval in ['2m', '5m', '15m', '30m'] and days > 60:
print("警告: 该间隔最多只能获取 60 天")
days = 60
all_data = []
end_date = datetime.now()
# 确定批次大小
if interval == '1m':
batch_days = 7
else:
batch_days = 60
# 分批下载
while days > 0:
batch_size = min(days, batch_days)
start_date = end_date - timedelta(days=batch_size)
print(f"下载 {start_date.date()} 到 {end_date.date()}")
data = yf.download(
symbol,
start=start_date.strftime('%Y-%m-%d'),
end=end_date.strftime('%Y-%m-%d'),
interval=interval,
progress=False
)
if not data.empty:
all_data.append(data)
end_date = start_date
days -= batch_size
if all_data:
combined = pd.concat(all_data)
combined = combined.sort_index()
# 移除重复的时间戳
combined = combined[~combined.index.duplicated(keep='first')]
return combined
return pd.DataFrame()
# 使用示例
data = get_extended_intraday('AAPL', days=30, interval='5m')
print(f"获取了 {len(data)} 个数据点")方案 2: 使用更长的时间间隔
# 如果 1 分钟太受限,使用 5 分钟或 1 小时
data_5m = yf.download('AAPL', period='60d', interval='5m') # 60天
data_1h = yf.download('AAPL', period='2y', interval='1h') # 730天
# 或者使用日线数据
data_daily = yf.download('AAPL', period='max', interval='1d') # 所有历史方案 3: 实时下载并保存
import pandas as pd
import time
from datetime import datetime
class MinuteDataCollector:
"""
实时收集并保存分钟数据
"""
def __init__(self, symbol, save_path='data'):
self.symbol = symbol
self.save_path = save_path
os.makedirs(save_path, exist_ok=True)
def collect_and_save(self):
"""每天收集一次数据并保存"""
today = datetime.now().strftime('%Y-%m-%d')
filename = f"{self.save_path}/{self.symbol}_{today}.csv"
# 获取最近 7 天的 1 分钟数据
data = yf.download(self.symbol, period='7d', interval='1m', progress=False)
if not data.empty:
data.to_csv(filename)
print(f"保存了 {len(data)} 条数据到 {filename}")
def load_historical(self, start_date, end_date):
"""从保存的文件中加载历史数据"""
import os
import glob
files = glob.glob(f"{self.save_path}/{self.symbol}_*.csv")
all_data = []
for file in files:
df = pd.read_csv(file, index_col=0, parse_dates=True)
all_data.append(df)
if all_data:
combined = pd.concat(all_data)
combined = combined.sort_index()
combined = combined[~combined.index.duplicated(keep='first')]
# 过滤日期范围
mask = (combined.index >= start_date) & (combined.index <= end_date)
return combined[mask]
return pd.DataFrame()
# 使用示例:
# 每天运行一次来收集数据
collector = MinuteDataCollector('AAPL')
collector.collect_and_save()
# 稍后加载历史数据
historical = collector.load_historical('2023-01-01', '2023-12-31')方案 4: 考虑付费数据源
yfinance 并不适合需要大量历史分钟数据的场景,如果你有这方面的需求,应该考虑付费数据源,推进我们的Infoway API,我们提供分钟级的历史数据 + 实时行情数据。
5.2 分钟数据缺少盘前盘后交易
默认情况下,yfinance 的分钟数据只包含常规交易时段(9:30-16:00 EST),不包含盘前(4:00-9:30)和盘后(16:00-20:00)数据。
何时出现:
- 需要分析盘前盘后交易活动
- 重大新闻发布在盘前/盘后时段
- 需要完整的 24 小时交易数据(加密货币除外)
具体症状:
data = yf.download('AAPL', period='1d', interval='1m')
print(data.index[0], data.index[-1])
# 输出类似:
# 2024-01-12 09:30:00-05:00 2024-01-12 16:00:00-05:00
# 只有常规交易时段解决方案:
方案 1: 使用 prepost=True 参数
# 包含盘前盘后数据
data = yf.download('AAPL', period='1d', interval='1m', prepost=True)
print(f"数据范围: {data.index[0]} 到 {data.index[-1]}")
# 现在应该包含盘前和盘后时段
# 使用 Ticker 对象
ticker = yf.Ticker('AAPL')
data = ticker.history(period='1d', interval='1m', prepost=True)方案 2: 分离常规、盘前、盘后时段
import pandas as pd
def separate_trading_sessions(data):
"""
将数据分为盘前、常规、盘后三个时段
"""
# 假设是美国东部时区
data.index = pd.to_datetime(data.index)
if data.index.tz is not None:
data.index = data.index.tz_convert('America/New_York')
# 提取时间部分
times = data.index.time
# 定义时段
pre_market = data[(times >= pd.Timestamp('04:00').time()) &
(times < pd.Timestamp('09:30').time())]
regular = data[(times >= pd.Timestamp('09:30').time()) &
(times < pd.Timestamp('16:00').time())]
after_hours = data[(times >= pd.Timestamp('16:00').time()) &
(times < pd.Timestamp('20:00').time())]
return {
'pre_market': pre_market,
'regular': regular,
'after_hours': after_hours
}
# 使用
data = yf.download('AAPL', period='1d', interval='1m', prepost=True)
sessions = separate_trading_sessions(data)
print(f"盘前: {len(sessions['pre_market'])} 条")
print(f"常规: {len(sessions['regular'])} 条")
print(f"盘后: {len(sessions['after_hours'])} 条")6. 多股票下载问题
6.1 MultiIndex 列难以处理
当下载多个股票时,yfinance 返回的 DataFrame 使用 MultiIndex 列格式,第一层是数据类型(Open, High, Low, Close 等),第二层是股票代码,这让数据访问变得复杂。
何时出现:
- 使用
yf.download(['AAPL', 'GOOGL', ...])下载多个股票 - 需要提取单个股票的所有数据
- 需要提取所有股票的单个指标(如收盘价)
具体症状:
data = yf.download(['AAPL', 'GOOGL', 'MSFT'], period='1mo')
print(data.columns)
# 输出:
# MultiIndex([( 'Adj Close', 'AAPL'),
# ( 'Adj Close', 'GOOGL'),
# ( 'Adj Close', 'MSFT'),
# ( 'Close', 'AAPL'),
# ( 'Close', 'GOOGL'),
# ...
# 访问变得复杂:
aapl_close = data['Close']['AAPL'] # 或
aapl_close = data[('Close', 'AAPL')]
# 容易混淆和出错根本原因:
- pandas MultiIndex 是处理多维数据的标准方式
- yfinance 为了同时返回多个股票的多个指标,使用了这种结构
- 对新手不友好
解决方案:
方案 1: 使用 group_by=’ticker’ 参数(推荐)
# 这会改变 MultiIndex 的顺序
data = yf.download(['AAPL', 'GOOGL', 'MSFT'], period='1mo', group_by='ticker')
print(data.columns)
# 输出:
# MultiIndex([('AAPL', 'Adj Close'),
# ('AAPL', 'Close'),
# ('AAPL', 'High'),
# ...
# ('GOOGL', 'Adj Close'),
# ...
# 现在访问更直观:
aapl_data = data['AAPL'] # 获取 AAPL 的所有列
aapl_close = data['AAPL']['Close']
# 遍历股票
for ticker in ['AAPL', 'GOOGL', 'MSFT']:
ticker_data = data[ticker]
print(f"{ticker} 最后收盘价: {ticker_data['Close'].iloc[-1]}")方案 2: 只提取需要的数据类型
data = yf.download(['AAPL', 'GOOGL', 'MSFT'], period='1mo')
# 只获取收盘价(返回单层索引 DataFrame)
close_prices = data['Close']
print(close_prices.head())
# AAPL GOOGL MSFT
# Date
# 2024-01-02 185.64 140.93 370.73
# 2024-01-03 184.25 138.65 369.03
# 只获取成交量
volumes = data['Volume']方案 3: 转换为字典格式
def multiindex_to_dict(data, tickers):
"""
将 MultiIndex DataFrame 转换为字典
"""
result = {}
for ticker in tickers:
try:
# 提取单个股票的所有数据
ticker_data = data.xs(ticker, level=1, axis=1)
result[ticker] = ticker_data
except KeyError:
print(f"警告: {ticker} 数据缺失")
return result
# 使用
tickers = ['AAPL', 'GOOGL', 'MSFT']
data = yf.download(tickers, period='1mo')
data_dict = multiindex_to_dict(data, tickers)
# 现在可以像这样访问:
aapl = data_dict['AAPL']
print(aapl['Close'].head())方案 4: 分别下载每个股票(最简单但慢)
import time
tickers = ['AAPL', 'GOOGL', 'MSFT']
data_dict = {}
for ticker in tickers:
data_dict[ticker] = yf.download(ticker, period='1mo', progress=False)
time.sleep(1) # 避免限流
# 访问非常直观:
aapl = data_dict['AAPL']
print(aapl['Close'].head())6.2 批量下载时部分股票失败导致数据不完整
当批量下载多个股票时,如果其中有一个或多个股票代码无效或数据不可用,整个下载不会失败,但会导致返回的 DataFrame 缺少某些股票的列,且没有明确的错误提示。
何时出现:
- 股票列表中包含已退市的股票
- 股票代码拼写错误
- 某些股票临时数据不可用
- 混合不同交易所的股票(如同时包含美股和港股)
具体症状:
tickers = ['AAPL', 'GOOGL', 'INVALID_TICKER', 'MSFT', 'DELISTED']
data = yf.download(tickers, period='1mo')
# 可能看到这样的输出:
# [*********************100%***********************] 5 of 5 completed
# 2 Failed downloads:
# - INVALID_TICKER: possibly delisted; no price data found
# - DELISTED: possibly delisted; no price data found
print(data.columns.levels[1]) # 只有 AAPL, GOOGL, MSFT
# INVALID_TICKER 和 DELISTED 不在结果中根本原因:
- yfinance 不会因为部分失败而抛出异常
- 失败的股票会被静默跳过
- 用户可能没注意到错误信息
解决方案:
方案 1: 预先验证所有股票代码
def validate_tickers(tickers):
"""
验证股票代码列表,返回有效和无效的股票
"""
valid = []
invalid = []
for ticker in tickers:
try:
t = yf.Ticker(ticker)
# 尝试获取少量数据来验证
hist = t.history(period='5d')
if not hist.empty:
valid.append(ticker)
else:
invalid.append(ticker)
except:
invalid.append(ticker)
return valid, invalid
# 使用
tickers = ['AAPL', 'GOOGL', 'INVALID', 'MSFT']
valid, invalid = validate_tickers(tickers)
print(f"有效: {valid}")
print(f"无效: {invalid}")
# 只下载有效的股票
if valid:
data = yf.download(valid, period='1mo')方案 2: 下载后检查缺失的股票
def download_and_check(tickers, **kwargs):
"""
下载并检查是否所有股票都存在
"""
data = yf.download(tickers, **kwargs)
if isinstance(tickers, str):
tickers = [tickers]
# 检查哪些股票在结果中
if 'Close' in data.columns:
if isinstance(data.columns, pd.MultiIndex):
present_tickers = data.columns.levels[1].tolist()
else:
present_tickers = [tickers[0]]
else:
present_tickers = []
missing_tickers = set(tickers) - set(present_tickers)
if missing_tickers:
print(f"警告: 以下股票未下载成功: {missing_tickers}")
return data, list(missing_tickers)
# 使用
tickers = ['AAPL', 'GOOGL', 'INVALID', 'MSFT']
data, missing = download_and_check(tickers, period='1mo')方案 3: 捕获并记录 yfinance 的警告信息
import warnings
import io
import sys
def download_with_warning_capture(tickers, **kwargs):
"""
捕获 yfinance 的警告信息
"""
# 捕获 stdout
old_stdout = sys.stdout
sys.stdout = captured_output = io.StringIO()
try:
data = yf.download(tickers, **kwargs)
# 获取输出
output = captured_output.getvalue()
# 解析失败的股票
failed = []
for line in output.split('\n'):
if 'Failed download' in line or 'delisted' in line:
# 提取股票代码
if ':' in line:
ticker = line.split(':')[0].strip('- ')
failed.append(ticker)
return data, failed
finally:
sys.stdout = old_stdout
# 使用
data, failed_tickers = download_with_warning_capture(
['AAPL', 'INVALID', 'GOOGL'],
period='1mo'
)
print(f"失败的股票: {failed_tickers}")