dashboard/signals_view.py

"""Evidence & Deltas — source-backed signals in one ranked, filterable stream.

Replaces the former MD&A / Earnings Call / Risks / Quality & Tone tabs:
their content is normalized by dashboard/signal_feed.py and rendered with
the single components.signal_card renderer. Also hosts the one global
semantic search (filings + transcripts) and the on-demand news loader.
"""
from __future__ import annotations

import streamlit as st

from dashboard import fmt_period
from dashboard.i18n import t
from dashboard.theme import (
    GREEN, RED, GRAY, AMBER,
    BG_MUTED, BORDER, TEXT, TEXT_MUTED, TEXT_FAINT,
    BULL_BG, BEAR_BG,
    FS_PAGE, FS_META, FS_EYEBROW,
)
from dashboard.components import signal_card, section_header
from dashboard import provenance
from dashboard.signal_feed import (
    KIND_PRIORITY, Signal, aggregate_sentiment, build_feed, global_band, rank_feed,
)

_FILTER_KEYS = (
    "signals_filter_type", "signals_filter_stance",
    "signals_filter_source", "signals_high_sig", "signals_search",
)

_SENT_SECTION_LABELS = {
    "metrics": "sent_metrics", "guidance": "sent_guidance", "mda": "sent_mda",
    "earnings_call": "sent_call", "news": "sent_news",
}

_TONE_STYLES = {
    "bull": (GREEN, BULL_BG), "bear": (RED, BEAR_BG), "neutral": (GRAY, BG_MUTED),
}


# ── Sentiment strip ───────────────────────────────────────────────────────────

def _gauge_html(score: int) -> str:
    """Red→green gradient bar with a dot at `score` (-2..+2)."""
    pct = (score + 2) / 4 * 90 + 5
    dot_color = GREEN if score > 0 else (RED if score < 0 else GRAY)
    return (
        f'<div style="margin:4px 0 2px;height:6px;border-radius:99px;position:relative;'
        f'background:linear-gradient(90deg,#fecaca 0%,#fde68a 38%,#d1fae5 62%,#6ee7b7 100%);">'
        f'<div style="position:absolute;top:50%;left:{pct:.1f}%;transform:translate(-50%,-50%);'
        f'width:12px;height:12px;border-radius:50%;background:{dot_color};'
        f'border:2px solid #fff;box-shadow:0 0 0 1.5px {dot_color};"></div>'
        f'</div>'
    )


def _render_sentiment_strip(sentiment: dict | None) -> None:
    if not sentiment:
        return
    agg = aggregate_sentiment(sentiment)
    if agg is None:
        return

    band_key, tone = global_band(agg)
    band_color, band_bg = _TONE_STYLES[tone]
    sign = "+" if agg >= 0 else ""

    cells_html = ""
    for key, label_key in _SENT_SECTION_LABELS.items():
        label = t(label_key)
        section = sentiment.get(key)
        if section is None or section.get("score") is None:
            cells_html += (
                f'<div style="padding:10px 12px;border-right:1px solid {BORDER};background:#fff;">'
                f'<div style="font-size:{FS_EYEBROW};font-weight:700;text-transform:uppercase;'
                f'letter-spacing:0.07em;color:{TEXT_MUTED};margin-bottom:6px;">{label}</div>'
                f'<div style="font-size:0.68rem;color:{TEXT_MUTED};">—</div>'
                f'</div>'
            )
        else:
            score = section["score"]
            fg = GREEN if score > 0 else (RED if score < 0 else GRAY)
            rationale = section.get("rationale", "")
            cells_html += (
                f'<div style="padding:10px 12px;border-right:1px solid {BORDER};background:#fff;">'
                f'<div style="font-size:{FS_EYEBROW};font-weight:700;text-transform:uppercase;'
                f'letter-spacing:0.07em;color:{TEXT_MUTED};margin-bottom:4px;">{label}</div>'
                f'{_gauge_html(score)}'
                f'<div style="font-size:0.68rem;font-weight:600;color:{fg};margin-bottom:4px;">'
                f'{section.get("label", "")}</div>'
                f'<div style="font-size:{FS_META};color:{TEXT_MUTED};line-height:1.4;">{rationale}</div>'
                f'</div>'
            )

    st.markdown(
        f'<div class="primer-card" style="display:grid;grid-template-columns:130px repeat(5,1fr);'
        f'border:1px solid {BORDER};border-radius:12px;overflow:hidden;margin-bottom:8px;">'
        f'<div style="background:{band_bg};border-right:2px solid {band_color};'
        f'padding:12px 14px;display:flex;flex-direction:column;justify-content:center;">'
        f'<div style="font-size:0.56rem;font-weight:700;text-transform:uppercase;'
        f'letter-spacing:0.09em;color:{band_color};margin-bottom:4px;">{t("print_sentiment")}</div>'
        f'<div style="font-size:1.75rem;font-weight:800;line-height:1;color:{band_color};">'
        f'{sign}{agg:.1f}</div>'
        f'<div style="font-size:0.72rem;font-weight:600;color:{band_color};margin-top:3px;">'
        f'{t(band_key)}</div>'
        f'</div>'
        f'{cells_html}'
        f'</div>',
        unsafe_allow_html=True,
    )


