Financial Datasets测评报告：解锁金融AI潜能的全方位指南

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from financial_datasets import FinancialDataLoader, CompanyFinancials
import warnings
warnings.filterwarnings('ignore')

class FinancialIndicatorAnalysis:
    """上市公司财务指标分析系统"""
    
    def __init__(self):
        self.loader = FinancialDataLoader(api_key="your_api_key_here")
        self.companies = []
        
    def load_company_data(self, symbols, start_date, end_date):
        """加载公司财务数据"""
        print("正在加载财务数据...")
        self.companies = []
        
        for symbol in symbols:
            try:
                # 获取公司基本信息
                company_info = self.loader.get_company_profile(symbol)
                
                # 获取财务报表数据
                balance_sheet = self.loader.get_balance_sheet(symbol, start_date, end_date)
                income_statement = self.loader.get_income_statement(symbol, start_date, end_date)
                cash_flow = self.loader.get_cash_flow(symbol, start_date, end_date)
                
                company_data = {
                    'symbol': symbol,
                    'name': company_info['name'],
                    'sector': company_info['sector'],
                    'balance_sheet': balance_sheet,
                    'income_statement': income_statement,
                    'cash_flow': cash_flow
                }
                
                self.companies.append(company_data)
                print(f"成功加载 {symbol} 数据")
                
            except Exception as e:
                print(f"加载 {symbol} 数据时出错: {str(e)}")
    
    def calculate_financial_ratios(self):
        """计算财务比率"""
        print("\n正在计算财务比率...")
        
        ratios_data = []
        
        for company in self.companies:
            try:
                symbol = company['symbol']
                name = company['name']
                sector = company['sector']
                balance = company['balance_sheet']
                income = company['income_statement']
                cash_flow = company['cash_flow']
                
                # 获取最新可用数据
                latest_balance = balance.iloc[-1] if not balance.empty else None
                latest_income = income.iloc[-1] if not income.empty else None
                latest_cash_flow = cash_flow.iloc[-1] if not cash_flow.empty else None
                
                if latest_balance is None or latest_income is None:
                    continue
                
                # 计算盈利能力比率
                revenue = latest_income.get('revenue', 0)
                net_income = latest_income.get('net_income', 0)
                total_assets = latest_balance.get('total_assets', 1)
                equity = latest_balance.get('total_equity', 1)
                
                roa = net_income / total_assets * 100 if total_assets != 0 else 0
                roe = net_income / equity * 100 if equity != 0 else 0
                profit_margin = net_income / revenue * 100 if revenue != 0 else 0
                
                # 计算偿债能力比率
                current_assets = latest_balance.get('current_assets', 0)
                current_liabilities = latest_balance.get('current_liabilities', 1)
                total_debt = latest_balance.get('total_debt', 0)
                
                current_ratio = current_assets / current_liabilities if current_liabilities != 0 else 0
                debt_to_equity = total_debt / equity * 100 if equity != 0 else 0
                
                # 计算运营效率比率
                cogs = latest_income.get('cost_of_goods_sold', 0)
                inventory = latest_balance.get('inventory', 1)
                
                inventory_turnover = cogs / inventory if inventory != 0 else 0
                
                ratios = {
                    'symbol': symbol,
                    'name': name,
                    'sector': sector,
                    'roa': roa,
                    'roe': roe,
                    'profit_margin': profit_margin,
                    'current_ratio': current_ratio,
                    'debt_to_equity': debt_to_equity,
                    'inventory_turnover': inventory_turnover
                }
                
                ratios_data.append(ratios)
                
            except Exception as e:
                print(f"计算 {company['symbol']} 财务比率时出错: {str(e)}")
        
        return pd.DataFrame(ratios_data)
    
    def identify_top_companies(self, ratios_df, metric, top_n=5):
        """根据指定指标识别顶级公司"""
        if ratios_df.empty:
            return pd.DataFrame()
            
        top_companies = ratios_df.nlargest(top_n, metric)
        return top_companies
    
    def visualize_analysis(self, ratios_df):
        """可视化分析结果"""
        if ratios_df.empty:
            print("没有可用的数据进行可视化")
            return
            
        fig, axes = plt.subplots(2, 2, figsize=(15, 12))
        fig.suptitle('上市公司财务指标分析', fontsize=16)
        
        # ROE分布
        axes[0, 0].barh(ratios_df['symbol'], ratios_df['roe'])
        axes[0, 0].set_title('净资产收益率(ROE) (%)')
        axes[0, 0].set_xlabel('ROE (%)')
        
        # 利润率分布
        axes[0, 1].barh(ratios_df['symbol'], ratios_df['profit_margin'])
        axes[0, 1].set_title('净利润率 (%)')
        axes[0, 1].set_xlabel('净利润率 (%)')
        
        # 流动比率分布
        axes[1, 0].barh(ratios_df['symbol'], ratios_df['current_ratio'])
        axes[1, 0].set_title('流动比率')
        axes[1, 0].set_xlabel('流动比率')
        
        # 负债权益比分布
        axes[1, 1].barh(ratios_df['symbol'], ratios_df['debt_to_equity'])
        axes[1, 1].set_title('负债权益比 (%)')
        axes[1, 1].set_xlabel('负债权益比 (%)')
        
        plt.tight_layout()
        plt.savefig('financial_analysis.png', dpi=300, bbox_inches='tight')
        plt.show()
    
    def generate_report(self, ratios_df):
        """生成分析报告"""
        if ratios_df.empty:
            print("没有可用的数据生成报告")
            return
            
        print("\n" + "="*60)
        print("           上市公司财务分析报告")
        print("="*60)
        
        # 找出各指标表现最好的公司
        metrics = ['roe', 'profit_margin', 'current_ratio', 'debt_to_equity']
        metric_names = ['净资产收益率', '净利润率', '流动比率', '负债权益比']
        
        for i, metric in enumerate(metrics):
            if metric in ratios_df.columns:
                best_company = ratios_df.loc[ratios_df[metric].idxmax()]
                print(f"{metric_names[i]}最佳: {best_company['name']} ({best_company['symbol']}) - {best_company[metric]:.2f}")
        
        print("\n行业分布:")
        sector_counts = ratios_df['sector'].value_counts()
        for sector, count in sector_counts.items():
            print(f"  {sector}: {count}家公司")
        
        print("\n关键指标统计:")
        print(f"  平均ROE: {ratios_df['roe'].mean():.2f}%")
        print(f"  平均净利润率: {ratios_df['profit_margin'].mean():.2f}%")
        print(f"  平均流动比率: {ratios_df['current_ratio'].mean():.2f}")
        print(f"  平均负债权益比: {ratios_df['debt_to_equity'].mean():.2f}%")

