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=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)