path: root/components/valuation.py

"""Valuation panel — key ratios, DCF model, comparable companies, analyst targets, earnings history."""
import pandas as pd
import plotly.graph_objects as go
import streamlit as st
from services.data_service import (
    get_company_info,
    get_free_cash_flow_series,
    get_analyst_price_targets,
    get_recommendations_summary,
    get_earnings_history,
    get_next_earnings_date,
    get_ebitda_from_income_stmt,
)
from services.fmp_service import (
    get_key_ratios,
    get_peers,
    get_ratios_for_tickers,
    get_historical_ratios,
    get_historical_key_metrics,
    get_analyst_estimates,
)
from services.valuation_service import run_dcf, run_ev_ebitda, compute_historical_growth_rate
from utils.formatters import fmt_ratio, fmt_pct, fmt_large, fmt_currency


FINANCIAL_SECTORS = {"Financial Services"}
FINANCIAL_INDUSTRY_KEYWORDS = (
    "bank",
    "insurance",
    "asset management",
    "capital markets",
    "financial data",
    "credit services",
    "mortgage",
    "reit",
)

INDUSTRY_PEER_MAP = {
    "consumer electronics": ["AAPL", "SONY", "DELL", "HPQ", "LOGI"],
    "software - infrastructure": ["MSFT", "ORCL", "CRM", "NOW", "SNOW"],
    "semiconductors": ["NVDA", "AMD", "AVGO", "QCOM", "INTC"],
    "internet content & information": ["GOOGL", "META", "PINS", "SNAP", "RDDT"],
    "banks - diversified": ["JPM", "BAC", "WFC", "C", "GS"],
    "credit services": ["V", "MA", "AXP", "DFS", "COF"],
    "insurance - diversified": ["BRK-B", "AIG", "ALL", "TRV", "CB"],
    "reit - industrial": ["PLD", "PSA", "EXR", "COLD", "REXR"],
}
SECTOR_PEER_MAP = {
    "Technology": ["AAPL", "MSFT", "NVDA", "ORCL", "ADBE"],
    "Communication Services": ["GOOGL", "META", "NFLX", "TMUS", "DIS"],
    "Consumer Cyclical": ["AMZN", "TSLA", "HD", "MCD", "NKE"],
    "Consumer Defensive": ["WMT", "COST", "PG", "KO", "PEP"],
    "Financial Services": ["JPM", "BAC", "WFC", "GS", "MS"],
    "Healthcare": ["LLY", "UNH", "JNJ", "MRK", "PFE"],
    "Industrials": ["GE", "CAT", "RTX", "UPS", "UNP"],
    "Energy": ["XOM", "CVX", "COP", "SLB", "EOG"],
    "Utilities": ["NEE", "DUK", "SO", "AEP", "XEL"],
    "Real Estate": ["PLD", "AMT", "EQIX", "O", "SPG"],
}


def _is_financial_company(info: dict) -> bool:
    sector = str(info.get("sector") or "").strip()
    industry = str(info.get("industry") or "").strip().lower()
    if sector in FINANCIAL_SECTORS:
        return True
    return any(keyword in industry for keyword in FINANCIAL_INDUSTRY_KEYWORDS)


def _suggest_peer_tickers(ticker: str, info: dict) -> list[str]:
    industry = str(info.get("industry") or "").strip().lower()
    sector = str(info.get("sector") or "").strip()

    candidates = []
    if industry in INDUSTRY_PEER_MAP:
        candidates.extend(INDUSTRY_PEER_MAP[industry])
    if not candidates and sector in SECTOR_PEER_MAP:
        candidates.extend(SECTOR_PEER_MAP[sector])

    candidates = [c.upper() for c in candidates if c.upper() != ticker.upper()]
    seen = set()
    deduped = []
    for c in candidates:
        if c not in seen:
            deduped.append(c)
            seen.add(c)
    return deduped[:8]


