"""Fundamentals Overlay v1.0 — Valuation, Quality & Growth Metrics Extracts PE, PEG, ROE, debt/equity, FCF yield, growth estimates from yfinance. Maps raw metrics to 0-100 scoring for the multi-factor engine. """ import yfinance as yf import numpy as np from typing import Dict, Optional from datetime import datetime # Sector median PE ratios (US market, approximate) SECTOR_MEDIAN_PE = { 'Technology': 25.0, 'Healthcare': 22.0, 'Financial Services': 15.0, 'Industrials': 18.0, 'Consumer Discretionary': 20.0, 'Consumer Staples': 20.0, 'Energy': 12.0, 'Utilities': 18.0, 'Real Estate': 18.0, 'Basic Materials': 14.0, 'Communication Services': 18.0, } # Default when sector unknown DEFAULT_SECTOR_PE = 18.0 class FundamentalsOverlay: """Pull fundamentals from yfinance info and score for multi-factor engine.""" def __init__(self): self._cache = {} # ticker -> (info_dict, timestamp) self._cache_ttl = 3600 # 1 hour def fetch_info(self, ticker: str) -> Optional[Dict]: """Fetch yfinance info with caching.""" now = datetime.now() if ticker in self._cache: info, ts = self._cache[ticker] if (now - ts).total_seconds() < self._cache_ttl: return info try: info = yf.Ticker(ticker).info if not info or info.get('trailingPE') is None and info.get('forwardPE') is None: return None self._cache[ticker] = (info, now) return info except Exception: return None def extract_metrics(self, ticker: str) -> Dict: """Extract all relevant fundamentals.""" info = self.fetch_info(ticker) if not info: return self._default_metrics(ticker) sector = info.get('sector', 'Unknown') industry = info.get('industry', 'Unknown') sector_pe = SECTOR_MEDIAN_PE.get(sector, DEFAULT_SECTOR_PE) # Valuation pe_trailing = info.get('trailingPE') pe_forward = info.get('forwardPE') peg_ratio = info.get('pegRatio') ps_ratio = info.get('priceToSalesTrailing12Months') pb_ratio = info.get('priceToBook') ev_ebitda = info.get('enterpriseToEbitda') # Quality roe = info.get('returnOnEquity') roa = info.get('returnOnAssets') debt_equity = info.get('debtToEquity') current_ratio = info.get('currentRatio') gross_margin = info.get('grossMargins') operating_margin = info.get('operatingMargins') profit_margin = info.get('profitMargins') # Growth revenue_growth = info.get('revenueGrowth') earnings_growth = info.get('earningsGrowth') earnings_qtr_growth = info.get('earningsQuarterlyGrowth') est_growth = info.get('earningsGrowth') # Forward estimate book_value_growth = None # Not directly available # Cash Flow fcf = info.get('freeCashflow') market_cap = info.get('marketCap') fcf_yield = (fcf / market_cap) if fcf and market_cap else None # Dividend div_yield = info.get('dividendYield') payout_ratio = info.get('payoutRatio') # Price & Performance price = info.get('currentPrice') or info.get('regularMarketPrice') fifty_two_high = info.get('fiftyTwoWeekHigh') fifty_two_low = info.get('fiftyTwoWeekLow') beta = info.get('beta') eps = info.get('trailingEps') # Insider / Institutional held_insiders = info.get('heldPercentInsiders') held_institutions = info.get('heldPercentInstitutions') short_ratio = info.get('shortRatio') short_pct = info.get('shortPercentOfFloat') return { 'ticker': ticker, 'sector': sector, 'industry': industry, 'sector_median_pe': sector_pe, # Valuation 'pe_trailing': pe_trailing, 'pe_forward': pe_forward, 'peg_ratio': peg_ratio, 'ps_ratio': ps_ratio, 'pb_ratio': pb_ratio, 'ev_ebitda': ev_ebitda, # Quality 'roe': roe, 'roa': roa, 'debt_equity': debt_equity, 'current_ratio': current_ratio, 'gross_margin': gross_margin, 'operating_margin': operating_margin, 'profit_margin': profit_margin, # Growth 'revenue_growth': revenue_growth, 'earnings_growth': earnings_growth, 'earnings_qtr_growth': earnings_qtr_growth, 'est_growth': est_growth, 'fcf_yield': fcf_yield, # Dividend 'dividend_yield': div_yield, 'payout_ratio': payout_ratio, # Risk 'beta': beta, 'price': price, 'fifty_two_week_high': fifty_two_high, 'fifty_two_week_low': fifty_two_low, 'eps': eps, # Ownership 'held_insiders': held_insiders, 'held_institutions': held_institutions, 'short_ratio': short_ratio, 'short_pct': short_pct, } def _default_metrics(self, ticker: str) -> Dict: """Default when yfinance info unavailable.""" return { 'ticker': ticker, 'sector': 'Unknown', 'pe_trailing': None, 'pe_forward': None, 'peg_ratio': None, 'ps_ratio': None, 'pb_ratio': None, 'roe': None, 'debt_equity': None, 'revenue_growth': None, 'est_growth': None, 'fcf_yield': None, 'beta': None, } def score_fundamentals(self, metrics: Dict) -> Dict: """Score fundamentals 0-100 for multi-factor engine.""" score = 50.0 sector_pe = metrics.get('sector_median_pe', DEFAULT_SECTOR_PE) # Valuation (30 points) pe = metrics.get('pe_forward') or metrics.get('pe_trailing') if pe: if pe < 10: score += 30 elif pe < sector_pe * 0.6: score += 25 elif pe < sector_pe * 0.8: score += 15 elif pe < sector_pe: score += 8 elif pe < sector_pe * 1.2: score += 0 elif pe < sector_pe * 1.5: score -= 15 else: score -= 25 peg = metrics.get('peg_ratio') if peg and peg > 0: if peg < 0.8: score += 20 elif peg < 1.0: score += 15 elif peg < 1.5: score += 5 elif peg > 2.5: score -= 20 elif peg > 2.0: score -= 10 pb = metrics.get('pb_ratio') if pb: if pb < 1.0: score += 10 elif pb < 2.0: score += 5 elif pb > 10: score -= 15 elif pb > 5: score -= 10 # Quality (30 points) roe = metrics.get('roe') if roe: if roe > 0.25: score += 30 elif roe > 0.20: score += 25 elif roe > 0.15: score += 20 elif roe > 0.10: score += 10 elif roe < 0.05: score -= 15 de = metrics.get('debt_equity') if de is not None: if de < 0.5: score += 15 elif de < 1.0: score += 10 elif de > 3.0: score -= 20 elif de > 2.0: score -= 15 elif de > 1.5: score -= 10 gm = metrics.get('gross_margin') if gm: if gm > 0.50: score += 10 elif gm > 0.30: score += 5 # Growth (25 points) rev_g = metrics.get('revenue_growth') if rev_g: if rev_g > 0.30: score += 25 elif rev_g > 0.20: score += 20 elif rev_g > 0.10: score += 15 elif rev_g > 0.05: score += 8 elif rev_g < 0: score -= 15 earn_g = metrics.get('est_growth') or metrics.get('earnings_growth') if earn_g: if earn_g > 0.25: score += 20 elif earn_g > 0.15: score += 15 elif earn_g > 0.05: score += 5 elif earn_g < -0.05: score -= 15 # Cash Flow (15 points) fcf_y = metrics.get('fcf_yield') if fcf_y: if fcf_y > 0.08: score += 15 elif fcf_y > 0.05: score += 10 elif fcf_y > 0.02: score += 5 elif fcf_y < 0: score -= 15 # Risk adjustment (beta penalty) beta = metrics.get('beta') if beta: if beta > 2.0: score -= 10 elif beta > 1.5: score -= 5 return { 'fundamental_score': max(0, min(100, round(score, 1))), 'metrics': metrics, 'category_scores': { 'valuation_raw': pe if pe else 0, 'peg_raw': peg if peg else 0, 'roe_raw': roe if roe else 0, 'growth_raw': earn_g if earn_g else 0, 'fcf_yield_raw': fcf_y if fcf_y else 0, } } def full_analysis(self, ticker: str) -> Dict: """Complete fundamentals pipeline for a ticker.""" metrics = self.extract_metrics(ticker) scored = self.score_fundamentals(metrics) # Interpretation score = scored['fundamental_score'] if score > 80: interpretation = 'Excellent fundamentals — strong valuation + quality + growth' elif score > 65: interpretation = 'Good fundamentals — attractive on at least two dimensions' elif score > 50: interpretation = 'Average fundamentals — fairly priced, no edge' elif score > 35: interpretation = 'Weak fundamentals — overvalued or declining quality' else: interpretation = 'Poor fundamentals — avoid or short candidate' scored['interpretation'] = interpretation scored['ticker'] = ticker scored['timestamp'] = datetime.now().isoformat() return scored if __name__ == '__main__': fo = FundamentalsOverlay() result = fo.full_analysis('AAPL') print(f"Fundamental Score: {result['fundamental_score']}/100") print(f"Interpretation: {result['interpretation']}") print(f"Sector: {result['metrics'].get('sector', 'N/A')}") print(f"PE: {result['metrics'].get('pe_forward', result['metrics'].get('pe_trailing', 'N/A'))}") print(f"ROE: {result['metrics'].get('roe', 'N/A')}") print(f"Growth: {result['metrics'].get('est_growth', result['metrics'].get('earnings_growth', 'N/A'))}")