fundamentals_overlay.py

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