def render_valuation(ticker: str):
    tabs = st.tabs([
        "Key Ratios",
        "Historical Ratios",
        "DCF Model",
        "Comps",
        "Forward Estimates",
        "Analyst Targets",
        "Earnings History",
    ])
    tab_ratios, tab_hist, tab_dcf, tab_comps, tab_fwd, tab_analyst, tab_earnings = tabs

    with tab_ratios:
        _render_ratios(ticker)

    with tab_hist:
        try:
            _render_historical_ratios(ticker)
        except Exception as e:
            st.error(f"Historical ratios unavailable: {e}")

    with tab_dcf:
        _render_dcf(ticker)

    with tab_comps:
        _render_comps(ticker)

    with tab_fwd:
        try:
            _render_forward_estimates(ticker)
        except Exception as e:
            st.error(f"Forward estimates unavailable: {e}")

    with tab_analyst:
        try:
            _render_analyst_targets(ticker)
        except Exception as e:
            st.error(f"Analyst targets unavailable: {e}")

    with tab_earnings:
        try:
            _render_earnings_history(ticker)
        except Exception as e:
            st.error(f"Earnings history unavailable: {e}")


# ── Key Ratios ───────────────────────────────────────────────────────────────

def _render_ratios(ticker: str):
    ratios = get_key_ratios(ticker)
    info = get_company_info(ticker)

    if not ratios and not info:
        st.info("Ratio data unavailable. Check your FMP API key.")
        return

    def r(fmp_key, yf_key=None, fmt=fmt_ratio):
        val = ratios.get(fmp_key) if ratios else None
        if val is None and yf_key and info:
            val = info.get(yf_key)
        return fmt(val) if val is not None else "—"

    # Compute EV/EBITDA from income statement EBITDA — yfinance's info["ebitda"]
    # is a known bad value for many tickers (miscalculated TTM aggregation).
    def _ev_ebitda() -> str:
        # Prefer FMP if available
        fmp_val = (ratios or {}).get("enterpriseValueMultipleTTM") or (ratios or {}).get("evToEBITDATTM")
        if fmp_val is not None:
            return fmt_ratio(fmp_val)
        ev = info.get("enterpriseValue")
        ebitda = get_ebitda_from_income_stmt(ticker)
        if ev and ebitda and ebitda > 0:
            return fmt_ratio(ev / ebitda)
        return "—"

    rows = [
        ("Valuation", [
            ("P/E (TTM)", r("peRatioTTM", "trailingPE")),
            ("Forward P/E", fmt_ratio(info.get("forwardPE")) if info.get("forwardPE") is not None else "—"),
            ("P/S (TTM)", r("priceToSalesRatioTTM", "priceToSalesTrailing12Months")),
            ("P/B", r("priceToBookRatioTTM", "priceToBook")),
            ("EV/EBITDA", _ev_ebitda()),
            ("EV/Revenue", r("evToSalesTTM", "enterpriseToRevenue")),
        ]),
        ("Profitability", [
            ("Gross Margin", r("grossProfitMarginTTM", "grossMargins", fmt_pct)),
            ("Operating Margin", r("operatingProfitMarginTTM", "operatingMargins", fmt_pct)),
            ("Net Margin", r("netProfitMarginTTM", "profitMargins", fmt_pct)),
            ("ROE", r("returnOnEquityTTM", "returnOnEquity", fmt_pct)),
            ("ROA", r("returnOnAssetsTTM", "returnOnAssets", fmt_pct)),
            ("ROIC", r("returnOnInvestedCapitalTTM", fmt=fmt_pct)),
        ]),
        ("Leverage & Liquidity", [
            ("Debt/Equity", r("debtToEquityRatioTTM", "debtToEquity")),
            ("Current Ratio", r("currentRatioTTM", "currentRatio")),
            ("Quick Ratio", r("quickRatioTTM", "quickRatio")),
            ("Interest Coverage", r("interestCoverageRatioTTM")),
            ("Dividend Yield", r("dividendYieldTTM", "dividendYield", fmt_pct)),
            ("Payout Ratio", r("dividendPayoutRatioTTM", "payoutRatio", fmt_pct)),
        ]),
    ]

    for section_name, metrics in rows:
        st.markdown(f"**{section_name}**")
        cols = st.columns(6)
        for col, (label, val) in zip(cols, metrics):
            col.metric(label, val)
        st.write("")


# ── DCF Model ────────────────────────────────────────────────────────────────

