path: root/services/data_service.py

"""yfinance wrapper — price history, financial statements, company info."""
import yfinance as yf
import pandas as pd
import streamlit as st


@st.cache_data(ttl=60)
def search_tickers(query: str) -> list[dict]:
    """Search for tickers by company name or symbol. Returns list of {symbol, name, exchange}."""
    if not query or len(query.strip()) < 2:
        return []
    try:
        results = yf.Search(query.strip(), max_results=8).quotes
        out = []
        for r in results:
            symbol = r.get("symbol", "")
            name = r.get("longname") or r.get("shortname") or symbol
            exchange = r.get("exchange") or r.get("exchDisp", "")
            if symbol:
                out.append({"symbol": symbol, "name": name, "exchange": exchange})
        return out
    except Exception:
        return []


@st.cache_data(ttl=300)
def get_company_info(ticker: str) -> dict:
    """Return company info dict from yfinance."""
    t = yf.Ticker(ticker.upper())
    info = t.info or {}
    return info


@st.cache_data(ttl=300)
def get_price_history(ticker: str, period: str = "1y") -> pd.DataFrame:
    """Return OHLCV price history."""
    t = yf.Ticker(ticker.upper())
    df = t.history(period=period)
    df.index = pd.to_datetime(df.index)
    return df


@st.cache_data(ttl=3600)
def get_income_statement(ticker: str, quarterly: bool = False) -> pd.DataFrame:
    t = yf.Ticker(ticker.upper())
    df = t.quarterly_income_stmt if quarterly else t.income_stmt
    return df if df is not None else pd.DataFrame()


@st.cache_data(ttl=3600)
def get_balance_sheet(ticker: str, quarterly: bool = False) -> pd.DataFrame:
    t = yf.Ticker(ticker.upper())
    df = t.quarterly_balance_sheet if quarterly else t.balance_sheet
    return df if df is not None else pd.DataFrame()


@st.cache_data(ttl=3600)
def get_cash_flow(ticker: str, quarterly: bool = False) -> pd.DataFrame:
    t = yf.Ticker(ticker.upper())
    df = t.quarterly_cashflow if quarterly else t.cashflow
    return df if df is not None else pd.DataFrame()


@st.cache_data(ttl=300)
def get_market_indices() -> dict:
    """Return latest price + day change % for major indices."""
    symbols = {
        "S&P 500": "^GSPC",
        "NASDAQ": "^IXIC",
        "DOW": "^DJI",
        "VIX": "^VIX",
    }
    result = {}
    for name, sym in symbols.items():
        try:
            t = yf.Ticker(sym)
            hist = t.history(period="2d")
            if len(hist) >= 2:
                prev_close = hist["Close"].iloc[-2]
                last = hist["Close"].iloc[-1]
                pct_change = (last - prev_close) / prev_close
            elif len(hist) == 1:
                last = hist["Close"].iloc[-1]
                pct_change = 0.0
            else:
                result[name] = {"price": None, "change_pct": None}
                continue
            result[name] = {"price": float(last), "change_pct": float(pct_change)}
        except Exception:
            result[name] = {"price": None, "change_pct": None}
    return result


@st.cache_data(ttl=3600)
def get_analyst_price_targets(ticker: str) -> dict:
    """Return analyst price target summary (keys: current, high, low, mean, median)."""
    try:
        t = yf.Ticker(ticker.upper())
        data = t.analyst_price_targets
        return data if isinstance(data, dict) and data else {}
    except Exception:
        return {}


@st.cache_data(ttl=3600)
def get_recommendations_summary(ticker: str) -> pd.DataFrame:
    """Return analyst recommendation counts by period.
    Columns: period, strongBuy, buy, hold, sell, strongSell.
    Row with period='0m' is the current month.
    """
    try:
        t = yf.Ticker(ticker.upper())
        df = t.recommendations_summary
        return df if df is not None and not df.empty else pd.DataFrame()
    except Exception:
        return pd.DataFrame()


@st.cache_data(ttl=3600)
def get_earnings_history(ticker: str) -> pd.DataFrame:
    """Return historical EPS actual vs estimate.
    Columns: epsActual, epsEstimate, epsDifference, surprisePercent.
    """
    try:
        t = yf.Ticker(ticker.upper())
        df = t.earnings_history
        return df if df is not None and not df.empty else pd.DataFrame()
    except Exception:
        return pd.DataFrame()


