data_fetcher.py

"""Utilities to download and normalize market data from Yahoo Finance."""

from __future__ import annotations

import logging
from typing import Any, Dict, List, Optional, TypedDict

import yfinance as yf

LOGGER = logging.getLogger(__name__)

# Discount rate for DCF intrinsic value (WACC proxy for broad equity)
DCF_DISCOUNT_RATE = 0.10
# Number of projection years for DCF
DCF_PROJECTION_YEARS = 10
# Terminal growth rate (perpetuity growth after projection)
DCF_TERMINAL_GROWTH = 0.025


class RawFinancialData(TypedDict, total=False):
    """Structure returned by :func:`fetch_raw` with normalized metrics."""

    price: Optional[float]
    shares: Optional[float]
    assets: Optional[float]
    liabilities: Optional[float]
    enterpriseValue: Optional[float]
    ebitda: Optional[float]
    netIncome: Optional[float]
    equity: Optional[float]
    pe: Optional[float]
    pb: Optional[float]
    ps: Optional[float]
    dividend: Optional[float]
    free_cashflow: Optional[float]
    earnings_growth: Optional[float]
    revenue_growth: Optional[float]
    peg: Optional[float]
    ev_ebitda: Optional[float]
    de_ratio: Optional[float]
    roe: Optional[float]
    intrinsic_value: Optional[float]
    current_ratio: Optional[float]
    sector_ps_median: Optional[float]
    sector_ev_ebitda_median: Optional[float]
    piotroski_f_score: Optional[int]


def _safe_get(frame: Any, key: str) -> Optional[float]:
    """Return ``frame[key]`` if available, otherwise ``None``.

    ``yfinance`` returns :class:`pandas.DataFrame` objects for financial
    statements. Accessing missing rows raises ``KeyError`` so this helper keeps
    :func:`fetch_raw` tidy without swallowing unrelated exceptions.
    """

    try:
        return float(frame.loc[key].iloc[0])
    except Exception:  # pragma: no cover - defensive guard
        return None


def _safe_get_col(frame: Any, key: str, col: int) -> Optional[float]:
    """Return ``frame.loc[key].iloc[col]`` safely for multi-period statements."""

    try:
        return float(frame.loc[key].iloc[col])
    except Exception:
        return None


def _compute_piotroski(
    financials: Any, balance: Any, cashflow: Any
) -> Optional[int]:
    """Compute the Piotroski F-Score (0-9) from financial statements.

    Requires at least two periods of data for year-over-year comparison.
    Returns ``None`` when data is insufficient.
    """

    try:
        # Current period (col 0) vs prior period (col 1)
        net_income_curr = _safe_get_col(financials, "Net Income", 0)
        net_income_prev = _safe_get_col(financials, "Net Income", 1)
        ocf_curr = _safe_get_col(cashflow, "Total Cash From Operating Activities", 0)
        ocf_prev = _safe_get_col(cashflow, "Total Cash From Operating Activities", 1)
        assets_curr = _safe_get_col(balance, "Total Assets", 0)
        assets_prev = _safe_get_col(balance, "Total Assets", 1)
        liab_curr = _safe_get_col(balance, "Total Liab", 0)
        liab_prev = _safe_get_col(balance, "Total Liab", 1)
        equity_curr = _safe_get_col(balance, "Total Stockholder Equity", 0)
        equity_prev = _safe_get_col(balance, "Total Stockholder Equity", 1)
        shares_curr = _safe_get_col(balance, "Common Stock", 0)
        shares_prev = _safe_get_col(balance, "Common Stock", 1)
        revenue_curr = _safe_get_col(financials, "Total Revenue", 0)
        revenue_prev = _safe_get_col(financials, "Total Revenue", 1)
        gross_profit_curr = _safe_get_col(financials, "Gross Profit", 0)
        gross_profit_prev = _safe_get_col(financials, "Gross Profit", 1)

        if any(v is None for v in [net_income_curr, ocf_curr, assets_curr, assets_prev]):
            return None

        score = 0

        # 1. Profitability: positive net income
        if net_income_curr > 0:
            score += 1

        # 2. Profitability: positive operating cash flow
        if ocf_curr > 0:
            score += 1

        # 3. Profitability: ROA improving
        if net_income_prev is not None and assets_prev and assets_prev > 0:
            roa_curr = net_income_curr / assets_curr
            roa_prev = net_income_prev / assets_prev
            if roa_curr > roa_prev:
                score += 1

        # 4. Quality: OCF > Net Income (accruals check)
        if ocf_curr > net_income_curr:
            score += 1

        # 5. Leverage: declining long-term debt ratio
        if liab_curr is not None and liab_prev is not None and assets_prev > 0:
            debt_ratio_curr = liab_curr / assets_curr
            debt_ratio_prev = liab_prev / assets_prev
            if debt_ratio_curr < debt_ratio_prev:
                score += 1

        # 6. Liquidity: current ratio improving (assets/liabilities as proxy)
        if equity_curr is not None and equity_prev is not None:
            if liab_curr and liab_prev and liab_curr > 0 and liab_prev > 0:
                cr_curr = (assets_curr - equity_curr + equity_curr) / liab_curr
                cr_prev = (assets_prev - equity_prev + equity_prev) / liab_prev
                # simplified: assets/liabilities
                cr_curr = assets_curr / liab_curr
                cr_prev = assets_prev / liab_prev
                if cr_curr > cr_prev:
                    score += 1

        # 7. Dilution: no new shares issued
        if shares_curr is not None and shares_prev is not None:
            if shares_curr <= shares_prev:
                score += 1
        else:
            score += 1  # benefit of doubt if data missing

        # 8. Efficiency: gross margin improving
        if (
            gross_profit_curr is not None
            and gross_profit_prev is not None
            and revenue_curr
            and revenue_prev
            and revenue_curr > 0
            and revenue_prev > 0
        ):
            gm_curr = gross_profit_curr / revenue_curr
            gm_prev = gross_profit_prev / revenue_prev
            if gm_curr > gm_prev:
                score += 1

        # 9. Efficiency: asset turnover improving
        if revenue_curr is not None and revenue_prev is not None and assets_prev > 0:
            at_curr = revenue_curr / assets_curr
            at_prev = revenue_prev / assets_prev
            if at_curr > at_prev:
                score += 1

        return score

    except Exception:
        return None