def _render_dcf(ticker: str):
    info = get_company_info(ticker)

    if _is_financial_company(info):
        st.warning(
            "DCF is disabled for financial companies in Prism. Free-cash-flow and capital-structure "
            "assumptions are not directly comparable for banks, insurers, and similar businesses."
        )
        st.caption(
            "Use ratios, comps, earnings history, and analyst targets instead. A bank-specific valuation "
            "framework can be added later."
        )
        return

    shares = info.get("sharesOutstanding") or info.get("floatShares")
    current_price = info.get("currentPrice") or info.get("regularMarketPrice")
    total_debt = info.get("totalDebt") or 0.0
    cash_and_equivalents = (
        info.get("totalCash")
        or info.get("cash")
        or info.get("cashAndCashEquivalents")
        or 0.0
    )
    preferred_equity = info.get("preferredStock") or 0.0
    minority_interest = info.get("minorityInterest") or 0.0

    if not shares:
        st.info("Shares outstanding not available — DCF cannot be computed.")
        return

    fcf_series = get_free_cash_flow_series(ticker)
    if fcf_series.empty:
        st.info("Free cash flow data unavailable.")
        return

    # Compute historical growth rate for slider default + caption reference
    hist_growth = compute_historical_growth_rate(fcf_series)
    hist_growth_pct = hist_growth * 100 if hist_growth is not None else 5.0
    slider_default = float(max(-20.0, min(30.0, hist_growth_pct)))

    st.markdown("**Assumptions**")
    col1, col2, col3, col4 = st.columns(4)
    with col1:
        wacc = st.slider("WACC (%)", min_value=5.0, max_value=20.0, value=10.0, step=0.5) / 100
    with col2:
        terminal_growth = st.slider(
            "Terminal Growth (%)", min_value=0.5, max_value=5.0, value=2.5, step=0.5
        ) / 100
    with col3:
        projection_years = st.slider("Projection Years", min_value=3, max_value=10, value=5, step=1)
    with col4:
        fcf_growth_pct = st.slider(
            "FCF Growth (%)",
            min_value=-20.0,
            max_value=30.0,
            value=round(slider_default, 1),
            step=0.5,
            help=f"Historical median: {hist_growth_pct:.1f}%. Drag to override.",
        )

    st.caption(f"Historical FCF growth (median): **{hist_growth_pct:.1f}%**")

    result = run_dcf(
        fcf_series=fcf_series,
        shares_outstanding=shares,
        wacc=wacc,
        terminal_growth=terminal_growth,
        projection_years=projection_years,
        growth_rate_override=fcf_growth_pct / 100,
        total_debt=total_debt,
        cash_and_equivalents=cash_and_equivalents,
        preferred_equity=preferred_equity,
        minority_interest=minority_interest,
    )

    if not result:
        st.warning("Insufficient data to run DCF model.")
        return
    if result.get("error"):
        st.warning(result["error"])
        return

    iv = result["intrinsic_value_per_share"]

    m1, m2, m3, m4 = st.columns(4)
    m1.metric("Equity Value / Share", fmt_currency(iv))
    if current_price:
        upside = (iv - current_price) / current_price
        m2.metric("Current Price", fmt_currency(current_price))
        m3.metric("Upside / Downside", f"{upside * 100:+.1f}%", delta=f"{upside * 100:+.1f}%")
    m4.metric("FCF Growth Used", f"{result['growth_rate_used'] * 100:.1f}%")

    st.caption(
        "DCF is modeled on firm-level free cash flow, so enterprise value is bridged to equity value "
        "using cash and debt before calculating per-share value."
    )

    bridge1, bridge2, bridge3, bridge4 = st.columns(4)
    bridge1.metric("Enterprise Value", fmt_large(result["enterprise_value"]))
    bridge2.metric("Net Debt", fmt_large(result["net_debt"]))
    bridge3.metric("Equity Value", fmt_large(result["equity_value"]))
    bridge4.metric("Terminal Value PV", fmt_large(result["terminal_value_pv"]))

    st.write("")

    years = [f"Year {y}" for y in result["years"]]
    discounted = result["discounted_fcfs"]
    terminal_pv = result["terminal_value_pv"]

    fig = go.Figure(go.Bar(
        x=years + ["Terminal Value"],
        y=[(v / 1e9) for v in discounted] + [terminal_pv / 1e9],
        marker_color=["#4F8EF7"] * len(years) + ["#F7A24F"],
        text=[f"${v / 1e9:.2f}B" for v in discounted] + [f"${terminal_pv / 1e9:.2f}B"],
        textposition="outside",
    ))
    fig.update_layout(
        title="Enterprise Value Build: PV of Forecast FCFs + Terminal Value (Billions)",
        yaxis_title="USD (Billions)",
        plot_bgcolor="rgba(0,0,0,0)",
        paper_bgcolor="rgba(0,0,0,0)",
        margin=dict(l=0, r=0, t=40, b=0),
        height=360,
    )
    st.plotly_chart(fig, use_container_width=True)

    # ── EV/EBITDA Valuation ───────────────────────────────────────────────────
    st.divider()
    st.markdown("**EV/EBITDA Valuation**")

    # Use income statement EBITDA — info["ebitda"] is unreliable in yfinance
    ebitda = get_ebitda_from_income_stmt(ticker) or info.get("ebitda")
    total_debt = info.get("totalDebt") or 0.0
    total_cash = info.get("totalCash") or 0.0
    ev_ebitda_current = info.get("enterpriseToEbitda")

    if not ebitda or ebitda <= 0:
        st.info("EBITDA not available or negative — EV/EBITDA valuation cannot be computed.")
    else:
        default_multiple = float(ev_ebitda_current) if ev_ebitda_current else 15.0
        default_multiple = max(1.0, min(50.0, round(default_multiple, 1)))

        ev_col1, ev_col2 = st.columns([1, 3])
        with ev_col1:
            help_text = (
                f"Current market multiple: {ev_ebitda_current:.1f}x"
                if ev_ebitda_current else "Current multiple unavailable"
            )
            target_multiple = st.slider(
                "Target EV/EBITDA",
                min_value=1.0,
                max_value=50.0,
                value=default_multiple,
                step=0.5,
                help=help_text,
            )

        ev_result = run_ev_ebitda(
            ebitda=float(ebitda),
            total_debt=float(total_debt),
            total_cash=float(total_cash),
            shares_outstanding=float(shares),
            target_multiple=target_multiple,
        )

        if ev_result:
            imp_price = ev_result["implied_price_per_share"]
            ev_m1, ev_m2, ev_m3, ev_m4 = st.columns(4)
            ev_m1.metric("Implied Price (EV/EBITDA)", fmt_currency(imp_price))
            if current_price:
                ev_upside = (imp_price - current_price) / current_price
                ev_m2.metric("Current Price", fmt_currency(current_price))
                ev_m3.metric(
                    "Upside / Downside",
                    f"{ev_upside * 100:+.1f}%",
                    delta=f"{ev_upside * 100:+.1f}%",
                )
            ev_m4.metric("Implied EV", fmt_large(ev_result["implied_ev"]))
            st.caption(
                f"EBITDA: {fmt_large(ebitda)}  ·  "
                f"Net Debt: {fmt_large(ev_result['net_debt'])}  ·  "
                f"Equity Value: {fmt_large(ev_result['equity_value'])}"
            )
        else:
            st.warning("Could not compute EV/EBITDA valuation.")