def _sentiment_display_allowed(brief: dict) -> bool:
    """Fail closed until aggregate sentiment has an explicit calibration flag."""
    policy = brief.get("display_policy") or {}
    return isinstance(policy, dict) and policy.get("sentiment_calibrated") is True


def _render_experimental_notice() -> None:
    st.markdown(
        f'<div style="background:#fffbeb;border:1px solid #fde68a;'
        f'border-left:3px solid {AMBER};border-radius:0 8px 8px 0;'
        f'padding:9px 12px;margin:0 0 16px;font-size:{FS_META};'
        f'line-height:1.5;color:{TEXT_MUTED};">'
        f'<strong style="color:{TEXT};">Validation required.</strong> '
        f'AI hypotheses are experimental; delta detectors are heuristic. '
        f'Validate both against the cited source text.</div>',
        unsafe_allow_html=True,
    )


# ── Filters ───────────────────────────────────────────────────────────────────

def _reset_filters_on_ticker_change(ticker: str) -> None:
    """Pills options are feed-derived; clear stale selections when the active
    ticker changes (must run before the filter widgets instantiate)."""
    if st.session_state.get("_signals_ticker") != ticker:
        for key in _FILTER_KEYS:
            st.session_state.pop(key, None)
        st.session_state["_signals_ticker"] = ticker


def _options_with_selection(present: list[str], state_key: str) -> list[str]:
    """Feed-derived options ∪ current selection — a selected value absent from
    the new options would otherwise raise inside st.pills."""
    selected = st.session_state.get(state_key) or []
    extras = [s for s in selected if s not in present]
    return present + extras


def _render_filters(feed: list[Signal]) -> tuple[list[str], list[str], list[str], bool]:
    kinds_present = sorted(
        {s.kind for s in feed}, key=lambda k: KIND_PRIORITY.get(k, 99)
    )
    stances_present = [
        s for s in ("bull", "bear", "neutral", "mixed")
        if any(sig.stance == s for sig in feed)
    ]
    sources_present = sorted({s.source for s in feed if s.source})

    sel_kinds = st.pills(
        t("filter_type"),
        options=_options_with_selection(kinds_present, "signals_filter_type"),
        selection_mode="multi",
        format_func=lambda k: t(f"kind_{k}"),
        key="signals_filter_type",
    )

    col_stance, col_source, col_sig = st.columns([2, 2, 1])
    with col_stance:
        sel_stances = st.pills(
            t("filter_stance"),
            options=_options_with_selection(stances_present, "signals_filter_stance"),
            selection_mode="multi",
            format_func=lambda s: t(f"stance_{s}"),
            key="signals_filter_stance",
        )
    with col_source:
        sel_sources = st.pills(
            t("filter_source"),
            options=_options_with_selection(sources_present, "signals_filter_source"),
            selection_mode="multi",
            key="signals_filter_source",
        )
    with col_sig:
        high_only = st.toggle(t("filter_high_sig"), key="signals_high_sig")

    return sel_kinds or [], sel_stances or [], sel_sources or [], bool(high_only)


def _apply_filters(
    feed: list[Signal],
    kinds: list[str], stances: list[str], sources: list[str], high_only: bool,
) -> list[Signal]:
    out = feed
    if kinds:
        out = [s for s in out if s.kind in kinds]
    if stances:
        out = [s for s in out if s.stance in stances]
    if sources:
        out = [s for s in out if s.source in sources]
    if high_only:
        out = [s for s in out if s.significance == "HIGH"]
    return out


# ── Global semantic search ────────────────────────────────────────────────────

