Greg Pomerantz
3fbf332b31
- fundlab/narrative.py: data-driven English prose per fund (performance, drivers tiered by fit, explicit 'what we do NOT know', bottom line) - fundlab/reportdata.py: static build -> reports/report_data.json - app.py Fund Lab Summary: at-a-glance table + per-fund expanders (narrative, equity curve, period table with fund-ref gap, drivers, reference mix, tax, cluster peers) - fundlab/report.py: narrative in the HTML report; forward-selected reference (weak-fit funds anchor to cash); SLEEVE_DESC exposure explanations - BUG: mix_series() never applied the betas (reference curves were raw sleeve sums; JLPSX 'reference' +407% vs fund +123%) - fixed and all reference curves/tables regenerated - reports/fund_report.html + report_data.json regenerated
346 lines
13 KiB
Python
346 lines
13 KiB
Python
"""Build reports/report_data.json: everything the web app (and the HTML
|
|
report) need for the per-fund report pages, in one structured file.
|
|
|
|
One entry per fund (11 alpha candidates + 13 shortlist funds):
|
|
meta name, verdict, group, order
|
|
narrative list of prose paragraphs (fundlab.narrative)
|
|
chart {dates, fund, ref, ivv} (downsampled price paths, rebased)
|
|
perf rows [period, fund, reference, ivv, fund-ref] (strings)
|
|
drivers markdown lines (reference model, signature, verdicts)
|
|
reference {rows [[loading, name, exposure]], weak, netcash, text}
|
|
tax plain text
|
|
peers {cluster, rows, proscons}
|
|
|
|
Run: .venv/bin/python -m fundlab.reportdata
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
|
|
from fundlab import report as _r
|
|
from fundlab import decompose
|
|
from fundlab.narrative import narrate
|
|
|
|
HERE = Path(__file__).parent
|
|
OUT = HERE.parent / "reports" / "report_data.json"
|
|
|
|
MAX_PTS = 1500
|
|
|
|
|
|
def _down(s: pd.Series) -> tuple[list, list]:
|
|
if s is None:
|
|
return [], []
|
|
if len(s) > MAX_PTS:
|
|
s = s.iloc[:: (len(s) // MAX_PTS + 1)]
|
|
return [str(d.date()) for d in s.index], [float(v) for v in s]
|
|
|
|
|
|
def _chart(sym: str, refs_curve: list[dict]) -> dict:
|
|
p = _r.price(sym)
|
|
fd, fv = _down(100 * p / p.iloc[0] if p is not None else None)
|
|
if refs_curve:
|
|
mix = _r.mix_series(refs_curve)
|
|
else:
|
|
b = _r.price("bil")
|
|
mix = 100 * b / b.iloc[0] if b is not None else None
|
|
md, mv = _down(mix)
|
|
iv = _r.price("ivv")
|
|
id_, ivv = _down(100 * iv / iv.iloc[0] if iv is not None else None)
|
|
return {"dates": fd, "fund": fv, "ref_dates": md, "ref": mv,
|
|
"ivv_dates": id_, "ivv": ivv}
|
|
|
|
|
|
def _perf_rows(sym: str, refs_curve: list[dict]) -> list[list[str]]:
|
|
p = _r.price(sym)
|
|
if p is None:
|
|
return []
|
|
mix = (_r.mix_series(refs_curve) if refs_curve
|
|
else _r.price("bil"))
|
|
iv = _r.price("ivv")
|
|
st = _r.perf_stats(p)
|
|
rows = []
|
|
|
|
def add(label, a, b):
|
|
fr_ = _r.window_ret(p, a, b)
|
|
mr = _r.window_ret(mix, a, b)
|
|
ir = _r.window_ret(iv, a, b)
|
|
gap = (fr_ - mr) if (fr_ is not None and mr is not None) else None
|
|
rows.append([label, _r.fmt_pct(fr_), _r.fmt_pct(mr),
|
|
_r.fmt_pct(ir), _r.fmt_pct(gap)])
|
|
|
|
add("Full history", st["start"], st["end"])
|
|
add("Last 5y", "2021-01-01", st["end"])
|
|
add("Last 1y", "2025-09-01", st["end"])
|
|
for lab, a, b in _r.EPISODES:
|
|
add(lab, a, b)
|
|
yr = p.resample("YE").last().dropna()
|
|
years = [str(yr.index[i].year) for i in range(1, len(yr))][-6:]
|
|
for i, y in enumerate(years):
|
|
a = f"{y}-01-01"
|
|
b = f"{int(y) + 1}-01-01" if i < len(years) - 1 else st["end"]
|
|
add(y, a, b)
|
|
return rows
|
|
|
|
|
|
def _drivers(sym: str, fr: dict, srow: dict, dr: dict) -> list[str]:
|
|
out = []
|
|
a5 = srow.get("alpha_ann_5y")
|
|
t5 = srow.get("alpha_t_5y")
|
|
r25 = srow.get("r2_5y")
|
|
if isinstance(r25, (int, float)) or isinstance(a5, (int, float)):
|
|
out.append(
|
|
f"**Reference model, last 5 years:** R² = "
|
|
f"{_r.fmt_r2(r25 if isinstance(r25, (int, float)) else None)}, "
|
|
f"alpha = "
|
|
f"{_r.fmt_pct(a5) if isinstance(a5, (int, float)) else '—'}"
|
|
f"{' (t = ' + format(t5, '+.1f') + ')' if isinstance(t5, (int, float)) else ''}")
|
|
a_f = (fr.get("full") or {}).get("alpha_ann")
|
|
t_f = (fr.get("full") or {}).get("alpha_t")
|
|
r2f = srow.get("r2_full")
|
|
if isinstance(r2f, (int, float)) or isinstance(a_f, (int, float)):
|
|
out.append(
|
|
f"**Reference model, full history:** R² = "
|
|
f"{_r.fmt_r2(r2f if isinstance(r2f, (int, float)) else None)}, "
|
|
f"alpha = "
|
|
f"{_r.fmt_pct(a_f) if isinstance(a_f, (int, float)) else '—'}"
|
|
f"{' (t = ' + format(t_f, '+.1f') + ')' if isinstance(t_f, (int, float)) else ''}.")
|
|
f = fr.get("rec5") or {}
|
|
betas = f.get("betas") or {}
|
|
top = sorted(betas.items(), key=lambda kv: -abs(kv[1]))[:6]
|
|
if top:
|
|
out.append("**Return-driver signature (34 sleeves):** "
|
|
+ ", ".join(f"`{s}` {b:+.2f}" for s, b in top)
|
|
+ f" - net cash {1 - sum(betas.values()):+.2f}.")
|
|
if dr and "verdict" in dr:
|
|
out.append(f"**Decomposition verdict:** {dr['verdict']}")
|
|
roll = dr.get("rolling") or {}
|
|
if roll.get("max_drift") is not None:
|
|
out.append(f"Weight stability: max 1y β-drift = "
|
|
f"{roll['max_drift']:.2f} (0 = perfectly stable, "
|
|
f">1 = unstable).")
|
|
if dr.get("note"):
|
|
out.append(f"*{dr['note']}*")
|
|
if srow.get("verdict"):
|
|
out.append(f"**Screen verdict:** {srow['verdict']}")
|
|
return out
|
|
|
|
|
|
def _reference(sym: str, fr: dict, refs: list[dict],
|
|
refs_curve: list[dict]) -> dict:
|
|
rows = []
|
|
for c in refs:
|
|
nm, desc = _r.sleeve_desc(c["sym"])
|
|
rows.append([f"{c['sym'].upper()} {c['beta']:+.2f}", nm, desc])
|
|
sb = sum(c["beta"] for c in refs)
|
|
cash = 1 - sb
|
|
if cash > 0.05:
|
|
net = f"loadings sum to {sb:.2f} → ~{cash:.0%} net cash"
|
|
elif cash < -0.05:
|
|
net = f"loadings sum to {sb:.2f} → ~{-cash:.0%} net levered"
|
|
else:
|
|
net = f"loadings sum to {sb:.2f} → fully invested"
|
|
weak = bool(refs) and not refs_curve
|
|
text = ("**Weak fit - read with care.** These loadings are each "
|
|
"individually significant but collectively explain little; "
|
|
"the performance table anchors to cash, not this mix."
|
|
if weak else
|
|
"The reference is the fund's own fitted sleeve mix; "
|
|
"'alpha' everywhere means outperformance vs this mix, in "
|
|
"excess of the T-bill rate.")
|
|
return {"rows": rows, "weak": weak, "netcash": net, "text": text}
|
|
|
|
|
|
def _tax(sym: str) -> str:
|
|
t = _r.tax_row(sym)
|
|
if not t:
|
|
return "No tax classification on file."
|
|
loc = t.get("location", "?")
|
|
basis = t.get("basis", "?")
|
|
conf = "from actual N-PORT holdings (high confidence)" \
|
|
if basis == "N-PORT" else "from the return-sleeve mix (model)"
|
|
s = (f"Character score {t.get('score', '?')} ({conf}). ")
|
|
if "MIXED" in loc:
|
|
s += "Model says MIXED - the last 1099-DIV is the arbiter."
|
|
elif "TAXABLE" in loc:
|
|
if "munis" in loc:
|
|
s += ("Keep in the **taxable** account "
|
|
"(munis - tax-exempt interest is wasted in an IRA).")
|
|
elif "defers" in loc:
|
|
s += ("Keep in the **taxable** account "
|
|
"(income mostly defers to the LTCG/ROC rate).")
|
|
else:
|
|
s += "Keep in the **taxable** account."
|
|
else:
|
|
s += f"Recommended account: **{loc}**."
|
|
if t.get("notes"):
|
|
s += f" _{t['notes']}_"
|
|
return s
|
|
|
|
|
|
def _peers(sym: str, fr: dict) -> dict:
|
|
if sym in _r.MEMBER:
|
|
cid, clabel, cn = _r.MEMBER[sym]
|
|
members = _r.KMEANS["clusters"][cid]["syms"]
|
|
else:
|
|
try:
|
|
cid, clabel = _r.cluster_of_row(_r.fund_loading_row(fr))
|
|
except Exception:
|
|
return {}
|
|
cn = _r.KMEANS["clusters"].get(cid, {}).get("n", 0)
|
|
members = _r.KMEANS["clusters"].get(cid, {}).get("syms", [])
|
|
peers = []
|
|
for s in members:
|
|
if s == sym:
|
|
continue
|
|
v = _r.search_row(s)
|
|
if isinstance(v.get("alpha_t_5y"), (int, float)):
|
|
peers.append((v["alpha_t_5y"], s))
|
|
peers.sort(key=lambda x: (x[0] > 0, x[0]), reverse=True)
|
|
pos = [p for p in peers if p[0] > 0]
|
|
peers = [s for _t, s in (pos + [p for p in peers if p[0] <= 0])[:4]]
|
|
|
|
def statline(s: str) -> list[str]:
|
|
p = _r.price(s)
|
|
if p is None:
|
|
return ["—", "—", "—", "—", "—", "n/a"]
|
|
st = _r.perf_stats(p)
|
|
v = _r.search_row(s)
|
|
t5 = _r.window_ret(p, "2021-01-01", st["end"])
|
|
a5 = v.get("alpha_ann_5y")
|
|
tt = v.get("alpha_t_5y")
|
|
tax = (_r.tax_row(s) or {}).get("location") or "n/a"
|
|
return [_r.fmt_pct(t5), _r.fmt_pct(st["cagr"]),
|
|
_r.fmt_pct(st["mdd"]),
|
|
_r.fmt_r2(v.get("r2_5y")),
|
|
(f"{_r.fmt_pct(a5)} (t={tt:+.1f})"
|
|
if isinstance(a5, (int, float)) else "—"), tax]
|
|
|
|
rows = [[f"{sym.upper()} (this fund)", *statline(sym)]]
|
|
for s in peers:
|
|
rows.append([s.upper(), *statline(s)])
|
|
|
|
# pros / cons
|
|
adv, dis = [], []
|
|
p = _r.price(sym)
|
|
if p is not None and peers:
|
|
st = _r.perf_stats(p)
|
|
t5 = _r.window_ret(p, "2021-01-01", st["end"])
|
|
vals = {}
|
|
for s in peers:
|
|
pp = _r.price(s)
|
|
if pp is None:
|
|
continue
|
|
ss = _r.perf_stats(pp)
|
|
vals[s] = (_r.window_ret(pp, "2021-01-01", ss["end"]),
|
|
ss["mdd"], ss["vol"])
|
|
if vals:
|
|
best_t5 = max(v[0] for v in vals.values() if v[0] is not None)
|
|
best_dd = max(v[1] for v in vals.values())
|
|
low_vol = min(v[2] for v in vals.values())
|
|
if t5 is not None and t5 >= best_t5 - 0.02:
|
|
adv.append("5y return at the top of the cluster")
|
|
elif t5 is not None and t5 < best_t5 - 0.10:
|
|
dis.append(f"5y return trails the best peer by "
|
|
f"{100 * (best_t5 - t5):.0f}pp")
|
|
if st["mdd"] < best_dd - 0.05:
|
|
adv.append("sharpest drawdown in the cluster")
|
|
elif st["mdd"] > best_dd + 0.05:
|
|
dis.append(f"deeper drawdown than the calmest peer "
|
|
f"({_r.fmt_pct(st['mdd'])} vs "
|
|
f"{_r.fmt_pct(best_dd)})")
|
|
if st["vol"] < low_vol - 0.02:
|
|
adv.append("lowest volatility in the cluster")
|
|
elif st["vol"] > low_vol + 0.05:
|
|
dis.append("meaningfully more volatile than the "
|
|
"calmest peer")
|
|
if adv:
|
|
pc = "**Advantages vs peers:** " + "; ".join(adv) + "."
|
|
elif dis:
|
|
pc = "**Disadvantages vs peers:** " + "; ".join(dis) + "."
|
|
else:
|
|
pc = ("Middle of the cluster - no decisive edge on return, "
|
|
"drawdown or volatility; the choice comes down to alpha "
|
|
"quality (t), tax fit and conviction in the strategy.")
|
|
if dis and adv:
|
|
pc += " **Disadvantages:** " + "; ".join(dis) + "."
|
|
return {"cluster": clabel, "n": cn,
|
|
"rows": rows, "proscons": pc}
|
|
|
|
|
|
def build_fund(sym: str, group: str, order: int) -> dict:
|
|
t0 = time.time()
|
|
fr = _r.factor_row(sym)
|
|
dr = _r.decomp_row(sym)
|
|
srow = _r.search_row(sym)
|
|
name = (srow.get("name") or _r.FONDS.get(sym, {}).get("name")
|
|
or (fr or {}).get("name") or sym.upper())
|
|
refs = _r.ref_components(sym, fr)
|
|
r2f_ = (fr.get("rec5") or fr.get("full") or {}).get("r2")
|
|
r25 = srow.get("r2_5y") if isinstance(srow.get("r2_5y"), (int, float)) \
|
|
else None
|
|
fit = r25 if r25 is not None else r2f_
|
|
refs_curve = refs if (fit is not None and fit >= 0.5) else []
|
|
p = _r.price(sym)
|
|
st_ = _r.perf_stats(p) if p is not None else {}
|
|
stats = {"t5y": _r.fmt_pct(_r.window_ret(p, "2021-01-01",
|
|
st_["end"])) if p is not None
|
|
else "—",
|
|
"cagr": _r.fmt_pct(st_.get("cagr")),
|
|
"mdd": _r.fmt_pct(st_.get("mdd")),
|
|
"r2_5y": _r.fmt_r2(srow.get("r2_5y")),
|
|
"alpha_5y": (f"{_r.fmt_pct(srow['alpha_ann_5y'])} "
|
|
f"(t={srow['alpha_t_5y']:+.1f})"
|
|
if isinstance(srow.get("alpha_ann_5y"),
|
|
(int, float)) else "—")}
|
|
d = {
|
|
"meta": {"sym": sym, "name": name, "group": group, "order": order,
|
|
"verdict": srow.get("verdict", ""),
|
|
"objective": (_r.FONDS.get(sym) or {}).get("objective",
|
|
"")},
|
|
"stats": stats,
|
|
"narrative": narrate(sym),
|
|
"chart": _chart(sym, refs_curve),
|
|
"perf": _perf_rows(sym, refs_curve),
|
|
"drivers": _drivers(sym, fr, srow, dr),
|
|
"reference": _reference(sym, fr, refs, refs_curve),
|
|
"tax": _tax(sym),
|
|
"peers": _peers(sym, fr),
|
|
}
|
|
print(f" {sym:8s} {time.time() - t0:5.1f}s", flush=True)
|
|
return d
|
|
|
|
|
|
def main() -> None:
|
|
from fundlab.decompose import ALIAS
|
|
from fundlab.nport import FUND_TOKENS
|
|
t0 = time.time()
|
|
cand = [v["sym"] for v in _r.SEARCH.values()
|
|
if isinstance(v, dict)
|
|
and str(v.get("verdict", "")).startswith("CANDIDATE")]
|
|
cand.sort(key=lambda s: -(_r.SEARCH[s]["alpha_t_5y"]
|
|
if isinstance(_r.SEARCH[s].get("alpha_t_5y"),
|
|
(int, float)) else -9))
|
|
short = [s for s in FUND_TOKENS if s not in ALIAS]
|
|
data = {"generated": time.strftime("%Y-%m-%d %H:%M"),
|
|
"funds": {}}
|
|
i = 0
|
|
for s in cand:
|
|
i += 1
|
|
data["funds"][s] = build_fund(s, "candidate", i)
|
|
for s in short:
|
|
i += 1
|
|
data["funds"][s] = build_fund(s, "shortlist", i)
|
|
OUT.parent.mkdir(exist_ok=True)
|
|
OUT.write_text(json.dumps(data))
|
|
print(f"wrote {OUT} in {time.time() - t0:.0f}s "
|
|
f"({OUT.stat().st_size / 1e6:.1f} MB, {len(data['funds'])} funds)")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|