# ── Comps Table ──────────────────────────────────────────────────────────────

def _render_comps(ticker: str):
    info = get_company_info(ticker)
    auto_peers = get_peers(ticker)
    suggested_peers = _suggest_peer_tickers(ticker, info)

    default_peer_string = ", ".join(auto_peers or suggested_peers)
    manual_peer_string = st.text_input(
        "Peer tickers",
        value=default_peer_string,
        help="Edit the comparable-company set manually. Comma-separated tickers.",
        key=f"peer_input_{ticker.upper()}",
    )

    if auto_peers:
        st.caption("Using FMP-discovered peers.")
    elif suggested_peers:
        st.caption("Using Prism fallback peers based on sector/industry. Edit them if you want a tighter comp set.")
    else:
        st.caption("No automatic peer set found. Enter peer tickers manually to build a comps table.")

    manual_peers = [p.strip().upper() for p in manual_peer_string.split(",") if p.strip()]
    peer_list = []
    seen = {ticker.upper()}
    for peer in manual_peers:
        if peer not in seen:
            peer_list.append(peer)
            seen.add(peer)

    all_tickers = [ticker.upper()] + peer_list[:9]

    with st.spinner("Loading comps..."):
        ratios_list = get_ratios_for_tickers(all_tickers)

    if not ratios_list:
        st.info("Could not load ratios for the selected peer companies.")
        return

    display_cols = {
        "symbol": "Ticker",
        "peRatioTTM": "P/E",
        "priceToSalesRatioTTM": "P/S",
        "priceToBookRatioTTM": "P/B",
        "enterpriseValueMultipleTTM": "EV/EBITDA",
        "evToEBITDATTM": "EV/EBITDA",
        "netProfitMarginTTM": "Net Margin",
        "returnOnEquityTTM": "ROE",
        "debtToEquityRatioTTM": "D/E",
    }

    df = pd.DataFrame(ratios_list)
    if "enterpriseValueMultipleTTM" not in df.columns and "evToEBITDATTM" in df.columns:
        df["enterpriseValueMultipleTTM"] = df["evToEBITDATTM"]
    if "debtToEquityRatioTTM" not in df.columns and "debtEquityRatioTTM" in df.columns:
        df["debtToEquityRatioTTM"] = df["debtEquityRatioTTM"]

    available = [c for c in ["symbol", "peRatioTTM", "priceToSalesRatioTTM", "priceToBookRatioTTM", "enterpriseValueMultipleTTM", "netProfitMarginTTM", "returnOnEquityTTM", "debtToEquityRatioTTM"] if c in df.columns]
    df = df[available].rename(columns=display_cols)

    pct_cols = {"Net Margin", "ROE"}
    for col in df.columns:
        if col == "Ticker":
            continue
        if col in pct_cols:
            df[col] = df[col].apply(lambda v: fmt_pct(v) if v is not None else "—")
        else:
            df[col] = df[col].apply(lambda v: fmt_ratio(v) if v is not None else "—")

    def highlight_subject(row):
        if row["Ticker"] == ticker.upper():
            return ["background-color: rgba(79,142,247,0.15)"] * len(row)
        return [""] * len(row)

    st.dataframe(
        df.style.apply(highlight_subject, axis=1),
        use_container_width=True,
        hide_index=True,
    )


