Add fundamentals overlay — valuation, quality, growth metrics from yfinance info

fundamentals_overlay.py

@@ -0,0 +1,290 @@
+"""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'))}")