@st.cache_data(ttl=3600)
def get_next_earnings_date(ticker: str) -> str | None:
    """Return the next expected earnings date as a string, or None.
    Uses t.calendar (no lxml dependency).
    """
    try:
        t = yf.Ticker(ticker.upper())
        cal = t.calendar
        dates = cal.get("Earnings Date", [])
        if dates:
            return str(dates[0])
        return None
    except Exception:
        return None


@st.cache_data(ttl=3600)
def get_insider_transactions(ticker: str) -> pd.DataFrame:
    """Return insider transactions from yfinance.
    Columns: Shares, URL, Text, Insider, Position, Transaction, Start Date, Ownership, Value
    """
    try:
        t = yf.Ticker(ticker.upper())
        df = t.insider_transactions
        return df if df is not None and not df.empty else pd.DataFrame()
    except Exception:
        return pd.DataFrame()


@st.cache_data(ttl=3600)
def compute_ttm_ratios(ticker: str) -> dict:
    """Compute all key financial ratios from raw yfinance quarterly statements.

    Returns a dict with FMP-compatible key names so existing rendering code
    doesn't need changes.  Income items use TTM (sum of last 4 quarters).
    Balance-sheet items use the most recent quarter.  Market data (price,
    market cap, EV) comes from yfinance's info dict.

    This replaces both FMP's /ratios-ttm and /key-metrics-ttm endpoints,
    saving ~2 API calls per ticker.
    """
    try:
        t = yf.Ticker(ticker.upper())
        info = t.info or {}
        inc_q = t.quarterly_income_stmt
        bal_q = t.quarterly_balance_sheet
        cf_q = t.quarterly_cashflow

        if inc_q is None or inc_q.empty:
            return {}

        ratios: dict = {}

        def ttm(label):
            """Sum last 4 quarters from quarterly income statement."""
            if label in inc_q.index:
                vals = inc_q.loc[label].iloc[:4].dropna()
                if len(vals) == 4:
                    return float(vals.sum())
            return None

        def ttm_cf(label):
            """Sum last 4 quarters from quarterly cash flow."""
            if cf_q is not None and not cf_q.empty and label in cf_q.index:
                vals = cf_q.loc[label].iloc[:4].dropna()
                if len(vals) == 4:
                    return float(vals.sum())
            return None

        def bs(label):
            """Most recent quarterly balance-sheet value."""
            if bal_q is not None and not bal_q.empty and label in bal_q.index:
                val = bal_q.loc[label].iloc[0]
                if pd.notna(val):
                    return float(val)
            return None

        # ── TTM income items ──────────────────────────────────────────────────
        revenue = ttm("Total Revenue")
        gross_profit = ttm("Gross Profit")
        operating_income = ttm("Operating Income")
        net_income = ttm("Net Income")
        ebit = ttm("EBIT")
        ebitda = ttm("EBITDA") or ttm("Normalized EBITDA")
        interest_expense = ttm("Interest Expense")
        tax_provision = ttm("Tax Provision")
        pretax_income = ttm("Pretax Income")

        # ── Balance-sheet items (most recent quarter) ────────────────────────
        equity = bs("Stockholders Equity") or bs("Common Stock Equity")
        total_assets = bs("Total Assets")
        total_debt = bs("Total Debt")
        current_assets = bs("Current Assets")
        current_liabilities = bs("Current Liabilities")
        inventory = bs("Inventory")
        cash = (
            bs("Cash And Cash Equivalents")
            or bs("Cash Cash Equivalents And Short Term Investments")
            or 0.0
        )

        # ── Market data (live) ────────────────────────────────────────────────
        market_cap = info.get("marketCap")
        ev = info.get("enterpriseValue")

        # ── Profitability ─────────────────────────────────────────────────────
        if revenue and revenue > 0:
            if gross_profit is not None:
                ratios["grossProfitMarginTTM"] = gross_profit / revenue
            if operating_income is not None:
                ratios["operatingProfitMarginTTM"] = operating_income / revenue
            if net_income is not None:
                ratios["netProfitMarginTTM"] = net_income / revenue

        if equity and equity > 0 and net_income is not None:
            ratios["returnOnEquityTTM"] = net_income / equity

        if total_assets and total_assets > 0 and net_income is not None:
            ratios["returnOnAssetsTTM"] = net_income / total_assets

        # ROIC = NOPAT / Invested Capital
        if ebit is not None and pretax_income and pretax_income != 0:
            effective_tax_rate = max(0.0, (tax_provision or 0) / pretax_income)
            nopat = ebit * (1 - effective_tax_rate)
            invested_capital = (equity or 0) + (total_debt or 0) - cash
            if invested_capital > 0:
                ratios["returnOnInvestedCapitalTTM"] = nopat / invested_capital

        # ── Valuation multiples ───────────────────────────────────────────────
        if market_cap and market_cap > 0:
            if net_income and net_income > 0:
                ratios["peRatioTTM"] = market_cap / net_income
            if revenue and revenue > 0:
                ratios["priceToSalesRatioTTM"] = market_cap / revenue
            if equity and equity > 0:
                ratios["priceToBookRatioTTM"] = market_cap / equity

        if ev and ev > 0:
            if revenue and revenue > 0:
                ratios["evToSalesTTM"] = ev / revenue
            if ebitda and ebitda > 0:
                ratios["enterpriseValueMultipleTTM"] = ev / ebitda

        # ── Leverage & Liquidity ──────────────────────────────────────────────
        if equity and equity > 0 and total_debt is not None:
            ratios["debtToEquityRatioTTM"] = total_debt / equity

        if current_liabilities and current_liabilities > 0 and current_assets is not None:
            ratios["currentRatioTTM"] = current_assets / current_liabilities
            inv = inventory if inventory is not None else 0.0
            ratios["quickRatioTTM"] = (current_assets - inv) / current_liabilities

        if ebit is not None and interest_expense:
            ie = abs(interest_expense)
            if ie > 0:
                ratios["interestCoverageRatioTTM"] = ebit / ie

        # ── Dividends (from cash flow statement) ─────────────────────────────
        dividends_paid = None
        for label in ("Cash Dividends Paid", "Common Stock Dividend Paid"):
            val = ttm_cf(label)
            if val is not None:
                dividends_paid = abs(val)
                break

        if dividends_paid and dividends_paid > 0:
            if market_cap and market_cap > 0:
                ratios["dividendYieldTTM"] = dividends_paid / market_cap
            if net_income and net_income > 0:
                ratios["dividendPayoutRatioTTM"] = dividends_paid / net_income

        return ratios
    except Exception:
        return {}


