Greg Pomerantz
e28f8fcf85
- per-fund report charts now use chart_widget.equity_chart_html (st.iframe) instead of static st.plotly_chart: every visible window re-bases each line to 1.0 at its left edge, so fund/reference/index compare in any zoom level - same semantics as the Equity curves tab - removed 'expand all' checkbox (24 expanded sections drowned the page) - added 'rest of Fund Lab continues below' marker after the report section; the other Fund Lab sections (alpha search, clusters, N-PORT, drawdown, tax, CEF) were never deleted and are unchanged
1263 lines
59 KiB
Python
1263 lines
59 KiB
Python
"""Streamlit UI for stock/portfolio analysis.
|
||
|
||
Run: streamlit run app.py
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import re
|
||
from pathlib import Path
|
||
|
||
import pandas as pd
|
||
import plotly.graph_objects as go
|
||
import streamlit as st
|
||
|
||
import metrics as m
|
||
from data import DEFAULT_ROOT, generation, load_bundle, search_symbols, up_to_date
|
||
from fundlab import nport
|
||
from portfolio import parse_items, parse_weights, portfolio_returns
|
||
from portfolios import Portfolio, delete as pf_delete, load_all as pf_load, save as pf_save
|
||
from tax import after_tax_portfolio
|
||
|
||
st.set_page_config(page_title="Stock & Portfolio Analyzer", layout="wide")
|
||
|
||
# ---------------------------------------------------------------- settings
|
||
# user inputs persist across reloads / server restarts / devices
|
||
SETTINGS_FILE = Path(__file__).parent / "settings.json"
|
||
_settings: dict = {}
|
||
try:
|
||
_settings = json.loads(SETTINGS_FILE.read_text())
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def _remember(**kw) -> None:
|
||
"""Persist current widget values (write only on change)."""
|
||
global _settings
|
||
changed = False
|
||
for k, v in kw.items():
|
||
if _settings.get(k) != v:
|
||
_settings[k] = v
|
||
changed = True
|
||
if changed:
|
||
try:
|
||
SETTINGS_FILE.write_text(json.dumps(_settings, indent=2))
|
||
except Exception:
|
||
pass
|
||
|
||
# ---------------------------------------------------------------- sidebar
|
||
st.sidebar.title("Stock & Portfolio Analyzer")
|
||
|
||
root = st.sidebar.text_input("Data root", str(DEFAULT_ROOT))
|
||
rebuild = st.sidebar.checkbox("Rebuild parquet cache", value=False)
|
||
with st.spinner("Loading data..."):
|
||
bundle = load_bundle(root=Path(root), rebuild=rebuild)
|
||
if not up_to_date(Path(root)):
|
||
st.caption("Data files changed since the last cache refresh — a background "
|
||
"thread is updating the cache; figures below may be a few minutes stale.")
|
||
|
||
# data symbols are stored lowercase; input is resolved case-insensitively
|
||
_LOWER2SYM = {s.lower(): s for s in bundle.adj.columns}
|
||
|
||
|
||
def _resolve_weights(w: dict) -> dict:
|
||
"""Map parsed symbols onto the data's symbol case (insensitive match)."""
|
||
return {_LOWER2SYM.get(s.lower(), s): v for s, v in w.items()}
|
||
|
||
|
||
def _replace_token(spec: str, old: str, new: str) -> str:
|
||
"""Replace the symbol entry `old` (with or without a ':weight') by `new`.
|
||
|
||
Grammar: spaces separate items, commas join elements of one item.
|
||
"""
|
||
def fix_item(item: str) -> str:
|
||
out = []
|
||
for p in item.split(","):
|
||
sym = p.rsplit(":", 1)[0]
|
||
if sym.lower() == old.lower():
|
||
p = new if ":" not in p else new + p.rsplit(":", 1)[1]
|
||
out.append(p)
|
||
return ",".join(out)
|
||
return " ".join(fix_item(it) for it in spec.split())
|
||
|
||
|
||
st.sidebar.subheader("Portfolio")
|
||
# hand-off from e.g. 'Load saved into field': must be applied before the
|
||
# spec widget is (re)instantiated
|
||
_handover = st.session_state.pop("spec_handover", None)
|
||
if _handover is not None:
|
||
st.session_state["spec"] = _handover
|
||
spec = st.sidebar.text_input(
|
||
"Symbol(s) or portfolio(s)", value=_settings.get("spec", ""), key="spec",
|
||
help=("One or more, space-separated — same grammar as the Benchmark "
|
||
"field.\n"
|
||
"MSFT <- single symbol\n"
|
||
"MSFT:0.6,V:0.4 <- portfolio (commas join its elements, bare "
|
||
"symbols are equal-weight)\n"
|
||
"MSFT V googl:0.5,amzn:0.5 <- three separate entries, each "
|
||
"analyzed on its own\n"
|
||
"The FIRST entry is the 'current' portfolio: it drives the tax "
|
||
"detail tab, the allocation tab and Save."))
|
||
|
||
saved = pf_load()
|
||
if saved:
|
||
st.sidebar.subheader("Saved portfolios")
|
||
lcol1, lcol2 = st.sidebar.columns([3, 1])
|
||
load_name = lcol1.selectbox("Load into field", [p.name for p in saved], key="load_name")
|
||
if lcol2.button("Load"):
|
||
s = next(p.spec for p in saved if p.name == load_name)
|
||
# old specs used ', ' between elements; that would now read as two items
|
||
st.session_state["spec_handover"] = s.replace(", ", ",")
|
||
st.rerun()
|
||
|
||
_scheme_labels = ["Buy & hold (drift)", "Weekly", "Monthly", "Quarterly", "Yearly"]
|
||
scheme = st.sidebar.selectbox(
|
||
"Rebalance scheme", _scheme_labels,
|
||
index=min(int(_settings.get("scheme_index", 3)), len(_scheme_labels) - 1))
|
||
freq = {
|
||
"Buy & hold (drift)": None, "Weekly": "1W", "Monthly": "1ME",
|
||
"Quarterly": "QE", "Yearly": "YE",
|
||
}[scheme]
|
||
cost_bps = st.sidebar.number_input("Trading cost (bps, one-way)", 0.0, 100.0,
|
||
float(_settings.get("cost_bps", 5.0)))
|
||
|
||
st.sidebar.subheader("Tax (after-tax analysis)")
|
||
lt_rate = st.sidebar.number_input("Long-term gains %", 0.0, 49.0,
|
||
float(_settings.get("lt_rate", 20.0))) / 100
|
||
st_rate = st.sidebar.number_input("Short-term gains %", 0.0, 49.0,
|
||
float(_settings.get("st_rate", 15.0))) / 100
|
||
div_rate = st.sidebar.number_input("Dividends %", 0.0, 49.0,
|
||
float(_settings.get("div_rate", 15.0))) / 100
|
||
|
||
# statistics available in the Statistics tab (names = metrics.summary keys)
|
||
_STATS_VALID = ("total_return", "return", "vol", "sharpe", "sortino",
|
||
"max_dd", "calmar", "beta", "alpha", "ann_return_bench")
|
||
_STATS_DEFAULT = ",".join(_STATS_VALID)
|
||
# older saved settings may still use the long names
|
||
_STATS_RENAME = {"ann_return": "return", "ann_vol": "vol", "alpha_ann": "alpha"}
|
||
|
||
st.sidebar.subheader("Period & benchmark")
|
||
_p = int(_settings.get("period", 2010))
|
||
period = st.sidebar.selectbox("From year", range(2002, 2026),
|
||
index=(min(max(_p, 2002), 2025) - 2002))
|
||
bench_spec = st.sidebar.text_input(
|
||
"Benchmark (optional)", value=_settings.get("bench_spec", ""), key="bench_spec",
|
||
help=("Same grammar as the symbol field: a single symbol (MSFT) or a "
|
||
"portfolio (MSFT:0.6,V:0.4 — commas join its elements).\n"
|
||
"Spaces separate DISTINCT benchmarks, e.g.\n"
|
||
"MSFT MSFT:0.6,V:0.4\n"
|
||
"Empty = none. Simulated with the same rebalance scheme, cost and "
|
||
"tax rules. Beta/alpha in the stats table refers to the first "
|
||
"benchmark."))
|
||
|
||
st.sidebar.subheader("Statistics")
|
||
stats_order = st.sidebar.text_input(
|
||
"Statistics (comma-separated, in display order)",
|
||
value=_settings.get("stats_order", _STATS_DEFAULT), key="stats_order",
|
||
help=("Which statistics to show on the Statistics tab, in the order they "
|
||
"appear.\n"
|
||
"Available: " + ", ".join(_STATS_VALID) + "\n"
|
||
"beta, alpha_ann and ann_return_bench only have a value when a "
|
||
"benchmark is set. Empty field = show all, default order."))
|
||
_stats_cols = [_STATS_RENAME.get(s.strip(), s.strip())
|
||
for s in stats_order.split(",") if s.strip()]
|
||
_unknown_stats = [s for s in _stats_cols if s not in _STATS_VALID]
|
||
if _unknown_stats:
|
||
st.sidebar.warning(f"Unknown statistic(s) ignored: {', '.join(_unknown_stats)}")
|
||
_stats_cols = [s for s in _stats_cols if s in _STATS_VALID]
|
||
if not _stats_cols:
|
||
_stats_cols = list(_STATS_VALID)
|
||
|
||
st.sidebar.subheader("Date range (display, all tabs)")
|
||
_win_labels = ["Max", "10Y", "5Y", "3Y", "1Y", "YTD", "3M", "1M"]
|
||
_dw = _settings.get("date_window", _settings.get("equity_window", "Max"))
|
||
win = st.sidebar.radio("Window", _win_labels, horizontal=True, key="date_window",
|
||
index=_win_labels.index(_dw) if _dw in _win_labels else 0,
|
||
help=("Standard window for the START of the display "
|
||
"range, counted back from the end. The window "
|
||
"applies to every tab.\n"
|
||
"Max = from the first data point."))
|
||
|
||
|
||
def _parse_date(s: str):
|
||
"""YYYYMMDD (or YYYY-MM-DD) -> Timestamp, '' -> None, garbage -> 'bad'."""
|
||
s = s.strip()
|
||
if not s:
|
||
return None
|
||
for fmt in ("%Y%m%d", "%Y-%m-%d"):
|
||
try:
|
||
return pd.to_datetime(s, format=fmt)
|
||
except Exception:
|
||
pass
|
||
return "bad"
|
||
|
||
|
||
range_start = st.sidebar.text_input(
|
||
"Start (YYYYMMDD, blank = window start)",
|
||
value=_settings.get("range_start", ""), key="range_start",
|
||
help="Absolute start date, e.g. 20230102 = Jan 2, 2023. Overrides the "
|
||
"Window. Blank = the Window decides.")
|
||
range_end = st.sidebar.text_input(
|
||
"End (YYYYMMDD, blank = latest data)",
|
||
value=_settings.get("range_end", ""), key="range_end",
|
||
help="Absolute end date, e.g. 20230102 = Jan 2, 2023. Blank = latest "
|
||
"available data.")
|
||
_start_ts = _parse_date(range_start)
|
||
_end_ts = _parse_date(range_end)
|
||
if _start_ts == "bad":
|
||
st.sidebar.warning(f"Invalid start date '{range_start.strip()}' "
|
||
"(use YYYYMMDD) — ignored.")
|
||
_start_ts = None
|
||
if _end_ts == "bad":
|
||
st.sidebar.warning(f"Invalid end date '{range_end.strip()}' "
|
||
"(use YYYYMMDD) — ignored.")
|
||
_end_ts = None
|
||
|
||
# remember the sidebar inputs now, before any validation st.stop()
|
||
_remember(spec=spec, bench_spec=bench_spec, scheme_index=_scheme_labels.index(scheme),
|
||
cost_bps=cost_bps, lt_rate=lt_rate * 100, st_rate=st_rate * 100,
|
||
div_rate=div_rate * 100, period=period, stats_order=stats_order,
|
||
date_window=win, range_start=range_start, range_end=range_end)
|
||
|
||
if not spec.strip():
|
||
st.info("Type symbols or portfolios in the sidebar — e.g. `MSFT` or "
|
||
"`MSFT:0.6,V:0.4`; space-separated entries are each analyzed "
|
||
"separately. The page updates as soon as the input is valid.")
|
||
st.stop()
|
||
|
||
try:
|
||
items = [_resolve_weights(w) for w in parse_items(spec)]
|
||
except ValueError as e:
|
||
st.error(f"Invalid input: {e}")
|
||
st.stop()
|
||
weights = items[0] # first entry = the 'current' portfolio
|
||
all_syms = {s for w in items for s in w}
|
||
|
||
unknown = [s for s in all_syms if s not in bundle.adj.columns]
|
||
if unknown:
|
||
st.error(f"Symbol field — not in the data: {', '.join(unknown)}")
|
||
for u in unknown[:2]:
|
||
cands = [s for s in search_symbols(bundle, u, limit=4) if s not in all_syms]
|
||
if cands:
|
||
st.caption(f"Did you mean {u}:")
|
||
cols = st.sidebar.columns(2)
|
||
for i, s in enumerate(cands):
|
||
if cols[i % 2].button(s, key=f"fix-{u}-{s}", use_container_width=True):
|
||
st.session_state["spec_handover"] = _replace_token(spec, u, s)
|
||
st.rerun()
|
||
st.stop()
|
||
|
||
start = f"{period}-01-01"
|
||
use_syms = list(weights)
|
||
|
||
# ---- save / delete / select saved portfolios -------------------------
|
||
st.sidebar.subheader("Saved portfolios")
|
||
if len(items) == 1:
|
||
pname = st.sidebar.text_input("Name to save current portfolio as", key="save_name")
|
||
if st.sidebar.button("Save current portfolio"):
|
||
pname = pname.strip()
|
||
if pname:
|
||
# commas without spaces: under the spec grammar that stays ONE portfolio
|
||
spec_str = ",".join(f"{s}:{w:g}" for s, w in weights.items())
|
||
pf_save(Portfolio(pname, spec_str, freq, cost_bps))
|
||
st.sidebar.success(f"Saved '{pname}'.")
|
||
st.rerun()
|
||
else:
|
||
st.sidebar.warning("Give it a name first (field below).")
|
||
else:
|
||
st.sidebar.caption("Save: the Symbol field must hold exactly ONE "
|
||
"portfolio to save (it would save the first entry).")
|
||
|
||
compare_names: list[str] = []
|
||
if saved:
|
||
compare_names = st.sidebar.multiselect(
|
||
"Compare on the chart", [p.name for p in saved],
|
||
default=[p.name for p in saved])
|
||
dcol1, dcol2 = st.sidebar.columns([3, 1])
|
||
del_name = dcol1.text_input("Delete by name", key="del_name")
|
||
if dcol2.button("Delete") and del_name.strip():
|
||
pf_delete(del_name.strip())
|
||
st.rerun()
|
||
|
||
# ------------------------------------------------------------ compute
|
||
# cache the heavy simulations: switching the chart window / curve toggle / tax
|
||
# rates only rebuilds the (cheap) HTML — the portfolio+tax runs happen once
|
||
@st.cache_data(show_spinner=False)
|
||
def _compute_portfolio(w_key: tuple, scheme: str | None, cost_bps: float,
|
||
start: str, lt: float, st_r: float, div: float,
|
||
root: str, gen: int):
|
||
w = dict(w_key)
|
||
r = portfolio_returns(bundle.adj, w, rebalance=scheme, cost_bps=cost_bps, start=start)
|
||
t = after_tax_portfolio(bundle.adj, bundle.div, bundle.capg, w, rebalance=scheme,
|
||
cost_bps=cost_bps, lt_rate=lt, st_rate=st_r,
|
||
div_rate=div, start=start)
|
||
return r, t
|
||
|
||
|
||
def build_result(name: str, w: dict, scheme: str | None, cost: float) -> dict:
|
||
r, t = _compute_portfolio(tuple(sorted(w.items())), scheme, cost, start,
|
||
lt_rate, st_rate, div_rate, str(root),
|
||
generation(Path(root)))
|
||
return {"name": name, "weights": w, "res": r, "tax": t}
|
||
|
||
results = []
|
||
for k, w in enumerate(items):
|
||
name = "Current" if len(items) == 1 else ", ".join(w)
|
||
results.append(build_result(name, w, freq, cost_bps))
|
||
for p in saved:
|
||
if p.name not in compare_names:
|
||
continue
|
||
try:
|
||
w = {s: v for s, v in parse_weights(p.spec).items() if s in bundle.adj.columns}
|
||
if not w:
|
||
st.sidebar.warning(f"'{p.name}': no known symbols, skipped.")
|
||
continue
|
||
results.append(build_result(p.name, w, p.scheme, p.cost_bps))
|
||
except Exception as e:
|
||
st.sidebar.warning(f"'{p.name}': {e}")
|
||
|
||
res, taxres = results[0]["res"], results[0]["tax"]
|
||
|
||
# benchmark: same spec format as the main field, simulated with the same
|
||
# rebalance scheme and cost; invalid/unknown input just disables it
|
||
# benchmarks: ';' separates independent specs; each is simulated exactly
|
||
# like the candidate portfolio (same scheme, cost and tax rules)
|
||
# each item (line, or space-separated on a line) is one benchmark
|
||
benchmarks: list[dict] = []
|
||
for i, item in enumerate(bench_spec.split(), 1):
|
||
try:
|
||
w = _resolve_weights(parse_weights(item))
|
||
except ValueError as e:
|
||
st.sidebar.warning(f"Benchmark {i} ignored: {e}")
|
||
continue
|
||
unknown = [s for s in w if s not in bundle.adj.columns]
|
||
if unknown:
|
||
st.sidebar.warning(f"Benchmark {i} ignored: not in the data: "
|
||
f"{', '.join(unknown)}")
|
||
continue
|
||
try:
|
||
r, t = _compute_portfolio(tuple(sorted(w.items())), freq, cost_bps, start,
|
||
lt_rate, st_rate, div_rate, str(root),
|
||
generation(Path(root)))
|
||
except ValueError as e:
|
||
st.sidebar.warning(f"Benchmark {i} ignored: {e}")
|
||
continue
|
||
benchmarks.append({
|
||
"label": ", ".join(w),
|
||
"price": r.equity.reindex(res.equity.index).ffill(),
|
||
"after": t.equity.reindex(res.equity.index).ffill(),
|
||
})
|
||
# first benchmark is the reference for beta/alpha in the stats table
|
||
bench_price = benchmarks[0]["price"] if benchmarks else None
|
||
bench_after = benchmarks[0]["after"] if benchmarks else None
|
||
bench_label = benchmarks[0]["label"] if benchmarks else None
|
||
|
||
# ------------------------------------------------------------ display
|
||
if len(items) == 1:
|
||
if len(use_syms) == 1:
|
||
st.title(f"{use_syms[0]} — single symbol")
|
||
else:
|
||
st.title(f"Portfolio: {', '.join(use_syms)}")
|
||
else:
|
||
st.title("Portfolios: " + " · ".join(", ".join(w) for w in items))
|
||
if len(results) > 1:
|
||
st.caption(f"Comparing: {', '.join(r['name'] for r in results)}")
|
||
st.caption(f"{scheme} · start {start} · cost {cost_bps} bps · "
|
||
f"tax LT/ST/div {lt_rate:.0%}/{st_rate:.0%}/{div_rate:.0%}"
|
||
+ (f" · benchmark: {' ; '.join(b['label'] for b in benchmarks)}"
|
||
if benchmarks else ""))
|
||
|
||
# global curve mode — applies to the stats table AND the chart
|
||
_mode_labels = ["Pre-tax", "After-tax", "Pre-tax + after-tax"]
|
||
mode = st.radio("Curve", _mode_labels,
|
||
horizontal=True, key="curve_mode",
|
||
index=_mode_labels.index(_settings["curve_mode"])
|
||
if _settings.get("curve_mode") in _mode_labels else 0)
|
||
show_pre, show_after, both = mode != "After-tax", mode != "Pre-tax", \
|
||
mode == "Pre-tax + after-tax"
|
||
|
||
# ------------------------------------------------------------ display range
|
||
# one global [lo, hi] window applied to every tab (display only — the
|
||
# simulation itself still starts at 'From year'). Start: the Start box wins,
|
||
# else the Window counted back from the end. End: the End box, else latest.
|
||
_data_idx = res.equity.index
|
||
_data_start, _data_end = _data_idx[0], _data_idx[-1]
|
||
hi = _end_ts if _end_ts is not None else _data_end
|
||
if _start_ts is not None:
|
||
lo = _start_ts
|
||
elif win == "YTD":
|
||
lo = pd.Timestamp(hi.year, 1, 1)
|
||
elif win != "Max":
|
||
unit = "years" if win.endswith("Y") else "months"
|
||
lo = hi - pd.DateOffset(**{unit: int(win[:-1])})
|
||
else:
|
||
lo = _data_start
|
||
if lo > hi or len(_data_idx[(_data_idx >= lo) & (_data_idx <= hi)]) < 2:
|
||
st.sidebar.warning("Date range covers no (or one) data point — "
|
||
"showing the full range.")
|
||
lo, hi = _data_start, _data_end
|
||
|
||
|
||
def rng(s):
|
||
"""Clip a price/return series (or frame) to the display range."""
|
||
return s[(s.index >= lo) & (s.index <= hi)]
|
||
|
||
tab_stats, tab_equity, tab_alloc, tab_tax, tab_corr, tab_fundlab = st.tabs(
|
||
["Statistics", "Equity curves", "Allocation", "Tax detail",
|
||
"Correlation", "Fund Lab"])
|
||
|
||
with tab_stats:
|
||
# one table per benchmark: same rows, but beta/alpha/bench-return are
|
||
# measured against THAT benchmark (no benchmarks -> single table
|
||
# without those columns). One row per candidate (the whole portfolio,
|
||
# not its components); the pre/after suffix only appears in 'both' mode.
|
||
candidates = [(r["name"], r["res"].equity, r["tax"].equity) for r in results]
|
||
|
||
def bench_block(bench: dict | None) -> None:
|
||
rows = [(n, rng(p), rng(a)) for n, p, a in candidates]
|
||
if bench is not None:
|
||
rows.append((f"benchmark: {bench['label']}", rng(bench["price"]),
|
||
rng(bench["after"])))
|
||
bm = (rng(bench["price"]).resample("ME").last()
|
||
if bench is not None else None)
|
||
|
||
def pick(d: dict) -> dict:
|
||
# user-configured selection + order; skip columns the summary
|
||
# lacks (e.g. beta without a benchmark)
|
||
out = {}
|
||
for k in _stats_cols:
|
||
if k in d:
|
||
out[k] = d[k]
|
||
return out
|
||
|
||
summaries = {}
|
||
for name, pre, after in rows:
|
||
if both:
|
||
summaries[f"{name} (pre-tax)"] = pick(m.summary(pre, bm))
|
||
summaries[f"{name} (after-tax)"] = pick(m.summary(after, bm))
|
||
elif show_pre:
|
||
summaries[name] = pick(m.summary(pre, bm))
|
||
else:
|
||
summaries[name] = pick(m.summary(after, bm))
|
||
st.dataframe(m.format_summary_table(summaries), width='stretch')
|
||
|
||
if benchmarks:
|
||
for b in benchmarks:
|
||
st.subheader(f"vs {b['label']}")
|
||
bench_block(b)
|
||
else:
|
||
bench_block(None)
|
||
|
||
with tab_equity:
|
||
# Self-contained plotly.js page (see chart_widget.py): full mouse zoom +
|
||
# pan, with every relayout instantly clamping x to the data and re-fitting
|
||
# y to the exact min/max of the visible data. No clipping, no blank space.
|
||
from chart_widget import equity_chart_html
|
||
|
||
_remember(curve_mode=mode)
|
||
idx = rng(res.equity).index
|
||
|
||
# labels carry the pre/after suffix only in 'both' mode
|
||
palette = ["#1f77b4", "#ff7f0e", "#2ca02c", "#d62728",
|
||
"#9467bd", "#17becf", "#e377c2", "#8c564b"]
|
||
|
||
def lab(name, which):
|
||
return f"{name} — {which}" if both else name
|
||
|
||
series: dict[str, pd.Series] = {}
|
||
styles: dict[str, dict] = {}
|
||
for k, r in enumerate(results):
|
||
color = palette[k % len(palette)]
|
||
if show_pre:
|
||
series[lab(r["name"], "pre-tax")] = rng(r["res"].equity)
|
||
styles[lab(r["name"], "pre-tax")] = {"color": color, "width": 2}
|
||
if show_after:
|
||
s = lab(r["name"], "after-tax")
|
||
series[s] = rng(r["tax"].equity)
|
||
# faded only in 'both' mode, to stay distinct from the pre-tax line
|
||
styles[s] = {"color": color, "width": 2,
|
||
"opacity": 0.55 if both else 1.0}
|
||
for b in benchmarks:
|
||
base = f"benchmark ({b['label']})"
|
||
if show_pre:
|
||
p = rng(b["price"])
|
||
s = lab(base, "pre-tax")
|
||
series[s] = p / p.iloc[0]
|
||
styles[s] = {"width": 1, "opacity": 0.6}
|
||
if show_after:
|
||
p = rng(b["after"])
|
||
s = lab(base, "after-tax")
|
||
series[s] = p / p.iloc[0]
|
||
styles[s] = {"width": 1, "opacity": 0.3 if both else 0.6}
|
||
|
||
html = equity_chart_html(series, idx, ytitle="growth (1.0 = start)", styles=styles)
|
||
st.iframe(html, height=600)
|
||
def final(name, pre, after):
|
||
if both:
|
||
return f"{name}: pre {pre:.2f}× / after {after:.2f}×"
|
||
return f"{name}: {(pre if show_pre else after):.2f}×"
|
||
|
||
finals = " · ".join(final(r["name"], rng(r["res"].equity).iloc[-1],
|
||
rng(r["tax"].equity).iloc[-1]) for r in results)
|
||
for b in benchmarks:
|
||
finals += " · " + final(f"benchmark ({b['label']})",
|
||
rng(b["price"]).iloc[-1],
|
||
rng(b["after"]).iloc[-1])
|
||
st.caption(f"Final: {finals}. "
|
||
"Drag to box-zoom, right-drag (or mode-bar hand) to pan, "
|
||
"scroll to zoom, double-click to reset.")
|
||
|
||
with tab_alloc:
|
||
st.caption(f"Allocation for the **{results[0]['name']}** portfolio "
|
||
"(first entry in the Symbol field).")
|
||
alloc = rng(res.allocation)
|
||
st.dataframe(alloc.tail(1).T, width='stretch')
|
||
fig = go.Figure(go.Bar(x=alloc.columns, y=alloc.iloc[-1]))
|
||
fig.update_layout(height=360, title=f"Drifted allocation @ {alloc.index[-1].date()}")
|
||
st.plotly_chart(fig, width='stretch')
|
||
st.caption(f"Final allocation: " +
|
||
", ".join(f"{k} {v:.1%}" for k, v in alloc.iloc[-1].items()))
|
||
|
||
with tab_tax:
|
||
st.caption(f"Tax detail for the **{results[0]['name']}** portfolio "
|
||
"(first entry in the Symbol field).")
|
||
# all values below are fractions of the starting account (1.0 = 100%)
|
||
yr_tax = rng(taxres.taxes).resample("YE").sum()
|
||
yr_tax.index = yr_tax.index.year
|
||
yr_tax = yr_tax.map(lambda v: f"{v:.2%}")
|
||
yr_real = rng(taxres.realized).resample("YE").sum()
|
||
yr_real.index = yr_real.index.year
|
||
yr_real = yr_real.map(lambda v: f"{v:+.2%}")
|
||
yr_liq = rng(taxres.liq_tax).resample("YE").last()
|
||
yr_liq.index = yr_liq.index.year
|
||
yr_liq = yr_liq.map(lambda v: f"{v:.2%}")
|
||
st.subheader("Taxes actually paid per year (distributions + rebalance sales)")
|
||
st.dataframe(yr_tax, width='stretch')
|
||
st.subheader("Realized gains/losses per year (at rebalances)")
|
||
st.dataframe(yr_real, width='stretch')
|
||
st.subheader("Unrealized-gain tax you would owe if you sold everything (year-end)")
|
||
st.dataframe(yr_liq, width='stretch')
|
||
_tax_total = rng(taxres.taxes)["total"].sum()
|
||
_final_after = rng(taxres.equity).iloc[-1]
|
||
st.caption(f"Total taxes paid to date: {_tax_total:.2%} of start · "
|
||
f"final after-tax value (sell everything): {_final_after:.2f}× "
|
||
f"= {_final_after - 1:+.1%} total return")
|
||
|
||
with tab_corr:
|
||
# cross-correlation (daily returns, pairwise over common history):
|
||
# per portfolio, its components vs the benchmarks; then all portfolios
|
||
# vs the benchmarks
|
||
def corr_block(title: str, series: dict[str, pd.Series]) -> None:
|
||
if len(series) < 2:
|
||
return
|
||
st.subheader(title)
|
||
tbl = m.xcorr(series)
|
||
# the table is symmetric: row names double as column names, so the
|
||
# columns are just numbered 1..n (column i = row i)
|
||
tbl.columns = range(1, len(tbl.columns) + 1)
|
||
st.dataframe(tbl, width='stretch')
|
||
|
||
for r in results:
|
||
if len(r["weights"]) < 2:
|
||
continue # single component: it already has a row in the
|
||
# all-portfolios table below
|
||
series = {s: rng(bundle.adj[s]) for s in r["weights"]}
|
||
series.update({f"benchmark: {b['label']}": rng(b["price"])
|
||
for b in benchmarks})
|
||
corr_block(f"{r['name']} — components vs benchmarks", series)
|
||
series = {r["name"]: rng(r["res"].equity) for r in results}
|
||
series.update({f"benchmark: {b['label']}": rng(b["price"])
|
||
for b in benchmarks})
|
||
corr_block("All portfolios vs benchmarks", series)
|
||
st.caption("Pearson correlation of daily returns; pairs correlate over their "
|
||
"common history. Benchmarks are prefixed with 'benchmark: '. "
|
||
"The table is symmetric — column number i is the series of row i.")
|
||
|
||
# ---------------------------------------------------------------- fund lab
|
||
# per-fund report mockup from N-PORT schedules of investments (fundlab)
|
||
with tab_fundlab:
|
||
try:
|
||
_FUNDS = json.loads((Path(__file__).parent / "funds.json").read_text())
|
||
except Exception:
|
||
_FUNDS = {}
|
||
_MAN = nport.manifest()
|
||
_ORDER = list(nport.FUND_TOKENS)
|
||
_FL_LABELS = {s: f"{s.upper()} — {_FUNDS[s]['name']}"
|
||
for s in _ORDER if s in _FUNDS}
|
||
_fl_pick = st.selectbox("Fund (shortlist)", list(_FL_LABELS),
|
||
format_func=lambda s: _FL_LABELS[s],
|
||
key="fundlab_pick")
|
||
|
||
# --- all-funds summary table -------------------------------------
|
||
from fundlab import decompose as _dc
|
||
try:
|
||
_DR = json.loads(_dc.RESULTS.read_text())
|
||
except Exception:
|
||
_DR = {}
|
||
# --- per-fund report: narrative + performance + drivers + peers ----
|
||
# static build (fundlab.reportdata) so the page stays fast
|
||
_RD = Path(__file__).parent / "reports" / "report_data.json"
|
||
if not _RD.exists():
|
||
with st.expander("Summary - per-fund reports (not built yet)"):
|
||
st.info("Run `.venv/bin/python -m fundlab.reportdata` in the "
|
||
"project root, then refresh.")
|
||
else:
|
||
try:
|
||
_rd = json.loads(_RD.read_text())
|
||
_rfunds = _rd["funds"]
|
||
_rorder = sorted(_rfunds,
|
||
key=lambda s: _rfunds[s]["meta"]["order"])
|
||
except Exception as _e: # noqa: BLE001
|
||
st.warning(f"report data unreadable: {_e}")
|
||
_rfunds = {}
|
||
if _rfunds:
|
||
st.subheader(f"Summary - per-fund reports "
|
||
f"({len(_rfunds)} funds)")
|
||
st.caption(
|
||
f"Generated {_rd['generated']}. Each fund: a narrative "
|
||
"discussion first (the performance, what drives it, and "
|
||
"what we do NOT know), then the equity curve, the period "
|
||
"table (fund vs fitted reference vs S&P 500, with the "
|
||
"fund-minus-reference gap), the return drivers, the "
|
||
"reference mix explained sleeve by sleeve, tax placement, "
|
||
"and the best peers from the same return-driver cluster. "
|
||
"All alphas are in excess of the 3-mo T-bill rate.")
|
||
_ag = []
|
||
for _s in _rorder:
|
||
_f = _rfunds[_s]
|
||
_m, _st = _f["meta"], _f.get("stats", {})
|
||
_ag.append({
|
||
"fund": f"{_m['sym'].upper()} — {_m['name'][:44]}",
|
||
"group": _m["group"],
|
||
"5y": _st.get("t5y", "—"),
|
||
"CAGR": _st.get("cagr", "—"),
|
||
"maxDD": _st.get("mdd", "—"),
|
||
"R² 5y": _st.get("r2_5y", "—"),
|
||
"alpha 5y": _st.get("alpha_5y", "—"),
|
||
"cluster": (_f.get("peers") or {}).get("cluster", ""),
|
||
})
|
||
st.dataframe(pd.DataFrame(_ag), width="stretch",
|
||
hide_index=True)
|
||
for _s in _rorder:
|
||
_f = _rfunds[_s]
|
||
_m, _st = _f["meta"], _f.get("stats", {})
|
||
_grp = "C" if _m["group"] == "candidate" else "S"
|
||
_title = (f"{_grp}{_m['order']:02d} · {_m['sym'].upper()} "
|
||
f"— {_m['name']}")
|
||
if _m.get("verdict"):
|
||
_title += f" [{_m['verdict'][:44]}]"
|
||
with st.expander(_title):
|
||
if _f.get("narrative"):
|
||
st.markdown("\n\n".join(_f["narrative"]))
|
||
# same re-basing-on-zoom widget as the Equity curves tab:
|
||
# every visible window re-bases each line to 1.0 at its
|
||
# left edge, so fund/reference/index are comparable no
|
||
# matter where you zoom
|
||
_c = _f.get("chart", {})
|
||
if _c.get("dates"):
|
||
from chart_widget import equity_chart_html
|
||
_ds = pd.to_datetime(_c["dates"])
|
||
_series = {_m["sym"].upper():
|
||
pd.Series(_c["fund"], _ds)}
|
||
if _c.get("ref"):
|
||
_series["fitted reference"] = pd.Series(
|
||
_c["ref"], pd.to_datetime(_c["ref_dates"]))
|
||
if _c.get("ivv"):
|
||
_series["S&P 500 (IVV)"] = pd.Series(
|
||
_c["ivv"], pd.to_datetime(_c["ivv_dates"]))
|
||
st.iframe(equity_chart_html(
|
||
{k: s / s.iloc[0]
|
||
for k, s in _series.items()},
|
||
_ds, height=460, ytitle="growth (1.0 = start)",
|
||
styles={
|
||
_m["sym"].upper(): {"width": 2},
|
||
"fitted reference":
|
||
{"width": 1, "dash": "dash",
|
||
"opacity": 0.8},
|
||
"S&P 500 (IVV)":
|
||
{"width": 1, "dash": "dot",
|
||
"opacity": 0.5},
|
||
}), height=490)
|
||
if _f.get("perf"):
|
||
st.dataframe(pd.DataFrame(
|
||
_f["perf"],
|
||
columns=["period", "fund", "reference",
|
||
"S&P 500", "fund − reference"]),
|
||
width="stretch", hide_index=True)
|
||
st.caption(
|
||
"fund − reference = period alpha/timing (the "
|
||
"part of that period the fitted mix does not "
|
||
"explain). For cash-anchored funds the "
|
||
"reference is the T-bill rate itself.")
|
||
if _f.get("drivers"):
|
||
st.markdown("\n\n".join(_f["drivers"]))
|
||
_ref = _f.get("reference")
|
||
if _ref and _ref.get("rows"):
|
||
st.markdown(f"**Reference mix** ({_ref['netcash']})")
|
||
st.dataframe(pd.DataFrame(
|
||
_ref["rows"],
|
||
columns=["loading", "what it is",
|
||
"what it exposes you to"]),
|
||
width="stretch", hide_index=True)
|
||
st.caption(_ref["text"])
|
||
if _f.get("tax"):
|
||
st.markdown(f"**Tax** — {_f['tax']}")
|
||
_pe = _f.get("peers")
|
||
if _pe and _pe.get("rows"):
|
||
st.markdown(
|
||
f"**Peers — cluster: {_pe['cluster']}** "
|
||
f"(n={_pe['n']})")
|
||
st.dataframe(pd.DataFrame(
|
||
_pe["rows"],
|
||
columns=["fund", "5y", "CAGR", "maxDD",
|
||
"R² 5y", "alpha 5y", "tax"]),
|
||
width="stretch", hide_index=True)
|
||
st.caption(_pe["proscons"])
|
||
st.caption("← the rest of the Fund Lab continues below: "
|
||
"alpha search, return-driver clusters, N-PORT "
|
||
"cross-check, drawdown resilience, tax location, "
|
||
"and the CEF ranking.")
|
||
|
||
# --- alpha search: all screened funds (shortlist + longlist + harvest)
|
||
with st.expander("Alpha search — all screened funds, ranked"):
|
||
# search_all (comprehensive overnight screen) is the superset;
|
||
# read it first so its rows win any sym collision.
|
||
_by_sym: dict[str, dict] = {}
|
||
for _src in ("search_all.json", "search_results.json",
|
||
"search_mined.json", "search_external.json"):
|
||
try:
|
||
_j = json.loads((_dc.RESULTS.parent / _src).read_text())
|
||
except Exception:
|
||
continue
|
||
for _k, _v in _j.items():
|
||
if isinstance(_v, dict) and _v.get("sym"):
|
||
_v.setdefault("source", _src)
|
||
_by_sym.setdefault(_v["sym"], _v)
|
||
_all_rows = list(_by_sym.values())
|
||
if _all_rows:
|
||
def _tier(r) -> int:
|
||
v = str(r.get("verdict", ""))
|
||
if v.startswith("CANDIDATE"):
|
||
return 0
|
||
if "correlated" in v:
|
||
return 1
|
||
if "not persistent" in v:
|
||
return 2
|
||
if "sleeve" in v:
|
||
return 3
|
||
if "weak" in v:
|
||
return 4
|
||
return 5
|
||
_all_rows.sort(key=lambda r: (_tier(r),
|
||
-(r.get("alpha_t_5y")
|
||
if isinstance(r.get("alpha_t_5y"),
|
||
(int, float)) else -9)))
|
||
_tbl = []
|
||
for _v in _all_rows:
|
||
_a5 = _v.get("alpha_ann_5y")
|
||
_tbl.append({
|
||
"fund": f"{_v['sym'].upper()} — {_v.get('name', '')[:50]}",
|
||
"bucket": _v.get("bucket", ""),
|
||
"R² 5y": (f"{_v['r2_5y']:.2f}"
|
||
if isinstance(_v.get("r2_5y"), (int, float))
|
||
else "—"),
|
||
"alpha 5y": (f"{_a5*100:+.1f}% (t={_v['alpha_t_5y']:+.1f})"
|
||
if isinstance(_a5, (int, float)) else "—"),
|
||
"reference (5y)": _v.get("ref_5y") or "—",
|
||
"corr port": (f"{_v['corr_portfolio']:.2f}"
|
||
if isinstance(_v.get("corr_portfolio"),
|
||
(int, float)) else "—"),
|
||
"6m + %": (f"{_v['alpha_pos_frac']:.0%}"
|
||
if isinstance(_v.get("alpha_pos_frac"),
|
||
(int, float)) else "—"),
|
||
"verdict": _v.get("verdict", _v.get("error", "")),
|
||
})
|
||
st.dataframe(pd.DataFrame(_tbl), width="stretch")
|
||
st.caption(
|
||
"Screen: daily total returns IN EXCESS OF THE 3-MO T-BILL "
|
||
"RATE (BIL) vs 21 broad sleeve axes (same set for every "
|
||
"fund); 'alpha 5y' = OLS intercept of the excess returns "
|
||
"over the last 5 years (t-stat) — i.e. outperformance vs "
|
||
"the fund's OWN fitted sleeve mix (the 'reference' column), "
|
||
"not vs one index; a cash position earns exactly the t-bill "
|
||
"rate and adds zero alpha. 'corr port' = correlation with "
|
||
"your current qspnx/pmaix portfolio; '6m + %' = share of "
|
||
"rolling 6-month windows where the fund beat its fitted "
|
||
"sleeve mix. CANDIDATE = R²5y < 0.6 (or < 0.85 with strong "
|
||
"residual alpha), t5y ≥ 2, t-full ≥ 1.25, ≥ 45% positive "
|
||
"windows, portfolio correlation < 0.3.")
|
||
|
||
# --- return-driver clusters: k-means on 35-sleeve loading vectors ---
|
||
with st.expander(
|
||
"Return-driver clusters — funds grouped by what drives "
|
||
"their returns"):
|
||
_ck = st.slider("clusters", 10, 60, 30, step=2, key="fl_clus_k")
|
||
try:
|
||
import numpy as _np
|
||
from fundlab import cluster as _cl
|
||
from fundlab import factors as _fac
|
||
_syms, _V = _cl.loading_matrix()
|
||
_fr = json.loads((_dc.RESULTS.parent /
|
||
"factor_results.json").read_text())
|
||
# distance on the cash-emphasized matrix (a pure cash fund's
|
||
# level axis is otherwise swallowed by the low-exposure cloud);
|
||
# labels below use the raw _V values
|
||
_lab = _cl.kmeans(_cl.emphasized(_V, _fac.AXES.index("cash")),
|
||
_ck)
|
||
_sumrows = []
|
||
for c in range(_ck):
|
||
idx = [i for i, l in enumerate(_lab) if l == c]
|
||
if not idx:
|
||
continue
|
||
med = _np.median(_V[idx], 0)
|
||
top = _np.argsort(-_np.abs(med))[:4]
|
||
prof = {_fac.AXES[t]: float(med[t]) for t in top
|
||
if abs(med[t]) >= 0.08}
|
||
lab = _cl.label(prof)
|
||
t5s = [_fr[_syms[i]].get("alpha_t_5y") for i in idx]
|
||
t5s = [t for t in t5s if isinstance(t, (int, float))]
|
||
n_sig = sum(1 for t in t5s if t >= 2)
|
||
best = max(idx, key=lambda i: abs(
|
||
_fr[_syms[i]].get("alpha_t_5y") or 0))
|
||
_sumrows.append({
|
||
"cluster": lab, "n": len(idx),
|
||
"t5≥2": n_sig,
|
||
"top by |t5|": (f"{_syms[best].upper()} "
|
||
f"({_fr[_syms[best]].get('name','')[:40]})")})
|
||
# remember cluster id with each summary row for the selectbox
|
||
for _i, _r in enumerate(_sumrows):
|
||
_r["_id"] = _i
|
||
_sumrows.sort(key=lambda r: -r["n"])
|
||
st.dataframe(pd.DataFrame(
|
||
{k: r[k] for k in ("cluster", "n", "t5≥2", "top by |t5|")}
|
||
for r in _sumrows), width="stretch")
|
||
_cpick = st.selectbox(
|
||
"expand a cluster",
|
||
[f"{r['cluster']} (n={r['n']}, t5≥2: {r['t5≥2']})"
|
||
for r in _sumrows], key="fl_clus_pick")
|
||
_wanted = _cpick.split(" (")[0]
|
||
_crows = [r for r in _sumrows if r["cluster"] == _wanted]
|
||
_cn = (_crows[0]["n"] if _crows else None)
|
||
_crows = [r for r in _sumrows
|
||
if r["cluster"] == _wanted and r["n"] == _cn]
|
||
_target = _crows[0]["cluster"] if _crows else None
|
||
_members = []
|
||
for c in range(_ck):
|
||
idx = [i for i, l in enumerate(_lab) if l == c]
|
||
if not idx:
|
||
continue
|
||
med = _np.median(_V[idx], 0)
|
||
top = _np.argsort(-_np.abs(med))[:4]
|
||
prof = {_fac.AXES[t]: float(med[t]) for t in top
|
||
if abs(med[t]) >= 0.08}
|
||
if _cl.label(prof) == _target:
|
||
_members = [(i, idx[0]) for i in idx]
|
||
break
|
||
_rows = []
|
||
for i, _b in _members:
|
||
v = _fr.get(_syms[i], {})
|
||
_a5 = v.get("alpha_ann_5y")
|
||
_t5 = (v.get("alpha_t_5y")
|
||
if isinstance(v.get("alpha_t_5y"), (int, float))
|
||
else -99)
|
||
_rows.append({
|
||
"fund": f"{_syms[i].upper()} — {v.get('name','')[:48]}",
|
||
"R²": (f"{v['full']['r2']:.2f}"
|
||
if isinstance(v.get("full"), dict) else "—"),
|
||
"alpha 5y": (f"{_a5*100:+.1f}% (t={_t5:+.1f})"
|
||
if isinstance(_a5, (int, float)) else "—"),
|
||
"corr port": (f"{v['corr_portfolio']:.2f}"
|
||
if isinstance(
|
||
v.get("corr_portfolio"),
|
||
(int, float)) else "—"),
|
||
"verdict": (v.get("verdict") or "")[:40],
|
||
"src": "local" if v.get("local") else "new",
|
||
"_t": _t5,
|
||
})
|
||
_rows.sort(key=lambda r: -r["_t"])
|
||
for r in _rows:
|
||
r.pop("_t")
|
||
st.dataframe(pd.DataFrame(_rows), width="stretch")
|
||
st.caption(
|
||
"Each fund's daily total returns IN EXCESS OF THE 3-MO "
|
||
"T-BILL RATE (BIL) are regressed on 34 sleeve axes (equity "
|
||
"styles/sizes/intl/EM, the duration ladder, IG/HY/muni/MBS/"
|
||
"preferred/EM-debt credit, sectors, gold/oil/commodities, "
|
||
"CTA); the loading vector is the fund's 'return-driver "
|
||
"signature' — the betas are its NET INVESTED mix — plus a "
|
||
"virtual CASH axis = 1 − Σβ (net cash/T-bill position; 1 = "
|
||
"fully in cash, < 0 = levered). k-means groups similar "
|
||
"signatures. 'corr port' is shown for reference only - "
|
||
"high-corr funds are REPLACEMENTS for current holdings, "
|
||
"not rejections.")
|
||
except Exception as e: # noqa: BLE001
|
||
st.warning(f"cluster view unavailable: {e}")
|
||
|
||
# ---- N-PORT cross-check: what the top candidates actually hold ----
|
||
with st.expander(
|
||
"N-PORT cross-check - what the top candidates actually hold"):
|
||
_XC = _dc.RESULTS.parent / "xcheck_report.json"
|
||
if not _XC.exists():
|
||
st.info("No cross-check on file yet "
|
||
"(run `python -m fundlab.xcheck`).")
|
||
else:
|
||
from fundlab.xcheck import _fnum as _xfnum
|
||
_x = json.loads(_XC.read_text())
|
||
_x = {k: v for k, v in _x.items() if not v.get("error")}
|
||
st.caption(
|
||
"Each candidate's returns said 'alpha'; this checks the "
|
||
"filing. Holdings pulled from the fund's own NPORT-P, "
|
||
"matched by exact series name (so a sibling fund's "
|
||
"book is never shown). Buckets are keyword guesses on "
|
||
"position names - read the top positions.")
|
||
_xrows = []
|
||
for _s, _r in _x.items():
|
||
_b = ", ".join(f"{b['name']} {b['pct']:.0f}%"
|
||
for b in _r.get("buckets", [])[:3])
|
||
_top = _r.get("top", [{}])[0]
|
||
_tn = (_top.get("name") or _top.get("title")
|
||
or _top.get("text") or "")[:40]
|
||
_xrows.append({
|
||
"fund": _s.upper(), "as of": _r.get("as_of", ""),
|
||
"n": _r.get("n_positions", 0),
|
||
"t5": _r.get("t5"),
|
||
"top bucket": _b or "—", "#1 position": _tn,
|
||
"note": _r.get("note", "")[:60]})
|
||
_xrows.sort(key=lambda r: -abs(r["t5"] or 0))
|
||
st.dataframe(pd.DataFrame(_xrows), width="stretch")
|
||
_xp = st.selectbox(
|
||
"fund holdings detail", list(_x.keys()),
|
||
key="fl_xc_pick")
|
||
_xr = _x[_xp]
|
||
st.markdown(f"**{_xp.upper()}** - {_xr.get('name','')} "
|
||
f"(as of {_xr.get('as_of')}, "
|
||
f"{_xr.get('n_positions')} positions, "
|
||
f"source: {_xr.get('src','')})")
|
||
if _xr.get("note"):
|
||
st.warning(_xr["note"])
|
||
if _xr.get("categories"):
|
||
st.dataframe(
|
||
pd.DataFrame(_xr["categories"]), width="stretch")
|
||
_xtop = []
|
||
for _p in _xr.get("top", []):
|
||
if "pct" in _p:
|
||
_pct = _xfnum(_p.get("pct"))
|
||
_xtop.append({
|
||
"pos": f"{_pct:+.2f}%" if _pct is not None else "?",
|
||
"name": (_p.get("name") or _p.get("title") or "")
|
||
[:60], "cat": _p.get("asset_cat", "")})
|
||
else:
|
||
_xtop.append({"pos": f"${_p['value']:,.0f}",
|
||
"name": _p.get("text", "")[:60],
|
||
"cat": _p.get("cat", "")})
|
||
st.dataframe(pd.DataFrame(_xtop), width="stretch")
|
||
|
||
# ---- drawdown resilience: who was positive when equities crashed ----
|
||
with st.expander(
|
||
"Drawdown resilience - who was positive when equities "
|
||
"crashed"):
|
||
_DDF = _dc.RESULTS.parent / "drawdown_results.json"
|
||
if not _DDF.exists():
|
||
st.info("No drawdown screen on file yet "
|
||
"(run `python -m fundlab.drawdown`).")
|
||
else:
|
||
_d = json.loads(_DDF.read_text())
|
||
_ep = _d["episodes"]
|
||
st.dataframe(pd.DataFrame([
|
||
{"scenario": e["label"], "peak": e["peak"],
|
||
"trough": e["trough"],
|
||
"index drop": f"{e['min_dd']*100:.1f}%"}
|
||
for e in _ep]), width="stretch")
|
||
st.caption(
|
||
"Funds' total return over each peak->trough window "
|
||
"(their own NAV, first print after the peak to the "
|
||
"trough). Scenarios are detected from the index, not "
|
||
"hard-coded. Sorted by # scenarios positive.")
|
||
_df = _d["funds"]
|
||
_cands = {s: v for s, v in _df.items()
|
||
if v["verdict"].startswith("CANDIDATE")}
|
||
_rows = []
|
||
for s, v in _cands.items():
|
||
if v["n_pos"] < 3:
|
||
continue
|
||
_rows.append({
|
||
"fund": s.upper(),
|
||
"name": v["name"][:44],
|
||
**{e["label"]: (f"{v['rets'][e['label']]*100:+.1f}%"
|
||
if e["label"] in v["rets"] else "n/a")
|
||
for e in _ep},
|
||
"# pos": f"{v['n_pos']}/{v['n_avail']}",
|
||
"worst": f"{v['min_ret']*100:+.1f}%",
|
||
"corr port": (f"{v['corr_port']:+.2f}"
|
||
if isinstance(v.get("corr_port"),
|
||
(int, float)) else "—"),
|
||
"alpha t5": (f"{v['t5']:+.1f}"
|
||
if isinstance(v.get("t5"),
|
||
(int, float)) else "—"),
|
||
"_k": (v["n_pos"], v["n_avail"], v["min_ret"]),
|
||
})
|
||
_rows.sort(key=lambda r: (-r["_k"][0], -r["_k"][1],
|
||
r["_k"][2]))
|
||
for r in _rows:
|
||
r.pop("_k")
|
||
st.dataframe(pd.DataFrame(_rows), width="stretch")
|
||
st.caption(
|
||
"Note: being positive in every equity drawdown is mostly "
|
||
"a duration property - the 5/5 group is all "
|
||
"ultra-short/cash. The interesting rows are the alpha "
|
||
"funds with 4/5 (merger arb, market-neutral, "
|
||
"securitized credit) that still earned their 5y alpha.")
|
||
|
||
# ---- tax location: taxable account vs IRA -------------------------
|
||
with st.expander(
|
||
"Tax location - which account (taxable vs IRA) for each fund"):
|
||
_TP = _dc.RESULTS.parent / "taxplan_results.json"
|
||
if not _TP.exists():
|
||
st.info("No tax-location screen on file yet "
|
||
"(run `python -m fundlab.taxplan`).")
|
||
else:
|
||
_tp = json.loads(_TP.read_text())
|
||
st.caption(
|
||
"Where each fund's distributions should live, given the "
|
||
"current LTCG rate < future ordinary rate. Basis: N-PORT "
|
||
"holdings (16 shortlist + 22 cross-checked) or return-"
|
||
"sleeve proxy (250 candidates). 'score' = estimated share "
|
||
"of distributions that are tax-favorable (qualified "
|
||
"dividends + LTCG + tax-exempt). MIXED funds: pull the "
|
||
"last 1099-DIV - it is the final arbiter.")
|
||
|
||
_sp = {}
|
||
_SPF = _dc.RESULTS.parent / "taxsplit_results.json"
|
||
if _SPF.exists():
|
||
_spj = json.loads(_SPF.read_text())
|
||
for _grp in _spj.values():
|
||
for _s, _v in _grp.items():
|
||
if _v:
|
||
_sp.setdefault(_s.upper(), _v)
|
||
|
||
def _tp_table(funds: dict) -> pd.DataFrame:
|
||
rows = []
|
||
for s, r in sorted(funds.items()):
|
||
_v = _sp.get(s.upper())
|
||
rows.append({
|
||
"fund": s,
|
||
"name": r["name"][:44],
|
||
"location": r["location"],
|
||
"5y price": (f"{_v['price']*100:+.0f}%"
|
||
if _v else ""),
|
||
"5y payout": (f"{_v['payout_a']*100:.1f}%/yr"
|
||
if _v else ""),
|
||
"12m payout": (f"{_v['payout_12m']*100:.1f}%"
|
||
if _v else ""),
|
||
"score": r["score"],
|
||
"basis": r["basis"],
|
||
"notes": r["notes"][:120],
|
||
})
|
||
order = {"TAXABLE": 0, "TAXABLE (defers to LTCG)": 1,
|
||
"TAXABLE (munis)": 2,
|
||
"MIXED (check 1099)": 3, "IRA": 4,
|
||
"FLEXIBLE (cash)": 5, "NO DATA": 9}
|
||
df = pd.DataFrame(rows)
|
||
df["_o"] = df["location"].map(
|
||
lambda x: order.get(x, 5))
|
||
df = df.sort_values(["_o", "score"],
|
||
ascending=[True, False])
|
||
return df.drop(columns="_o")
|
||
|
||
st.markdown("**16-fund shortlist**")
|
||
st.dataframe(_tp_table(_tp["shortlist"]), width="stretch")
|
||
st.markdown("**22 cross-checked**")
|
||
st.dataframe(_tp_table(_tp["xcheck"]), width="stretch")
|
||
_sel = st.selectbox(
|
||
"Candidates (250)",
|
||
["All locations", "TAXABLE", "TAXABLE (defers to LTCG)",
|
||
"TAXABLE (munis)",
|
||
"MIXED (check 1099)", "IRA", "FLEXIBLE (cash)"],
|
||
key="_tp_loc")
|
||
_c = _tp["candidates"]
|
||
if _sel != "All locations":
|
||
_c = {s: r for s, r in _c.items()
|
||
if r["location"] == _sel}
|
||
st.dataframe(_tp_table(_c), width="stretch")
|
||
st.caption(
|
||
"Reading the table: TAXABLE = income is mostly qualified "
|
||
"dividends/LTCG (or tax-exempt) - the taxable account's "
|
||
"low LTCG rate is the benefit. IRA = ordinary interest / "
|
||
"STCG / non-qualified - deferral is the benefit. FLEXIBLE "
|
||
"= cash, no placement value either way. Note the merger-"
|
||
"arb trap: equity-looking books that distribute mostly "
|
||
"SHORT-TERM gains (HMEZX, MERVX) belong in the IRA.")
|
||
|
||
_f = _FUNDS.get(_fl_pick, {})
|
||
_man = _MAN.get(_fl_pick, {})
|
||
st.subheader(f"{_f.get('name', _fl_pick)} · {_fl_pick.upper()}")
|
||
st.write(_f.get("objective") or "_no objective on file_")
|
||
_src_bits = []
|
||
if _f.get("category"):
|
||
_src_bits.append(f"category: **{_f['category']}**")
|
||
if _man.get("filed"):
|
||
_src_bits.append(f"N-PORT filed {_man['filed']}")
|
||
if _man.get("url"):
|
||
_src_bits.append(f"[source filing]({_man['url']})")
|
||
if _src_bits:
|
||
st.caption(" · ".join(_src_bits))
|
||
|
||
snap = nport.build(_fl_pick, nport.FUND_TOKENS[_fl_pick])
|
||
if snap is None:
|
||
st.info("Holdings not available: "
|
||
+ (_man.get("note") or "no cached N-PORT filing for this fund."))
|
||
else:
|
||
c1, c2, c3, c4 = st.columns(4)
|
||
c1.metric("As of", snap.get("as_of") or "?")
|
||
c2.metric("Net assets",
|
||
f"${snap['net_assets']:,.0f}" if snap.get("net_assets") else "?")
|
||
c3.metric("Positions parsed", f"{snap['n_positions']}")
|
||
c4.metric("Categories", f"{len(snap['categories'])}")
|
||
|
||
lc, rc = st.columns([3, 2])
|
||
with lc:
|
||
st.markdown("**Reported composition** — the fund's own "
|
||
"category/percentage lines, in filing order")
|
||
if snap["categories"]:
|
||
cats = pd.DataFrame(snap["categories"])
|
||
fig = go.Figure(go.Bar(
|
||
x=cats["pct"].head(20)[::-1],
|
||
y=cats["name"].head(20)[::-1],
|
||
orientation="h", marker_color="#4c78a8"))
|
||
fig.update_layout(height=380, margin=dict(l=0, r=0, t=8, b=8),
|
||
xaxis_title="% of net assets")
|
||
st.plotly_chart(fig, width="stretch")
|
||
with st.expander("All category lines"):
|
||
st.dataframe(cats, width="stretch")
|
||
else:
|
||
st.caption("No category lines could be parsed from this filing.")
|
||
with rc:
|
||
st.markdown("**Rough keyword buckets** — my coarse parse of the "
|
||
"position lines, *not* the fund's own classification")
|
||
if snap["buckets"]:
|
||
bks = pd.DataFrame(snap["buckets"])
|
||
bks["value"] = bks["value"].map(lambda v: f"${v:,.0f}")
|
||
st.dataframe(bks, width="stretch")
|
||
else:
|
||
st.caption("No position values could be parsed.")
|
||
if snap["top"]:
|
||
with st.expander(f"Top positions ({len(snap['top'])})"):
|
||
top = pd.DataFrame(snap["top"])
|
||
top["value"] = top["value"].map(lambda v: f"${v:,.0f}")
|
||
st.dataframe(top, width="stretch")
|
||
|
||
# ---- returns-based strategy decomposition -------------------------
|
||
st.markdown("**Strategy decomposition** — which benchmark sleeves explain "
|
||
"the fund's returns (OLS forward selection, BIC-gated, "
|
||
"|t|>2; full history + last 5 years)")
|
||
_dr = _DR.get(_fl_pick, {})
|
||
if not _dr or "verdict" not in _dr:
|
||
st.info("No decomposition available for this fund.")
|
||
elif "r2" not in _dr:
|
||
st.info(_dr["verdict"])
|
||
if _dr.get("note"):
|
||
st.markdown("**Holdings cross-check:** " + _dr["note"])
|
||
else:
|
||
_rec = _dr.get("recent", {})
|
||
_v = _dr["verdict"]
|
||
if _v.startswith("static") or _v.startswith("mostly stable"):
|
||
st.success(_v)
|
||
elif "no component" in _v or "not a static" in _v:
|
||
st.warning(_v)
|
||
else:
|
||
st.info(_v)
|
||
_c = st.columns(6)
|
||
_c[0].metric("R² full", f"{_dr.get('r2', float('nan')):.3f}")
|
||
_c[1].metric("R² 5y",
|
||
f"{_rec['r2']:.3f}" if "r2" in _rec else "n/a")
|
||
_c[2].metric("alpha 5y",
|
||
f"{_rec['alpha_ann']*100:+.1f}% (t={_rec['alpha_t']:+.1f})"
|
||
if "alpha_ann" in _rec else "n/a")
|
||
_c[3].metric("trk err 5y",
|
||
f"{_rec['tracking_err_ann']*100:.1f}%"
|
||
if "tracking_err_ann" in _rec else "n/a")
|
||
_c[4].metric("max β-drift",
|
||
f"{_dr.get('rolling', {}).get('max_drift', 0.0):.2f}")
|
||
_c[5].metric("sample", f"{_dr['n_obs']}d", help=f"{_dr['start']} .. {_dr['end']}")
|
||
_rows = []
|
||
_full_b = {c["sym"]: c for c in _dr.get("components", [])}
|
||
_rec_b = {c["sym"]: c for c in _rec.get("components", [])}
|
||
for _sym in list(dict.fromkeys(list(_full_b) + list(_rec_b))):
|
||
_frow = _full_b.get(_sym)
|
||
_rrow = _rec_b.get(_sym)
|
||
_rows.append({
|
||
"component": f"{_sym} — {_frow['label'] if _frow else _rrow['label']}",
|
||
"β full": f"{_frow['beta']:+.2f} (t={_frow['t']:+.1f})" if _frow else "—",
|
||
"β 5y": f"{_rrow['beta']:+.2f} (t={_rrow['t']:+.1f})" if _rrow else "—",
|
||
})
|
||
if _rows:
|
||
st.dataframe(pd.DataFrame(_rows), width="stretch")
|
||
if _full_b:
|
||
_bar = pd.DataFrame([
|
||
{"component": f"{s} — {c['label']}", "full": c["beta"]}
|
||
for s, c in _full_b.items()])
|
||
fig = go.Figure(go.Bar(
|
||
x=_bar["full"], y=_bar["component"], orientation="h",
|
||
marker_color=["#2ca02c" if v >= 0 else "#d62728"
|
||
for v in _bar["full"]]))
|
||
fig.update_layout(height=30 + 24 * len(_bar),
|
||
margin=dict(l=0, r=0, t=8, b=8),
|
||
xaxis_title="β (full sample)")
|
||
st.plotly_chart(fig, width="stretch")
|
||
if _dr.get("note"):
|
||
st.markdown("**Holdings cross-check:** " + _dr["note"])
|
||
st.caption(
|
||
"Method: daily total returns IN EXCESS OF THE 3-MO T-BILL RATE "
|
||
"(BIL); greedy forward selection on BIC (add only if ΔBIC ≥ 2 and "
|
||
"|t| > 2); candidate sleeves curated per fund from the strategy "
|
||
"text + N-PORT. β = exposure, not a literal "
|
||
"holding weight; 'drift' = mean |rolling 1y β − full β| relative to "
|
||
"the full β. High-R² + low drift ≈ static mix; low R² with positive "
|
||
"alpha ≈ market-neutral/alpha strategy.")
|
||
|
||
if _f.get("strategy"):
|
||
st.markdown("**Strategy (excerpt from the prospectus)**")
|
||
st.write(_f["strategy"])
|
||
st.caption(
|
||
"Mockup: category lines are taken verbatim from the filing and "
|
||
"parsed heuristically (each family formats its schedule "
|
||
"differently), so expect the odd mis-parse. Positions are only "
|
||
"counted where the layout exposes a dollar value; the bucket "
|
||
"table is a rough keyword classification. N-PORT holdings are "
|
||
"quarterly and up to ~60 days stale.")
|
||
|
||
# --- closed-end funds (tax-arb, actual distribution character) ------
|
||
import pandas as _pd
|
||
_fl_dir = Path(__file__).parent / "fundlab"
|
||
_rank_path = _fl_dir / "cef_rank_all.json"
|
||
if _rank_path.exists():
|
||
_rank = json.loads(_rank_path.read_text())
|
||
_univ = json.loads((_fl_dir / "cef_universe.json").read_text())
|
||
_rows = []
|
||
for _score, _s, _ch, _t5, _vol, _npos, _tend, _disc in _rank:
|
||
_rows.append({
|
||
"sym": _s,
|
||
"name": (_univ.get(_s, {}).get("name") or "")[:44],
|
||
"char": _ch,
|
||
"disc%": None if _disc is None else round(100 * _disc, 1),
|
||
"t5%": None if _t5 is None else round(100 * _t5, 1),
|
||
"vol5%": None if _vol is None else round(100 * _vol, 1),
|
||
"pos_scen": _npos,
|
||
"tenders": _tend,
|
||
"score": _score})
|
||
_cef_df = _pd.DataFrame(_rows)
|
||
st.divider()
|
||
st.subheader("Closed-end funds — tax-arb ranking")
|
||
st.dataframe(_cef_df, width="stretch", height=460)
|
||
st.caption(
|
||
f"{len(_cef_df)} CEFs with VERIFIED per-share distribution "
|
||
"character (divs + gains + ROC == total distributions per year, "
|
||
"plus the NAV chain, from the fund's own report). char 0–1: "
|
||
"0 = all ordinary income, 1 = all capital-gains/ROC (taxable "
|
||
"income that defers to the LTCG/ROC rate in a taxable account). "
|
||
"score = char × (5y total return + 0.4 × 5y vol): harvestable "
|
||
"character weighted by how much damage a forced sale could do. "
|
||
"tenders = N-23C repurchase/tender filings (sponsor capital-"
|
||
"management pressure). disc% = FY-end market vs NAV.")
|