""" 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")