# ── Analyst Targets ──────────────────────────────────────────────────────────

def _render_analyst_targets(ticker: str):
    targets = get_analyst_price_targets(ticker)
    recs = get_recommendations_summary(ticker)

    if not targets and (recs is None or recs.empty):
        st.info("Analyst data unavailable for this ticker.")
        return

    if targets:
        st.markdown("**Analyst Price Targets**")
        current = targets.get("current")
        mean_t = targets.get("mean")

        t1, t2, t3, t4, t5 = st.columns(5)
        t1.metric("Low", fmt_currency(targets.get("low")))
        t2.metric("Mean", fmt_currency(mean_t))
        t3.metric("Median", fmt_currency(targets.get("median")))
        t4.metric("High", fmt_currency(targets.get("high")))
        if current and mean_t:
            upside = (mean_t - current) / current
            t5.metric("Upside to Mean", f"{upside * 100:+.1f}%", delta=f"{upside * 100:+.1f}%")
        else:
            t5.metric("Current Price", fmt_currency(current))

        st.write("")

    if recs is not None and not recs.empty:
        st.markdown("**Analyst Recommendations (Current Month)**")

        current_row = recs[recs["period"] == "0m"] if "period" in recs.columns else pd.DataFrame()
        if current_row.empty:
            current_row = recs.iloc[[0]]

        row = current_row.iloc[0]
        counts = {
            "Strong Buy": int(row.get("strongBuy", 0)),
            "Buy": int(row.get("buy", 0)),
            "Hold": int(row.get("hold", 0)),
            "Sell": int(row.get("sell", 0)),
            "Strong Sell": int(row.get("strongSell", 0)),
        }
        total = sum(counts.values())

        cols = st.columns(5)
        for col, (label, count) in zip(cols, counts.items()):
            pct = f"{count / total * 100:.0f}%" if total > 0 else "—"
            col.metric(label, str(count), delta=pct, delta_color="off")

        st.write("")

        colors = ["#2ecc71", "#82e0aa", "#f0b27a", "#e59866", "#e74c3c"]
        fig = go.Figure(go.Bar(
            x=list(counts.keys()),
            y=list(counts.values()),
            marker_color=colors,
            text=list(counts.values()),
            textposition="outside",
        ))
        fig.update_layout(
            title="Analyst Recommendation Distribution",
            yaxis_title="# Analysts",
            plot_bgcolor="rgba(0,0,0,0)",
            paper_bgcolor="rgba(0,0,0,0)",
            margin=dict(l=0, r=0, t=40, b=0),
            height=280,
        )
        st.plotly_chart(fig, use_container_width=True)


