"""
components.py
All Streamlit UI components โ shared between Option A and Option B.
"""
import streamlit as st
import pandas as pd
import numpy as np
def show_freshness_status(freshness: dict):
if freshness.get("fresh"):
st.success(freshness["message"])
else:
st.warning(freshness["message"])
def show_availability_warnings(availability: dict):
for etf, info in availability.items():
if not info.get("available"):
st.warning(info.get("message", f"โ ๏ธ {etf} data unavailable."))
def show_signal_banner(etf: str, hold_period: int, next_date,
net_score: float, in_cash: bool,
next_etf: str = None):
is_cash = in_cash or etf == "CASH"
if is_cash:
st.warning(f"๐ก๏ธ **DRAWDOWN PROTECTION ACTIVE** โ Staying in CASH on {next_date}")
else:
bg = "linear-gradient(135deg, #00d1b2 0%, #00a896 100%)"
label = f"๐ฏ {next_date} โ {etf}"
sub = f"Next Trading Day Signal"
st.markdown(f"""
P2-ETF FORECASTER ยท NEXT TRADING DAY SIGNAL
{label}
{sub}
""", unsafe_allow_html=True)
def show_etf_scores_table(scores: dict, arima_results: dict,
run_scores: dict, active_etfs: list,
hold_periods: list, fee_bps: int):
st.subheader("๐ ETF Signal Breakdown โ Option A")
st.caption(f"Transaction cost: **{fee_bps} bps** applied to net scores")
rows = []
for etf in active_etfs:
arima = arima_results.get(etf, {})
run = run_scores.get(etf, {})
err = arima.get("error")
direction_str = "โ Up" if arima.get("direction", 0) == 1 else "โ Down"
order_str = str(arima.get("order", "N/A"))
pressure = run.get("pressure", 0.0)
run_len = run.get("run_length", 0)
run_dir = run.get("direction", "")
row = {
"ETF": etf,
"ARIMA Order": order_str,
"Direction": direction_str if not err else f"โ ๏ธ {err}",
"Run Pressure": f"{pressure:.2f} ({run_len}d {run_dir})",
}
for h in hold_periods:
row[f"{h}d Score"] = f"{scores.get(etf, {}).get(h, 0.0):.4f}"
rows.append(row)
df_table = pd.DataFrame(rows).reset_index(drop=True)
def _highlight_best(row):
try:
vals = [float(r["1d Score"]) for r in rows]
if float(row["1d Score"]) == max(vals):
return ["background-color: rgba(0,200,150,0.15); font-weight:bold"] * len(row)
except Exception:
pass
return [""] * len(row)
styled = (
df_table.style
.apply(_highlight_best, axis=1)
.set_properties(**{"text-align": "center", "font-size": "13px"})
.set_table_styles([
{"selector": "th", "props": [("font-size", "13px"),
("font-weight", "bold"),
("text-align", "center")]},
{"selector": "td", "props": [("padding", "8px")]},
])
)
st.dataframe(styled, use_container_width=True)
def show_hold_period_rationale(best_etf: str, best_h: int,
scores: dict, hold_periods: list, fee_bps: int):
etf_scores = scores.get(best_etf, {})
if not etf_scores:
return
st.caption(
f"**Hold Period Selection for {best_etf}** "
f"(fee = {fee_bps} bps โ higher hold = lower relative cost):"
)
cols = st.columns(len(hold_periods))
for col, h in zip(cols, hold_periods):
s = etf_scores.get(h, 0.0)
is_best = h == best_h
color = "#00d1b2" if is_best else "#888"
badge = "โ
SELECTED" if is_best else ""
bg = "rgba(0,209,178,0.08)" if is_best else "#f8f8f8"
col.markdown(f"""
{h}-DAY HOLD {badge}
{s:.4f}
net score
""", unsafe_allow_html=True)
def show_metrics_row(result: dict, tbill_rate: float, spy_ann: float = None):
c1, c2, c3, c4, c5 = st.columns(5)
if spy_ann is not None:
diff = (result["ann_return"] - spy_ann) * 100
sign = "+" if diff >= 0 else ""
delta_str = f"vs SPY: {sign}{diff:.2f}%"
else:
delta_str = f"vs T-bill: {(result['ann_return'] - tbill_rate)*100:.2f}%"
c1.metric("๐ Ann. Return", f"{result['ann_return']*100:.2f}%", delta=delta_str)
c2.metric("๐ Sharpe", f"{result['sharpe']:.2f}",
delta="Strong" if result["sharpe"] > 1 else "Weak")
c3.metric("๐ฏ Hit Ratio 15d", f"{result['hit_ratio']*100:.0f}%",
delta="Good" if result["hit_ratio"] > 0.55 else "Weak")
c4.metric("๐ Max Drawdown", f"{result['max_dd']*100:.2f}%",
delta="Peak to Trough")
worst_date = result.get("max_daily_date", "N/A")
dd_delta = f"on {worst_date}" if worst_date != "N/A" else "Worst Single Day"
c5.metric("โ ๏ธ Max Daily DD", f"{result['max_daily_dd']*100:.2f}%", delta=dd_delta)
def show_audit_trail(audit_trail: list):
"""Option A audit trail โ shows Hold period column."""
if not audit_trail:
st.info("No audit trail data available.")
return
df = pd.DataFrame(audit_trail).tail(20)
if "In_Cash" in df.columns:
if df["In_Cash"].any():
df["In_Cash"] = df["In_Cash"].map({True: "๐ก๏ธ CASH", False: ""})
else:
df = df.drop(columns=["In_Cash"])
cols = [c for c in ["Date", "Signal", "Hold", "Net_Return", "In_Cash"]
if c in df.columns]
df = df[cols]
def _color_ret(val):
return ("color: #00c896; font-weight:bold" if val > 0
else "color: #ff4b4b; font-weight:bold")
styled = (
df.style
.map(_color_ret, subset=["Net_Return"])
.format({"Net_Return": "{:.2%}"})
.set_properties(**{"font-size": "14px", "text-align": "center"})
.set_table_styles([
{"selector": "th", "props": [("font-size", "13px"),
("font-weight", "bold"),
("text-align", "center")]},
{"selector": "td", "props": [("padding", "10px")]},
])
)
st.dataframe(styled, use_container_width=True, height=500)
def show_audit_trail_b(audit_trail: list):
"""Option B audit trail โ shows momentum Score instead of Hold period."""
if not audit_trail:
st.info("No audit trail data available.")
return
df = pd.DataFrame(audit_trail).tail(20)
if "In_Cash" in df.columns:
if df["In_Cash"].any():
df["In_Cash"] = df["In_Cash"].map({True: "๐ก๏ธ CASH", False: ""})
else:
df = df.drop(columns=["In_Cash"])
cols = [c for c in ["Date", "Signal", "Rank Score", "Net_Return", "In_Cash"]
if c in df.columns]
df = df[cols]
def _color_ret(val):
return ("color: #00c896; font-weight:bold" if val > 0
else "color: #ff4b4b; font-weight:bold")
fmt = {"Net_Return": "{:.2%}"}
if "Rank Score" in df.columns:
# Rank Score may be "โ" for CASH rows so only format numeric
fmt["Rank Score"] = lambda x: f"{x:.2f}" if isinstance(x, float) else x
styled = (
df.style
.map(_color_ret, subset=["Net_Return"])
.format(fmt)
.set_properties(**{"font-size": "14px", "text-align": "center"})
.set_table_styles([
{"selector": "th", "props": [("font-size", "13px"),
("font-weight", "bold"),
("text-align", "center")]},
{"selector": "td", "props": [("padding", "10px")]},
])
)
st.dataframe(styled, use_container_width=True, height=500)
def show_momentum_scores_table(momentum_scores: dict, active_etfs: list,
current_etf: str,
lb_short_days: int = 21,
lb_mid_days: int = 63,
lb_long_days: int = 126):
"""Option B โ per-ETF rank-based momentum breakdown with correct lookback labels."""
def _days_to_label(days):
months = round(days / 21)
return f"{months}M"
s_lbl = _days_to_label(lb_short_days)
m_lbl = _days_to_label(lb_mid_days)
l_lbl = _days_to_label(lb_long_days)
st.subheader("๐ ETF Momentum Rankings โ Option B")
st.caption(
f"Lookback windows: **{s_lbl}** (short) ยท **{m_lbl}** (mid) ยท **{l_lbl}** (long) ยท "
"Rank 1 = strongest ยท Lowest composite rank wins"
)
rows = []
for etf in active_etfs:
info = momentum_scores.get(etf, {})
rows.append({
"Selected": "โญ" if etf == current_etf else "",
"ETF": etf,
f"{s_lbl} Ret": info.get("ret_1m", 0.0),
f"{s_lbl} Rank": info.get("rank_1m", 0),
f"{m_lbl} Ret": info.get("ret_3m", 0.0),
f"{m_lbl} Rank": info.get("rank_3m", 0),
f"{l_lbl} Ret": info.get("ret_6m", 0.0),
f"{l_lbl} Rank": info.get("rank_6m", 0),
"Mom Rank": info.get("momentum_rank", 0.0),
"RS/SPY": info.get("rs_spy", 0.0),
"RS Rank": info.get("rs_rank", 0),
"MA Slope": info.get("ma_slope", 0.0),
"MA Rank": info.get("ma_rank", 0),
"Comp Rank": info.get("rank_score", 0.0),
})
rows = sorted(rows, key=lambda x: x["Comp Rank"])
ret_cols = [f"{s_lbl} Ret", f"{m_lbl} Ret", f"{l_lbl} Ret",
"RS/SPY", "MA Slope"]
col_order = ["Selected", "ETF",
f"{s_lbl} Ret", f"{s_lbl} Rank",
f"{m_lbl} Ret", f"{m_lbl} Rank",
f"{l_lbl} Ret", f"{l_lbl} Rank",
"Mom Rank", "RS/SPY", "RS Rank",
"MA Slope", "MA Rank", "Comp Rank"]
df_table = pd.DataFrame(rows).reset_index(drop=True)[col_order].reset_index(drop=True)
def _color_ret(val):
if isinstance(val, float):
return ("color: #00c896; font-weight:bold" if val > 0
else "color: #ff4b4b" if val < 0 else "")
return ""
def _highlight_top(row):
if row["ETF"] == current_etf:
return ["background-color: rgba(0,200,150,0.15); font-weight:bold"] * len(row)
return [""] * len(row)
fmt = {"Comp Rank": "{:.2f}", "Mom Rank": "{:.2f}",
"RS Rank": "{:.2f}", "MA Rank": "{:.0f}"}
for c in ret_cols:
fmt[c] = "{:.2%}"
styled = (
df_table.style
.apply(_highlight_top, axis=1)
.map(_color_ret, subset=ret_cols)
.format(fmt)
.set_properties(**{"text-align": "center", "font-size": "13px"})
.set_table_styles([
{"selector": "th", "props": [("font-size", "12px"),
("font-weight", "bold"),
("text-align", "center")]},
{"selector": "td", "props": [("padding", "6px")]},
])
)
st.dataframe(styled, use_container_width=True)
st.caption("**Comp Rank** = 50% Momentum Rank + 25% RS/SPY Rank + 25% MA Slope Rank ยท Lower = stronger signal")
def show_methodology():
"""Option B strategy methodology section."""
st.divider()
st.subheader("๐ Strategy Methodology โ Option B")
with st.expander("How does this strategy work?", expanded=False):
st.markdown("""
### Cross-Sectional Momentum Rotation
This strategy answers one question every trading day: **which ETF has the strongest
relative momentum right now?** It then holds that ETF until a better opportunity
or a drawdown protection event occurs.
---
#### Step 1 โ Daily Momentum Ranking
For each ETF (TLT, TBT, VNQ, SLV, GLD), three trailing return windows are computed:
- **Short window** (e.g. 2m for a 6m lookback)
- **Mid window** (e.g. 4m)
- **Long window** = the full lookback period you select in the sidebar
Each ETF is then **ranked 1 to 5** within each window (1 = highest return = best).
The **Composite Rank** is the average of these three individual ranks.
The ETF with the **lowest composite rank** (closest to 1) is selected.
This rank-based approach is intentionally relative โ it asks *"which ETF is
strongest vs its peers?"* not *"which ETF is up the most in absolute terms?"*
This prevents a single runaway ETF from dominating purely due to an extreme move.
---
#### Step 2 โ Why only 3 to 18 months lookback?
- **Less than 3 months**: Too noisy. Daily return noise dominates over genuine
trend signal. Short-term reversals are common and would cause excessive switching.
- **More than 18 months**: Too slow. A regime change (e.g. rates rising, gold
falling out of favour) takes many months to show up in the signal โ by which
time the strategy has already suffered significant losses staying in the wrong ETF.
- **Sweet spot 6โ12 months**: Academic research on cross-sectional momentum
(Jegadeesh & Titman 1993, and subsequent ETF studies) consistently finds
6โ12 month formation periods produce the strongest risk-adjusted returns.
The 9m lookback tends to offer the best Sharpe in this ETF universe.
---
#### Step 3 โ Drawdown Protection (Stop Loss)
A **2-day compound drawdown** trigger protects against sudden crashes:
> If (1 + Dayโโ return) ร (1 + Dayโโ return) โ 1 โค โ10%, move to **CASH**
Key design choices:
- **2-day window**: A single bad day can be noise (e.g. flash crash that reverses).
Two consecutive bad days suggest a genuine adverse move.
- **โ10% threshold**: Chosen to catch meaningful drawdowns (like Jan 30, 2026's
โ28.5% SLV crash) without triggering on routine daily volatility.
- **CASH earns T-bill rate** while protection is active โ not zero.
- The check happens at the **start of the trading day** using prior days' data,
so the signal is always actionable before markets open.
---
#### Step 4 โ Re-entry Condition (Z-score โฅ 1.2ฯ)
Once in CASH, the strategy does **not** simply re-enter after a fixed number of days.
Instead it waits for a statistically meaningful recovery signal:
> Re-enter when the top-ranked ETF's most recent daily return is โฅ **1.2 standard
> deviations above its 63-day rolling mean**
Why Z-score and not just "two up days"?
- After a crash, prices often bounce weakly before rolling over again (dead cat bounce).
- A Z-score โฅ 1.2ฯ means the recent return is in the **top ~12% of that ETF's
own distribution** โ not just slightly positive, but genuinely above-trend.
- The 63-day window (โ 3 months) is long enough to capture the post-crash
distribution including the crash itself, making the bar appropriately high.
- 1.2ฯ is a deliberate balance: strict enough to avoid re-entering on weak bounces,
lenient enough not to miss the early stages of a genuine recovery.
---
#### Transaction Costs
Fees are charged **only on switches** between ETFs, not on hold days.
At 10 bps per switch, and with typical holding periods of several weeks,
the annual fee drag is modest โ usually 20โ50 bps/year depending on regime.
""")