Spaces:
Running
Running
| """ | |
| Financial analysis engine for fundamental metrics. | |
| Calculates growth, margins, returns, and cash flow metrics. | |
| """ | |
| import pandas as pd | |
| import numpy as np | |
| from typing import Dict, Optional, Tuple, List | |
| from datetime import datetime | |
| class FinancialAnalyzer: | |
| """Analyzes financial statements and calculates key metrics""" | |
| def __init__(self, financial_data: Dict): | |
| """ | |
| Initialize analyzer with fetched financial data | |
| Args: | |
| financial_data: Complete dataset from FinancialDataFetcher | |
| """ | |
| self.data = financial_data | |
| self.ticker = financial_data.get('ticker') | |
| self.metrics = financial_data.get('metrics', {}) | |
| self.statements = financial_data.get('financial_statements', {}) | |
| self.company_info = financial_data.get('company_info', {}) | |
| def analyze_growth(self) -> Dict: | |
| """ | |
| Analyze revenue and earnings growth trends | |
| Returns: | |
| Dictionary with growth metrics | |
| """ | |
| results = { | |
| 'revenue_growth_ttm': self.metrics.get('revenue_growth'), | |
| 'earnings_growth_ttm': self.metrics.get('earnings_growth'), | |
| } | |
| # Calculate historical growth from income statement | |
| try: | |
| income_stmt = self.statements.get('income_statement') | |
| if income_stmt is not None and not income_stmt.empty: | |
| # Revenue growth (most recent vs 1 year ago) | |
| if 'Total Revenue' in income_stmt.index: | |
| revenues = income_stmt.loc['Total Revenue'].values | |
| if len(revenues) >= 2: | |
| results['revenue_growth_yoy'] = ((revenues[0] - revenues[1]) / abs(revenues[1])) if revenues[1] != 0 else None | |
| # 3-year CAGR if available | |
| if len(revenues) >= 3: | |
| years = min(len(revenues) - 1, 3) | |
| cagr = (revenues[0] / revenues[years]) ** (1/years) - 1 if revenues[years] != 0 else None | |
| results['revenue_cagr_3y'] = cagr | |
| # Net income growth | |
| if 'Net Income' in income_stmt.index: | |
| net_incomes = income_stmt.loc['Net Income'].values | |
| if len(net_incomes) >= 2: | |
| # Handle negative values | |
| if net_incomes[1] != 0: | |
| results['net_income_growth_yoy'] = ((net_incomes[0] - net_incomes[1]) / abs(net_incomes[1])) | |
| except Exception as e: | |
| print(f"Error calculating historical growth: {e}") | |
| # Growth assessment | |
| revenue_growth = results.get('revenue_growth_yoy') or results.get('revenue_growth_ttm') | |
| earnings_growth = results.get('net_income_growth_yoy') or results.get('earnings_growth_ttm') | |
| results['growth_quality'] = self._assess_growth_quality(revenue_growth, earnings_growth) | |
| return results | |
| def analyze_margins(self) -> Dict: | |
| """ | |
| Analyze profitability margins and trends | |
| Returns: | |
| Dictionary with margin metrics | |
| """ | |
| results = { | |
| 'gross_margin': self.metrics.get('gross_margin'), | |
| 'operating_margin': self.metrics.get('operating_margin'), | |
| 'profit_margin': self.metrics.get('profit_margin'), | |
| 'ebitda_margin': self.metrics.get('ebitda_margin'), | |
| } | |
| # Calculate margin trends from income statement | |
| try: | |
| income_stmt = self.statements.get('income_statement') | |
| if income_stmt is not None and not income_stmt.empty: | |
| # Calculate margins for multiple periods | |
| if 'Total Revenue' in income_stmt.index: | |
| revenues = income_stmt.loc['Total Revenue'].values | |
| # Gross margin trend | |
| if 'Gross Profit' in income_stmt.index and len(revenues) >= 2: | |
| gross_profits = income_stmt.loc['Gross Profit'].values | |
| margins = [gp / rev if rev != 0 else None for gp, rev in zip(gross_profits, revenues)] | |
| margins = [m for m in margins if m is not None] | |
| if len(margins) >= 2: | |
| results['gross_margin_current'] = margins[0] | |
| results['gross_margin_trend'] = margins[0] - margins[1] | |
| results['gross_margin_stable'] = abs(margins[0] - margins[1]) < 0.02 # Within 2% | |
| # Operating margin trend | |
| if 'Operating Income' in income_stmt.index and len(revenues) >= 2: | |
| op_incomes = income_stmt.loc['Operating Income'].values | |
| margins = [oi / rev if rev != 0 else None for oi, rev in zip(op_incomes, revenues)] | |
| margins = [m for m in margins if m is not None] | |
| if len(margins) >= 2: | |
| results['operating_margin_current'] = margins[0] | |
| results['operating_margin_trend'] = margins[0] - margins[1] | |
| results['operating_leverage'] = margins[0] > margins[1] # Expanding margins | |
| # Net margin trend | |
| if 'Net Income' in income_stmt.index and len(revenues) >= 2: | |
| net_incomes = income_stmt.loc['Net Income'].values | |
| margins = [ni / rev if rev != 0 else None for ni, rev in zip(net_incomes, revenues)] | |
| margins = [m for m in margins if m is not None] | |
| if len(margins) >= 2: | |
| results['net_margin_current'] = margins[0] | |
| results['net_margin_trend'] = margins[0] - margins[1] | |
| except Exception as e: | |
| print(f"Error calculating margin trends: {e}") | |
| # Margin assessment | |
| results['margin_quality'] = self._assess_margin_quality(results) | |
| return results | |
| def analyze_returns(self) -> Dict: | |
| """ | |
| Analyze return on capital metrics | |
| Returns: | |
| Dictionary with return metrics | |
| """ | |
| results = { | |
| 'roe': self.metrics.get('return_on_equity'), | |
| 'roa': self.metrics.get('return_on_assets'), | |
| } | |
| # Calculate ROIC (Return on Invested Capital) | |
| try: | |
| income_stmt = self.statements.get('income_statement') | |
| balance_sheet = self.statements.get('balance_sheet') | |
| if income_stmt is not None and balance_sheet is not None: | |
| if not income_stmt.empty and not balance_sheet.empty: | |
| # NOPAT = Net Operating Profit After Tax | |
| if 'Operating Income' in income_stmt.index and 'Tax Provision' in income_stmt.index: | |
| op_income = income_stmt.loc['Operating Income'].iloc[0] | |
| total_revenue = income_stmt.loc['Total Revenue'].iloc[0] if 'Total Revenue' in income_stmt.index else 1 | |
| tax_provision = income_stmt.loc['Tax Provision'].iloc[0] | |
| # Estimate tax rate | |
| pretax_income = income_stmt.loc['Pretax Income'].iloc[0] if 'Pretax Income' in income_stmt.index else op_income | |
| tax_rate = abs(tax_provision / pretax_income) if pretax_income != 0 else 0.21 | |
| nopat = op_income * (1 - tax_rate) | |
| # Invested Capital = Total Debt + Total Equity - Cash | |
| total_debt = self.metrics.get('total_debt', 0) | |
| total_assets = balance_sheet.loc['Total Assets'].iloc[0] if 'Total Assets' in balance_sheet.index else 0 | |
| total_liabilities = balance_sheet.loc['Total Liabilities Net Minority Interest'].iloc[0] if 'Total Liabilities Net Minority Interest' in balance_sheet.index else 0 | |
| equity = total_assets - total_liabilities | |
| cash = self.metrics.get('total_cash', 0) | |
| invested_capital = total_debt + equity - cash | |
| if invested_capital > 0: | |
| results['roic'] = nopat / invested_capital | |
| except Exception as e: | |
| print(f"Error calculating ROIC: {e}") | |
| # Returns assessment | |
| results['returns_quality'] = self._assess_returns_quality(results) | |
| return results | |
| def analyze_cash_flow(self) -> Dict: | |
| """ | |
| Analyze cash flow metrics | |
| Returns: | |
| Dictionary with cash flow metrics | |
| """ | |
| results = { | |
| 'operating_cash_flow': self.metrics.get('operating_cash_flow', 0), | |
| 'free_cash_flow': self.metrics.get('free_cash_flow', 0), | |
| } | |
| # Calculate FCF margin and conversion | |
| try: | |
| income_stmt = self.statements.get('income_statement') | |
| cashflow_stmt = self.statements.get('cash_flow') | |
| if income_stmt is not None and not income_stmt.empty: | |
| revenue = income_stmt.loc['Total Revenue'].iloc[0] if 'Total Revenue' in income_stmt.index else 0 | |
| net_income = income_stmt.loc['Net Income'].iloc[0] if 'Net Income' in income_stmt.index else 0 | |
| if revenue > 0: | |
| results['fcf_margin'] = results['free_cash_flow'] / revenue | |
| results['ocf_margin'] = results['operating_cash_flow'] / revenue | |
| if net_income != 0: | |
| results['fcf_conversion'] = results['free_cash_flow'] / net_income | |
| # Cash flow trend | |
| if cashflow_stmt is not None and not cashflow_stmt.empty: | |
| if 'Free Cash Flow' in cashflow_stmt.index: | |
| fcf_values = cashflow_stmt.loc['Free Cash Flow'].values | |
| if len(fcf_values) >= 2: | |
| results['fcf_growth'] = ((fcf_values[0] - fcf_values[1]) / abs(fcf_values[1])) if fcf_values[1] != 0 else None | |
| results['fcf_positive_trend'] = fcf_values[0] > fcf_values[1] | |
| except Exception as e: | |
| print(f"Error calculating cash flow metrics: {e}") | |
| # Cash flow assessment | |
| results['cash_flow_quality'] = self._assess_cash_flow_quality(results) | |
| return results | |
| def analyze_financial_health(self) -> Dict: | |
| """ | |
| Analyze financial health and leverage | |
| Returns: | |
| Dictionary with financial health metrics | |
| """ | |
| results = { | |
| 'total_cash': self.metrics.get('total_cash', 0), | |
| 'total_debt': self.metrics.get('total_debt', 0), | |
| 'debt_to_equity': self.metrics.get('debt_to_equity'), | |
| 'current_ratio': self.metrics.get('current_ratio'), | |
| 'quick_ratio': self.metrics.get('quick_ratio'), | |
| } | |
| # Calculate net debt | |
| results['net_debt'] = results['total_debt'] - results['total_cash'] | |
| # Calculate interest coverage | |
| try: | |
| income_stmt = self.statements.get('income_statement') | |
| if income_stmt is not None and not income_stmt.empty: | |
| if 'Operating Income' in income_stmt.index and 'Interest Expense' in income_stmt.index: | |
| op_income = income_stmt.loc['Operating Income'].iloc[0] | |
| interest = abs(income_stmt.loc['Interest Expense'].iloc[0]) | |
| if interest > 0: | |
| results['interest_coverage'] = op_income / interest | |
| except Exception as e: | |
| print(f"Error calculating interest coverage: {e}") | |
| # Health assessment | |
| results['financial_health_quality'] = self._assess_financial_health(results) | |
| return results | |
| def _assess_growth_quality(self, revenue_growth: Optional[float], earnings_growth: Optional[float]) -> str: | |
| """Assess growth quality""" | |
| if revenue_growth is None or earnings_growth is None: | |
| return "Insufficient Data" | |
| if revenue_growth > 0.15 and earnings_growth > 0.15: | |
| return "Strong" | |
| elif revenue_growth > 0.08 and earnings_growth > 0.08: | |
| return "Good" | |
| elif revenue_growth > 0 and earnings_growth > 0: | |
| return "Moderate" | |
| else: | |
| return "Weak" | |
| def _assess_margin_quality(self, margins: Dict) -> str: | |
| """Assess margin quality""" | |
| operating_margin = margins.get('operating_margin_current') or margins.get('operating_margin') | |
| margin_trend = margins.get('operating_margin_trend') | |
| if operating_margin is None: | |
| return "Insufficient Data" | |
| if operating_margin > 0.20 and (margin_trend is None or margin_trend >= 0): | |
| return "Excellent" | |
| elif operating_margin > 0.10 and (margin_trend is None or margin_trend >= 0): | |
| return "Good" | |
| elif operating_margin > 0.05: | |
| return "Moderate" | |
| else: | |
| return "Weak" | |
| def _assess_returns_quality(self, returns: Dict) -> str: | |
| """Assess returns quality""" | |
| roe = returns.get('roe') | |
| roic = returns.get('roic') | |
| primary_return = roic if roic is not None else roe | |
| if primary_return is None: | |
| return "Insufficient Data" | |
| if primary_return > 0.20: | |
| return "Excellent" | |
| elif primary_return > 0.15: | |
| return "Good" | |
| elif primary_return > 0.10: | |
| return "Moderate" | |
| else: | |
| return "Weak" | |
| def _assess_cash_flow_quality(self, cash_flow: Dict) -> str: | |
| """Assess cash flow quality""" | |
| fcf = cash_flow.get('free_cash_flow', 0) | |
| fcf_conversion = cash_flow.get('fcf_conversion') | |
| if fcf <= 0: | |
| return "Negative" | |
| if fcf_conversion and fcf_conversion > 1.0: | |
| return "Excellent" | |
| elif fcf_conversion and fcf_conversion > 0.8: | |
| return "Good" | |
| elif fcf > 0: | |
| return "Moderate" | |
| else: | |
| return "Weak" | |
| def _assess_financial_health(self, health: Dict) -> str: | |
| """Assess financial health""" | |
| net_debt = health.get('net_debt', 0) | |
| current_ratio = health.get('current_ratio') | |
| interest_coverage = health.get('interest_coverage') | |
| # Net cash position is excellent | |
| if net_debt < 0: | |
| return "Excellent" | |
| # Check liquidity and debt coverage | |
| healthy_liquidity = current_ratio and current_ratio > 1.5 | |
| healthy_coverage = interest_coverage and interest_coverage > 3 | |
| if healthy_liquidity and healthy_coverage: | |
| return "Good" | |
| elif (current_ratio and current_ratio > 1.0) or (interest_coverage and interest_coverage > 1.5): | |
| return "Moderate" | |
| else: | |
| return "Weak" | |
| def generate_summary(self) -> Dict: | |
| """ | |
| Generate comprehensive analysis summary | |
| Returns: | |
| Complete analysis results | |
| """ | |
| return { | |
| 'ticker': self.ticker, | |
| 'company_name': self.company_info.get('company_name', self.ticker), | |
| 'sector': self.company_info.get('sector', 'Unknown'), | |
| 'industry': self.company_info.get('industry', 'Unknown'), | |
| 'analysis_date': datetime.now().isoformat(), | |
| 'growth_analysis': self.analyze_growth(), | |
| 'margin_analysis': self.analyze_margins(), | |
| 'returns_analysis': self.analyze_returns(), | |
| 'cash_flow_analysis': self.analyze_cash_flow(), | |
| 'financial_health': self.analyze_financial_health() | |
| } | |
| if __name__ == "__main__": | |
| # Test with sample data | |
| print("This module is meant to be imported and used with data from data_fetcher.py") | |