"""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) VALUATION_CACHE = TTLCache(maxsize=128, ttl=3600) HIST_RATIOS_CACHE: TTLCache = TTLCache(maxsize=128, ttl=3600) RATIOS_ENDPOINT_CACHE: TTLCache = TTLCache(maxsize=128, ttl=3600) PERIODS = {"1m", "3m", "6m", "1y", "2y", "5y"} YF_PERIOD_MAP = {"1m": "1mo", "3m": "3mo", "6m": "6mo", "1y": "1y", "2y": "2y", "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() try: if pd.isna(value): return None except (TypeError, ValueError): 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")), ], } _GROWTH_FLOOR = -0.50 _GROWTH_CAP = 0.50 _GROWTH_MIN_BASE = 1e-9 def _cap_growth(value: float) -> float: return max(_GROWTH_FLOOR, min(_GROWTH_CAP, float(value))) def _dcf_capped_growth_rate(fcf_series: "pd.Series") -> float | None: historical = fcf_series.sort_index().dropna().astype(float).values if len(historical) < 2: return None rates = [] for i in range(1, len(historical)): prev, curr = float(historical[i - 1]), float(historical[i]) if abs(prev) < _GROWTH_MIN_BASE: continue if prev <= 0 or curr <= 0: continue rates.append((curr - prev) / prev) if not rates: return None raw = float(pd.Series(rates).median()) return _cap_growth(raw) def _build_fcf_series(cf_annual: "pd.DataFrame") -> "pd.Series | None": if cf_annual is None or cf_annual.empty: return None op_labels = ("Operating Cash Flow", "Cash Flow From Continuing Operating Activities") op_row = None for label in op_labels: if label in cf_annual.index: op_row = pd.to_numeric(cf_annual.loc[label], errors="coerce") break if op_row is None or "Capital Expenditure" not in cf_annual.index: return None capex_row = pd.to_numeric(cf_annual.loc["Capital Expenditure"], errors="coerce") fcf = (op_row + capex_row).dropna().sort_index() return fcf if not fcf.empty else None def _build_multiple_result(raw: dict) -> dict: if not raw: return {"available": False} return { "available": True, "implied_price_per_share": raw.get("implied_price_per_share"), "implied_ev": raw.get("implied_ev"), "equity_value": raw.get("equity_value"), "net_debt": raw.get("net_debt"), "multiple_used": raw.get("target_multiple_used"), } def _run_dcf( fcf_series: "pd.Series", shares_outstanding: float, wacc: float = 0.10, terminal_growth: float = 0.03, projection_years: int = 5, total_debt: float = 0.0, cash_and_equivalents: float = 0.0, preferred_equity: float = 0.0, minority_interest: float = 0.0, ) -> dict: if fcf_series.empty or shares_outstanding <= 0: return {} historical = fcf_series.sort_index().dropna().astype(float).values if len(historical) < 2: return {} if wacc <= 0: return {"error": "WACC must be greater than 0%."} if terminal_growth >= wacc: return {"error": "Terminal growth must be lower than WACC."} growth_rate = _dcf_capped_growth_rate(fcf_series) if growth_rate is None: growth_rate = 0.05 base_fcf = float(historical[-1]) if base_fcf <= 0: return { "error": ( "DCF is not meaningful with zero or negative base free cash flow. " "Use comps, EV/EBITDA, or adjust the model after underwriting a credible FCF turnaround." ) } projected = [base_fcf * ((1 + growth_rate) ** yr) for yr in range(1, projection_years + 1)] discounted = [fcf / ((1 + wacc) ** i) for i, fcf in enumerate(projected, start=1)] fcf_pv_sum = float(sum(discounted)) terminal_fcf = float(projected[-1]) * (1 + terminal_growth) terminal_value = terminal_fcf / (wacc - terminal_growth) terminal_value_pv = terminal_value / ((1 + wacc) ** projection_years) enterprise_value = fcf_pv_sum + terminal_value_pv total_debt = float(total_debt or 0.0) cash_and_equivalents = float(cash_and_equivalents or 0.0) preferred_equity = float(preferred_equity or 0.0) minority_interest = float(minority_interest or 0.0) net_debt = total_debt - cash_and_equivalents equity_value = enterprise_value - net_debt - preferred_equity - minority_interest intrinsic_value_per_share = equity_value / shares_outstanding return { "intrinsic_value_per_share": intrinsic_value_per_share, "enterprise_value": enterprise_value, "equity_value": equity_value, "net_debt": net_debt, "cash_and_equivalents": cash_and_equivalents, "total_debt": total_debt, "terminal_value_pv": terminal_value_pv, "fcf_pv_sum": fcf_pv_sum, "growth_rate_used": growth_rate, "base_fcf": base_fcf, } def _run_ev_ebitda( ebitda: float, total_debt: float, total_cash: float, preferred_equity: float, minority_interest: float, shares_outstanding: float, target_multiple: float, ) -> dict: if not ebitda or ebitda <= 0: return {} if not shares_outstanding or shares_outstanding <= 0: return {} if not target_multiple or target_multiple <= 0: return {} implied_ev = ebitda * target_multiple net_debt = (total_debt or 0.0) - (total_cash or 0.0) other_claims = (preferred_equity or 0.0) + (minority_interest or 0.0) equity_value = implied_ev - net_debt - other_claims return { "implied_ev": implied_ev, "net_debt": net_debt, "equity_value": equity_value, "implied_price_per_share": equity_value / shares_outstanding, "target_multiple_used": target_multiple, } def _run_ev_revenue( revenue: float, total_debt: float, total_cash: float, preferred_equity: float, minority_interest: float, shares_outstanding: float, target_multiple: float, ) -> dict: if not revenue or revenue <= 0: return {} if not shares_outstanding or shares_outstanding <= 0: return {} if not target_multiple or target_multiple <= 0: return {} implied_ev = revenue * target_multiple net_debt = (total_debt or 0.0) - (total_cash or 0.0) other_claims = (preferred_equity or 0.0) + (minority_interest or 0.0) equity_value = implied_ev - net_debt - other_claims return { "implied_ev": implied_ev, "net_debt": net_debt, "equity_value": equity_value, "implied_price_per_share": equity_value / shares_outstanding, "target_multiple_used": target_multiple, } def _run_price_to_book(book_value_per_share: float, target_multiple: float) -> dict: if not book_value_per_share or book_value_per_share <= 0: return {} if not target_multiple or target_multiple <= 0: return {} return { "implied_price_per_share": float(book_value_per_share) * float(target_multiple), "target_multiple_used": float(target_multiple), "book_value_per_share": float(book_value_per_share), } @cached(VALUATION_CACHE) def get_valuation(symbol: str) -> dict: sym = normalize_symbol(symbol) cf_annual = get_cash_flow(sym, quarterly=False) inc_q = get_income_statement(sym, quarterly=True) bal_q = get_balance_sheet(sym, quarterly=True) info = get_company_info(sym) shares = get_shares_outstanding(sym) current_price = _safe_float(info.get("currentPrice")) total_debt = _balance_value(bal_q, "Total Debt") or 0.0 cash = _balance_value( bal_q, "Cash And Cash Equivalents", "Cash Cash Equivalents And Short Term Investments" ) or 0.0 preferred = _balance_value(bal_q, "Preferred Stock") or 0.0 minority = _balance_value(bal_q, "Minority Interest") or 0.0 equity = _balance_value(bal_q, "Stockholders Equity", "Common Stock Equity") ebitda_ttm = _statement_ttm(inc_q, "EBITDA", "Normalized EBITDA") revenue_ttm = _statement_ttm(inc_q, "Total Revenue") book_value_per_share: float | None = None if equity is not None and shares is not None and shares > 0: book_value_per_share = equity / shares ev_ebitda_multiple = _safe_float(info.get("enterpriseToEbitda")) ev_revenue_multiple = _safe_float(info.get("enterpriseToRevenue")) pb_multiple = _safe_float(info.get("priceToBook")) fcf_series = _build_fcf_series(cf_annual) dcf_raw: dict = {} if fcf_series is not None and shares is not None and shares > 0: dcf_raw = _run_dcf( fcf_series=fcf_series, shares_outstanding=shares, total_debt=total_debt, cash_and_equivalents=cash, preferred_equity=preferred, minority_interest=minority, ) if not dcf_raw: dcf_out: dict = {"available": False, "wacc": 0.10, "terminal_growth": 0.03} elif "error" in dcf_raw: dcf_out = {"available": True, "error": dcf_raw["error"], "wacc": 0.10, "terminal_growth": 0.03} else: dcf_out = { "available": True, "intrinsic_value_per_share": dcf_raw.get("intrinsic_value_per_share"), "enterprise_value": dcf_raw.get("enterprise_value"), "equity_value": dcf_raw.get("equity_value"), "net_debt": dcf_raw.get("net_debt"), "cash_and_equivalents": dcf_raw.get("cash_and_equivalents"), "total_debt": dcf_raw.get("total_debt"), "terminal_value_pv": dcf_raw.get("terminal_value_pv"), "fcf_pv_sum": dcf_raw.get("fcf_pv_sum"), "growth_rate_used": dcf_raw.get("growth_rate_used"), "base_fcf": dcf_raw.get("base_fcf"), "wacc": 0.10, "terminal_growth": 0.03, } common = dict( total_debt=total_debt, total_cash=cash, preferred_equity=preferred, minority_interest=minority, shares_outstanding=shares or 0.0, ) ev_ebitda_out = _build_multiple_result( _run_ev_ebitda(ebitda=ebitda_ttm, target_multiple=ev_ebitda_multiple, **common) if ebitda_ttm and ev_ebitda_multiple and shares else {} ) ev_revenue_out = _build_multiple_result( _run_ev_revenue(revenue=revenue_ttm, target_multiple=ev_revenue_multiple, **common) if revenue_ttm and ev_revenue_multiple and shares else {} ) pb_out = _build_multiple_result( _run_price_to_book( book_value_per_share=book_value_per_share, target_multiple=pb_multiple, ) if book_value_per_share and pb_multiple else {} ) return { "symbol": sym, "current_price": current_price, "shares_outstanding": shares, "dcf": dcf_out, "ev_ebitda": ev_ebitda_out, "ev_revenue": ev_revenue_out, "price_to_book": pb_out, } @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 _find_price_at_date(price_history: list[dict], target: "pd.Timestamp") -> float | None: """Return closing price from price_history nearest to target date (within 45 days).""" if not price_history: return None best_price: float | None = None best_delta = float("inf") for pt in price_history: try: delta = abs((pd.Timestamp(pt["date"]) - target).days) if delta < best_delta: best_delta = delta best_price = _safe_float(pt.get("close")) except Exception: continue return best_price if best_delta <= 45 else None @cached(HIST_RATIOS_CACHE) def compute_historical_ratios(symbol: str) -> dict[str, list[float | None]]: """Per-fiscal-year ratios from annual statements, oldest-first (up to 4 points).""" sym = normalize_symbol(symbol) inc_a = get_income_statement(sym, quarterly=False) bal_a = get_balance_sheet(sym, quarterly=False) cf_a = get_cash_flow(sym, quarterly=False) if inc_a is None or inc_a.empty: return {} years = list(inc_a.columns[: min(len(inc_a.columns), 4)]) price_history = get_price_history(sym, period="5y") current_shares = get_shares_outstanding(sym) try: shares_history_raw = yf.Ticker(sym).get_shares_full(start="2000-01-01") if isinstance(shares_history_raw, pd.Series): shares_history = pd.to_numeric(shares_history_raw, errors="coerce").dropna().sort_index() else: shares_history = pd.Series(dtype=float) except Exception: shares_history = pd.Series(dtype=float) result: dict[str, list[float | None]] = {k: [] for k in [ "gross_margin", "operating_margin", "net_margin", "ebitda_margin", "roe", "roa", "debt_to_equity", "current_ratio", "trailing_pe", "ev_to_ebitda", "price_to_book", "price_to_sales", ]} def _balance_shares(period_date: pd.Timestamp) -> float | None: if bal_a is None or bal_a.empty or period_date not in bal_a.columns: return None for label in _SHARE_LABELS: if label not in bal_a.index: continue shares_value = _safe_float(bal_a.loc[label, period_date]) if shares_value is not None and shares_value > 0: return shares_value return None def _historical_shares_for_date(period_date: pd.Timestamp) -> float | None: direct_balance_shares = _balance_shares(period_date) if direct_balance_shares is not None: return direct_balance_shares if not shares_history.empty: target = pd.Timestamp(period_date) index = shares_history.index if getattr(index, "tz", None) is not None and target.tzinfo is None: target = target.tz_localize(index.tz) elif getattr(index, "tz", None) is None and target.tzinfo is not None: target = target.tz_localize(None) deltas = pd.Series(index - target, index=index).abs() if not deltas.empty: nearest_idx = deltas.idxmin() if abs(pd.Timestamp(nearest_idx) - target) <= pd.Timedelta(days=180): shares_value = _safe_float(shares_history.loc[nearest_idx]) if shares_value is not None and shares_value > 0: return shares_value return current_shares for col in years: col_dt = pd.Timestamp(col) def _inc(label: str) -> float | None: if label not in inc_a.index: return None return _safe_float(inc_a.loc[label, col]) if col in inc_a.columns else None def _bal(label: str) -> float | None: if bal_a is None or bal_a.empty or label not in bal_a.index: return None return _safe_float(bal_a.loc[label, col]) if col in bal_a.columns else None revenue = _inc("Total Revenue") gross_profit = _inc("Gross Profit") operating_income = _inc("Operating Income") net_income = _inc("Net Income") ebitda = _inc("EBITDA") or _inc("Normalized EBITDA") 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") current_assets = _bal("Current Assets") current_liabilities = _bal("Current Liabilities") cash = _bal("Cash And Cash Equivalents") or _bal("Cash Cash Equivalents And Short Term Investments") or 0.0 period_shares = _historical_shares_for_date(col_dt) rev = revenue if revenue and revenue > 0 else None result["gross_margin"].append(_cap_ratio(gross_profit / rev, -5, 5) if rev and gross_profit is not None else None) result["operating_margin"].append(_cap_ratio(operating_income / rev, -5, 5) if rev and operating_income is not None else None) result["net_margin"].append(_cap_ratio(net_income / rev, -5, 5) if rev and net_income is not None else None) result["ebitda_margin"].append(_cap_ratio(ebitda / rev, -5, 5) if rev and ebitda is not None else None) result["roe"].append(_cap_ratio(net_income / equity, -10, 10) if equity and equity > 0 and net_income is not None else None) result["roa"].append(_cap_ratio(net_income / total_assets, -10, 10) if total_assets and total_assets > 0 and net_income is not None else None) result["debt_to_equity"].append(_cap_ratio(total_debt / equity, -1, 100) if equity and equity > 0 and total_debt is not None else None) result["current_ratio"].append(current_assets / current_liabilities if current_liabilities and current_liabilities > 0 and current_assets is not None else None) price = _find_price_at_date(price_history, col_dt) market_cap = price * period_shares if price and period_shares else None ev = market_cap + (total_debt or 0.0) - cash if market_cap else None result["trailing_pe"].append(_cap_ratio(market_cap / net_income, 0, 500) if market_cap and net_income and net_income > 0 else None) result["ev_to_ebitda"].append(_cap_ratio(ev / ebitda, 0, 500) if ev and ebitda and ebitda > 1e6 else None) result["price_to_book"].append(_cap_ratio(market_cap / equity, 0, 100) if market_cap and equity and equity > 0 else None) result["price_to_sales"].append(_cap_ratio(market_cap / revenue, 0, 100) if market_cap and revenue and revenue > 0 else None) return {k: list(reversed(v)) for k, v in result.items()} @cached(RATIOS_ENDPOINT_CACHE) def get_ratios(symbol: str) -> dict: """Build the full RatiosResponse dict for the /ratios endpoint.""" sym = normalize_symbol(symbol) ttm = compute_ttm_ratios(sym) hist = compute_historical_ratios(sym) info = get_company_info(sym) income = get_income_statement(sym, quarterly=True) balance = get_balance_sheet(sym, quarterly=True) cf = get_cash_flow(sym, quarterly=True) ebitda = _statement_ttm(income, "EBITDA", "Normalized EBITDA") revenue = _statement_ttm(income, "Total Revenue") current_assets = _balance_value(balance, "Current Assets") current_liabilities = _balance_value(balance, "Current Liabilities") inventory = _balance_value(balance, "Inventory") ebit = _statement_ttm(income, "EBIT") interest_expense = _statement_ttm(income, "Interest Expense") op_cf = _statement_ttm(cf, "Operating Cash Flow", "Cash From Operations") capex_raw = _statement_ttm(cf, "Capital Expenditure") capex = abs(capex_raw) if capex_raw is not None else None fcf = (op_cf - capex) if op_cf is not None and capex is not None else None market_cap = ttm.get("market_cap") quick_ratio: float | None = None if current_liabilities and current_liabilities > 0 and current_assets is not None: quick_ratio = (current_assets - (inventory or 0.0)) / current_liabilities interest_coverage: float | None = None if interest_expense and ebit is not None: ie = abs(interest_expense) if ie > 0 and ebit > 0: interest_coverage = _cap_ratio(ebit / ie, 0, 1000) ebitda_margin = _cap_ratio(ebitda / revenue, -5, 5) if revenue and revenue > 0 and ebitda is not None else None fcf_margin = _cap_ratio(fcf / revenue, -5, 5) if revenue and revenue > 0 and fcf is not None else None p_fcf = _cap_ratio(market_cap / fcf, 0, 1000) if market_cap and fcf and fcf > 0 else None fwd_pe = _safe_float(info.get("forwardPE")) if info else None forward_pe = fwd_pe if fwd_pe and 0 < fwd_pe < 500 else None def point(ttm_key: str | None, hist_key: str | None, override: float | None = None) -> dict: val = override if override is not None else (ttm.get(ttm_key) if ttm_key else None) spark = hist.get(hist_key, []) if hist_key else [] return {"value": val, "spark": spark, "vs_sector": None} return { "pe_ttm": point("trailing_pe", "trailing_pe"), "ev_ebitda": point("ev_to_ebitda", "ev_to_ebitda"), "gross_margin": point("gross_margin_ttm", "gross_margin"), "net_margin": point("net_margin_ttm", "net_margin"), "price_to_book": point("price_to_book", "price_to_book"), "price_to_sales": point("price_to_sales", "price_to_sales"), "ev_to_sales": point("ev_to_sales", None), "p_fcf": point(None, None, p_fcf), "forward_pe": point(None, None, forward_pe), "operating_margin": point("operating_margin_ttm", "operating_margin"), "ebitda_margin": point(None, "ebitda_margin", ebitda_margin), "fcf_margin": point(None, None, fcf_margin), "roe": point("roe_ttm", "roe"), "roa": point("roa_ttm", "roa"), "roic": point("roic_ttm", None), "debt_to_equity": point("debt_to_equity", "debt_to_equity"), "current_ratio": point("current_ratio", "current_ratio"), "quick_ratio": point(None, None, quick_ratio), "interest_coverage": point(None, None, interest_coverage), "dividend_yield": point("dividend_yield_ttm", None), "dividend_payout": point("dividend_payout_ratio_ttm", 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} @cached(BETA_CACHE) def compute_beta(symbol: str) -> float | None: """Compute trailing 2-year beta against SPY from weekly returns.""" sym = normalize_symbol(symbol) if sym == "SPY": return 1.0 ticker_history = get_price_history(sym, period="2y") spy_history = get_price_history("SPY", period="2y") if not ticker_history or not spy_history: return None try: ticker_closes = {row["date"]: row["close"] for row in ticker_history if row.get("close") is not None} spy_closes = {row["date"]: row["close"] for row in spy_history if row.get("close") is not None} ticker_series = pd.Series(ticker_closes, dtype=float) ticker_series.index = pd.to_datetime(ticker_series.index) ticker_series = ticker_series.sort_index() spy_series = pd.Series(spy_closes, dtype=float) spy_series.index = pd.to_datetime(spy_series.index) spy_series = spy_series.sort_index() ticker_weekly = ticker_series.resample("W").last().pct_change(fill_method=None).dropna() spy_weekly = spy_series.resample("W").last().pct_change(fill_method=None).dropna() aligned = pd.concat([ticker_weekly, spy_weekly], axis=1, join="inner").dropna() aligned.columns = ["ticker", "spy"] if len(aligned) < 52: return None spy_var = aligned["spy"].var() if spy_var == 0: return None beta = aligned["ticker"].cov(aligned["spy"]) / spy_var beta = max(-3.0, min(3.0, beta)) return round(beta, 4) except Exception: return None @cached(SHORT_CACHE) def get_fmp_short_interest(symbol: str) -> dict[str, Any]: """Fetch short interest data from FMP as a fallback when yfinance returns nothing.""" sym = normalize_symbol(symbol) fmp_key = os.getenv("FMP_API_KEY") if not fmp_key: return {} try: with httpx.Client(timeout=3.0) as client: res = client.get( "https://financialmodelingprep.com/api/v4/short-of-float-symbol", params={"symbol": sym, "apikey": fmp_key}, ) rows = res.json() if not isinstance(rows, list) or not rows: return {} row = rows[0] or {} result: dict[str, Any] = {} short_pct = _safe_float(row.get("shortPercent")) if short_pct is not None: result["short_percent_of_float"] = short_pct short_ratio = _safe_float(row.get("shortRatio")) if short_ratio is not None: result["short_ratio"] = short_ratio shares_short = _safe_int(row.get("shortsVolume")) if shares_short is not None: result["shares_short"] = shares_short return result except Exception: return {} def _build_quote_and_stats( sym: str, 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"))) if beta is None: beta = compute_beta(sym) if beta is not None: field_sources["stats.beta"] = "computed" 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(sym, 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 all(v is None for v in short_interest.values()): fmp_short = get_fmp_short_interest(sym) if fmp_short: short_interest.update(fmp_short) 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, }, }