@st.cache_data(ttl=3600)
def get_ebitda_from_income_stmt(ticker: str) -> float | None:
    """Return the most recent annual EBITDA from the income statement.

    yfinance's info['ebitda'] can be badly wrong for companies with large
    stock-based compensation (e.g. it may deduct SBC, leaving near-zero EBITDA
    even when the income statement EBITDA line is hundreds of millions).
    The income statement 'EBITDA' row is the standard EBIT + D&A figure.
    """
    try:
        t = yf.Ticker(ticker.upper())
        inc = t.income_stmt
        for label in ("EBITDA", "Normalized EBITDA"):
            if label in inc.index:
                val = inc.loc[label].iloc[0]
                if val is not None and pd.notna(val):
                    return float(val)
        return None
    except Exception:
        return None


@st.cache_data(ttl=900)
def get_options_chain(ticker: str) -> dict:
    """Return options chain data for the nearest available expirations.

    Returns:
        {
            "expirations": [str, ...],        # all available expiry dates
            "chains": [
                {"expiry": str, "calls": DataFrame, "puts": DataFrame},
                ...
            ]
        }
    """
    try:
        t = yf.Ticker(ticker.upper())
        expirations = t.options
        if not expirations:
            return {}
        chains = []
        for exp in expirations[:8]:
            try:
                chain = t.option_chain(exp)
                chains.append({"expiry": exp, "calls": chain.calls, "puts": chain.puts})
            except Exception:
                pass
        return {"expirations": list(expirations), "chains": chains}
    except Exception:
        return {}


@st.cache_data(ttl=3600)
def get_sec_filings(ticker: str) -> list[dict]:
    """Return SEC filings from yfinance.
    Each dict has: date, type, title, edgarUrl, exhibits.
    """
    try:
        t = yf.Ticker(ticker.upper())
        filings = t.sec_filings
        return filings if filings else []
    except Exception:
        return []