# ── Earnings History ──────────────────────────────────────────────────────────

def _render_earnings_history(ticker: str):
    eh = get_earnings_history(ticker)
    next_date = get_next_earnings_date(ticker)

    if next_date:
        st.info(f"Next earnings date: **{next_date}**")

    if eh is None or eh.empty:
        st.info("Earnings history unavailable for this ticker.")
        return

    st.markdown("**Historical EPS: Actual vs. Estimate**")

    df = eh.copy().sort_index(ascending=False)
    df.index = df.index.astype(str)
    df.index.name = "Quarter"

    display = pd.DataFrame(index=df.index)
    display["EPS Actual"] = df["epsActual"].apply(fmt_currency)
    display["EPS Estimate"] = df["epsEstimate"].apply(fmt_currency)
    display["Surprise"] = df["epsDifference"].apply(
        lambda v: f"{'+' if float(v) >= 0 else ''}{fmt_currency(v)}"
        if pd.notna(v) else "—"
    )
    display["Surprise %"] = df["surprisePercent"].apply(
        lambda v: f"{float(v) * 100:+.2f}%" if pd.notna(v) else "—"
    )

    def highlight_surprise(row):
        try:
            pct_str = row["Surprise %"].replace("%", "").replace("+", "")
            val = float(pct_str)
            color = "rgba(46,204,113,0.15)" if val >= 0 else "rgba(231,76,60,0.15)"
            return ["", "", f"background-color: {color}", f"background-color: {color}"]
        except Exception:
            return [""] * len(row)

    st.dataframe(
        display.style.apply(highlight_surprise, axis=1),
        use_container_width=True,
        hide_index=False,
    )
    st.download_button(
        "Download CSV",
        display.to_csv().encode(),
        file_name=f"{ticker.upper()}_earnings_history.csv",
        mime="text/csv",
        key=f"dl_earnings_{ticker}",
    )

    # EPS chart — oldest to newest
    df_chart = eh.sort_index()
    fig = go.Figure()
    fig.add_trace(go.Scatter(
        x=df_chart.index.astype(str),
        y=df_chart["epsActual"],
        name="Actual EPS",
        mode="lines+markers",
        line=dict(color="#4F8EF7", width=2),
    ))
    fig.add_trace(go.Scatter(
        x=df_chart.index.astype(str),
        y=df_chart["epsEstimate"],
        name="Estimated EPS",
        mode="lines+markers",
        line=dict(color="#F7A24F", width=2, dash="dash"),
    ))
    fig.update_layout(
        title="EPS: Actual vs. Estimate",
        yaxis_title="EPS ($)",
        plot_bgcolor="rgba(0,0,0,0)",
        paper_bgcolor="rgba(0,0,0,0)",
        margin=dict(l=0, r=0, t=40, b=0),
        height=280,
        legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1),
    )
    st.plotly_chart(fig, use_container_width=True)


# ── Historical Ratios ────────────────────────────────────────────────────────

_HIST_RATIO_OPTIONS = {
    "P/E": ("peRatio", "priceToEarningsRatio", None),
    "P/B": ("priceToBookRatio", None, None),
    "P/S": ("priceToSalesRatio", None, None),
    "EV/EBITDA": ("enterpriseValueMultiple", "evToEBITDA", None),
    "Net Margin": ("netProfitMargin", None, "pct"),
    "Operating Margin": ("operatingProfitMargin", None, "pct"),
    "Gross Margin": ("grossProfitMargin", None, "pct"),
    "ROE": ("returnOnEquity", None, "pct"),
    "ROA": ("returnOnAssets", None, "pct"),
    "Debt/Equity": ("debtEquityRatio", None, None),
}

_CHART_COLORS = [
    "#4F8EF7", "#F7A24F", "#2ecc71", "#e74c3c",
    "#9b59b6", "#1abc9c", "#f39c12", "#e67e22",
]


def _extract_hist_series(rows: list[dict], primary: str, alt: str | None) -> dict[str, float]:
    """Extract {year: value} from FMP historical rows."""
    out = {}
    for row in rows:
        date = str(row.get("date", ""))[:4]
        val = row.get(primary)
        if val is None and alt:
            val = row.get(alt)
        if val is not None:
            try:
                out[date] = float(val)
            except (TypeError, ValueError):
                pass
    return out