def _compute_dcf_intrinsic(
    free_cashflow: float,
    shares: float,
    growth_rate: Optional[float],
) -> Optional[float]:
    """Compute per-share intrinsic value using a two-stage DCF model.

    Stage 1: Project FCF for ``DCF_PROJECTION_YEARS`` using ``growth_rate``
    (capped between -5% and +20% to avoid extreme projections).
    Stage 2: Terminal value using Gordon Growth Model.
    """

    if shares <= 0 or free_cashflow <= 0:
        return None

    # Cap growth rate to realistic bounds
    g = growth_rate if growth_rate is not None else 0.05
    g = max(-0.05, min(g, 0.20))

    r = DCF_DISCOUNT_RATE
    tg = DCF_TERMINAL_GROWTH

    pv_fcfs = 0.0
    projected_fcf = free_cashflow
    for year in range(1, DCF_PROJECTION_YEARS + 1):
        projected_fcf *= (1 + g)
        pv_fcfs += projected_fcf / ((1 + r) ** year)

    # Terminal value (Gordon Growth Model)
    terminal_fcf = projected_fcf * (1 + tg)
    terminal_value = terminal_fcf / (r - tg)
    pv_terminal = terminal_value / ((1 + r) ** DCF_PROJECTION_YEARS)

    enterprise_value = pv_fcfs + pv_terminal
    return enterprise_value / shares


def _fetch_sector_peers(
    stock: Any, info: Dict[str, Any]
) -> tuple[Optional[float], Optional[float]]:
    """Fetch sector median P/S and EV/EBITDA from Yahoo Finance sector data.

    Uses the stock's sector and industry to find comparable companies.
    Returns (median_ps, median_ev_ebitda).
    """

    try:
        sector = info.get("sector")
        industry = info.get("industry")
        if not industry:
            return None, None

        # yfinance Sector/Industry screener
        screener = yf.Sector(sector) if sector else None
        if screener is None:
            return None, None

        # Get top companies overview for the sector
        overview = screener.overview
        if overview is None or overview.empty:
            return None, None

        # Try to get industry-level data for tighter comparison
        try:
            ind = yf.Industry(industry)
            top_companies = ind.top_companies
            if top_companies is not None and not top_companies.empty:
                median_ps = None
                median_ev = None
                if "Price to Sales" in top_companies.columns:
                    ps_vals = top_companies["Price to Sales"].dropna()
                    if len(ps_vals) >= 3:
                        median_ps = float(ps_vals.median())
                if "Enterprise Value to EBITDA" in top_companies.columns:
                    ev_vals = top_companies["Enterprise Value to EBITDA"].dropna()
                    if len(ev_vals) >= 3:
                        median_ev = float(ev_vals.median())
                if median_ps is not None or median_ev is not None:
                    return median_ps, median_ev
        except Exception:
            pass

        return None, None

    except Exception:
        LOGGER.debug("Could not fetch sector peers for %s", info.get("symbol", "?"))
        return None, None


