path: root/backend/app/services/data_service.py

"""yfinance wrapper for Prism v2 Overview data."""
from __future__ import annotations

import math
import os
import statistics
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)

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)
    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,
        },
    }