数据获取
数据源选择
财报数据有几个主要来源:
| 数据源 | 类型 | 优点 | 缺点 | 价格 | 推荐度 |
|---|---|---|---|---|---|
| yfinance | 开源库 | 免费、简单 | 数据可能延迟 | 免费 | ⭐⭐⭐⭐⭐ |
| Alpha Vantage | API | 数据质量高 | 免费版限制多 | 免费/付费 | ⭐⭐⭐⭐ |
| Finnhub | API | 实时数据、免费额度大 | 需要注册 | 免费/付费 | ⭐⭐⭐⭐⭐ |
| IEX Cloud | API | 稳定 | 免费版很少 | $9/月起 | ⭐⭐⭐ |
| SEC EDGAR | 官方 | 最权威 | 格式复杂 | 免费 | ⭐⭐⭐ |
推荐方案:
- 学习阶段:yfinance(最简单)
- 生产环境:Finnhub(免费额度够用)
安装依赖
bash
# yfinance
pip install yfinance
# Finnhub
pip install finnhub-python
# 数据处理
pip install pandas numpyyfinance 快速开始
基础用法
python
import yfinance as yf
import pandas as pd
# 获取股票数据
ticker = yf.Ticker("AAPL")
# 获取历史价格
hist = ticker.history(period="1y")
print(hist.head())
# 获取公司信息
info = ticker.info
print(f"公司名称: {info['longName']}")
print(f"行业: {info['industry']}")
print(f"市值: {info['marketCap']}")获取财务报表
python
# 获取利润表
financials = ticker.financials
print("年度利润表:")
print(financials)
# 获取季度利润表
quarterly_financials = ticker.quarterly_financials
print("\n季度利润表:")
print(quarterly_financials)
# 获取资产负债表
balance_sheet = ticker.balance_sheet
print("\n资产负债表:")
print(balance_sheet)
# 获取现金流量表
cash_flow = ticker.cashflow
print("\n现金流量表:")
print(cash_flow)提取关键指标
python
def extract_key_metrics(ticker_symbol):
"""提取关键财务指标"""
ticker = yf.Ticker(ticker_symbol)
info = ticker.info
metrics = {
# 基本信息
"公司名称": info.get("longName"),
"股票代码": ticker_symbol,
"行业": info.get("industry"),
"市值": info.get("marketCap"),
# 估值指标
"市盈率": info.get("trailingPE"),
"市净率": info.get("priceToBook"),
"市销率": info.get("priceToSalesTrailing12Months"),
# 盈利能力
"毛利率": info.get("profitMargins"),
"ROE": info.get("returnOnEquity"),
"ROA": info.get("returnOnAssets"),
# 财务健康
"债务股权比": info.get("debtToEquity"),
"流动比率": info.get("currentRatio"),
"速动比率": info.get("quickRatio"),
# 增长指标
"营收增长率": info.get("revenueGrowth"),
"净利润增长率": info.get("earningsGrowth"),
}
return metrics
# 使用
metrics = extract_key_metrics("AAPL")
for key, value in metrics.items():
print(f"{key}: {value}")Finnhub API
注册和获取 API Key
- 访问 finnhub.io
- 注册账号
- 在 Dashboard 获取 API Key
基础用法
python
import finnhub
# 初始化
finnhub_client = finnhub.Client(api_key="your-api-key")
# 获取公司基本信息
profile = finnhub_client.company_profile2(symbol='AAPL')
print(f"公司: {profile['name']}")
print(f"行业: {profile['gind']}")
# 获取实时报价
quote = finnhub_client.quote('AAPL')
print(f"当前价格: {quote['c']}")
print(f"涨跌: {quote['d']}")
print(f"涨跌幅: {quote['dp']}%")
# 获取财务指标
metrics = finnhub_client.company_basic_financials('AAPL', 'all')
print(f"PE 比率: {metrics['metric']['10DayAverageTradingVolume']}")获取财报数据
python
# 获取财报列表
financials = finnhub_client.financials(
symbol='AAPL',
statement='bs', # bs:资产负债表, ic:利润表, cf:现金流量表
freq='annual' # annual:年度, quarterly:季度
)
# 解析财报数据
for statement in financials['financials']:
year = statement['year']
revenue = statement.get('revenue', {})
print(f"{year}年营收: {revenue}")获取 SEC 财报
安装依赖
bash
pip install sec-edgar-downloader下载财报
python
from secedgar import filings, FilingType
# 下载最新的 10-K(年度报告)
my_filings = filings(
cik_lookup="AAPL",
filing_type=FilingType.FILING_10K,
count=1,
user_agent="Your email" # SEC 要求
)
# 下载到本地
my_filings.save("/path/to/download")数据处理
清洗数据
python
def clean_financial_data(df):
"""清洗财务数据"""
# 处理缺失值
df = df.fillna(0)
# 转换为数值类型
numeric_columns = df.select_dtypes(include=['object']).columns
df[numeric_columns] = df[numeric_columns].apply(pd.to_numeric, errors='ignore')
# 排序(按日期)
if 'Date' in df.columns:
df = df.sort_values('Date')
return df计算财务比率
python
def calculate_financial_ratios(financial_data):
"""计算财务比率"""
ratios = {}
# 流动比率 = 流动资产 / 流动负债
ratios['流动比率'] = (
financial_data['流动资产'] / financial_data['流动负债']
)
# 速动比率 = (流动资产 - 存货) / 流动负债
ratios['速动比率'] = (
(financial_data['流动资产'] - financial_data['存货']) /
financial_data['流动负债']
)
# 资产负债率 = 总负债 / 总资产
ratios['资产负债率'] = (
financial_data['总负债'] / financial_data['总资产']
)
# 毛利率 = (营收 - 营业成本) / 营收
ratios['毛利率'] = (
(financial_data['营收'] - financial_data['营业成本']) /
financial_data['营收']
)
return ratios同比增长计算
python
def calculate_yoy_growth(current_year, previous_year):
"""计算同比增长率"""
if previous_year == 0:
return None
growth = ((current_year - previous_year) / previous_year) * 100
return round(growth, 2)
# 示例
revenue_2024 = 100000
revenue_2023 = 80000
revenue_growth = calculate_yoy_growth(revenue_2024, revenue_2023)
print(f"营收同比增长: {revenue_growth}%")缓存策略
避免频繁调用 API:
python
import pickle
import os
from datetime import datetime, timedelta
def get_cached_data(symbol, cache_days=7):
"""获取缓存的数据"""
cache_file = f"cache/{symbol}.pkl"
# 检查缓存是否存在且未过期
if os.path.exists(cache_file):
file_time = datetime.fromtimestamp(os.path.getmtime(cache_file))
if datetime.now() - file_time < timedelta(days=cache_days):
with open(cache_file, 'rb') as f:
return pickle.load(f)
# 缓存不存在或过期,重新获取
ticker = yf.Ticker(symbol)
data = ticker.history(period="1y")
# 保存到缓存
os.makedirs("cache", exist_ok=True)
with open(cache_file, 'wb') as f:
pickle.dump(data, f)
return data错误处理
python
def safe_get_stock_data(symbol):
"""安全获取股票数据,处理异常"""
try:
ticker = yf.Ticker(symbol)
hist = ticker.history(period="1y")
if hist.empty:
print(f"警告: {symbol} 没有数据")
return None
return hist
except Exception as e:
print(f"获取 {symbol} 数据时出错: {str(e)}")
return None数据验证
确保数据的准确性:
python
def validate_financial_data(metrics):
"""验证财务数据的合理性"""
warnings = []
# 检查异常值
if metrics.get('市盈率', 0) < 0:
warnings.append("警告: 市盈率为负,公司可能亏损")
if metrics.get('债务股权比', 0) > 200:
warnings.append("警告: 债务水平过高")
if metrics.get('流动比率', 0) < 1:
warnings.append("警告: 流动比率低于1,可能存在流动性风险")
return warnings下一步
数据获取完成,开始进行数据分析。