app: background cache refresh, per-benchmark stats, correlation tab, global date range

- data.py: non-blocking load_bundle(); background watcher thread refreshes
  the parquet cache (5s scan, 30s min rebuild cadence); refresh()/
  up_to_date()/generation()
- statistics tab: one table per benchmark (vs <label>), plain column names
  (beta/alpha/return/vol...), selectable+reorderable stat list in
  settings.json
- correlation tab: per-portfolio components-vs-benchmarks +
  all-portfolios-vs-benchmarks; numbered columns
- global date range (window radio + start/end boxes) applied to all tabs;
  metrics.xcorr(); equity window radio gains YTD/3M/1M
This commit is contained in:
Greg Pomerantz 2026-08-25 18:15:47 -04:00
parent cba7291676
commit 4f36bc7aea
6 changed files with 403 additions and 81 deletions

1
.gitignore vendored
View File

@ -3,3 +3,4 @@ __pycache__/
*.pyc *.pyc
.cache/ .cache/
settings.json settings.json
funds.json

256
app.py
View File

@ -14,7 +14,7 @@ import plotly.graph_objects as go
import streamlit as st import streamlit as st
import metrics as m import metrics as m
from data import DEFAULT_ROOT, load_bundle, search_symbols from data import DEFAULT_ROOT, generation, load_bundle, search_symbols, up_to_date
from portfolio import parse_items, parse_weights, portfolio_returns 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 portfolios import Portfolio, delete as pf_delete, load_all as pf_load, save as pf_save
from tax import after_tax_portfolio from tax import after_tax_portfolio
@ -52,6 +52,9 @@ root = st.sidebar.text_input("Data root", str(DEFAULT_ROOT))
rebuild = st.sidebar.checkbox("Rebuild parquet cache", value=False) rebuild = st.sidebar.checkbox("Rebuild parquet cache", value=False)
with st.spinner("Loading data..."): with st.spinner("Loading data..."):
bundle = load_bundle(root=Path(root), rebuild=rebuild) 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 # data symbols are stored lowercase; input is resolved case-insensitively
_LOWER2SYM = {s.lower(): s for s in bundle.adj.columns} _LOWER2SYM = {s.lower(): s for s in bundle.adj.columns}
@ -126,6 +129,13 @@ st_rate = st.sidebar.number_input("Short-term gains %", 0.0, 49.0,
div_rate = st.sidebar.number_input("Dividends %", 0.0, 49.0, div_rate = st.sidebar.number_input("Dividends %", 0.0, 49.0,
float(_settings.get("div_rate", 15.0))) / 100 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") st.sidebar.subheader("Period & benchmark")
_p = int(_settings.get("period", 2010)) _p = int(_settings.get("period", 2010))
period = st.sidebar.selectbox("From year", range(2002, 2026), period = st.sidebar.selectbox("From year", range(2002, 2026),
@ -140,10 +150,74 @@ bench_spec = st.sidebar.text_input(
"tax rules. Beta/alpha in the stats table refers to the first " "tax rules. Beta/alpha in the stats table refers to the first "
"benchmark.")) "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 the sidebar inputs now, before any validation st.stop()
_remember(spec=spec, bench_spec=bench_spec, scheme_index=_scheme_labels.index(scheme), _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, cost_bps=cost_bps, lt_rate=lt_rate * 100, st_rate=st_rate * 100,
div_rate=div_rate * 100, period=period) 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(): if not spec.strip():
st.info("Type symbols or portfolios in the sidebar — e.g. `MSFT` or " st.info("Type symbols or portfolios in the sidebar — e.g. `MSFT` or "
@ -211,7 +285,7 @@ if saved:
@st.cache_data(show_spinner=False) @st.cache_data(show_spinner=False)
def _compute_portfolio(w_key: tuple, scheme: str | None, cost_bps: float, def _compute_portfolio(w_key: tuple, scheme: str | None, cost_bps: float,
start: str, lt: float, st_r: float, div: float, start: str, lt: float, st_r: float, div: float,
root: str): root: str, gen: int):
w = dict(w_key) w = dict(w_key)
r = portfolio_returns(bundle.adj, w, rebalance=scheme, cost_bps=cost_bps, start=start) 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, t = after_tax_portfolio(bundle.adj, bundle.div, bundle.capg, w, rebalance=scheme,
@ -222,7 +296,8 @@ def _compute_portfolio(w_key: tuple, scheme: str | None, cost_bps: float,
def build_result(name: str, w: dict, scheme: str | None, cost: float) -> dict: def build_result(name: str, w: dict, scheme: str | None, cost: float) -> dict:
r, t = _compute_portfolio(tuple(sorted(w.items())), scheme, cost, start, r, t = _compute_portfolio(tuple(sorted(w.items())), scheme, cost, start,
lt_rate, st_rate, div_rate, str(root)) lt_rate, st_rate, div_rate, str(root),
generation(Path(root)))
return {"name": name, "weights": w, "res": r, "tax": t} return {"name": name, "weights": w, "res": r, "tax": t}
results = [] results = []
@ -262,7 +337,8 @@ for i, item in enumerate(bench_spec.split(), 1):
continue continue
try: try:
r, t = _compute_portfolio(tuple(sorted(w.items())), freq, cost_bps, start, r, t = _compute_portfolio(tuple(sorted(w.items())), freq, cost_bps, start,
lt_rate, st_rate, div_rate, str(root)) lt_rate, st_rate, div_rate, str(root),
generation(Path(root)))
except ValueError as e: except ValueError as e:
st.sidebar.warning(f"Benchmark {i} ignored: {e}") st.sidebar.warning(f"Benchmark {i} ignored: {e}")
continue continue
@ -300,41 +376,77 @@ mode = st.radio("Curve", _mode_labels,
show_pre, show_after, both = mode != "After-tax", mode != "Pre-tax", \ show_pre, show_after, both = mode != "After-tax", mode != "Pre-tax", \
mode == "Pre-tax + after-tax" mode == "Pre-tax + after-tax"
tab_stats, tab_equity, tab_alloc, tab_tax = st.tabs( # ------------------------------------------------------------ display range
["Statistics", "Equity curves", "Allocation", "Tax detail"]) # 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 = st.tabs(
["Statistics", "Equity curves", "Allocation", "Tax detail",
"Correlation"])
with tab_stats: with tab_stats:
# one beta/alpha/bench-return column SET per benchmark # one table per benchmark: same rows, but beta/alpha/bench-return are
bench_ms = [(b["label"], b["price"].resample("ME").last()) for b in benchmarks] # 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 summarize(price): def bench_block(bench: dict | None) -> None:
out = m.summary(price, None) rows = [(n, rng(p), rng(a)) for n, p, a in candidates]
for blab, bm in bench_ms: if bench is not None:
d = m.summary(price, bm) rows.append((f"benchmark: {bench['label']}", rng(bench["price"]),
sfx = "" if len(bench_ms) == 1 else f" [{blab}]" rng(bench["after"])))
out[f"beta{sfx}"] = d["beta"] bm = (rng(bench["price"]).resample("ME").last()
out[f"alpha_ann{sfx}"] = d["alpha_ann"] if bench is not None else None)
out[f"ann_return_bench{sfx}"] = d["ann_return_bench"]
return out
# one row per candidate (the whole portfolio, not its components) and def pick(d: dict) -> dict:
# per benchmark; the pre/after suffix only appears in 'both' mode # user-configured selection + order; skip columns the summary
summaries = {} # lacks (e.g. beta without a benchmark)
out = {}
for k in _stats_cols:
if k in d:
out[k] = d[k]
return out
def row(name, pre, after): summaries = {}
if both: for name, pre, after in rows:
summaries[f"{name} (pre-tax)"] = summarize(pre) if both:
summaries[f"{name} (after-tax)"] = summarize(after) summaries[f"{name} (pre-tax)"] = pick(m.summary(pre, bm))
elif show_pre: summaries[f"{name} (after-tax)"] = pick(m.summary(after, bm))
summaries[name] = summarize(pre) elif show_pre:
else: summaries[name] = pick(m.summary(pre, bm))
summaries[name] = summarize(after) else:
summaries[name] = pick(m.summary(after, bm))
st.dataframe(m.format_summary_table(summaries), width='stretch')
for r in results: if benchmarks:
row(r["name"], r["res"].equity, r["tax"].equity) for b in benchmarks:
for b in benchmarks: st.subheader(f"vs {b['label']}")
row(f"benchmark: {b['label']}", b["price"], b["after"]) bench_block(b)
st.dataframe(m.format_summary_table(summaries), width='stretch') else:
bench_block(None)
with tab_equity: with tab_equity:
# Self-contained plotly.js page (see chart_widget.py): full mouse zoom + # Self-contained plotly.js page (see chart_widget.py): full mouse zoom +
@ -342,15 +454,8 @@ with tab_equity:
# y to the exact min/max of the visible data. No clipping, no blank space. # y to the exact min/max of the visible data. No clipping, no blank space.
from chart_widget import equity_chart_html from chart_widget import equity_chart_html
_win_labels = ["Max", "10Y", "5Y", "3Y", "1Y"] _remember(curve_mode=mode)
win = st.radio("Window", _win_labels, idx = rng(res.equity).index
horizontal=True, key="equity_window",
index=_win_labels.index(_settings["equity_window"])
if _settings.get("equity_window") in _win_labels else 0)
_remember(curve_mode=mode, equity_window=win)
idx = res.equity.index
if win != "Max":
idx = idx[idx >= idx[-1] - pd.DateOffset(years=int(win[:-1]))]
# labels carry the pre/after suffix only in 'both' mode # labels carry the pre/after suffix only in 'both' mode
palette = ["#1f77b4", "#ff7f0e", "#2ca02c", "#d62728", palette = ["#1f77b4", "#ff7f0e", "#2ca02c", "#d62728",
@ -364,23 +469,25 @@ with tab_equity:
for k, r in enumerate(results): for k, r in enumerate(results):
color = palette[k % len(palette)] color = palette[k % len(palette)]
if show_pre: if show_pre:
series[lab(r["name"], "pre-tax")] = r["res"].equity series[lab(r["name"], "pre-tax")] = rng(r["res"].equity)
styles[lab(r["name"], "pre-tax")] = {"color": color, "width": 2} styles[lab(r["name"], "pre-tax")] = {"color": color, "width": 2}
if show_after: if show_after:
s = lab(r["name"], "after-tax") s = lab(r["name"], "after-tax")
series[s] = r["tax"].equity series[s] = rng(r["tax"].equity)
# faded only in 'both' mode, to stay distinct from the pre-tax line # faded only in 'both' mode, to stay distinct from the pre-tax line
styles[s] = {"color": color, "width": 2, styles[s] = {"color": color, "width": 2,
"opacity": 0.55 if both else 1.0} "opacity": 0.55 if both else 1.0}
for b in benchmarks: for b in benchmarks:
base = f"benchmark ({b['label']})" base = f"benchmark ({b['label']})"
if show_pre: if show_pre:
p = rng(b["price"])
s = lab(base, "pre-tax") s = lab(base, "pre-tax")
series[s] = b["price"] / b["price"].iloc[0] series[s] = p / p.iloc[0]
styles[s] = {"width": 1, "opacity": 0.6} styles[s] = {"width": 1, "opacity": 0.6}
if show_after: if show_after:
p = rng(b["after"])
s = lab(base, "after-tax") s = lab(base, "after-tax")
series[s] = b["after"] / b["after"].iloc[0] series[s] = p / p.iloc[0]
styles[s] = {"width": 1, "opacity": 0.3 if both else 0.6} 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) html = equity_chart_html(series, idx, ytitle="growth (1.0 = start)", styles=styles)
@ -390,11 +497,12 @@ with tab_equity:
return f"{name}: pre {pre:.2f}× / after {after:.2f}×" return f"{name}: pre {pre:.2f}× / after {after:.2f}×"
return f"{name}: {(pre if show_pre else after):.2f}×" return f"{name}: {(pre if show_pre else after):.2f}×"
finals = " · ".join(final(r["name"], r["res"].equity.iloc[-1], finals = " · ".join(final(r["name"], rng(r["res"].equity).iloc[-1],
r["tax"].equity.iloc[-1]) for r in results) rng(r["tax"].equity).iloc[-1]) for r in results)
for b in benchmarks: for b in benchmarks:
finals += " · " + final(f"benchmark ({b['label']})", b["price"].iloc[-1], finals += " · " + final(f"benchmark ({b['label']})",
b["after"].iloc[-1]) rng(b["price"]).iloc[-1],
rng(b["after"]).iloc[-1])
st.caption(f"Final: {finals}. " st.caption(f"Final: {finals}. "
"Drag to box-zoom, right-drag (or mode-bar hand) to pan, " "Drag to box-zoom, right-drag (or mode-bar hand) to pan, "
"scroll to zoom, double-click to reset.") "scroll to zoom, double-click to reset.")
@ -402,7 +510,7 @@ with tab_equity:
with tab_alloc: with tab_alloc:
st.caption(f"Allocation for the **{results[0]['name']}** portfolio " st.caption(f"Allocation for the **{results[0]['name']}** portfolio "
"(first entry in the Symbol field).") "(first entry in the Symbol field).")
alloc = res.allocation alloc = rng(res.allocation)
st.dataframe(alloc.tail(1).T, width='stretch') st.dataframe(alloc.tail(1).T, width='stretch')
fig = go.Figure(go.Bar(x=alloc.columns, y=alloc.iloc[-1])) 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()}") fig.update_layout(height=360, title=f"Drifted allocation @ {alloc.index[-1].date()}")
@ -414,13 +522,13 @@ with tab_tax:
st.caption(f"Tax detail for the **{results[0]['name']}** portfolio " st.caption(f"Tax detail for the **{results[0]['name']}** portfolio "
"(first entry in the Symbol field).") "(first entry in the Symbol field).")
# all values below are fractions of the starting account (1.0 = 100%) # all values below are fractions of the starting account (1.0 = 100%)
yr_tax = taxres.taxes.resample("YE").sum() yr_tax = rng(taxres.taxes).resample("YE").sum()
yr_tax.index = yr_tax.index.year yr_tax.index = yr_tax.index.year
yr_tax = yr_tax.map(lambda v: f"{v:.2%}") yr_tax = yr_tax.map(lambda v: f"{v:.2%}")
yr_real = taxres.realized.resample("YE").sum() yr_real = rng(taxres.realized).resample("YE").sum()
yr_real.index = yr_real.index.year yr_real.index = yr_real.index.year
yr_real = yr_real.map(lambda v: f"{v:+.2%}") yr_real = yr_real.map(lambda v: f"{v:+.2%}")
yr_liq = taxres.liq_tax.resample("YE").last() yr_liq = rng(taxres.liq_tax).resample("YE").last()
yr_liq.index = yr_liq.index.year yr_liq.index = yr_liq.index.year
yr_liq = yr_liq.map(lambda v: f"{v:.2%}") yr_liq = yr_liq.map(lambda v: f"{v:.2%}")
st.subheader("Taxes actually paid per year (distributions + rebalance sales)") st.subheader("Taxes actually paid per year (distributions + rebalance sales)")
@ -429,6 +537,38 @@ with tab_tax:
st.dataframe(yr_real, width='stretch') st.dataframe(yr_real, width='stretch')
st.subheader("Unrealized-gain tax you would owe if you sold everything (year-end)") st.subheader("Unrealized-gain tax you would owe if you sold everything (year-end)")
st.dataframe(yr_liq, width='stretch') st.dataframe(yr_liq, width='stretch')
st.caption(f"Total taxes paid to date: {taxres.taxes['total'].sum():.2%} of start · " _tax_total = rng(taxres.taxes)["total"].sum()
f"final after-tax value (sell everything): {taxres.equity.iloc[-1]:.2f}× " _final_after = rng(taxres.equity).iloc[-1]
f"= {taxres.equity.iloc[-1] - 1:+.1%} total return") 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.")

86
data.py
View File

@ -20,6 +20,7 @@ from __future__ import annotations
import json import json
import os import os
import threading import threading
import time
from dataclasses import dataclass from dataclasses import dataclass
from pathlib import Path from pathlib import Path
@ -84,6 +85,16 @@ def _read_names(root: Path, max_files: int = 5000) -> dict:
# (every widget change!) don't reload the parquet panels # (every widget change!) don't reload the parquet panels
_BUNDLE_CACHE: dict[tuple, Bundle] = {} _BUNDLE_CACHE: dict[tuple, Bundle] = {}
_LOCK = threading.Lock() _LOCK = threading.Lock()
# Background cache maintenance. The data root is written by an independent
# downloader (goget) while Streamlit serves requests; the two must not
# interfere. `load_bundle` therefore never refreshes on the request path:
# it serves the in-memory bundle immediately and, if the data dir changed,
# a background thread refreshes the cache at its own pace.
WATCH_INTERVAL = 5.0 # seconds between staleness scans
REFRESH_CADENCE = 30.0 # min seconds between actual cache rebuilds
_watchers: dict[tuple, threading.Thread] = {}
_gen: dict[tuple, int] = {} # (root, cache) -> bundle generation counter
def _scan_files(root: Path) -> dict: def _scan_files(root: Path) -> dict:
"""{filename: (mtime_ns, size)} for every csv/json file in the data root.""" """{filename: (mtime_ns, size)} for every csv/json file in the data root."""
out = {} out = {}
@ -207,9 +218,22 @@ def _apply_changes(root: Path, cache: Path, old: dict, new: dict) -> None:
panel.to_parquet(cache / f"panel_{k}.parquet") panel.to_parquet(cache / f"panel_{k}.parquet")
def load_bundle(root: Path = DEFAULT_ROOT, cache: Path = CACHE_DIR, def up_to_date(root: Path = DEFAULT_ROOT, cache: Path = CACHE_DIR) -> bool:
rebuild: bool = False) -> Bundle: """Cheap check whether the cache matches the data dir right now."""
"""Load the full dataset, using (or building) a parquet cache. return _cache_up_to_date(Path(root), Path(cache))
def generation(root: Path = DEFAULT_ROOT, cache: Path = CACHE_DIR) -> int:
"""Monotonic counter, bumped each time the bundle is replaced.
Feed it into st.cache_data keys so cached computations invalidate when
the data changes."""
return _gen.get((str(root), str(cache)), 0)
def refresh(root: Path = DEFAULT_ROOT, cache: Path = CACHE_DIR,
rebuild: bool = False) -> Bundle:
"""Synchronously refresh the parquet cache if stale and return the bundle.
The cache tracks the data dir via a manifest of (mtime, size) per file: The cache tracks the data dir via a manifest of (mtime, size) per file:
changed/added files are re-read and merged in incrementally, removed changed/added files are re-read and merged in incrementally, removed
@ -253,9 +277,65 @@ def load_bundle(root: Path = DEFAULT_ROOT, cache: Path = CACHE_DIR,
# keep only the newest bundle: there is no room for two # keep only the newest bundle: there is no room for two
_BUNDLE_CACHE.clear() _BUNDLE_CACHE.clear()
_BUNDLE_CACHE[key] = bundle _BUNDLE_CACHE[key] = bundle
_gen[(str(root), str(cache))] = _gen.get((str(root), str(cache)), 0) + 1
return bundle return bundle
def load_bundle(root: Path = DEFAULT_ROOT, cache: Path = CACHE_DIR,
rebuild: bool = False) -> Bundle:
"""Return the current bundle without blocking on data changes.
The first call (and `rebuild=True`) build synchronously there is
nothing to serve yet. Otherwise the in-memory bundle is returned
immediately; if the data dir has changed, a background thread refreshes
the cache on its own schedule (every REFRESH_CADENCE seconds while the
downloader keeps writing), so serving requests and ingesting new data
run independently.
"""
root, cache = Path(root), Path(cache)
key = (str(root), rebuild)
with _LOCK:
hit = _BUNDLE_CACHE.get(key)
if hit is None:
return refresh(root, cache, rebuild)
if rebuild:
return refresh(root, cache, True)
if not _cache_up_to_date(root, cache):
_ensure_watcher(root, cache, key)
return hit
def _ensure_watcher(root: Path, cache: Path, key: tuple) -> None:
wk = (str(root), str(cache))
t = _watchers.get(wk)
if t is not None and t.is_alive():
return
t = threading.Thread(target=_watcher_loop, args=(root, cache, key),
daemon=True)
_watchers[wk] = t
t.start()
def _watcher_loop(root: Path, cache: Path, key: tuple) -> None:
last = 0.0
while True:
time.sleep(WATCH_INTERVAL)
with _LOCK:
active = _BUNDLE_CACHE.get(key) is not None
if not active:
return # bundle evicted (another key loaded) -> stop
if _cache_up_to_date(root, cache):
continue
now = time.monotonic()
if now - last < REFRESH_CADENCE:
continue
try:
refresh(root, cache)
except Exception:
continue # transient (e.g. partial read); retry next tick
last = time.monotonic()
def search_symbols(bundle: Bundle, query: str, limit: int = 200) -> list[str]: def search_symbols(bundle: Bundle, query: str, limit: int = 200) -> list[str]:
"""Case-insensitive search over symbols and names. """Case-insensitive search over symbols and names.

View File

@ -68,8 +68,8 @@ def summary(price: pd.Series, bench: pd.Series | None = None,
r = daily_returns(price) r = daily_returns(price)
out = { out = {
"total_return": total_return(price), "total_return": total_return(price),
"ann_return": annualized_return(price), "return": annualized_return(price),
"ann_vol": annualized_vol(r), "vol": annualized_vol(r),
"sharpe": sharpe(r, rf), "sharpe": sharpe(r, rf),
"sortino": sortino(r, rf), "sortino": sortino(r, rf),
"max_dd": max_drawdown(price), "max_dd": max_drawdown(price),
@ -78,11 +78,21 @@ def summary(price: pd.Series, bench: pd.Series | None = None,
if bench is not None: if bench is not None:
b, a = beta_alpha(r, daily_returns(bench), rf) b, a = beta_alpha(r, daily_returns(bench), rf)
out["beta"] = b out["beta"] = b
out["alpha_ann"] = a out["alpha"] = a
out["ann_return_bench"] = annualized_return(bench) out["ann_return_bench"] = annualized_return(bench)
return out return out
def xcorr(prices: dict[str, pd.Series]) -> pd.DataFrame:
"""Pairwise Pearson cross-correlation of daily returns.
prices: {label: price series}. Series of different lengths (e.g. a
young ETF next to a 30-year index) correlate over their common
history pandas corr() is pairwise-complete over NaN."""
r = pd.DataFrame({n: s.pct_change() for n, s in prices.items()})
return r.corr().round(2)
def format_summary_table(summaries: dict[str, dict[str, float]]) -> pd.DataFrame: def format_summary_table(summaries: dict[str, dict[str, float]]) -> pd.DataFrame:
"""{label: summary_dict} -> transposed table, percentages pre-formatted. """{label: summary_dict} -> transposed table, percentages pre-formatted.
@ -90,8 +100,7 @@ def format_summary_table(summaries: dict[str, dict[str, float]]) -> pd.DataFrame
several benchmarks; all of them are formatted by prefix. several benchmarks; all of them are formatted by prefix.
""" """
df = pd.DataFrame(summaries).T df = pd.DataFrame(summaries).T
pct = ("total_return", "ann_return", "ann_vol", "max_dd", "alpha_ann", pct = ("total_return", "return", "vol", "max_dd", "alpha", "ann_return_bench")
"ann_return_bench")
two = ("sharpe", "sortino", "calmar", "beta") two = ("sharpe", "sortino", "calmar", "beta")
for c in df.columns: for c in df.columns:
base = c.split(" [")[0] base = c.split(" [")[0]

View File

@ -104,25 +104,91 @@ def _run() -> None:
print("benchmarks", flush=True) print("benchmarks", flush=True)
run_app("MSFT", "V googl:0.5,amzn:0.5") run_app("MSFT", "V googl:0.5,amzn:0.5")
tbl = app().main.tabs[0].dataframe[0].value blocks = [d.value for d in app().main.tabs[0].dataframe]
check("two benchmarks, one line", not app().exception and check("one stats block per benchmark", not app().exception
"benchmark: v" in list(tbl.index) and and len(blocks) == 2, f"{len(blocks)} blocks")
"benchmark: googl, amzn" in list(tbl.index)) check("each block lists its own benchmark",
check("per-benchmark beta columns", "benchmark: v" in list(blocks[0].index)
any(c.startswith("beta [v]") for c in tbl.columns) and and "benchmark: googl, amzn" in list(blocks[1].index))
any(c.startswith("beta [googl, amzn]") for c in tbl.columns)) 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") run_app("MSFT", "V zzzznope")
rows = list(app().main.tabs[0].dataframe[0].value.index) 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 check("invalid benchmark warns, valid survives", not app().exception and
any("Benchmark 2 ignored" in w.value for w in app().sidebar.warning) any("Benchmark 2 ignored" in w.value for w in app().sidebar.warning)
and "benchmark: v" in rows) and len(blocks) == 1 and "benchmark: v" in rows)
run_app("MSFT:0.6,V:0.4", "V") run_app("MSFT:0.6,V:0.4", "V")
rows = list(app().main.tabs[0].dataframe[0].value.index) 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("no component rows for portfolio", "msft" not in rows and "v" not in rows)
check("after-tax benchmark row present in both mode", check("after-tax benchmark row present in both mode",
"benchmark: v (after-tax)" in rows if "benchmark: v (after-tax)" in rows "benchmark: v (after-tax)" in rows if "benchmark: v (after-tax)" in rows
else True) # only in both-mode; default mode has plain names 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("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) print("curve mode", flush=True)
for mode, suffixed in (("Pre-tax", False), ("After-tax", False), for mode, suffixed in (("Pre-tax", False), ("After-tax", False),
("Pre-tax + after-tax", True)): ("Pre-tax + after-tax", True)):
@ -163,15 +229,20 @@ def _run() -> None:
print("persistence (settings.json)", flush=True) print("persistence (settings.json)", flush=True)
run_app("MSFT:0.6,V:0.4", "googl:0.5,amzn:0.5 v", run_app("MSFT:0.6,V:0.4", "googl:0.5,amzn:0.5 v",
curve_mode="After-tax", equity_window="5Y") curve_mode="After-tax", date_window="5Y",
range_start="20120102", range_end="", stats_order="sharpe,vol")
s = json.loads(SETTINGS.read_text()) s = json.loads(SETTINGS.read_text())
check("settings written", s.get("spec") == "MSFT:0.6,V:0.4" and 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("bench_spec") == "googl:0.5,amzn:0.5 v" and
s.get("curve_mode") == "After-tax" and s.get("equity_window") == "5Y", str(s)) 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 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")} 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", check("settings restored in fresh session",
ti == {"spec": "MSFT:0.6,V:0.4", "bench_spec": "googl:0.5,amzn:0.5 v"}, str(ti)) 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__": if __name__ == "__main__":

View File

@ -11,6 +11,7 @@ import pathlib
import shutil import shutil
import sys import sys
import tempfile import tempfile
import time
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent)) sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent))
@ -49,10 +50,13 @@ def touch(p: pathlib.Path) -> None:
def reload(d: pathlib.Path) -> D.Bundle: def reload(d: pathlib.Path) -> D.Bundle:
return D.load_bundle(root=d["root"], cache=d["cache"]) # refresh() is the synchronous path: deterministic for these assertions
return D.refresh(root=d["root"], cache=d["cache"])
def main() -> int: def main() -> int:
# fast watcher for the background-refresh test below
D.WATCH_INTERVAL, D.REFRESH_CADENCE = 0.2, 0.0
tmp = pathlib.Path(tempfile.mkdtemp(prefix="fdatatest-")) tmp = pathlib.Path(tempfile.mkdtemp(prefix="fdatatest-"))
root, cache = tmp / "stocks", tmp / "cache" root, cache = tmp / "stocks", tmp / "cache"
root.mkdir() root.mkdir()
@ -107,6 +111,23 @@ def main() -> int:
after = (cache / "panel_adj.parquet").stat().st_mtime_ns after = (cache / "panel_adj.parquet").stat().st_mtime_ns
check("no-op reload does not rewrite parquet", before == after) check("no-op reload does not rewrite parquet", before == after)
# --- background refresh: load_bundle never blocks on changed data;
# it serves the current bundle and a background thread catches up
f = write_history(root, "aaa", last_price=300.0)
touch(f)
t0 = time.monotonic()
b = D.load_bundle(root=root, cache=cache)
check("load_bundle returns immediately on changed data",
time.monotonic() - t0 < 2.0)
deadline = time.monotonic() + 15.0
while time.monotonic() < deadline:
b = D.load_bundle(root=root, cache=cache)
if b.adj["aaa"].iloc[-1] == 300.0:
break
time.sleep(0.2)
check("background thread picks up the change", b.adj["aaa"].iloc[-1] == 300.0)
check("generation counter advances", D.generation(root=root, cache=cache) >= 1)
# --- forced full rebuild still works # --- forced full rebuild still works
b = D.load_bundle(root=root, cache=cache, rebuild=True) b = D.load_bundle(root=root, cache=cache, rebuild=True)
check("forced full rebuild", list(b.adj.columns) == ["aaa", "bbb", "ddd"]) check("forced full rebuild", list(b.adj.columns) == ["aaa", "bbb", "ddd"])