Greg Pomerantz
d5a7bb70b0
- Pass the benchmark's DAILY price series to metrics.summary() instead of a month-end-resampled one; summary() derives daily returns and annualizes with 252d, so the old resample made beta/alpha regress the fund's month-end daily returns against whole-month benchmark returns (and skewed ann_return_bench). - New sidebar setting 'Risk-free rate %' (default 4%, persisted) passed through to summary(), so Sharpe, Sortino and CAPM alpha are computed in excess of rf; noted in the page caption and stats help.
277 lines
12 KiB
Python
277 lines
12 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", flush=True)
|
|
run_app("MSFT", "V", stats_order="sharpe,sortino,beta,alpha,ann_return_bench")
|
|
t0 = app().main.tabs[0].dataframe[0].value
|
|
alpha0 = t0.loc["Current", "alpha"]
|
|
sharpe0 = t0.loc["Current", "sharpe"]
|
|
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}")
|
|
rf_widget = next(n for n in app().sidebar.number_input
|
|
if "Risk-free" in n.label)
|
|
check("default risk-free rate is 4%", rf_widget.value == 4.0,
|
|
str(rf_widget.value))
|
|
rf_widget.set_value(10.0).run()
|
|
t1 = app().main.tabs[0].dataframe[0].value
|
|
check("alpha adjusts for the risk-free rate",
|
|
not app().exception and t1.loc["Current", "alpha"] != alpha0,
|
|
f"{t1.loc['Current', 'alpha']} vs {alpha0}")
|
|
check("sharpe adjusts for the risk-free rate",
|
|
t1.loc["Current", "sharpe"] != sharpe0,
|
|
f"{t1.loc['Current', 'sharpe']} vs {sharpe0}")
|
|
rf_widget.set_value(4.0).run()
|
|
|
|
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())
|