"""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_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)