财务报表介绍
OpenBB 平台数据扩展提供对季度或年度财务报表的访问。
还有用于比率和其他常见非 GAAP 指标的端点。
大多数数据提供商需要订阅才能访问所有数据,请参考特定提供商的网站了解权限和覆盖范围的详细信息。
财务报表函数归类在 obb.equity.fundamental
模块下。
信息
首先,将 OpenBB 平台导入 Python 会话:
from openbb import obb
财务报表
典型的财务报表包含三个端点:
- 资产负债表:
obb.equity.fundamental.balance()
- 损益表:
obb.equity.fundamental.income()
- 现金流量表:
obb.equity.fundamental.cash()
主要参数包括:
symbol
:公司的股票代码period
:'annual'(年度)或 'quarter'(季度)。默认为 'annual'limit
:限制返回的结果数量,从最新开始。默认为 5。从历史角度看,150 可以追溯到 1985 年。历史记录的数量因提供商而异
字段名称
信息
- 每个数据提供商都有自己解析和组织三大财务报表的方式
- 每个报表中的项目因数据源和报告公司类型而异
- 行项目的名称因数据源而异
- "日期"值可能不同,因为它们来自期间开始/结束或报告日期
此示例突出显示了不同提供商对公司事实的不同标签。
import pandas as pd
# 比较不同提供商的资产负债表数据
df = pd.DataFrame()
# 获取 yfinance 提供商的数据
yfinance_data = obb.equity.fundamental.balance("TGT", provider="yfinance", limit=3)
df["yfinance"] = yfinance_data.to_dataframe().columns.tolist()
# 获取其他提供商的数据进行比较
try:
fmp_data = obb.equity.fundamental.balance("TGT", provider="fmp", limit=3)
df["fmp"] = fmp_data.to_dataframe().columns.tolist()
except:
print("FMP 提供商不可用")
print("不同提供商的字段名称对比:")
print(df.head(10))
基本财务报表获取
资产负债表
# 获取年度资产负债表
balance_sheet = obb.equity.fundamental.balance(
symbol="AAPL",
period="annual",
limit=5,
provider="yfinance"
)
df_balance = balance_sheet.to_dataframe()
print("苹果公司资产负债表:")
print(df_balance.head())
# 查看主要资产项目
if 'total_assets' in df_balance.columns:
print(f"\n总资产趋势:")
print(df_balance[['period_ending', 'total_assets']].head())
损益表
# 获取季度损益表
income_statement = obb.equity.fundamental.income(
symbol="MSFT",
period="quarter",
limit=8, # 获取最近8个季度
provider="yfinance"
)
df_income = income_statement.to_dataframe()
print("微软公司损益表:")
print(df_income.head())
# 分析收入趋势
if 'total_revenue' in df_income.columns:
print(f"\n收入趋势:")
print(df_income[['period_ending', 'total_revenue']].head())
现金流量表
# 获取现金流量表
cash_flow = obb.equity.fundamental.cash(
symbol="GOOGL",
period="annual",
limit=5,
provider="yfinance"
)
df_cash = cash_flow.to_dataframe()
print("谷歌现金流量表:")
print(df_cash.head())
# 查看经营现金流
if 'operating_cash_flow' in df_cash.columns:
print(f"\n经营现金流趋势:")
print(df_cash[['period_ending', 'operating_cash_flow']].head())
财务比率分析
def analyze_financial_ratios(symbol, provider="yfinance"):
"""
分析公司的财务比率
"""
try:
# 获取最新的财务数据
balance = obb.equity.fundamental.balance(
symbol=symbol,
period="annual",
limit=1,
provider=provider
).to_dataframe()
income = obb.equity.fundamental.income(
symbol=symbol,
period="annual",
limit=1,
provider=provider
).to_dataframe()
if balance.empty or income.empty:
print(f"无法获取 {symbol} 的财务数据")
return None
# 计算基本比率
ratios = {}
# 流动比率 = 流动资产 / 流动负债
if 'current_assets' in balance.columns and 'current_liabilities' in balance.columns:
current_assets = balance['current_assets'].iloc[0]
current_liabilities = balance['current_liabilities'].iloc[0]
if current_liabilities != 0:
ratios['流动比率'] = current_assets / current_liabilities
# 资产负债率 = 总负债 / 总资产
if 'total_liabilities' in balance.columns and 'total_assets' in balance.columns:
total_liabilities = balance['total_liabilities'].iloc[0]
total_assets = balance['total_assets'].iloc[0]
if total_assets != 0:
ratios['资产负债率'] = total_liabilities / total_assets
# 净利润率 = 净利润 / 总收入
if 'net_income' in income.columns and 'total_revenue' in income.columns:
net_income = income['net_income'].iloc[0]
total_revenue = income['total_revenue'].iloc[0]
if total_revenue != 0:
ratios['净利润率'] = net_income / total_revenue
return ratios
except Exception as e:
print(f"分析 {symbol} 财务比率时出错: {e}")
return None
# 使用示例
companies = ['AAPL', 'MSFT', 'GOOGL', 'AMZN']
for company in companies:
print(f"\n{company} 财务比率分析:")
ratios = analyze_financial_ratios(company)
if ratios:
for ratio_name, value in ratios.items():
print(f" {ratio_name}: {value:.4f}")
多期间财务分析
def multi_period_analysis(symbol, periods=5):
"""
多期间财务分析
"""
try:
# 获取多年度数据
income_data = obb.equity.fundamental.income(
symbol=symbol,
period="annual",
limit=periods,
provider="yfinance"
).to_dataframe()
if income_data.empty:
print(f"无法获取 {symbol} 的收入数据")
return None
# 按日期排序
income_data = income_data.sort_values('period_ending')
analysis = {
'symbol': symbol,
'periods_analyzed': len(income_data)
}
# 收入增长分析
if 'total_revenue' in income_data.columns:
revenues = income_data['total_revenue'].tolist()
if len(revenues) > 1:
revenue_growth = []
for i in range(1, len(revenues)):
if revenues[i-1] != 0:
growth = (revenues[i] - revenues[i-1]) / revenues[i-1] * 100
revenue_growth.append(growth)
analysis['平均收入增长率'] = sum(revenue_growth) / len(revenue_growth) if revenue_growth else 0
analysis['最新收入'] = revenues[-1]
# 净利润分析
if 'net_income' in income_data.columns:
net_incomes = income_data['net_income'].tolist()
if len(net_incomes) > 1:
profit_growth = []
for i in range(1, len(net_incomes)):
if net_incomes[i-1] != 0:
growth = (net_incomes[i] - net_incomes[i-1]) / net_incomes[i-1] * 100
profit_growth.append(growth)
analysis['平均净利润增长率'] = sum(profit_growth) / len(profit_growth) if profit_growth else 0
analysis['最新净利润'] = net_incomes[-1]
return analysis
except Exception as e:
print(f"多期间分析 {symbol} 时出错: {e}")
return None
# 分析多家公司
tech_companies = ['AAPL', 'MSFT', 'GOOGL', 'META']
analysis_results = []
for company in tech_companies:
result = multi_period_analysis(company, periods=5)
if result:
analysis_results.append(result)
# 显示结果
print("科技公司财务分析结果:")
for result in analysis_results:
print(f"\n{result['symbol']}:")
for key, value in result.items():
if key != 'symbol':
if isinstance(value, float):
print(f" {key}: {value:.2f}")
else:
print(f" {key}: {value}")
行业比较分析
def industry_comparison(symbols, industry_name=""):
"""
行业比较分析
"""
comparison_data = []
for symbol in symbols:
try:
# 获取最新财务数据
balance = obb.equity.fundamental.balance(
symbol=symbol,
limit=1,
provider="yfinance"
).to_dataframe()
income = obb.equity.fundamental.income(
symbol=symbol,
limit=1,
provider="yfinance"
).to_dataframe()
if not balance.empty and not income.empty:
company_data = {'symbol': symbol}
# 提取关键指标
if 'total_assets' in balance.columns:
company_data['总资产'] = balance['total_assets'].iloc[0]
if 'total_revenue' in income.columns:
company_data['总收入'] = income['total_revenue'].iloc[0]
if 'net_income' in income.columns:
company_data['净利润'] = income['net_income'].iloc[0]
# 计算净利润率
if '总收入' in company_data and '净利润' in company_data:
if company_data['总收入'] != 0:
company_data['净利润率'] = company_data['净利润'] / company_data['总收入'] * 100
comparison_data.append(company_data)
except Exception as e:
print(f"获取 {symbol} 数据失败: {e}")
if comparison_data:
import pandas as pd
df = pd.DataFrame(comparison_data)
print(f"{industry_name}行业比较:")
print(df.to_string(index=False))
# 排名分析
if '净利润率' in df.columns:
df_sorted = df.sort_values('净利润率', ascending=False)
print(f"\n按净利润率排名:")
for i, row in df_sorted.iterrows():
print(f"{row['symbol']}: {row['净利润率']:.2f}%")
return comparison_data
# 银行业比较
bank_stocks = ['JPM', 'BAC', 'WFC', 'C']
bank_comparison = industry_comparison(bank_stocks, "银行")
# 科技股比较
tech_stocks = ['AAPL', 'MSFT', 'GOOGL', 'META']
tech_comparison = industry_comparison(tech_stocks, "科技")
财务健康度评估
def financial_health_score(symbol):
"""
财务健康度评分
"""
try:
# 获取财务数据
balance = obb.equity.fundamental.balance(
symbol=symbol,
limit=1,
provider="yfinance"
).to_dataframe()
income = obb.equity.fundamental.income(
symbol=symbol,
limit=1,
provider="yfinance"
).to_dataframe()
cash_flow = obb.equity.fundamental.cash(
symbol=symbol,
limit=1,
provider="yfinance"
).to_dataframe()
score = 0
max_score = 0
details = []
# 盈利能力 (30分)
max_score += 30
if not income.empty and 'net_income' in income.columns:
net_income = income['net_income'].iloc[0]
if net_income > 0:
score += 30
details.append("✓ 盈利能力: 公司盈利 (+30分)")
else:
details.append("✗ 盈利能力: 公司亏损 (+0分)")
# 流动性 (25分)
max_score += 25
if not balance.empty:
if 'current_assets' in balance.columns and 'current_liabilities' in balance.columns:
current_ratio = balance['current_assets'].iloc[0] / balance['current_liabilities'].iloc[0]
if current_ratio > 2:
score += 25
details.append(f"✓ 流动性: 流动比率 {current_ratio:.2f} (+25分)")
elif current_ratio > 1:
score += 15
details.append(f"△ 流动性: 流动比率 {current_ratio:.2f} (+15分)")
else:
details.append(f"✗ 流动性: 流动比率 {current_ratio:.2f} (+0分)")
# 负债水平 (25分)
max_score += 25
if not balance.empty:
if 'total_liabilities' in balance.columns and 'total_assets' in balance.columns:
debt_ratio = balance['total_liabilities'].iloc[0] / balance['total_assets'].iloc[0]
if debt_ratio < 0.3:
score += 25
details.append(f"✓ 负债水平: 资产负债率 {debt_ratio:.2f} (+25分)")
elif debt_ratio < 0.6:
score += 15
details.append(f"△ 负债水平: 资产负债率 {debt_ratio:.2f} (+15分)")
else:
details.append(f"✗ 负债水平: 资产负债率 {debt_ratio:.2f} (+0分)")
# 现金流 (20分)
max_score += 20
if not cash_flow.empty and 'operating_cash_flow' in cash_flow.columns:
operating_cf = cash_flow['operating_cash_flow'].iloc[0]
if operating_cf > 0:
score += 20
details.append("✓ 现金流: 经营现金流为正 (+20分)")
else:
details.append("✗ 现金流: 经营现金流为负 (+0分)")
# 计算最终评分
final_score = (score / max_score) * 100 if max_score > 0 else 0
# 评级
if final_score >= 80:
rating = "优秀"
elif final_score >= 60:
rating = "良好"
elif final_score >= 40:
rating = "一般"
else:
rating = "较差"
return {
'symbol': symbol,
'score': final_score,
'rating': rating,
'details': details
}
except Exception as e:
print(f"评估 {symbol} 财务健康度时出错: {e}")
return None
# 评估多家公司
companies_to_evaluate = ['AAPL', 'TSLA', 'AMZN', 'NFLX']
print("财务健康度评估结果:")
print("=" * 50)
for company in companies_to_evaluate:
result = financial_health_score(company)
if result:
print(f"\n{result['symbol']} - 评分: {result['score']:.1f}/100 ({result['rating']})")
for detail in result['details']:
print(f" {detail}")
最佳实践
- 数据验证:始终检查数据的完整性和准确性
- 多源对比:使用多个数据提供商验证关键数据
- 时间序列分析:分析多个时期的趋势而非单点数据
- 行业比较:将公司数据与同行业公司进行比较
- 比率分析:使用财务比率进行深入分析
# 完整的财务分析工作流
def comprehensive_financial_analysis(symbol):
"""
综合财务分析
"""
print(f"开始分析 {symbol} 的财务状况...")
print("=" * 60)
# 1. 基本财务数据
print("1. 获取基本财务数据...")
try:
balance = obb.equity.fundamental.balance(symbol=symbol, limit=3, provider="yfinance")
income = obb.equity.fundamental.income(symbol=symbol, limit=3, provider="yfinance")
cash_flow = obb.equity.fundamental.cash(symbol=symbol, limit=3, provider="yfinance")
print(" ✓ 财务数据获取成功")
except Exception as e:
print(f" ✗ 财务数据获取失败: {e}")
return
# 2. 财务比率分析
print("2. 财务比率分析...")
ratios = analyze_financial_ratios(symbol)
if ratios:
for ratio_name, value in ratios.items():
print(f" {ratio_name}: {value:.4f}")
# 3. 多期间趋势分析
print("3. 多期间趋势分析...")
trend_analysis = multi_period_analysis(symbol, periods=5)
if trend_analysis:
for key, value in trend_analysis.items():
if key != 'symbol' and isinstance(value, (int, float)):
print(f" {key}: {value:.2f}")
# 4. 财务健康度评估
print("4. 财务健康度评估...")
health_score = financial_health_score(symbol)
if health_score:
print(f" 综合评分: {health_score['score']:.1f}/100 ({health_score['rating']})")
print(f"\n{symbol} 财务分析完成!")
# 使用示例
comprehensive_financial_analysis("AAPL")