import threading import time import streamlit as st from dotenv import load_dotenv load_dotenv() from langchain_core.messages import HumanMessage from agent.llm import RunConfig, classify_llm_error from dashboard import i18n from dashboard import model_picker from dashboard.i18n import t from dashboard.runtime_env import is_hosted_space from dashboard.theme import inject_global_css, LOGO_SVG from dashboard.nav import NAV_ITEMS, LEGACY_NAV, VALID_KEYS, render as render_nav from dashboard import reasoning as reasoning_panel import dashboard.agent_graph as agent_graph st.set_page_config(page_title="Amplegest", page_icon="πŸ“Š", layout="wide", initial_sidebar_state="expanded") inject_global_css() # Eager-load the reranker once per process to avoid cold-start on first search. if not st.session_state.get("_reranker_warmed"): from storage.reranker import warmup as _warmup_reranker _warmup_reranker() st.session_state["_reranker_warmed"] = True # ── UI language β€” default to English (finance lingua franca) ─────────────────── st.session_state.setdefault("ui_lang", "en") # ── Navigation state β€” default to Company Overview (the onboarding view) ─────── st.session_state.setdefault("nav_key", "company") # Remap stale route keys from the pre-Decision-Stack navigation (open sessions). _nk = st.session_state["nav_key"] if _nk not in VALID_KEYS: st.session_state["nav_key"] = LEGACY_NAV.get(_nk, "company") # ── Generation state ──────────────────────────────────────────────────────────── st.session_state.setdefault("gen", { "running": False, "ticker": None, "trace": [], "brief": None, "error": None, "all_messages": [], "config": None, }) # ── Sidebar ───────────────────────────────────────────────────────────────────── with st.sidebar: st.markdown( f"""
{LOGO_SVG}
Amplegest
AI equity research
""", unsafe_allow_html=True, ) # ── Language selector β€” top of sidebar so first-time visitors set it before ── # anything else. Endonym labels (English / FranΓ§ais / EspaΓ±ol / Deutsch) are # self-identifying regardless of the current display language. _lang_labels = list(i18n.LANGUAGES.keys()) _current_lang_label = next( (lbl for lbl, code in i18n.LANGUAGES.items() if code == st.session_state.get("ui_lang", "en")), _lang_labels[0], ) _selected_lang_label = st.selectbox( t("language"), options=_lang_labels, index=_lang_labels.index(_current_lang_label), key="_lang_selectbox", label_visibility="collapsed", ) # Persist and rerun when the language changes so all t() calls refresh. _new_lang_code = i18n.LANGUAGES[_selected_lang_label] if _new_lang_code != st.session_state.get("ui_lang", "en"): st.session_state["ui_lang"] = _new_lang_code st.rerun() # ── Model / API key picker ───────────────────────────────────────────────── run_cfg, _key_source = model_picker.render() # ── Ticker input β€” single company at a time ─────────────────────────────── # Coverage clicks set a pending ticker; consume it BEFORE the widget # instantiates (writing a widget key after instantiation raises). _pending_ticker = st.session_state.pop("_pending_ticker", None) if _pending_ticker: st.session_state["ticker_input_box"] = _pending_ticker _raw_ticker = st.text_input( t("ticker_label"), placeholder=t("ticker_placeholder"), key="ticker_input_box", ).strip().upper() st.caption(t("ticker_caption")) # Graceful multi-ticker: if the user types "NVDA AAPL", take the first as the # active ticker and ignore the rest with a note. _ticker_tokens = [tok for tok in _raw_ticker.replace(",", " ").split() if tok] ticker_input = _ticker_tokens[0] if _ticker_tokens else "" if len(_ticker_tokens) > 1: st.info(t("ticker_multi_warning")) generate_clicked = st.button( t("generate_brief"), type="primary", use_container_width=True, disabled=not ticker_input or st.session_state["gen"]["running"] or run_cfg is None, ) if run_cfg is None: st.caption(t("model_blocked_caption")) st.divider() # Resolve any pending nav redirect (e.g. auto-navigate to Verdict after generation) if "_nav_pending_key" in st.session_state: st.session_state["nav_key"] = st.session_state.pop("_nav_pending_key") # ── Auto-load persisted brief from briefs.db β€” no regeneration needed ───── if ( ticker_input and not st.session_state["gen"]["running"] and st.session_state.get("brief_ticker") != ticker_input ): try: from storage import briefs_db _persisted = briefs_db.get_brief(ticker_input) if _persisted: st.session_state["brief"] = _persisted st.session_state["brief_ticker"] = ticker_input except Exception: pass # Inline brief lookup β€” _get_cached_brief() is defined after this block _sb = st.session_state.get("brief") _sidebar_brief = _sb if (_sb and ticker_input and st.session_state.get("brief_ticker") == ticker_input) else None render_nav(NAV_ITEMS, st.session_state["nav_key"], brief=_sidebar_brief) # ── Coverage β€” saved briefs, click to load without regenerating ─────────── try: from storage import briefs_db _coverage = briefs_db.list_briefs() except Exception: _coverage = [] if _coverage: st.divider() st.markdown( f'
{t("coverage_title")}
', unsafe_allow_html=True, ) for _row in _coverage: _cov_label = ( f'{_row["ticker"]} Β· {_row["filing_date"]}' if _row["filing_date"] else _row["ticker"] ) if st.button(_cov_label, key=f'cov_{_row["ticker"]}', use_container_width=True): st.session_state["_pending_ticker"] = _row["ticker"] st.rerun() st.caption(t("coverage_caption")) # Cost footer cost_state = st.session_state.get("_last_cost") cost_ticker = st.session_state.get("_last_cost_ticker") if cost_state and cost_ticker == ticker_input: c = cost_state st.divider() st.caption( f"Tokens: {c['input_tokens']:,} in / {c['output_tokens']:,} out\n" f"Est. cost: ${c['cost_usd']:.4f} USD" ) # ── Helpers ───────────────────────────────────────────────────────────────────── def _get_cached_brief() -> dict | None: cached = st.session_state.get("brief") if cached and st.session_state.get("brief_ticker") == ticker_input: return cached return None def _brief_placeholder(tab_name: str) -> None: st.info(f"{t('section_placeholder')} {tab_name}.") def _absorb_translation_cost(ticker: str, usage: dict | None, model: str) -> None: """Merge translation token usage into the session cost footer.""" if not usage: return try: from agent.cost_log import log_usage log_usage(ticker, model, usage["input_tokens"], usage["output_tokens"]) existing = st.session_state.get("_last_cost") if existing and st.session_state.get("_last_cost_ticker") == ticker: st.session_state["_last_cost"] = { "input_tokens": existing["input_tokens"] + usage["input_tokens"], "output_tokens": existing["output_tokens"] + usage["output_tokens"], "cost_usd": round(existing["cost_usd"] + usage["cost_usd"], 6), } else: st.session_state["_last_cost"] = usage st.session_state["_last_cost_ticker"] = ticker except Exception: pass def _get_display_brief(brief: dict | None, ticker: str, config: RunConfig | None) -> dict | None: """Return a brief translated to the current UI language if needed. Never raises. Falls back to original brief on any error. Uses session cache + DB cache to avoid re-translating on every rerun. """ if not brief: return brief target = i18n.report_language() if (brief.get("language") or "English") == target: return brief # No key available (yet) β€” skip translation silently rather than erroring; # the brief still renders in its original (English) language. if config is None: return brief # Session cache key: (ticker, target_lang, filing_date) cache = st.session_state.setdefault("_display_brief_cache", {}) failed = st.session_state.setdefault("_translate_failed", set()) cache_key = (ticker, target, brief.get("filing_date") or "") if cache_key in cache: return cache[cache_key] if cache_key in failed: st.caption(t("translation_failed")) return brief # Check persistent DB cache first try: from storage import briefs_db persisted = briefs_db.get_translation(ticker, target) except Exception: persisted = None if persisted: cache[cache_key] = persisted return persisted # Call the translator from agent.translate import translate_brief with st.spinner(t("translating_brief")): translated, usage = translate_brief(brief, target, config=config) if (translated.get("language") or "English") != target: failed.add(cache_key) st.caption(t("translation_failed")) return brief try: from storage import briefs_db briefs_db.save_translation(ticker, translated.get("language", target), translated) except Exception: pass cache[cache_key] = translated _absorb_translation_cost(ticker, usage, config.model) return translated def _reliability_counts(brief: dict) -> dict[str, int]: """Aggregate HIGH/MED/LOW counts across all SourcedFact fields in the brief.""" counts: dict[str, int] = {"HIGH": 0, "MEDIUM": 0, "LOW": 0} def _tally(obj: object) -> None: if isinstance(obj, dict): rel = obj.get("reliability") if rel in counts: counts[rel] += 1 for v in obj.values(): _tally(v) elif isinstance(obj, list): for item in obj: _tally(item) _tally(brief) return counts def _reliability_bar_html(brief: dict) -> str: """A compact stacked bar (HIGH/MEDIUM/LOW) with the breakdown as a hover tooltip β€” replaces a wall of "HIGH 73% Β· MEDIUM 25% Β· LOW 2%" text.""" from dashboard.theme import GREEN, AMBER, TEXT_FAINT counts = _reliability_counts(brief) total = sum(counts.values()) if total == 0: return "" pct = {k: round(v * 100 / total) for k, v in counts.items()} colors = {"HIGH": GREEN, "MEDIUM": AMBER, "LOW": "#9ca3af"} segments = "".join( f'' for k in ("HIGH", "MEDIUM", "LOW") if pct[k] > 0 ) title = " Β· ".join(f"{k} {pct[k]}%" for k in ("HIGH", "MEDIUM", "LOW") if pct[k] > 0) return ( f'' f'{t("sources_prefix")}' f'{segments}' f'' ) def _display_policy_allows(brief: dict, capability: str) -> bool: """Return True only when a display capability is explicitly validated. Persisted briefs predate ``display_policy`` and therefore fail closed. A truthy string or an LLM-invented value is deliberately insufficient: the producer must set the exact boolean ``True`` after the relevant alignment or calibration check has passed. """ policy = brief.get("display_policy") or {} return isinstance(policy, dict) and policy.get(capability) is True def _render_ticker_bar(ticker: str, brief: dict) -> None: """Render the sticky context bar above the main content area. Ticker, company and filing date are always safe to show. Event returns and aggregate reliability are hidden unless upstream validation explicitly enables them via alignment flags or ``display_policy``, respectively. """ from dashboard.theme import GREEN, RED, TEXT_MUTED company = brief.get("company_name", ticker) filing_date = brief.get("filing_date", "") me = brief.get("market_expectations") or {} event_returns_aligned = ( isinstance(me, dict) and _display_policy_allows(brief, "event_returns_aligned") and me.get("event_aligned") is True and me.get("event_comparison_allowed") is True ) d1 = ( me.get("d1_price_reaction_pct") if event_returns_aligned else None ) # D1 reaction d1_html = "" if d1 is not None: color = GREEN if d1 >= 0 else RED prefix = "+" if d1 >= 0 else "" d1_html = ( f'= 0 else "#fef2f2"};' f'color:{color};border:1px solid {color}33;' f'border-radius:4px;padding:2px 8px;font-size:0.75rem;font-weight:600;">' f'D1 {prefix}{d1:.1f}%' ) # Filing date and generation date β€” kept visually distinct so "as of" never # gets misread as "when this brief was produced" (the two can differ by # weeks if a persisted brief is reloaded, or the filing itself is old). date_html = "" if filing_date: date_html = ( f'{t("as_of")} {filing_date}' ) generated_at = str(brief.get("generated_at") or "")[:10] generated_html = "" if generated_at and generated_at != filing_date: generated_html = ( f'{t("generated_prefix")} {generated_at}' ) # Reliability breakdown β€” mini stacked bar (hover for the exact split) rel_html = ( _reliability_bar_html(brief) if _display_policy_allows(brief, "aggregate_reliability_meaningful") else "" ) # Dot separator helper sep = f'Β·' inner_parts = [ f'{ticker}', f'{company}', ] if date_html: inner_parts.append(date_html) if generated_html: inner_parts.append(generated_html) if d1_html: inner_parts.append(d1_html) if rel_html: inner_parts.append(rel_html) inner = f" {sep} ".join(inner_parts) st.markdown( f'
' f'
' f'{inner}' f'
' f'
', unsafe_allow_html=True, ) # ── Background generation thread ───────────────────────────────────────────────── def _run_brief_thread(ticker: str, gen: dict, language: str, config: RunConfig) -> None: """Run brief generation in a background thread. `config` is a frozen snapshot taken in the main thread at click time β€” this function never reads st.session_state, so a mid-run change to the sidebar picker cannot affect an in-flight run. """ initial = { "ticker": ticker, "messages": [HumanMessage(content=f"Generate a research brief for {ticker}.")], "tool_round_count": 0, "nudge_fired": False, "profile_payloads": None, "language": language, "brief": None, "brief_markdown": None, "synthesis_error": None, } try: from agent.graph import create_graph graph = create_graph(config) for event in graph.stream(initial, stream_mode="updates"): for node_name, output in event.items(): if output is None: continue msgs = output.get("messages") or [] if node_name == "agent": reasoning_panel.absorb_agent_messages(gen["trace"], msgs) gen["all_messages"].extend(msgs) # If the agent stopped emitting tool_calls, synthesis is next last = msgs[-1] if msgs else None if last and not getattr(last, "tool_calls", None): reasoning_panel.mark_synthesis(gen["trace"], "in_progress") elif node_name == "tools": reasoning_panel.absorb_tool_messages(gen["trace"], msgs) gen["all_messages"].extend(msgs) elif node_name == "nudge": for msg in msgs: text = getattr(msg, "content", "") if text: gen["trace"].append({"kind": "assistant_text", "text": f"⚠ Coverage nudge: {text}"}) elif node_name == "synthesis": gen["brief"] = output.get("brief") gen["error"] = output.get("synthesis_error") reasoning_panel.mark_synthesis(gen["trace"], "done", error=gen["error"]) except Exception as exc: import sys print(f"[gen thread error] {exc}", file=sys.stderr) gen["error"] = classify_llm_error(exc, config.provider) or str(exc) finally: gen["running"] = False if generate_clicked and ticker_input and not st.session_state["gen"]["running"] and run_cfg is not None: gen = st.session_state["gen"] gen.update( running=True, ticker=ticker_input, trace=[], brief=None, error=None, all_messages=[], config=run_cfg, ) _chosen_language = i18n.report_language() # NB: never name this `t` β€” it would shadow the i18n t() helper. _thread_args = (ticker_input, gen, _chosen_language, run_cfg) try: from streamlit.runtime.scriptrunner import add_script_run_ctx _gen_thread = threading.Thread(target=_run_brief_thread, args=_thread_args, daemon=True) add_script_run_ctx(_gen_thread) _gen_thread.start() except Exception: # Fallback: run synchronously if thread context attachment fails _gen_thread = threading.Thread(target=_run_brief_thread, args=_thread_args, daemon=True) _gen_thread.start() # ── Live trace poller (polls every 400ms while generation runs) ────────────────── def _live_trace_fragment() -> None: gen = st.session_state.get("gen", {}) if not gen.get("running") and not gen.get("trace"): return trace = gen["trace"] progress_val, progress_label = reasoning_panel.estimate_progress(trace) pct = int(progress_val * 100) ticker_label = gen.get("ticker", "") st.markdown( f'
' f'
' f'
' f'{t("generating_brief")} β€” {ticker_label}' f'
' f'
{pct}%
' f'
' f'
{progress_label}
' f'
' f'
' f'
' f'
', unsafe_allow_html=True, ) agent_graph.render(trace) with st.expander(t("detailed_reasoning"), expanded=False): reasoning_panel.render_trace_body(trace) if gen.get("error") and not gen.get("brief"): st.error(f"{t('gen_error_prefix')} {gen['error']}") if gen.get("running"): # Still generating β€” poll again in 400 ms (server-side timer avoids HF Spaces proxy race) time.sleep(0.4) st.rerun() else: # Thread has finished β€” persist results and navigate if gen.get("brief"): try: from agent.cost_log import compute_run_cost, log_usage from agent.llm import ANTHROPIC_DEFAULT_MODEL # Bill the model that actually ran this generation β€” the frozen # snapshot taken at click time, not whatever the sidebar picker # shows now (the visitor may have switched provider mid-run). _cfg = gen.get("config") run_model = _cfg.model if _cfg else ANTHROPIC_DEFAULT_MODEL cost = compute_run_cost(gen["all_messages"], run_model) log_usage(gen["ticker"], run_model, cost["input_tokens"], cost["output_tokens"]) st.session_state["_last_cost"] = cost st.session_state["_last_cost_ticker"] = gen["ticker"] except Exception: pass st.session_state["brief"] = gen["brief"] st.session_state["brief_ticker"] = gen["ticker"] # Invalidate translation cache for the fresh brief st.session_state.pop("_display_brief_cache", None) st.session_state.pop("_translate_failed", None) # Persist brief for the Coverage list + reload across sessions # (non-critical β€” never blocks UI) try: from storage import briefs_db briefs_db.save_brief(gen["ticker"], gen["brief"]) except Exception: pass st.session_state.setdefault("reasoning_trace", {})[gen["ticker"]] = list(gen["trace"]) st.session_state["_nav_pending_key"] = "verdict" gen.update(running=False, trace=[], brief=None, error=None, all_messages=[]) st.rerun() else: # Failure: store error message, clean up, return to normal UI err_msg = gen.get("error") or t("no_brief_error").format(ticker=gen.get("ticker", "")) st.session_state["_gen_error"] = (gen.get("ticker", ""), err_msg) gen.update(running=False, trace=[], brief=None, error=None, all_messages=[]) st.rerun() # ── Main content routing ────────────────────────────────────────────────────────── def _display_company_profile(profile: dict, config: RunConfig | None) -> dict: target_language = i18n.report_language() if target_language == "English": return profile profile_id = profile.get("_profile_id") if not profile_id: return profile try: from storage import company_profiles cached = company_profiles.get_translation(profile_id, target_language) if cached: return cached if config is None: return profile from agent.company_profile import translate_company_profile with st.spinner(t("translating_brief")): translated, usage = translate_company_profile(profile, target_language, config) if translated is not profile and translated.get("language") == target_language: company_profiles.save_translation(profile_id, target_language, translated) _absorb_translation_cost(str(profile.get("ticker") or ""), usage, config.model) return translated except Exception: pass st.caption(t("translation_failed")) return profile def _render_company_route(ticker: str, config: RunConfig | None) -> None: from agent.company_profile import profile_source_coverage, source_fingerprint from storage import company_profiles fingerprint = source_fingerprint(ticker) profile = company_profiles.get_latest_profile( ticker, source_fingerprint=fingerprint, language="English", ) stale = False if profile is None: profile = company_profiles.get_latest_profile(ticker, language="English") stale = profile is not None if profile is None: st.markdown( f"""
{t("overview_empty_title")}
{t("overview_empty_body")}
""", unsafe_allow_html=True, ) return if stale: st.caption(t("overview_stale")) coverage = profile_source_coverage(ticker) if ( coverage.get("business_sections_attempted", 0) < 1 or coverage.get("segments_geography_sections_attempted", 0) < 1 ): warning_key = ( "primer_backfill_warning_hosted" if is_hosted_space() else "primer_backfill_warning" ) st.warning(t(warning_key).format(ticker=ticker)) display_profile = _display_company_profile(profile, config) company_name = str(display_profile.get("company_name") or ticker) from analytics.company_news import fetch_company_news from analytics.company_market import fetch_company_market with st.spinner(t("primer_loading_market")): news_items, news_error = fetch_company_news(ticker, company_name) market, market_error = fetch_company_market(ticker, company_name) import copy from analysis.company_attention import attach_attention_stats display_profile = attach_attention_stats(copy.deepcopy(display_profile), ticker, news_items) from dashboard.company_primer import render as render_company_primer render_company_primer( display_profile, ticker, market=market, news_items=news_items, news_error=news_error, ) if market_error: st.caption(market_error) gen_error = st.session_state.pop("_gen_error", None) if gen_error: err_ticker, err_msg = gen_error st.error(f"{t('brief_error_prefix')} **{err_ticker}**: {err_msg}") gen = st.session_state["gen"] active_key = st.session_state.get("nav_key", "company") if gen["running"] or gen.get("trace"): _live_trace_fragment() elif not ticker_input: st.markdown( f"""
{t("landing_title")}
{t("landing_body")}
""", unsafe_allow_html=True, ) else: # Active ticker, not generating β€” show context bar then route to section brief = _get_display_brief(_get_cached_brief(), ticker_input, run_cfg) if brief: _render_ticker_bar(ticker_input, brief) if active_key == "company": _render_company_route(ticker_input, run_cfg) elif active_key == "verdict": if brief: from dashboard.verdict import render as render_verdict render_verdict(brief) else: st.markdown( f"""
{t("snapshot_empty_title")}
{t("snapshot_empty_body")}
""", unsafe_allow_html=True, ) elif active_key == "signals": if brief: from dashboard.signals_view import render as render_signals render_signals(brief, ticker_input) else: _brief_placeholder(t("nav_signals")) elif active_key == "financials": from dashboard.financials import render as render_financials render_financials(ticker_input, brief) elif active_key == "chat": from dashboard.chat import render as render_chat render_chat(ticker_input, brief=brief, config=run_cfg) # ── Global data disclaimer ───────────────────────────────────────────────────── if ticker_input and not gen.get("running") and not gen.get("trace"): st.markdown( f'
' f'{t("disclaimer")}' f'
', unsafe_allow_html=True, )