# 使用示例
if __name__ == "__main__":
    # 创建分析实例
    analyzer = FinancialIndicatorAnalysis()
    
    # 定义要分析的公司列表
    company_symbols = ['AAPL', 'MSFT', 'GOOGL', 'AMZN', 'TSLA', 'JPM', 'JNJ', 'WMT']
    
    # 加载数据（时间范围：最近5年）
    analyzer.load_company_data(company_symbols, '2018-01-01', '2023-12-31')
    
    # 计算财务比率
    ratios_df = analyzer.calculate_financial_ratios()
    
    if not ratios_df.empty:
        # 显示ROE最高的公司
        top_roe_companies = analyzer.identify_top_companies(ratios_df, 'roe', 3)
        print("\nROE最高的公司:")
        print(top_roe_companies[['symbol', 'name', 'roe']])
        
        # 生成可视化
        analyzer.visualize_analysis(ratios_df)
        
        # 生成分析报告
        analyzer.generate_report(ratios_df)
    else:
        print("未能计算出有效的财务比率数据")

# 示例优化代码 - 添加缓存机制
from functools import lru_cache
import json

class OptimizedFinancialAnalysis(FinancialIndicatorAnalysis):
    """优化版的财务分析系统"""
    
    @lru_cache(maxsize=100)
    def cached_company_data(self, symbol, start_date, end_date):
        """带缓存的数据加载方法"""
        cache_key = f"{symbol}_{start_date}_{end_date}"
        cache_file = f"cache/{cache_key}.json"
        
        # 检查缓存是否存在
        try:
            with open(cache_file, 'r') as f:
                return json.load(f)
        except FileNotFoundError:
            # 缓存不存在，从API加载
            data = self.loader.get_company_financials(symbol, start_date, end_date)
            # 保存到缓存
            with open(cache_file, 'w') as f:
                json.dump(data, f)
            return data