"""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''
)
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''
)
else:
score = section["score"]
fg = GREEN if score > 0 else (RED if score < 0 else GRAY)
rationale = section.get("rationale", "")
cells_html += (
f''
f'
{label}
'
f'{_gauge_html(score)}'
f'
'
f'{section.get("label", "")}
'
f'
{rationale}
'
f'
'
f'
'
f'
{t("print_sentiment")}
'
f'
'
f'{sign}{agg:.1f}
'
f'
'
f'{t(band_key)}
'
f'
'
f'{cells_html}'
f'
'
f'Validation required. '
f'AI hypotheses are experimental; delta detectors are heuristic. '
f'Validate both against the cited source text.
',
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'{r["text"]}
',
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''
f'
'
f'Evidence & Deltas
'
f'
{subtitle}
'
f'