"""yfinance wrapper for Prism v2 Overview data.""" from __future__ import annotations import math import os from typing import Any import httpx import pandas as pd import yfinance as yf from cachetools import TTLCache, cached SEARCH_CACHE = TTLCache(maxsize=128, ttl=60) INFO_CACHE = TTLCache(maxsize=256, ttl=300) FAST_INFO_CACHE = TTLCache(maxsize=256, ttl=300) PROFILE_ENRICH_CACHE = TTLCache(maxsize=256, ttl=300) PRICE_CACHE = TTLCache(maxsize=256, ttl=300) HISTORY_CACHE = TTLCache(maxsize=256, ttl=300) INTRADAY_CACHE = TTLCache(maxsize=128, ttl=60) MARKET_CACHE = TTLCache(maxsize=8, ttl=300) STATEMENT_CACHE = TTLCache(maxsize=256, ttl=3600) INCOME_CACHE = TTLCache(maxsize=256, ttl=3600) BALANCE_CACHE = TTLCache(maxsize=256, ttl=3600) CF_CACHE = TTLCache(maxsize=256, ttl=3600) SHARES_CACHE = TTLCache(maxsize=256, ttl=3600) RATIO_CACHE = TTLCache(maxsize=256, ttl=3600) BETA_CACHE = TTLCache(maxsize=256, ttl=3600) SHORT_CACHE = TTLCache(maxsize=256, ttl=3600) FINANCIALS_CACHE = TTLCache(maxsize=128, ttl=3600) PERIODS = {"1m", "3m", "6m", "1y", "5y"} YF_PERIOD_MAP = {"1m": "1mo", "3m": "3mo", "6m": "6mo", "1y": "1y", "5y": "5y"} _XMAP = {"NYQ": "NYSE", "NMS": "NASDAQ", "NGM": "NASDAQ", "NCM": "NASDAQ", "ASE": "AMEX"} _SHARE_LABELS = ( "Ordinary Shares Number", "Share Issued", "Common Stock Shares Outstanding", ) def normalize_symbol(symbol: str) -> str: return str(symbol or "").strip().upper() def _safe_float(value: Any) -> float | None: try: n = float(value) except (TypeError, ValueError): return None if math.isnan(n) or math.isinf(n): return None return n def _safe_int(value: Any) -> int | None: n = _safe_float(value) return int(round(n)) if n is not None else None def _json_value(value: Any) -> Any: if value is None: return None if isinstance(value, pd.Timestamp): return value.isoformat() if pd.isna(value): return None if hasattr(value, "item"): return _json_value(value.item()) return value def _cap_ratio(value: float | None, lower: float, upper: float) -> float | None: if value is None or value <= lower or value >= upper: return None return value def _fmt_col(ts: Any, quarterly: bool) -> str: t = pd.Timestamp(ts) if quarterly: q = (t.month - 1) // 3 + 1 return f"Q{q} {t.year}" return f"FY {t.year}" def _row_vals(frame: pd.DataFrame, label: str, n: int) -> list[float | None]: if frame is None or frame.empty or label not in frame.index: return [None] * n series = pd.to_numeric(frame.loc[label], errors="coerce") return [_safe_float(series.iloc[i]) if i < len(series) else None for i in range(n)] def _row_vals_multi(frame: pd.DataFrame, n: int, *labels: str) -> list[float | None]: for label in labels: vals = _row_vals(frame, label, n) if any(v is not None for v in vals): return vals return [None] * n def _fin_row(label: str, indent: int, is_total: bool, values: list[float | None]) -> dict: return {"label": label, "indent": indent, "is_total": is_total, "is_section": False, "is_margin": False, "values": values} def _fin_section(label: str) -> dict: return {"label": label, "indent": 0, "is_total": False, "is_section": True, "is_margin": False, "values": []} def _fin_margin(label: str, values: list[float | None]) -> dict: return {"label": label, "indent": 1, "is_total": False, "is_section": False, "is_margin": True, "values": values} def _safe_ratio(num: float | None, den: float | None) -> float | None: if num is None or den is None or den == 0: return None return num / den def _build_income(frame: pd.DataFrame, frame_q: pd.DataFrame, quarterly: bool) -> dict: if frame is None or frame.empty: return {"columns": [], "rows": []} n = min(len(frame.columns), 8 if quarterly else 4) col_labels = [_fmt_col(c, quarterly) for c in frame.columns[:n]] if not quarterly: col_labels.append("TTM") def v(label: str) -> list[float | None]: base = _row_vals(frame, label, n) return base + ([_statement_ttm(frame_q, label)] if not quarterly else []) def vm(*labels: str) -> list[float | None]: base = _row_vals_multi(frame, n, *labels) if not quarterly: ttm = None for lbl in labels: ttm = _statement_ttm(frame_q, lbl) if ttm is not None: break base = base + [ttm] return base rev = v("Total Revenue") gross = v("Gross Profit") net = v("Net Income") return { "columns": col_labels, "rows": [ _fin_row("Total Revenue", 0, True, rev), _fin_row("Cost of Revenue", 1, False, v("Cost Of Revenue")), _fin_row("Gross Profit", 0, True, gross), _fin_margin("gross margin", [_safe_ratio(g, r) for g, r in zip(gross, rev)]), _fin_row("Operating Expenses", 1, False, v("Operating Expense")), _fin_row("Operating Income", 0, True, v("Operating Income")), _fin_row("EBITDA", 1, False, vm("EBITDA", "Normalized EBITDA")), _fin_row("Interest Expense", 1, False, v("Interest Expense")), _fin_row("Pretax Income", 0, False, v("Pretax Income")), _fin_row("Tax Provision", 1, False, v("Tax Provision")), _fin_row("Net Income", 0, True, net), _fin_margin("net margin", [_safe_ratio(ni, r) for ni, r in zip(net, rev)]), _fin_row("EPS Basic", 1, False, v("Basic EPS")), ], } def _build_balance(frame: pd.DataFrame, frame_q: pd.DataFrame, quarterly: bool) -> dict: if frame is None or frame.empty: return {"columns": [], "rows": []} n = min(len(frame.columns), 8 if quarterly else 4) col_labels = [_fmt_col(c, quarterly) for c in frame.columns[:n]] if not quarterly: col_labels.append("MRQ") def v(*labels: str) -> list[float | None]: base = _row_vals_multi(frame, n, *labels) if not quarterly: val = None for lbl in labels: val = _balance_value(frame_q, lbl) if val is not None: break base = base + [val] return base return { "columns": col_labels, "rows": [ _fin_section("ASSETS"), _fin_row("Current Assets", 0, True, v("Current Assets")), _fin_row("Cash & Equivalents", 1, False, v("Cash And Cash Equivalents", "Cash Cash Equivalents And Short Term Investments")), _fin_row("Short Term Investments", 1, False, v("Other Short Term Investments", "Short Term Investments")), _fin_row("Receivables", 1, False, v("Receivables", "Net Receivables")), _fin_row("Inventory", 1, False, v("Inventory")), _fin_row("Total Assets", 0, True, v("Total Assets")), _fin_section("LIABILITIES"), _fin_row("Current Liabilities", 0, True, v("Current Liabilities")), _fin_row("Accounts Payable", 1, False, v("Payables And Accrued Expenses", "Accounts Payable")), _fin_row("Short Term Debt", 1, False, v("Current Debt", "Short Term Debt And Capital Lease Obligation")), _fin_row("Long Term Debt", 1, False, v("Long Term Debt", "Long Term Debt And Capital Lease Obligation")), _fin_row("Total Liabilities", 0, True, v("Total Liabilities Net Minority Interest", "Total Liabilities")), _fin_section("EQUITY"), _fin_row("Stockholders Equity", 0, True, v("Stockholders Equity", "Common Stock Equity")), ], } def _build_cash_flow(cf: pd.DataFrame, cf_q: pd.DataFrame, inc: pd.DataFrame, inc_q: pd.DataFrame, quarterly: bool) -> dict: if cf is None or cf.empty: return {"columns": [], "rows": []} n = min(len(cf.columns), 8 if quarterly else 4) col_labels = [_fmt_col(c, quarterly) for c in cf.columns[:n]] if not quarterly: col_labels.append("TTM") def cv(*labels: str) -> list[float | None]: base = _row_vals_multi(cf, n, *labels) if not quarterly: ttm = None for lbl in labels: ttm = _statement_ttm(cf_q, lbl) if ttm is not None: break base = base + [ttm] return base def iv(*labels: str) -> list[float | None]: base = _row_vals_multi(inc, n, *labels) if not quarterly: ttm = None for lbl in labels: ttm = _statement_ttm(inc_q, lbl) if ttm is not None: break base = base + [ttm] return base op_cf = cv("Operating Cash Flow", "Cash Flow From Continuing Operating Activities") capex = cv("Capital Expenditure") # CapEx is negative in yfinance; FCF = Operating CF + CapEx fcf = [a + b if a is not None and b is not None else None for a, b in zip(op_cf, capex)] rev = iv("Total Revenue") return { "columns": col_labels, "rows": [ _fin_section("OPERATING"), _fin_row("Net Income", 1, False, iv("Net Income")), _fin_row("D&A", 1, False, cv("Depreciation And Amortization", "Reconciled Depreciation")), _fin_row("Changes in Working Capital", 1, False, cv("Change In Working Capital")), _fin_row("Operating Cash Flow", 0, True, op_cf), _fin_section("INVESTING"), _fin_row("CapEx", 1, False, capex), _fin_row("Free Cash Flow", 0, True, fcf), _fin_margin("FCF margin", [_safe_ratio(f, r) for f, r in zip(fcf, rev)]), _fin_row("Investing Cash Flow", 0, True, cv("Investing Cash Flow", "Cash Flow From Continuing Investing Activities")), _fin_section("FINANCING"), _fin_row("Dividends Paid", 1, False, cv("Cash Dividends Paid", "Common Stock Dividend Paid")), _fin_row("Buybacks", 1, False, cv("Repurchase Of Capital Stock", "Common Stock Repurchase")), _fin_row("Financing Cash Flow", 0, True, cv("Financing Cash Flow", "Cash Flow From Continuing Financing Activities")), _fin_row("Net Change in Cash", 0, True, cv("Changes In Cash", "End Cash Position")), ], } @cached(FINANCIALS_CACHE) def get_financials(symbol: str, period: str = "annual") -> dict: sym = normalize_symbol(symbol) quarterly = period == "quarterly" inc = get_income_statement(sym, quarterly=quarterly) bal = get_balance_sheet(sym, quarterly=quarterly) cf = get_cash_flow(sym, quarterly=quarterly) inc_q = get_income_statement(sym, quarterly=True) if not quarterly else inc bal_q = get_balance_sheet(sym, quarterly=True) if not quarterly else bal cf_q = get_cash_flow(sym, quarterly=True) if not quarterly else cf return { "period": period, "income": _build_income(inc, inc_q, quarterly), "balance": _build_balance(bal, bal_q, quarterly), "cash_flow": _build_cash_flow(cf, cf_q, inc, inc_q, quarterly), } def _balance_value(frame: pd.DataFrame, *labels: str) -> float | None: if frame is None or frame.empty: return None for label in labels: if label not in frame.index: continue series = pd.to_numeric(frame.loc[label], errors="coerce").dropna() if series.empty: continue value = _safe_float(series.iloc[0]) if value is not None: return value return None def _statement_ttm(frame: pd.DataFrame, *labels: str) -> float | None: if frame is None or frame.empty: return None for label in labels: if label not in frame.index: continue series = pd.to_numeric(frame.loc[label].iloc[:4], errors="coerce").dropna() if len(series) == 4: value = _safe_float(series.sum()) if value is not None: return value return None def _latest_share_count(balance_sheet: pd.DataFrame) -> float | None: shares = _balance_value(balance_sheet, *_SHARE_LABELS) return shares if shares is not None and shares > 0 else None def _pick_search_match(symbol: str) -> dict[str, Any]: sym = normalize_symbol(symbol) results = search_tickers(sym) for row in results: if normalize_symbol(row.get("symbol")) == sym: return row return {} @cached(SEARCH_CACHE) def search_tickers(query: str) -> list[dict[str, Any]]: """Search for tickers by company name or symbol.""" q = str(query or "").strip() if len(q) < 2: return [] try: results = yf.Search(q, max_results=8).quotes out: list[dict[str, Any]] = [] for row in results: symbol = row.get("symbol", "") if not symbol: continue out.append( { "symbol": normalize_symbol(symbol), "name": row.get("longname") or row.get("shortname") or symbol, "exchange": row.get("exchange") or row.get("exchDisp") or None, } ) return out except Exception: return [] @cached(INFO_CACHE) def get_company_info(symbol: str) -> dict[str, Any]: """Return a JSON-safe company info dict from yfinance.""" sym = normalize_symbol(symbol) try: info = yf.Ticker(sym).info or {} if not isinstance(info, dict): return {} return {str(k): _json_value(v) for k, v in info.items()} except Exception: return {} @cached(FAST_INFO_CACHE) def get_fast_info(symbol: str) -> dict[str, Any]: """Return a JSON-safe subset of yfinance fast_info.""" sym = normalize_symbol(symbol) try: fast_info = yf.Ticker(sym).fast_info keys = [ "currency", "dayHigh", "dayLow", "exchange", "fiftyDayAverage", "lastPrice", "lastVolume", "marketCap", "open", "previousClose", "regularMarketPreviousClose", "shares", "tenDayAverageVolume", "threeMonthAverageVolume", "timezone", "twoHundredDayAverage", "yearChange", "yearHigh", "yearLow", ] return {key: _json_value(fast_info.get(key)) for key in keys} except Exception: return {} @cached(PRICE_CACHE) def get_latest_price(symbol: str) -> float | None: """Return latest close price, falling back to quote fields in info.""" sym = normalize_symbol(symbol) try: hist = yf.Ticker(sym).history(period="5d") if hist is not None and not hist.empty and "Close" in hist.columns: close = pd.to_numeric(hist["Close"], errors="coerce").dropna() if not close.empty: return _safe_float(close.iloc[-1]) info = get_company_info(sym) for key in ("currentPrice", "regularMarketPrice", "previousClose"): price = _safe_float(info.get(key)) if price is not None: return price return None except Exception: return None @cached(HISTORY_CACHE) def get_price_history(symbol: str, period: str = "1y") -> list[dict[str, Any]]: """Return JSON-safe OHLCV history.""" if period not in PERIODS: period = "1y" try: df = yf.Ticker(normalize_symbol(symbol)).history(period=YF_PERIOD_MAP[period]) if df is None or df.empty: return [] df.index = pd.to_datetime(df.index) return _history_rows(df, include_time=False) except Exception: return [] @cached(INTRADAY_CACHE) def get_intraday_history(symbol: str, period: str, interval: str) -> list[dict[str, Any]]: """Return intraday JSON-safe OHLCV history.""" try: df = yf.Ticker(normalize_symbol(symbol)).history(period=period, interval=interval) if df is None or df.empty: return [] df.index = pd.to_datetime(df.index) try: df = df.between_time("09:30", "16:00") except Exception: pass return _history_rows(df, include_time=True) except Exception: return [] @cached(INCOME_CACHE) def get_income_statement(symbol: str, quarterly: bool = False) -> pd.DataFrame: try: ticker = yf.Ticker(normalize_symbol(symbol)) frame = ticker.quarterly_income_stmt if quarterly else ticker.income_stmt return frame if isinstance(frame, pd.DataFrame) else pd.DataFrame() except Exception: return pd.DataFrame() @cached(BALANCE_CACHE) def get_balance_sheet(symbol: str, quarterly: bool = False) -> pd.DataFrame: try: ticker = yf.Ticker(normalize_symbol(symbol)) frame = ticker.quarterly_balance_sheet if quarterly else ticker.balance_sheet return frame if isinstance(frame, pd.DataFrame) else pd.DataFrame() except Exception: return pd.DataFrame() @cached(CF_CACHE) def get_cash_flow(symbol: str, quarterly: bool = False) -> pd.DataFrame: try: ticker = yf.Ticker(normalize_symbol(symbol)) frame = ticker.quarterly_cashflow if quarterly else ticker.cashflow return frame if isinstance(frame, pd.DataFrame) else pd.DataFrame() except Exception: return pd.DataFrame() @cached(SHARES_CACHE) def get_shares_outstanding(symbol: str) -> float | None: sym = normalize_symbol(symbol) info = get_company_info(sym) for key in ("sharesOutstanding", "impliedSharesOutstanding"): shares = _safe_float(info.get(key)) if shares is not None and shares > 0: return shares fast_info = get_fast_info(sym) shares = _safe_float(fast_info.get("shares")) if shares is not None and shares > 0: return shares balance_sheet = get_balance_sheet(sym, quarterly=True) shares = _latest_share_count(balance_sheet) if shares is not None: return shares try: history = yf.Ticker(sym).get_shares_full(start="2000-01-01") if isinstance(history, pd.Series): values = pd.to_numeric(history, errors="coerce").dropna() if not values.empty: latest = _safe_float(values.iloc[-1]) if latest is not None and latest > 0: return latest except Exception: pass return None def get_market_cap_computed(symbol: str, price: float | None = None, shares: float | None = None) -> float | None: latest_price = price if price is not None else get_latest_price(symbol) share_count = shares if shares is not None else get_shares_outstanding(symbol) if latest_price is not None and latest_price > 0 and share_count is not None and share_count > 0: return latest_price * share_count return None def _history_rows(df: pd.DataFrame, include_time: bool) -> list[dict[str, Any]]: rows: list[dict[str, Any]] = [] for idx, row in df.iterrows(): dt = pd.Timestamp(idx) date = dt.strftime("%Y-%m-%dT%H:%M:%S") if include_time else dt.strftime("%Y-%m-%d") rows.append( { "date": date, "open": _safe_float(row.get("Open")), "high": _safe_float(row.get("High")), "low": _safe_float(row.get("Low")), "close": _safe_float(row.get("Close")), "volume": _safe_float(row.get("Volume")), } ) return rows @cached(MARKET_CACHE) def get_market_indices() -> list[dict[str, Any]]: """Return latest price and day change percent for major indices.""" symbols = { "S&P 500": "^GSPC", "NASDAQ": "^IXIC", "DOW": "^DJI", "VIX": "^VIX", } result: list[dict[str, Any]] = [] for name, sym in symbols.items(): price: float | None = None pct_change: float | None = None try: hist = yf.Ticker(sym).history(period="2d") if len(hist) >= 2: prev_close = _safe_float(hist["Close"].iloc[-2]) last = _safe_float(hist["Close"].iloc[-1]) if prev_close and last is not None: price = last pct_change = (last - prev_close) / prev_close elif len(hist) == 1: price = _safe_float(hist["Close"].iloc[-1]) pct_change = 0.0 except Exception: pass result.append({"name": name, "price": price, "change_pct": pct_change}) return result def build_quote(info: dict[str, Any], symbol: str) -> dict[str, Any]: price = _safe_float(info.get("currentPrice") or info.get("regularMarketPrice")) or get_latest_price(symbol) prev_close = _safe_float(info.get("regularMarketPreviousClose") or info.get("previousClose")) change = None change_pct = None if price is not None and prev_close and prev_close > 0: change = price - prev_close change_pct = change / prev_close return {"price": price, "prev_close": prev_close, "change": change, "change_pct": change_pct} def build_signals(info: dict[str, Any], ratios: dict[str, Any]) -> list[dict[str, str]]: signals: list[dict[str, str]] = [] pe = _safe_float(info.get("trailingPE")) if pe is None: pe = _safe_float(ratios.get("trailing_pe")) if pe is not None and pe > 0: if pe < 15: signals.append({"key": "Valuation", "state": "pos", "value": f"P/E {pe:.1f}x", "description": "Attractive multiple"}) elif pe < 30: signals.append({"key": "Valuation", "state": "warn", "value": f"P/E {pe:.1f}x", "description": "Middle of range"}) else: signals.append({"key": "Valuation", "state": "neg", "value": f"P/E {pe:.1f}x", "description": "Premium multiple"}) else: signals.append({"key": "Valuation", "state": "neu", "value": "P/E unavailable", "description": "No trailing earnings"}) _ratio_signal(signals, "Growth", info.get("revenueGrowth"), 0.10, 0.0, "Strong top-line growth", "Low but positive growth", "Contracting revenue") profit = _safe_float(info.get("profitMargins")) if profit is None: profit = _safe_float(ratios.get("net_margin_ttm")) _ratio_signal(signals, "Profit", profit, 0.15, 0.05, "High net margin", "Moderate net margin", "Thin or negative margin") debt_to_equity = _safe_float(info.get("debtToEquity")) if debt_to_equity is not None: debt_to_equity = debt_to_equity / 100.0 else: debt_to_equity = _safe_float(ratios.get("debt_to_equity")) if debt_to_equity is not None: if debt_to_equity < 0.5: state, desc = "pos", "Low leverage" elif debt_to_equity < 2.0: state, desc = "warn", "Moderate leverage" else: state, desc = "neg", "High leverage" signals.append({"key": "Leverage", "state": state, "value": f"D/E {debt_to_equity:.2f}x", "description": desc}) return signals def _ratio_signal( signals: list[dict[str, str]], key: str, value: Any, positive_threshold: float, warn_threshold: float, positive_desc: str, warn_desc: str, negative_desc: str, ) -> None: ratio = _safe_float(value) if ratio is None: return if ratio > positive_threshold: state, desc = "pos", positive_desc elif ratio >= warn_threshold: state, desc = "warn", warn_desc else: state, desc = "neg", negative_desc formatted = f"{ratio * 100:+.0f}%" if key == "Growth" else f"{ratio * 100:.0f}%" signals.append({"key": key, "state": state, "value": formatted, "description": desc}) def _field(source_map: dict[str, dict[str, Any]], field_sources: dict[str, str], name: str, *candidates: tuple[str, str]) -> Any: for source_name, key in candidates: source = source_map.get(source_name) or {} value = source.get(key) if value is None: continue if isinstance(value, str) and not value.strip(): continue field_sources[name] = source_name return value return None def _history_snapshot(history: list[dict[str, Any]]) -> dict[str, Any]: if not history: return {} closes = [_safe_float(row.get("close")) for row in history] closes = [value for value in closes if value is not None] volumes = [_safe_float(row.get("volume")) for row in history] volumes = [value for value in volumes if value is not None] latest = history[-1] previous = history[-2] if len(history) > 1 else None return { "lastPrice": _safe_float(latest.get("close")), "previousClose": _safe_float(previous.get("close")) if previous else None, "lastVolume": _safe_float(latest.get("volume")), "yearHigh": max(closes) if closes else None, "yearLow": min(closes) if closes else None, "averageVolume": (sum(volumes) / len(volumes)) if volumes else None, } @cached(PROFILE_ENRICH_CACHE) def get_profile_enrichment(symbol: str) -> dict[str, Any]: sym = normalize_symbol(symbol) fmp_key = os.getenv("FMP_API_KEY") if fmp_key: try: with httpx.Client(timeout=3.0) as client: res = client.get( "https://financialmodelingprep.com/api/v3/profile/" + sym, params={"apikey": fmp_key}, ) rows = res.json() if isinstance(rows, list) and rows: row = rows[0] or {} return { "sector": row.get("sector"), "industry": row.get("industry"), "website": row.get("website"), "summary": row.get("description"), } except Exception: pass finnhub_key = os.getenv("FINNHUB_API_KEY") if finnhub_key: try: with httpx.Client(timeout=3.0) as client: res = client.get( "https://finnhub.io/api/v1/stock/profile2", params={"symbol": sym, "token": finnhub_key}, ) row = res.json() if isinstance(row, dict) and row: return { "industry": row.get("finnhubIndustry"), "website": row.get("weburl"), "name": row.get("name"), "exchange": row.get("exchange"), } except Exception: pass return {} def _build_profile(sym: str, info: dict[str, Any], fast_info: dict[str, Any], search_match: dict[str, Any], field_sources: dict[str, str]) -> dict[str, Any]: enrichment = get_profile_enrichment(sym) source_map = { "info": info, "fast_info": fast_info, "search": search_match, "enrichment": enrichment, } name = _field( source_map, field_sources, "profile.name", ("info", "longName"), ("info", "shortName"), ("enrichment", "name"), ("search", "name"), ) exchange = _field( source_map, field_sources, "profile.exchange", ("info", "exchange"), ("enrichment", "exchange"), ("fast_info", "exchange"), ("search", "exchange"), ) if exchange is not None: exchange = _XMAP.get(str(exchange), exchange) return { "symbol": sym, "name": str(name or sym), "sector": _field(source_map, field_sources, "profile.sector", ("info", "sector"), ("enrichment", "sector")), "industry": _field(source_map, field_sources, "profile.industry", ("info", "industry"), ("enrichment", "industry")), "exchange": exchange, "website": _field(source_map, field_sources, "profile.website", ("info", "website"), ("enrichment", "website")), "summary": _field(source_map, field_sources, "profile.summary", ("info", "longBusinessSummary"), ("enrichment", "summary")), } @cached(RATIO_CACHE) def compute_ttm_ratios(symbol: str) -> dict[str, Any]: sym = normalize_symbol(symbol) info = get_company_info(sym) price = _safe_float(info.get("currentPrice") or info.get("regularMarketPrice")) or get_latest_price(sym) shares = get_shares_outstanding(sym) income = get_income_statement(sym, quarterly=True) balance = get_balance_sheet(sym, quarterly=True) cash_flow = get_cash_flow(sym, quarterly=True) if income is None or income.empty: return {} revenue = _statement_ttm(income, "Total Revenue") gross_profit = _statement_ttm(income, "Gross Profit") operating_income = _statement_ttm(income, "Operating Income") net_income = _statement_ttm(income, "Net Income") ebit = _statement_ttm(income, "EBIT") ebitda = _statement_ttm(income, "EBITDA", "Normalized EBITDA") tax_provision = _statement_ttm(income, "Tax Provision") pretax_income = _statement_ttm(income, "Pretax Income") equity = _balance_value(balance, "Stockholders Equity", "Common Stock Equity") total_assets = _balance_value(balance, "Total Assets") total_debt = _balance_value(balance, "Total Debt", "Long Term Debt And Capital Lease Obligation") current_assets = _balance_value(balance, "Current Assets") current_liabilities = _balance_value(balance, "Current Liabilities") cash = _balance_value(balance, "Cash And Cash Equivalents", "Cash Cash Equivalents And Short Term Investments") or 0.0 market_cap = get_market_cap_computed(sym, price=price, shares=shares) trailing_eps = None if shares is not None and shares > 0 and net_income is not None: trailing_eps = net_income / shares ratios: dict[str, Any] = {} ratios["market_cap"] = market_cap ratios["trailing_eps"] = trailing_eps if revenue and revenue > 0: if gross_profit is not None: ratios["gross_margin_ttm"] = gross_profit / revenue if operating_income is not None: ratios["operating_margin_ttm"] = operating_income / revenue if net_income is not None: ratios["net_margin_ttm"] = net_income / revenue if equity and equity > 0 and net_income is not None: roe = net_income / equity if abs(roe) < 10: ratios["roe_ttm"] = roe if total_assets and total_assets > 0 and net_income is not None: roa = net_income / total_assets if abs(roa) < 10: ratios["roa_ttm"] = roa if ebit is not None and pretax_income not in (None, 0): effective_tax_rate = max(0.0, (tax_provision or 0.0) / pretax_income) invested_capital = (equity or 0.0) + (total_debt or 0.0) - cash if invested_capital > 0: roic = (ebit * (1 - effective_tax_rate)) / invested_capital if abs(roic) < 10: ratios["roic_ttm"] = roic if market_cap and market_cap > 0: if net_income and net_income > 0: ratios["trailing_pe"] = market_cap / net_income if revenue and revenue > 0: ratios["price_to_sales"] = _cap_ratio(market_cap / revenue, 0, 100) if equity and equity > 0: ratios["price_to_book"] = _cap_ratio(market_cap / equity, 0, 100) enterprise_value = market_cap + (total_debt or 0.0) - cash if revenue and revenue > 0: ratios["ev_to_sales"] = _cap_ratio(enterprise_value / revenue, 0, 100) if ebitda and ebitda > 1e6: ratios["ev_to_ebitda"] = _cap_ratio(enterprise_value / ebitda, 0, 500) if equity and equity > 0 and total_debt is not None: ratios["debt_to_equity"] = _cap_ratio(total_debt / equity, -1, 100) if current_liabilities and current_liabilities > 0 and current_assets is not None: ratios["current_ratio"] = current_assets / current_liabilities dividends_paid = _statement_ttm(cash_flow, "Cash Dividends Paid", "Common Stock Dividend Paid") if dividends_paid is not None: dividends_paid = abs(dividends_paid) if market_cap and market_cap > 0: div_yield = dividends_paid / market_cap if 0 <= div_yield < 1: ratios["dividend_yield_ttm"] = div_yield if net_income and net_income > 0: payout = dividends_paid / net_income if 0 <= payout < 10: ratios["dividend_payout_ratio_ttm"] = payout return {key: value for key, value in ratios.items() if value is not None} def _build_quote_and_stats( info: dict[str, Any], fast_info: dict[str, Any], month_history: list[dict[str, Any]], year_history: list[dict[str, Any]], computed: dict[str, Any], field_sources: dict[str, str], ) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any]]: month_snapshot = _history_snapshot(month_history) year_snapshot = _history_snapshot(year_history) source_map = { "info": info, "fast_info": fast_info, "history_recent": month_snapshot, "history_year": year_snapshot, "computed": computed, } price = _safe_float( _field( source_map, field_sources, "quote.price", ("info", "currentPrice"), ("info", "regularMarketPrice"), ("fast_info", "lastPrice"), ("history_recent", "lastPrice"), ) ) prev_close = _safe_float( _field( source_map, field_sources, "quote.prev_close", ("info", "regularMarketPreviousClose"), ("info", "previousClose"), ("fast_info", "regularMarketPreviousClose"), ("fast_info", "previousClose"), ("history_recent", "previousClose"), ) ) change = None change_pct = None if price is not None and prev_close is not None and prev_close > 0: change = price - prev_close change_pct = change / prev_close volume = _safe_float( _field( source_map, field_sources, "stats.volume", ("info", "volume"), ("fast_info", "lastVolume"), ("history_recent", "lastVolume"), ) ) average_volume = _safe_float( _field( source_map, field_sources, "stats.average_volume", ("info", "averageVolume"), ("fast_info", "threeMonthAverageVolume"), ("fast_info", "tenDayAverageVolume"), ("history_recent", "averageVolume"), ) ) market_cap = _safe_float( _field(source_map, field_sources, "stats.market_cap", ("info", "marketCap"), ("fast_info", "marketCap"), ("computed", "market_cap")) ) trailing_pe = _safe_float(_field(source_map, field_sources, "stats.trailing_pe", ("info", "trailingPE"), ("computed", "trailing_pe"))) trailing_eps = _safe_float(_field(source_map, field_sources, "stats.trailing_eps", ("info", "trailingEps"), ("computed", "trailing_eps"))) beta = _safe_float(_field(source_map, field_sources, "stats.beta", ("info", "beta"))) range_low = _safe_float( _field( source_map, field_sources, "range_52w.low", ("info", "fiftyTwoWeekLow"), ("fast_info", "yearLow"), ("history_year", "yearLow"), ) ) range_high = _safe_float( _field( source_map, field_sources, "range_52w.high", ("info", "fiftyTwoWeekHigh"), ("fast_info", "yearHigh"), ("history_year", "yearHigh"), ) ) return ( {"price": price, "prev_close": prev_close, "change": change, "change_pct": change_pct}, { "market_cap": market_cap, "trailing_pe": trailing_pe, "trailing_eps": trailing_eps, "volume": volume, "average_volume": average_volume, "beta": beta, }, {"low": range_low, "high": range_high, "price": price}, ) def _build_ratios(computed: dict[str, Any], field_sources: dict[str, str]) -> dict[str, Any]: ratios: dict[str, Any] = {} keys = ( "price_to_book", "price_to_sales", "ev_to_sales", "ev_to_ebitda", "gross_margin_ttm", "operating_margin_ttm", "net_margin_ttm", "roe_ttm", "roa_ttm", "roic_ttm", "debt_to_equity", "current_ratio", "dividend_yield_ttm", "dividend_payout_ratio_ttm", ) for key in keys: value = _safe_float(computed.get(key)) ratios[key] = value if value is not None: field_sources[f"ratios.{key}"] = "computed" return ratios def _has_any_overview_data( profile: dict[str, Any], quote: dict[str, Any], stats: dict[str, Any], ratios: dict[str, Any], range_52w: dict[str, Any], short_interest: dict[str, Any], field_sources: dict[str, str], ) -> bool: for bucket in (profile, quote, stats, ratios, range_52w, short_interest): for key, value in bucket.items(): if key == "symbol": continue if key == "name" and bucket is profile and "profile.name" not in field_sources: continue if isinstance(value, str) and value.strip(): return True if value is not None and not isinstance(value, str): return True return False def get_ticker_overview(symbol: str) -> dict[str, Any] | None: sym = normalize_symbol(symbol) info = get_company_info(sym) search_match = _pick_search_match(sym) fast_info = get_fast_info(sym) month_history = get_price_history(sym, period="1m") year_history = get_price_history(sym, period="1y") computed = compute_ttm_ratios(sym) field_sources: dict[str, str] = {} profile = _build_profile(sym, info, fast_info, search_match, field_sources) quote, stats, range_52w = _build_quote_and_stats(info, fast_info, month_history, year_history, computed, field_sources) ratios = _build_ratios(computed, field_sources) short = _safe_int(info.get("sharesShort")) short_prior = _safe_int(info.get("sharesShortPriorMonth")) short_delta = None if short is not None and short_prior and short_prior > 0: short_delta = (short - short_prior) / short_prior short_interest = { "short_percent_of_float": _safe_float(info.get("shortPercentOfFloat")), "short_ratio": _safe_float(info.get("shortRatio")), "shares_short": short, "shares_short_prior_month": short_prior, "shares_short_delta_pct": short_delta, } if not _has_any_overview_data(profile, quote, stats, ratios, range_52w, short_interest, field_sources): return None field_availability = { "profile.name": bool(profile.get("name")), "profile.exchange": profile.get("exchange") is not None, "profile.sector": profile.get("sector") is not None, "profile.industry": profile.get("industry") is not None, "profile.website": profile.get("website") is not None, "profile.summary": profile.get("summary") is not None, "quote.price": quote.get("price") is not None, "quote.prev_close": quote.get("prev_close") is not None, "stats.market_cap": stats.get("market_cap") is not None, "stats.trailing_pe": stats.get("trailing_pe") is not None, "stats.trailing_eps": stats.get("trailing_eps") is not None, "stats.volume": stats.get("volume") is not None, "stats.average_volume": stats.get("average_volume") is not None, "stats.beta": stats.get("beta") is not None, "ratios.price_to_book": ratios.get("price_to_book") is not None, "ratios.price_to_sales": ratios.get("price_to_sales") is not None, "ratios.ev_to_sales": ratios.get("ev_to_sales") is not None, "ratios.ev_to_ebitda": ratios.get("ev_to_ebitda") is not None, "ratios.gross_margin_ttm": ratios.get("gross_margin_ttm") is not None, "ratios.operating_margin_ttm": ratios.get("operating_margin_ttm") is not None, "ratios.net_margin_ttm": ratios.get("net_margin_ttm") is not None, "ratios.roe_ttm": ratios.get("roe_ttm") is not None, "ratios.roa_ttm": ratios.get("roa_ttm") is not None, "ratios.roic_ttm": ratios.get("roic_ttm") is not None, "ratios.debt_to_equity": ratios.get("debt_to_equity") is not None, "ratios.current_ratio": ratios.get("current_ratio") is not None, "ratios.dividend_yield_ttm": ratios.get("dividend_yield_ttm") is not None, "ratios.dividend_payout_ratio_ttm": ratios.get("dividend_payout_ratio_ttm") is not None, "range_52w.low": range_52w.get("low") is not None, "range_52w.high": range_52w.get("high") is not None, } is_partial = not all(field_availability.values()) return { "profile": profile, "quote": quote, "signals": build_signals(info, computed), "stats": stats, "ratios": ratios, "range_52w": range_52w, "short_interest": short_interest, "meta": { "status": "partial" if is_partial else "complete", "is_partial": is_partial, "field_availability": field_availability, "sources": field_sources, }, }