@st.cache_data(ttl=86400)
def get_historical_ratios_yfinance(ticker: str) -> list[dict]:
    """Compute annual historical ratios from yfinance financial statements.

    Returns dicts with the same field names used by FMP's /ratios and /key-metrics
    endpoints so callers can use either source interchangeably.
    Covers: margins, ROE, ROA, D/E, P/E, P/B, P/S (price-based ratios are
    approximate — they use average price near each fiscal year-end date).
    """
    try:
        t = yf.Ticker(ticker.upper())
        income = t.income_stmt       # rows=metrics, cols=fiscal-year dates
        balance = t.balance_sheet
        info = t.info or {}

        if income is None or income.empty:
            return []

        # One year of monthly price history per fiscal year going back 10 years
        hist = t.history(period="10y", interval="1mo")
        shares = info.get("sharesOutstanding") or info.get("impliedSharesOutstanding")

        rows: list[dict] = []
        for date in income.columns:
            row: dict = {"date": str(date)[:10]}

            # Pull income-statement items (may be NaN)
            def _inc(label):
                try:
                    v = income.loc[label, date]
                    return float(v) if pd.notna(v) else None
                except KeyError:
                    return None

            total_rev = _inc("Total Revenue")
            gross_profit = _inc("Gross Profit")
            operating_income = _inc("Operating Income")
            net_income = _inc("Net Income")
            ebitda_raw = _inc("EBITDA") or _inc("Normalized EBITDA")

            # Margins
            if total_rev and total_rev > 0:
                if gross_profit is not None:
                    row["grossProfitMargin"] = gross_profit / total_rev
                if operating_income is not None:
                    row["operatingProfitMargin"] = operating_income / total_rev
                if net_income is not None:
                    row["netProfitMargin"] = net_income / total_rev

            # Balance-sheet items
            equity = None
            total_assets = None
            total_debt = None
            if balance is not None and not balance.empty and date in balance.columns:
                def _bal(label):
                    try:
                        v = balance.loc[label, date]
                        return float(v) if pd.notna(v) else None
                    except KeyError:
                        return None

                equity = _bal("Stockholders Equity") or _bal("Common Stock Equity")
                total_assets = _bal("Total Assets")
                total_debt = _bal("Total Debt") or _bal("Long Term Debt And Capital Lease Obligation")
                total_cash = _bal("Cash And Cash Equivalents") or _bal("Cash Cash Equivalents And Short Term Investments") or 0.0

                if equity and equity > 0:
                    if net_income is not None:
                        row["returnOnEquity"] = net_income / equity
                    if total_debt is not None:
                        row["debtEquityRatio"] = total_debt / equity

                if total_assets and total_assets > 0 and net_income is not None:
                    row["returnOnAssets"] = net_income / total_assets

            # Price-based ratios — average closing price in ±45-day window around year-end
            if shares and not hist.empty:
                try:
                    date_ts = pd.Timestamp(date)
                    # Normalize timezones: yfinance history index may be tz-aware
                    hist_idx = hist.index
                    if hist_idx.tz is not None:
                        date_ts = date_ts.tz_localize(hist_idx.tz)
                    mask = (
                        (hist_idx >= date_ts - pd.DateOffset(days=45)) &
                        (hist_idx <= date_ts + pd.DateOffset(days=45))
                    )
                    window = hist.loc[mask, "Close"]
                    if not window.empty:
                        price = float(window.mean())
                        market_cap = price * shares

                        if net_income and net_income > 0:
                            row["peRatio"] = market_cap / net_income
                        if equity and equity > 0:
                            row["priceToBookRatio"] = market_cap / equity
                        if total_rev and total_rev > 0:
                            row["priceToSalesRatio"] = market_cap / total_rev

                        # EV/EBITDA — approximate
                        if ebitda_raw and ebitda_raw > 0 and total_debt is not None:
                            ev = market_cap + (total_debt or 0) - (total_cash or 0)
                            row["enterpriseValueMultiple"] = ev / ebitda_raw
                except Exception:
                    pass

            if len(row) > 1:
                rows.append(row)

        return rows
    except Exception:
        return []


@st.cache_data(ttl=3600)
def get_free_cash_flow_series(ticker: str) -> pd.Series:
    """Return annual Free Cash Flow series (most recent first)."""
    t = yf.Ticker(ticker.upper())
    cf = t.cashflow
    if cf is None or cf.empty:
        return pd.Series(dtype=float)
    if "Free Cash Flow" in cf.index:
        return cf.loc["Free Cash Flow"].dropna()
    # Compute from operating CF - capex
    try:
        op = cf.loc["Operating Cash Flow"]
        capex = cf.loc["Capital Expenditure"]
        return (op + capex).dropna()
    except KeyError:
        return pd.Series(dtype=float)