def _render_historical_ratios(ticker: str):
    with st.spinner("Loading historical ratios…"):
        ratio_rows = get_historical_ratios(ticker)
        metric_rows = get_historical_key_metrics(ticker)

    if not ratio_rows and not metric_rows:
        st.info("Historical ratio data unavailable.")
        return

    # Merge both lists by date
    combined: dict[str, dict] = {}
    for row in ratio_rows + metric_rows:
        date = str(row.get("date", ""))[:4]
        if date:
            combined.setdefault(date, {}).update(row)

    merged_rows = [{"date": d, **v} for d, v in sorted(combined.items(), reverse=True)]

    selected = st.multiselect(
        "Metrics to plot",
        options=list(_HIST_RATIO_OPTIONS.keys()),
        default=["P/E", "EV/EBITDA", "Net Margin", "ROE"],
    )

    if not selected:
        st.info("Select at least one metric to plot.")
        return

    fig = go.Figure()
    for i, label in enumerate(selected):
        primary, alt, fmt = _HIST_RATIO_OPTIONS[label]
        series = _extract_hist_series(merged_rows, primary, alt)
        if not series:
            continue
        years = sorted(series.keys())
        values = [series[y] * (100 if fmt == "pct" else 1) for y in years]
        y_label = f"{label} (%)" if fmt == "pct" else label
        fig.add_trace(go.Scatter(
            x=years,
            y=values,
            name=y_label,
            mode="lines+markers",
            line=dict(color=_CHART_COLORS[i % len(_CHART_COLORS)], width=2),
        ))

    fig.update_layout(
        title="Historical Ratios & Metrics",
        xaxis_title="Year",
        plot_bgcolor="rgba(0,0,0,0)",
        paper_bgcolor="rgba(0,0,0,0)",
        margin=dict(l=0, r=0, t=40, b=0),
        height=380,
        legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1),
        hovermode="x unified",
    )
    st.plotly_chart(fig, use_container_width=True)

    # Raw data table
    with st.expander("Raw data"):
        display_cols = {}
        for label in selected:
            primary, alt, fmt = _HIST_RATIO_OPTIONS[label]
            display_cols[label] = (primary, alt, fmt)

        table_rows = []
        for row in merged_rows:
            r: dict = {"Year": str(row.get("date", ""))[:4]}
            for label, (primary, alt, fmt) in display_cols.items():
                val = row.get(primary) or (row.get(alt) if alt else None)
                if val is not None:
                    try:
                        v = float(val)
                        r[label] = f"{v * 100:.2f}%" if fmt == "pct" else f"{v:.2f}x"
                    except (TypeError, ValueError):
                        r[label] = "—"
                else:
                    r[label] = "—"
            table_rows.append(r)

        if table_rows:
            st.dataframe(pd.DataFrame(table_rows), use_container_width=True, hide_index=True)


# ── Forward Estimates ────────────────────────────────────────────────────────

