Greg Pomerantz
bdb887dee8
- No fixed risk-free rate: the statistics page now nets Sharpe, Sortino and CAPM alpha against BIL (SPDR 1-3 Month T-Bill) daily total returns from the data bundle — the same reference the Fund Lab uses for all its alphas (pre-2007 dates fill 0, as in fundlab.decompose). - metrics: sharpe/sortino/beta_alpha/summary accept a daily rf SERIES (or scalar annual rate as before) via a shared excess() helper. - Also fixes a latent double-count: the old scalar-rf alpha subtracted rf twice (once in the returns, once in the intercept term); alpha is now mean(excess fund) - beta * mean(excess bench), the standard CAPM intercept on excess returns. - Page caption states the T-bill reference (or warns if BIL is absent). - Tests: series-rf identities (sharpe/sortino/beta/alpha), benchmark self-row beta 1 / alpha 0, caption check.
284 lines
13 KiB
Python
284 lines
13 KiB
Python
"""App-level tests (Streamlit AppTest — no browser needed).
|
|
|
|
Run: .venv/bin/python tests/test_app.py
|
|
Exit code 0 = all passed.
|
|
|
|
Memory note: one data-bundle load is ~2.3 GB and the server holds one copy,
|
|
so this test process may keep at most ONE AppTest alive at a time —
|
|
always go through run_app() (it releases the previous handle first).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import gc
|
|
import json
|
|
import logging
|
|
import pathlib
|
|
import sys
|
|
|
|
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent))
|
|
|
|
logging.getLogger("streamlit").setLevel(logging.ERROR)
|
|
from streamlit.testing.v1 import AppTest # noqa: E402
|
|
|
|
import portfolios as pf # noqa: E402
|
|
|
|
ROOT = pathlib.Path(__file__).resolve().parent.parent
|
|
SETTINGS = ROOT / "settings.json"
|
|
|
|
PASS, FAIL = 0, 0
|
|
_A: AppTest | None = None
|
|
|
|
|
|
def check(name: str, cond: bool, extra: str = "") -> None:
|
|
global PASS, FAIL
|
|
if cond:
|
|
PASS += 1
|
|
print(f" ok {name}", flush=True)
|
|
else:
|
|
FAIL += 1
|
|
print(f" FAIL {name} {extra}", flush=True)
|
|
|
|
|
|
def app() -> AppTest:
|
|
"""The single live AppTest (set by run_app)."""
|
|
return _A
|
|
|
|
|
|
def run_app(spec: str | None = None, bench: str | None = None,
|
|
**extra_state) -> AppTest:
|
|
"""Run the app with the given committed inputs.
|
|
|
|
spec/bench=None leaves the widget default (restored from settings.json).
|
|
Releases the previous AppTest first to bound memory.
|
|
"""
|
|
global _A
|
|
_A = None
|
|
gc.collect()
|
|
_A = AppTest.from_file(str(ROOT / "app.py"), default_timeout=120)
|
|
if spec is not None:
|
|
_A.session_state["spec"] = spec
|
|
if bench is not None:
|
|
_A.session_state["bench_spec"] = bench
|
|
for k, v in extra_state.items():
|
|
_A.session_state[k] = v
|
|
_A.run()
|
|
return _A
|
|
|
|
|
|
def main() -> int:
|
|
saved_settings = SETTINGS.read_text() if SETTINGS.exists() else None
|
|
SETTINGS.unlink(missing_ok=True)
|
|
try:
|
|
_run()
|
|
finally:
|
|
if saved_settings is not None:
|
|
SETTINGS.write_text(saved_settings)
|
|
for name in ("ttest", "oldfmt"):
|
|
pf.delete(name)
|
|
print(f"\n{PASS} passed, {FAIL} failed")
|
|
return 1 if FAIL else 0
|
|
|
|
|
|
def _run() -> None:
|
|
# NOTE: never bind app() to a local that outlives the next run_app() —
|
|
# each live AppTest pins a ~2.3 GB bundle; there may be at most one.
|
|
print("spec parsing", flush=True)
|
|
run_app("MSFT")
|
|
check("single symbol renders", not app().exception and
|
|
any("single symbol" in t.value for t in app().title))
|
|
run_app("MSFT:0.6,V:0.4")
|
|
check("comma portfolio renders", not app().exception and
|
|
any(t.value == "Portfolio: msft, v" for t in app().title))
|
|
run_app("MSFT V googl:0.5,amzn:0.5")
|
|
check("multiple entries render", not app().exception and
|
|
any("Portfolios:" in t.value for t in app().title))
|
|
run_app("MSFT:xyz")
|
|
check("bad weight -> error", not app().exception and
|
|
any("Invalid input" in e.value for e in app().error))
|
|
run_app("ZZZNOPE")
|
|
check("unknown symbol -> field-labelled error", not app().exception and
|
|
any("Symbol field" in e.value and "not in the data" in e.value
|
|
for e in app().error))
|
|
run_app("")
|
|
check("empty -> info, no crash", not app().exception and bool(app().info))
|
|
|
|
print("benchmarks", flush=True)
|
|
run_app("MSFT", "V googl:0.5,amzn:0.5")
|
|
blocks = [d.value for d in app().main.tabs[0].dataframe]
|
|
check("one stats block per benchmark", not app().exception
|
|
and len(blocks) == 2, f"{len(blocks)} blocks")
|
|
check("each block lists its own benchmark",
|
|
"benchmark: v" in list(blocks[0].index)
|
|
and "benchmark: googl, amzn" in list(blocks[1].index))
|
|
check("per-block beta columns (no suffix)",
|
|
"beta" in blocks[0].columns and "beta" in blocks[1].columns
|
|
and not any("[" in c for c in blocks[0].columns))
|
|
corr_blocks = [d.value for d in app().main.tabs[4].dataframe]
|
|
check("single-component portfolio: no components table",
|
|
len(corr_blocks) == 1,
|
|
f"{len(corr_blocks)} blocks")
|
|
check("all-portfolios cross-correlation table",
|
|
"Current" in corr_blocks[0].index and "benchmark: v" in corr_blocks[0].index
|
|
and list(corr_blocks[0].columns) == [1, 2, 3]
|
|
and abs(float(corr_blocks[0].loc["Current", 1]) - 1.0) < 1e-9)
|
|
run_app("MSFT:0.5,goog:0.5", "V")
|
|
corr_blocks = [d.value for d in app().main.tabs[4].dataframe]
|
|
corr, allcorr = corr_blocks[0], corr_blocks[-1]
|
|
check("multi-component portfolio gets a components table",
|
|
len(corr_blocks) == 2
|
|
and "msft" in corr.index and "goog" in corr.index
|
|
and "benchmark: v" in corr.index
|
|
and list(corr.columns) == [1, 2, 3]
|
|
and abs(float(corr.loc["msft", 1]) - 1.0) < 1e-9)
|
|
run_app("MSFT", "V zzzznope")
|
|
blocks = [d.value for d in app().main.tabs[0].dataframe]
|
|
rows = list(blocks[0].index) if blocks else []
|
|
check("invalid benchmark warns, valid survives", not app().exception and
|
|
any("Benchmark 2 ignored" in w.value for w in app().sidebar.warning)
|
|
and len(blocks) == 1 and "benchmark: v" in rows)
|
|
run_app("MSFT:0.6,V:0.4", "V")
|
|
blocks = [d.value for d in app().main.tabs[0].dataframe]
|
|
rows = list(blocks[0].index) if blocks else []
|
|
check("no component rows for portfolio", "msft" not in rows and "v" not in rows)
|
|
check("after-tax benchmark row present in both mode",
|
|
"benchmark: v (after-tax)" in rows if "benchmark: v (after-tax)" in rows
|
|
else True) # only in both-mode; default mode has plain names
|
|
|
|
print("statistics order", flush=True)
|
|
run_app("MSFT", "V", stats_order="sharpe,calmar")
|
|
cols = list(app().main.tabs[0].dataframe[0].value.columns)
|
|
check("stats columns follow configured order",
|
|
not app().exception and cols == ["sharpe", "calmar"], str(cols))
|
|
run_app("MSFT", "V", stats_order="sharpe,nonsense")
|
|
cols = list(app().main.tabs[0].dataframe[0].value.columns)
|
|
check("unknown statistic warned and ignored",
|
|
cols == ["sharpe"]
|
|
and any("nonsense" in w.value for w in app().sidebar.warning), str(cols))
|
|
run_app("MSFT", "V", stats_order="")
|
|
cols = list(app().main.tabs[0].dataframe[0].value.columns)
|
|
check("empty field shows all in default order",
|
|
len(cols) == 10 and cols[0] == "total_return" and "beta" in cols
|
|
and "return" in cols and "vol" in cols and "alpha" in cols
|
|
and "ann_return" not in cols and "ann_vol" not in cols,
|
|
str(cols))
|
|
run_app("MSFT", "V", stats_order="ann_return,ann_vol,alpha_ann")
|
|
cols = list(app().main.tabs[0].dataframe[0].value.columns)
|
|
check("old statistic names migrated", cols == ["return", "vol", "alpha"],
|
|
str(cols))
|
|
|
|
print("risk-free rate (BIL)", flush=True)
|
|
import metrics as _m
|
|
import numpy as _np
|
|
import pandas as _pd
|
|
idx = _pd.bdate_range("2020-01-01", periods=252)
|
|
_rng = _np.random.default_rng(7)
|
|
r = _pd.Series(_rng.normal(0.0005, 0.01, len(idx)), idx)
|
|
b = _pd.Series(_rng.normal(0.0002, 0.008, len(idx)), idx)
|
|
c = 0.0002
|
|
rf_s = _pd.Series(c, idx)
|
|
check("series rf: sharpe equals rf-shifted series",
|
|
abs(_m.sharpe(r, rf_s) - _m.sharpe(r - c)) < 1e-12)
|
|
check("series rf: sortino equals rf-shifted series",
|
|
abs(_m.sortino(r, rf_s) - _m.sortino(r - c)) < 1e-12)
|
|
check("series rf: beta invariant, alpha = shifted-series alpha",
|
|
abs(_m.beta_alpha(r, b, rf_s)[0] - _m.beta_alpha(r - c, b - c, 0.0)[0]) < 1e-12
|
|
and abs(_m.beta_alpha(r, b, rf_s)[1] - _m.beta_alpha(r - c, b - c, 0.0)[1]) < 1e-12)
|
|
check("scalar rf equals daily-constant series rf",
|
|
abs(_m.sharpe(r, c * _m.ANN) - _m.sharpe(r, rf_s)) < 1e-12)
|
|
|
|
run_app("MSFT", "V", stats_order="sharpe,sortino,beta,alpha,ann_return_bench")
|
|
t0 = app().main.tabs[0].dataframe[0].value
|
|
beta_bench = float(t0.loc["benchmark: v", "beta"])
|
|
alpha_bench = float(t0.loc["benchmark: v", "alpha"][:-1]) / 100
|
|
# the benchmark vs ITSELF must be beta 1 / alpha 0 — catches the
|
|
# monthly-resampled benchmark (daily-vs-monthly returns mismatch)
|
|
check("benchmark row is beta 1 / alpha 0 (daily-aligned regression)",
|
|
not app().exception and abs(beta_bench - 1.0) < 1e-9
|
|
and abs(alpha_bench) < 1e-9,
|
|
f"beta={beta_bench} alpha={alpha_bench}")
|
|
caps = " ".join(c.value for c in app().main.caption)
|
|
check("caption states the T-bill reference",
|
|
"3-mo T-bill (BIL)" in caps, caps[:400])
|
|
|
|
print("date range", flush=True)
|
|
# total_return (cumulative) is range-sensitive; the annualized 'return'
|
|
# can coincidentally match at 1-decimal precision
|
|
_so = "total_return" # pin: earlier tests persist other selections
|
|
run_app("MSFT", "V", stats_order=_so)
|
|
full_ret = app().main.tabs[0].dataframe[0].value.loc["Current", "total_return"]
|
|
run_app("MSFT", "V", stats_order=_so, range_end="20151231")
|
|
ret = app().main.tabs[0].dataframe[0].value.loc["Current", "total_return"]
|
|
check("end date limits the statistics", not app().exception
|
|
and ret != full_ret, f"{ret} vs {full_ret}")
|
|
run_app("MSFT", "V", stats_order=_so, range_start="20120102")
|
|
ret = app().main.tabs[0].dataframe[0].value.loc["Current", "total_return"]
|
|
check("start date overrides the window", not app().exception
|
|
and ret != full_ret, f"{ret} vs {full_ret}")
|
|
run_app("MSFT", "V", stats_order=_so, date_window="1Y")
|
|
ret = app().main.tabs[0].dataframe[0].value.loc["Current", "total_return"]
|
|
check("window applies to the statistics", not app().exception
|
|
and ret != full_ret, f"{ret} vs {full_ret}")
|
|
run_app("MSFT", "V", stats_order=_so, range_start="notadate")
|
|
check("invalid start date warns and is ignored",
|
|
not app().exception
|
|
and any("Invalid start date" in w.value for w in app().sidebar.warning))
|
|
|
|
print("curve mode", flush=True)
|
|
for mode, suffixed in (("Pre-tax", False), ("After-tax", False),
|
|
("Pre-tax + after-tax", True)):
|
|
run_app("MSFT:0.6,V:0.4", "V", curve_mode=mode)
|
|
rows = list(app().main.tabs[0].dataframe[0].value.index)
|
|
if suffixed:
|
|
check(f"mode {mode}: suffixed names",
|
|
any(x.endswith(" (pre-tax)") for x in rows) and
|
|
any(x.endswith(" (after-tax)") for x in rows))
|
|
else:
|
|
check(f"mode {mode}: plain names", "Current" in rows and
|
|
not any(x.endswith(" (pre-tax)") for x in rows))
|
|
|
|
print("save / load / delete", flush=True)
|
|
run_app("msft:0.6,v:0.4")
|
|
app().sidebar.text_input(key="save_name").set_value("ttest").run()
|
|
next(b for b in app().sidebar.button if "Save" in b.label).click()
|
|
app().run()
|
|
specs = {p.name: p.spec for p in pf.load_all()}
|
|
check("save writes comma-no-space spec",
|
|
specs.get("ttest") == "msft:0.6,v:0.4", str(specs))
|
|
run_app() # fresh session (releases the previous AppTest)
|
|
next(s for s in app().sidebar.selectbox if s.key == "load_name").set_value("ttest").run()
|
|
next(b for b in app().sidebar.button if b.label == "Load").click()
|
|
app().run()
|
|
check("load fills field with valid spec", app().session_state["spec"] == "msft:0.6,v:0.4")
|
|
app().sidebar.text_input(key="del_name").set_value("ttest").run()
|
|
next(b for b in app().sidebar.button if b.label == "Delete").click()
|
|
app().run()
|
|
check("delete removes it", all(p.name != "ttest" for p in pf.load_all()))
|
|
pf.save(pf.Portfolio("oldfmt", "msft:0.6, v:0.4", None, 0.0))
|
|
run_app()
|
|
next(s for s in app().sidebar.selectbox if s.key == "load_name").set_value("oldfmt").run()
|
|
next(b for b in app().sidebar.button if b.label == "Load").click()
|
|
app().run()
|
|
check("old-format spec normalized on load",
|
|
app().session_state["spec"] == "msft:0.6,v:0.4")
|
|
|
|
print("persistence (settings.json)", flush=True)
|
|
run_app("MSFT:0.6,V:0.4", "googl:0.5,amzn:0.5 v",
|
|
curve_mode="After-tax", date_window="5Y",
|
|
range_start="20120102", range_end="", stats_order="sharpe,vol")
|
|
s = json.loads(SETTINGS.read_text())
|
|
check("settings written", s.get("spec") == "MSFT:0.6,V:0.4" and
|
|
s.get("bench_spec") == "googl:0.5,amzn:0.5 v" and
|
|
s.get("curve_mode") == "After-tax" and s.get("date_window") == "5Y" and
|
|
s.get("range_start") == "20120102" and s.get("stats_order") == "sharpe,vol",
|
|
str(s))
|
|
run_app() # fresh session -> widgets must restore from settings.json
|
|
ti = {t.key: t.value for t in app().sidebar.text_input
|
|
if t.key in ("spec", "bench_spec", "range_start")}
|
|
check("settings restored in fresh session",
|
|
ti == {"spec": "MSFT:0.6,V:0.4", "bench_spec": "googl:0.5,amzn:0.5 v",
|
|
"range_start": "20120102"}, str(ti))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|