"""yfinance wrapper for Prism v2 Overview data.""" from __future__ import annotations import datetime import math import os import statistics from collections import defaultdict 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) SECTOR_BENCHMARK_CACHE: TTLCache = TTLCache(maxsize=128, ttl=3600) INSIDERS_CACHE: TTLCache = TTLCache(maxsize=256, ttl=3600) FILINGS_CACHE: TTLCache = TTLCache(maxsize=256, 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, "projection_years": 5} elif "error" in dcf_raw: dcf_out = {"available": True, "error": dcf_raw["error"], "wacc": 0.10, "terminal_growth": 0.03, "projection_years": 5} 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, "projection_years": 5, } 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) sector_bench = compute_sector_ratio_benchmarks(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, sector_key: str | 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 [] skey = sector_key if sector_key is not None else ttm_key vs_sector = sector_bench.get(skey) if skey else None return {"value": val, "spark": spark, "vs_sector": vs_sector} 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, "trailing_pe"), "operating_margin": point("operating_margin_ttm", "operating_margin"), "ebitda_margin": point(None, "ebitda_margin", ebitda_margin, "operating_margin_ttm"), "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, "current_ratio"), "interest_coverage": point(None, None, interest_coverage), "dividend_yield": point("dividend_yield_ttm", None), "dividend_payout": point("dividend_payout_ratio_ttm", None), } @cached(SECTOR_BENCHMARK_CACHE) def compute_sector_ratio_benchmarks(symbol: str) -> dict[str, float]: """Median TTM ratio benchmarks from same-sector peers (FMP-backed when available).""" sym = normalize_symbol(symbol) fmp_key = os.getenv("FMP_API_KEY") info = get_company_info(sym) sector_raw = info.get("sector") if isinstance(info, dict) else None sector = str(sector_raw or "").strip() if not sector: enrichment = get_profile_enrichment(sym) sector = str((enrichment or {}).get("sector") or "").strip() if not sector: return {} peer_symbols: list[str] = [] if fmp_key: try: with httpx.Client(timeout=3.5) as client: res = client.get( "https://financialmodelingprep.com/api/v3/stock-screener", params={ "sector": sector, "isEtf": "false", "isActivelyTrading": "true", "limit": 12, "apikey": fmp_key, }, ) rows = res.json() if isinstance(rows, list): for row in rows: psym = normalize_symbol((row or {}).get("symbol")) if not psym or psym == sym: continue peer_symbols.append(psym) except Exception: peer_symbols = [] # No-key or FMP failure fallback: search by sector term, then filter by exact sector. if not peer_symbols: try: candidates = search_tickers(sector) except Exception: candidates = [] target_sector = sector.lower() for row in candidates[:24]: psym = normalize_symbol((row or {}).get("symbol")) if not psym or psym == sym: continue pinfo = get_company_info(psym) psector = str((pinfo or {}).get("sector") or "").strip().lower() if psector and psector == target_sector: peer_symbols.append(psym) if not peer_symbols: return {} keys = [ "trailing_pe", "ev_to_ebitda", "gross_margin_ttm", "net_margin_ttm", "price_to_book", "price_to_sales", "ev_to_sales", "operating_margin_ttm", "roe_ttm", "roa_ttm", "roic_ttm", "debt_to_equity", "current_ratio", "dividend_yield_ttm", "dividend_payout_ratio_ttm", ] buckets: dict[str, list[float]] = {k: [] for k in keys} for psym in peer_symbols[:6]: try: ratios = compute_ttm_ratios(psym) except Exception: continue if not isinstance(ratios, dict): continue for key in keys: val = _safe_float(ratios.get(key)) if val is not None: buckets[key].append(val) out: dict[str, float] = {} for key, values in buckets.items(): if values: out[key] = float(statistics.median(values)) return out 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, }, } def _classify_insider(text: str) -> str: t = str(text or "").lower() if any(k in t for k in ("sale", "sold", "disposition")): return "sell" if any(k in t for k in ("purchase", "bought", "acquisition", "grant", "award", "exercise")): return "buy" return "other" @cached(INSIDERS_CACHE) def get_insider_transactions(symbol: str) -> dict: sym = normalize_symbol(symbol) try: t = yf.Ticker(sym) df = t.insider_transactions except Exception: df = None if df is None or (hasattr(df, "empty") and df.empty): return { "summary": {"buy_count": 0, "sell_count": 0, "buy_value": 0.0, "sell_value": 0.0}, "monthly_chart": [], "transactions": [], } df = df.copy() df["direction"] = df["Text"].apply(_classify_insider) def _to_dt(val: object) -> "pd.Timestamp | None": try: return pd.to_datetime(val) except Exception: return None df["_date"] = df["Start Date"].apply(_to_dt) cutoff = pd.Timestamp.now() - pd.Timedelta(days=180) recent = df[df["_date"] >= cutoff] def _total_value(subset: "pd.DataFrame") -> float: try: return float(subset["Value"].dropna().astype(float).sum()) except Exception: return 0.0 buys = recent[recent["direction"] == "buy"] sells = recent[recent["direction"] == "sell"] summary = { "buy_count": int(len(buys)), "sell_count": int(len(sells)), "buy_value": _total_value(buys), "sell_value": _total_value(sells), } # Monthly chart: last 6 months, buys positive, sells negative (in $M) monthly: dict[str, dict[str, float]] = {} for _, row in recent.iterrows(): dt = row["_date"] if dt is None or pd.isna(dt): continue key = dt.strftime("%Y-%m") monthly.setdefault(key, {"buy": 0.0, "sell": 0.0}) try: val = float(row["Value"]) if pd.notna(row["Value"]) else 0.0 except (TypeError, ValueError): val = 0.0 if row["direction"] in ("buy", "sell"): monthly[key][row["direction"]] += val monthly_chart = [ {"month": m, "buy": monthly[m]["buy"] / 1e6, "sell": monthly[m]["sell"] / 1e6} for m in sorted(monthly.keys()) ] # All transactions (newest first) transactions = [] for _, row in df.sort_values("_date", ascending=False).iterrows(): dt = row["_date"] transactions.append({ "date": dt.strftime("%Y-%m-%d") if dt is not None and not pd.isna(dt) else None, "insider": str(row.get("Insider", "") or ""), "position": str(row.get("Position", "") or ""), "direction": row["direction"], "shares": int(row["Shares"]) if pd.notna(row.get("Shares")) else None, "value": _safe_float(row.get("Value")), }) return { "summary": summary, "monthly_chart": monthly_chart, "transactions": transactions, } _FORM_DESCRIPTIONS: dict[str, str] = { "10-K": "Annual report", "10-Q": "Quarterly report", "8-K": "Material event disclosure", "DEF 14A": "Proxy statement", "S-1": "IPO registration", "S-3": "Securities registration", "4": "Insider ownership change", "SC 13G": "Beneficial ownership (passive)", "SC 13D": "Beneficial ownership (active)", } @cached(FILINGS_CACHE) def get_sec_filings(symbol: str) -> dict: """Fetch and pre-process SEC filings for a ticker. Returns a dict matching the FilingsResponse schema. Data source: yfinance `sec_filings`. FMP fallback if FMP_API_KEY is set and yfinance returns empty. """ sym = normalize_symbol(symbol) raw: list[dict] = [] # --- yfinance primary --- try: t = yf.Ticker(sym) yf_filings = t.sec_filings if yf_filings: raw = list(yf_filings) except Exception: pass # --- FMP fallback --- fmp_key = os.environ.get("FMP_API_KEY", "") if not raw and fmp_key: try: url = f"https://financialmodelingprep.com/api/v3/sec_filings/{sym}?limit=100&apikey={fmp_key}" resp = httpx.get(url, timeout=10) if resp.status_code == 200: raw = resp.json() or [] except Exception: pass # --- Normalize --- items: list[dict] = [] for r in raw: # yfinance shape: {date, type, title, edgarUrl, exhibits: {}} # FMP shape: {date, type, link, ...} date_raw = r.get("date") or r.get("filingDate") or "" form = str(r.get("type") or r.get("form") or "").strip() title = str(r.get("title") or _FORM_DESCRIPTIONS.get(form, form)).strip() # URL: prefer primary form exhibit, fall back to edgarUrl/link exhibits = r.get("exhibits") or {} filing_url = exhibits.get(form) or r.get("edgarUrl") or r.get("link") or None if not date_raw or not form: continue # Normalize date to YYYY-MM-DD try: if isinstance(date_raw, int): date_str = str(date_raw) else: date_str = str(date_raw)[:10] # validate datetime.date.fromisoformat(date_str) except (ValueError, TypeError): continue items.append({"date": date_str, "form": form, "title": title, "url": filing_url}) # Sort descending by date items.sort(key=lambda x: x["date"], reverse=True) # --- KPIs --- total = len(items) count_10k = sum(1 for f in items if f["form"] == "10-K") count_10q = sum(1 for f in items if f["form"] == "10-Q") count_8k = sum(1 for f in items if f["form"] == "8-K") distinct_forms = len({f["form"] for f in items}) kpis = { "total": total, "count_10k": count_10k, "count_10q": count_10q, "count_8k": count_8k, "distinct_forms": distinct_forms, } # --- Cadence (monthly, stacked by form bucket) --- month_map: dict[str, dict[str, int]] = defaultdict(lambda: {"count_10k": 0, "count_10q": 0, "count_8k": 0, "count_other": 0}) for f in items: m = f["date"][:7] if f["form"] == "10-K": month_map[m]["count_10k"] += 1 elif f["form"] == "10-Q": month_map[m]["count_10q"] += 1 elif f["form"] == "8-K": month_map[m]["count_8k"] += 1 else: month_map[m]["count_other"] += 1 cadence = [{"month": m, **counts} for m, counts in sorted(month_map.items())] # --- Form mix (top 8 by count) --- form_counts: dict[str, int] = defaultdict(int) for f in items: form_counts[f["form"]] += 1 sorted_forms = sorted(form_counts.items(), key=lambda x: x[1], reverse=True)[:8] form_mix = [ {"form": form, "count": cnt, "pct": round(cnt / total * 100, 1) if total else None} for form, cnt in sorted_forms ] # --- Readout --- if not items: readout = "No SEC filings found." else: dominant = sorted_forms[0][0] if sorted_forms else "—" latest = items[0]["date"] readout = ( f"Most common form: {dominant} · " f"Latest filing: {latest} · " f"{distinct_forms} distinct form type{'s' if distinct_forms != 1 else ''} on record" ) return { "kpis": kpis, "cadence": cadence, "form_mix": form_mix, "readout": readout, "filings": items, }