def _render_forward_estimates(ticker: str):
    with st.spinner("Loading forward estimates…"):
        estimates = get_analyst_estimates(ticker)

    annual = estimates.get("annual", [])
    quarterly = estimates.get("quarterly", [])

    if not annual and not quarterly:
        st.info("Forward estimates unavailable. Requires FMP API key.")
        return

    info = get_company_info(ticker)
    current_price = info.get("currentPrice") or info.get("regularMarketPrice")

    tab_ann, tab_qtr = st.tabs(["Annual", "Quarterly"])

    def _build_estimates_table(rows: list[dict]) -> pd.DataFrame:
        table = []
        for row in sorted(rows, key=lambda r: str(r.get("date", ""))):
            date = str(row.get("date", ""))[:7]
            # FMP stable endpoint uses revenueAvg / epsAvg (no "estimated" prefix)
            rev_avg = row.get("revenueAvg") or row.get("estimatedRevenueAvg")
            rev_lo = row.get("revenueLow") or row.get("estimatedRevenueLow")
            rev_hi = row.get("revenueHigh") or row.get("estimatedRevenueHigh")
            eps_avg = row.get("epsAvg") or row.get("estimatedEpsAvg")
            eps_lo = row.get("epsLow") or row.get("estimatedEpsLow")
            eps_hi = row.get("epsHigh") or row.get("estimatedEpsHigh")
            ebitda_avg = row.get("ebitdaAvg") or row.get("estimatedEbitdaAvg")
            num_analysts = row.get("numAnalystsRevenue") or row.get("numAnalystsEps") or row.get("numberAnalystEstimatedRevenue") or row.get("numberAnalysts")
            table.append({
                "Period": date,
                "Rev Low": fmt_large(rev_lo) if rev_lo else "—",
                "Rev Avg": fmt_large(rev_avg) if rev_avg else "—",
                "Rev High": fmt_large(rev_hi) if rev_hi else "—",
                "EPS Low": fmt_currency(eps_lo) if eps_lo else "—",
                "EPS Avg": fmt_currency(eps_avg) if eps_avg else "—",
                "EPS High": fmt_currency(eps_hi) if eps_hi else "—",
                "EBITDA Avg": fmt_large(ebitda_avg) if ebitda_avg else "—",
                "# Analysts": str(int(num_analysts)) if num_analysts else "—",
            })
        return pd.DataFrame(table)

    def _render_eps_chart(rows: list[dict], title: str):
        """Overlay historical EPS actuals with forward estimates."""
        eh = get_earnings_history(ticker)
        fwd_dates, fwd_eps = [], []
        for row in sorted(rows, key=lambda r: str(r.get("date", ""))):
            date = str(row.get("date", ""))[:7]
            eps = row.get("epsAvg") or row.get("estimatedEpsAvg")
            eps_lo = row.get("epsLow") or row.get("estimatedEpsLow")
            eps_hi = row.get("epsHigh") or row.get("estimatedEpsHigh")
            if eps is not None:
                fwd_dates.append(date)
                fwd_eps.append(float(eps))

        fig = go.Figure()

        if eh is not None and not eh.empty:
            hist = eh.sort_index()
            fig.add_trace(go.Scatter(
                x=hist.index.astype(str),
                y=hist["epsActual"],
                name="EPS Actual",
                mode="lines+markers",
                line=dict(color="#4F8EF7", width=2),
            ))

        if fwd_dates:
            # Low/high band
            fwd_lo = [float(r.get("epsLow") or r.get("estimatedEpsLow")) for r in sorted(rows, key=lambda r: str(r.get("date", "")))
                      if (r.get("epsLow") or r.get("estimatedEpsLow")) is not None]
            fwd_hi = [float(r.get("epsHigh") or r.get("estimatedEpsHigh")) for r in sorted(rows, key=lambda r: str(r.get("date", "")))
                      if (r.get("epsHigh") or r.get("estimatedEpsHigh")) is not None]

            if fwd_lo and fwd_hi and len(fwd_lo) == len(fwd_dates):
                fig.add_trace(go.Scatter(
                    x=fwd_dates + fwd_dates[::-1],
                    y=fwd_hi + fwd_lo[::-1],
                    fill="toself",
                    fillcolor="rgba(247,162,79,0.15)",
                    line=dict(color="rgba(0,0,0,0)"),
                    name="Est. Range",
                    hoverinfo="skip",
                ))

            fig.add_trace(go.Scatter(
                x=fwd_dates,
                y=fwd_eps,
                name="EPS Est. (Avg)",
                mode="lines+markers",
                line=dict(color="#F7A24F", width=2, dash="dash"),
            ))

        fig.update_layout(
            title=title,
            yaxis_title="EPS ($)",
            plot_bgcolor="rgba(0,0,0,0)",
            paper_bgcolor="rgba(0,0,0,0)",
            margin=dict(l=0, r=0, t=40, b=0),
            height=320,
            hovermode="x unified",
            legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1),
        )
        st.plotly_chart(fig, use_container_width=True)

    with tab_ann:
        if annual:
            df = _build_estimates_table(annual)
            st.dataframe(df, use_container_width=True, hide_index=True)
            st.write("")
            _render_eps_chart(annual, "Annual EPS: Historical Actuals + Forward Estimates")
        else:
            st.info("No annual estimates available.")

    with tab_qtr:
        if quarterly:
            df = _build_estimates_table(quarterly)
            st.dataframe(df, use_container_width=True, hide_index=True)
            st.write("")
            _render_eps_chart(quarterly, "Quarterly EPS: Historical Actuals + Forward Estimates")
        else:
            st.info("No quarterly estimates available.")