"""Macro tab — market-wide panels: Index Performance, Yield Curve, Sector Heatmap.""" import streamlit as st import streamlit.components.v1 as _html import plotly.graph_objects as go import yfinance as yf import pandas as pd from datetime import date as _date _INDICES = { "S&P 500": "^GSPC", "NASDAQ": "^IXIC", "DOW": "^DJI", "Russell": "^RUT", "VIX": "^VIX", } # (symbol, divisor) — yfinance encodes yields as rate×divisor _YIELDS = { "3M": ("^IRX", 100), "5Y": ("^FVX", 10), "10Y": ("^TNX", 10), "30Y": ("^TYX", 10), } _SECTORS = { "Technology": "XLK", "Financials": "XLF", "Healthcare": "XLV", "Energy": "XLE", "Industrials": "XLI", "Cons. Disc.": "XLY", "Cons. Stap.": "XLP", "Utilities": "XLU", "Real Estate": "XLRE", "Materials": "XLB", "Comm. Svcs.": "XLC", } # ── Data fetch ──────────────────────────────────────────────────────────────── @st.cache_data(ttl=300) def _get_macro_data() -> dict: indices = {} for name, sym in _INDICES.items(): try: hist = yf.Ticker(sym).history(period="1y") if hist.empty: indices[name] = None continue closes = hist["Close"].dropna() last = float(closes.iloc[-1]) def _pct(n): if len(closes) > n: return float((closes.iloc[-1] - closes.iloc[-(n + 1)]) / closes.iloc[-(n + 1)] * 100) return None p1d = _pct(1) p1w = _pct(5) p1m = _pct(21) today = _date.today() jan1 = pd.Timestamp(today.year, 1, 1) idx = closes.index if idx.tz is not None: jan1 = jan1.tz_localize(idx.tz) ytd_slice = closes[idx.normalize() >= jan1.normalize()] pytd = float((last - float(ytd_slice.iloc[0])) / float(ytd_slice.iloc[0]) * 100) if not ytd_slice.empty else None indices[name] = {"price": last, "1d": p1d, "1w": p1w, "1m": p1m, "ytd": pytd} except Exception: indices[name] = None yields = {} for label, (sym, div) in _YIELDS.items(): try: hist = yf.Ticker(sym).history(period="5d") if hist.empty: yields[label] = None continue closes = hist["Close"].dropna() curr = float(closes.iloc[-1]) / div prev = float(closes.iloc[-2]) / div if len(closes) >= 2 else None yields[label] = {"rate": curr, "change_1d": (curr - prev) if prev is not None else None} except Exception: yields[label] = None sectors = {} for name, etf in _SECTORS.items(): try: hist = yf.Ticker(etf).history(period="2d") if len(hist) >= 2: pct = float((hist["Close"].iloc[-1] - hist["Close"].iloc[-2]) / hist["Close"].iloc[-2] * 100) price = float(hist["Close"].iloc[-1]) elif len(hist) == 1: pct = 0.0 price = float(hist["Close"].iloc[-1]) else: sectors[name] = None continue sectors[name] = {"etf": etf, "pct": pct, "price": price} except Exception: sectors[name] = None return {"indices": indices, "yields": yields, "sectors": sectors} # ── Helpers ─────────────────────────────────────────────────────────────────── def _pct_cell(v) -> str: """Return an HTML with color class for a % value.""" if v is None: return "" sign = "+" if v >= 0 else "" cls = "pos" if v > 0.005 else ("neg" if v < -0.005 else "flat") return "" + sign + f"{v:.2f}%" + "" def _sector_bg(pct: float) -> str: """Interpolate background color between neutral and pos/neg based on magnitude.""" neutral = (24, 29, 38) pos_bg = (21, 36, 26) neg_bg = (42, 21, 23) intensity = min(1.0, abs(pct) / 3.0) target = pos_bg if pct >= 0 else neg_bg r = int(neutral[0] + (target[0] - neutral[0]) * intensity) g = int(neutral[1] + (target[1] - neutral[1]) * intensity) b = int(neutral[2] + (target[2] - neutral[2]) * intensity) return "rgb(" + str(r) + "," + str(g) + "," + str(b) + ")" # ── Panel renderers ─────────────────────────────────────────────────────────── def _render_index_table(indices: dict) -> None: css = ( "" ) head = ( "" "" "" ) rows = "" for name, d in indices.items(): if d is None: rows += "" continue price = f"{d['price']:,.2f}" rows += ( "" "" "" "" "" "" "" ) _html.html(css + head + rows + "
IndexPrice1D1W1MYTD
" + name + "
" + name + "" + price + "" + _pct_cell(d["1d"]) + "" + _pct_cell(d["1w"]) + "" + _pct_cell(d["1m"]) + "" + _pct_cell(d["ytd"]) + "
", height=290, scrolling=False) def _render_yield_curve(yields: dict) -> None: labels = list(yields.keys()) rates = [yields[l]["rate"] if yields[l] else None for l in labels] valid = [(l, r) for l, r in zip(labels, rates) if r is not None] fig = go.Figure() if valid: xl, yl = zip(*valid) fig.add_trace(go.Scatter( x=list(xl), y=list(yl), mode="lines+markers", line=dict(color="#C2AA7A", width=2), marker=dict(size=7, color="#C2AA7A", line=dict(width=1, color="#8F7A50")), hovertemplate="%{x}: %{y:.2f}%", )) fig.update_layout( xaxis_title="Maturity", yaxis=dict(tickformat=".2f", ticksuffix="%"), height=240, margin=dict(l=52, r=16, t=16, b=40), ) st.plotly_chart(fig, use_container_width=True) def _render_yield_table(yields: dict) -> None: css = ( "" ) head = ( "" "" "" ) rows = "" for label, d in yields.items(): if d is None: rows += "" continue rate = f"{d['rate']:.2f}%" chg = d["change_1d"] sign = "+" if chg is not None and chg >= 0 else "" chg_s = (sign + f"{chg:.2f}%") if chg is not None else "—" cls = "pos" if (chg or 0) > 0.001 else ("neg" if (chg or 0) < -0.001 else "flat") rows += ( "" "" "" "" ) _html.html(css + head + rows + "
MaturityYield1D Chg
" + label + "
" + label + "" + rate + "" + chg_s + "
", height=240, scrolling=False) def _render_sector_heatmap(sectors: dict) -> None: css = ( "" "
" ) cells = "" for name, d in sectors.items(): if d is None: cells += ( "
" "
" + name + "
" "
" "
" ) continue pct = d["pct"] bg = _sector_bg(pct) sign = "+" if pct >= 0 else "" cls = "pos" if pct > 0.005 else ("neg" if pct < -0.005 else "flat") cells += ( "
" "
" + name + "
" "
" + d["etf"] + "
" "
" + sign + f"{pct:.2f}%" + "
" "
" ) _html.html(css + cells + "
", height=340, scrolling=False) # ── Public entry point ──────────────────────────────────────────────────────── def render_macro() -> None: data = _get_macro_data() st.markdown( "
Index Performance
", unsafe_allow_html=True, ) _render_index_table(data["indices"]) st.markdown("
", unsafe_allow_html=True) col1, col2 = st.columns([3, 2]) with col1: st.markdown( "
Yield Curve
", unsafe_allow_html=True, ) _render_yield_curve(data["yields"]) with col2: st.markdown( "
Treasury Yields
", unsafe_allow_html=True, ) _render_yield_table(data["yields"]) st.markdown("
", unsafe_allow_html=True) st.markdown( "
Sector Performance — Today
", unsafe_allow_html=True, ) _render_sector_heatmap(data["sectors"])