def fetch_raw(ticker: str) -> Optional[RawFinancialData]:
    """Return raw financial metrics for ``ticker``.

    The function consolidates primary metrics and a handful of derived
    indicators consumed by :mod:`evaluator`. Failures are logged and ``None`` is
    returned so callers can decide how to handle missing data.
    """

    try:
        stock = yf.Ticker(ticker)
        info: Dict[str, Any] = stock.info or {}
        balance = stock.balance_sheet
        cashflow = stock.cashflow
        financials = stock.financials

        price = info.get("regularMarketPrice")
        shares = info.get("sharesOutstanding")
        enterprise_value = info.get("enterpriseValue")
        ebitda = info.get("ebitda")
        trailing_pe = info.get("trailingPE")
        price_to_book = info.get("priceToBook")
        price_to_sales = info.get("priceToSalesTrailing12Months")
        dividend = info.get("dividendRate", 0)
        # earningsGrowth is a decimal (e.g. 0.15 = 15% growth)
        earnings_growth = info.get("earningsGrowth")
        revenue_growth = info.get("revenueGrowth")
        current_ratio = info.get("currentRatio")

        net_income = _safe_get(financials, "Net Income")
        assets = _safe_get(balance, "Total Assets")
        liabilities = _safe_get(balance, "Total Liab")
        equity = _safe_get(balance, "Total Stockholder Equity")

        operating_cf = _safe_get(cashflow, "Total Cash From Operating Activities")
        capital_expenditures = _safe_get(cashflow, "Capital Expenditures")
        free_cashflow: Optional[float]
        if operating_cf is not None and capital_expenditures is not None:
            # Yahoo Finance reports CapEx as negative (cash outflow convention).
            # FCF = Operating CF - |CapEx|, i.e. OCF + negative_capex.
            free_cashflow = (
                operating_cf + capital_expenditures
                if capital_expenditures < 0
                else operating_cf - capital_expenditures
            )
        else:
            free_cashflow = None

        # Compute Piotroski F-Score from financial statement data
        piotroski = _compute_piotroski(financials, balance, cashflow)

        # Fetch sector peer medians for relative valuation
        sector_ps_median, sector_ev_ebitda_median = _fetch_sector_peers(stock, info)

        raw: RawFinancialData = {
            "price": price,
            "shares": shares,
            "assets": assets,
            "liabilities": liabilities,
            "enterpriseValue": enterprise_value,
            "ebitda": ebitda,
            "netIncome": net_income,
            "equity": equity,
            "pe": trailing_pe,
            "pb": price_to_book,
            "ps": price_to_sales,
            "dividend": dividend,
            "free_cashflow": free_cashflow,
            "earnings_growth": earnings_growth,
            "revenue_growth": revenue_growth,
            "current_ratio": current_ratio,
            "sector_ps_median": sector_ps_median,
            "sector_ev_ebitda_median": sector_ev_ebitda_median,
            "piotroski_f_score": piotroski,
        }

        # PEG: P/E divided by earnings growth rate (in %).
        # earningsGrowth from Yahoo is a decimal (0.15 = 15%), so PEG = PE / (growth * 100).
        if trailing_pe and earnings_growth and earnings_growth > 0:
            raw["peg"] = trailing_pe / (earnings_growth * 100)
        else:
            raw["peg"] = None

        if enterprise_value and ebitda and ebitda != 0:
            raw["ev_ebitda"] = enterprise_value / ebitda
        else:
            raw["ev_ebitda"] = None

        if liabilities is not None and equity and equity != 0:
            raw["de_ratio"] = liabilities / equity
        else:
            raw["de_ratio"] = None

        if net_income is not None and equity and equity != 0:
            raw["roe"] = net_income / equity
        else:
            raw["roe"] = None

        # DCF-based intrinsic value replaces naive 10x FCF multiple
        if free_cashflow is not None and shares and shares > 0:
            raw["intrinsic_value"] = _compute_dcf_intrinsic(
                free_cashflow, shares, earnings_growth
            )
        else:
            raw["intrinsic_value"] = None

        return raw

    except Exception:  # pragma: no cover - network failures
        LOGGER.exception("Failed to fetch data for %s", ticker)
        return None