def _render_search(ticker: str) -> None:
    st.markdown(section_header(t("search_label"), accent=AMBER), unsafe_allow_html=True)

    from storage.metrics_db import get_all_metrics
    rows = get_all_metrics(ticker)
    periods = [r.get("period", "") for r in rows if r.get("period")]

    col_q, col_p = st.columns([3, 2])
    with col_q:
        query = st.text_input(
            t("search_label"), placeholder=t("search_placeholder"),
            key="signals_search", label_visibility="collapsed",
        )
    with col_p:
        period_options = [""] + periods
        selected_period = st.selectbox(
            t("filter_source"), options=period_options,
            format_func=lambda p: fmt_period(p) if p else "—",
            key="signals_search_period", label_visibility="collapsed",
        )

    if not query:
        return

    from storage import vector_store
    from storage.reranker import rerank

    with st.spinner(t("chat_searching")):
        union: list[dict] = []
        for collection in ("filings", "transcripts"):
            hits = vector_store.search(
                collection, query, ticker, n_results=4,
                period=selected_period or None,
            )
            for h in hits:
                h["collection"] = collection
            union += hits
        # vector_store results carry no scores — a second cross-encoder pass
        # over the union is the only way to merge the two collections.
        merged = rerank(query, union, top_k=6)

    if not merged:
        st.warning(t("search_no_results"))
        return

    for r in merged:
        m = r.get("metadata", {})
        icon = "📄" if r.get("collection") == "filings" else "🎙️"
        context = m.get("chunk_context") or " · ".join(
            str(v) for v in (
                m.get("source", r.get("collection", "")),
                m.get("section", ""),
                m.get("filing_date", "") or m.get("date", ""),
            ) if v
        )
        with st.expander(f"{icon} {context}", expanded=False):
            st.markdown(
                f'<div style="font-size:0.85rem;line-height:1.7;color:{TEXT};">{r["text"]}</div>',
                unsafe_allow_html=True,
            )


# ── News (on-demand, Tavily) ──────────────────────────────────────────────────

def _load_news(ticker: str) -> None:
    import os
    try:
        from tavily import TavilyClient
        from storage.metrics_db import get_all_metrics

        api_key = os.environ.get("TAVILY_API_KEY")
        if not api_key:
            st.error("TAVILY_API_KEY not set.")
            return

        rows = get_all_metrics(ticker)
        company_name = rows[0]["company_name"] if rows else ticker
        latest_filing_date = rows[0]["filing_date"] if rows else None

        with st.spinner(t("chat_searching")):
            client = TavilyClient(api_key=api_key)
            results = client.search(
                query=f"{ticker} {company_name} earnings results guidance",
                max_results=6,
                search_depth="basic",
                days=60,
                topic="news",
            ).get("results", [])

        if latest_filing_date:
            results = [r for r in results if r.get("published_date", "") >= latest_filing_date]

        if not results:
            st.info(t("news_none"))
            return

        for item in results:
            signal_card(Signal(
                kind="news",
                headline=item.get("title", ""),
                body=(item.get("content", "") or "")[:300],
                source="news",
                significance="LOW",
                extra={
                    "url": item.get("url", ""),
                    "date": (item.get("published_date", "") or "")[:10],
                },
            ))

    except Exception as e:
        st.error(f"Failed to load news: {e}")


# ── Main render entrypoint ────────────────────────────────────────────────────

def render(brief: dict, ticker: str) -> None:
    filing_date = brief.get("filing_date", "")
    company_name = brief.get("company_name", "")

    subtitle_parts: list[str] = []
    if company_name:
        subtitle_parts.append(company_name)
    if ticker and ticker not in company_name:
        subtitle_parts.append(ticker)
    if filing_date:
        subtitle_parts.append(filing_date)
    subtitle_parts.append(t("signals_subtitle"))
    subtitle = " · ".join(subtitle_parts)

    # ── 1. Page header ───────────────────────────────────────────────────────
    st.markdown(
        f'<div style="margin-bottom:16px;padding-bottom:12px;border-bottom:1px solid {BORDER};">'
        f'<div style="font-size:{FS_PAGE};font-weight:700;color:{TEXT};letter-spacing:-0.02em;">'
        f'Evidence &amp; Deltas</div>'
        f'<div style="font-size:{FS_META};color:{TEXT_FAINT};margin-top:4px;">{subtitle}</div>'
        f'</div>',
        unsafe_allow_html=True,
    )

    # ── 2. Validation notice; aggregate sentiment remains fail-closed ────────
    provenance.render_card(ticker, brief)
    _render_experimental_notice()
    if _sentiment_display_allowed(brief):
        _render_sentiment_strip(brief.get("sentiment"))

    # ── 3. Filters + ranked stream ───────────────────────────────────────────
    _reset_filters_on_ticker_change(ticker)
    feed = rank_feed(build_feed(brief))

    if feed:
        kinds, stances, sources, high_only = _render_filters(feed)
        filtered = _apply_filters(feed, kinds, stances, sources, high_only)

        st.caption(f"{len(filtered)} / {len(feed)}")
        if filtered:
            for sig in filtered:
                signal_card(sig)
        else:
            st.info(t("signals_empty"))

    # ── 4. Global semantic search ────────────────────────────────────────────
    _render_search(ticker)

    # ── 5. Latest news (on demand) ───────────────────────────────────────────
    st.markdown(section_header(t("news_title"), accent=AMBER), unsafe_allow_html=True)
    if st.button(t("load_news"), key="signals_news_btn"):
        _load_news(ticker)