From 4f36bc7aead7bf4d289e06cd808a9313e03dd675 Mon Sep 17 00:00:00 2001
From: Greg Pomerantz <gmp@wow.st>
Date: Tue, 25 Aug 2026 18:15:47 -0400
Subject: [PATCH] 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
---
 .gitignore         |   1 +
 app.py             | 256 +++++++++++++++++++++++++++++++++++----------
 data.py            |  86 ++++++++++++++-
 metrics.py         |  19 +++-
 tests/test_app.py  |  99 +++++++++++++++---
 tests/test_data.py |  23 +++-
 6 files changed, 403 insertions(+), 81 deletions(-)

diff --git a/.gitignore b/.gitignore
index ab7120e..75a6dde 100644
--- a/.gitignore
+++ b/.gitignore
@@ -3,3 +3,4 @@ __pycache__/
 *.pyc
 .cache/
 settings.json
+funds.json
diff --git a/app.py b/app.py
index 56ba96e..cda97c1 100644
--- a/app.py
+++ b/app.py
@@ -14,7 +14,7 @@ import plotly.graph_objects as go
 import streamlit as st
 
 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 portfolios import Portfolio, delete as pf_delete, load_all as pf_load, save as pf_save
 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)
 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}
@@ -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,
                                    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),
@@ -140,10 +150,74 @@ bench_spec = st.sidebar.text_input(
           "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)
+          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 "
@@ -211,7 +285,7 @@ if saved:
 @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):
+                       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,
@@ -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:
     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}
 
 results = []
@@ -262,7 +337,8 @@ for i, item in enumerate(bench_spec.split(), 1):
         continue
     try:
         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:
         st.sidebar.warning(f"Benchmark {i} ignored: {e}")
         continue
@@ -300,41 +376,77 @@ mode = st.radio("Curve", _mode_labels,
 show_pre, show_after, both = mode != "After-tax", mode != "Pre-tax", \
     mode == "Pre-tax + after-tax"
 
-tab_stats, tab_equity, tab_alloc, tab_tax = st.tabs(
-    ["Statistics", "Equity curves", "Allocation", "Tax detail"])
+# ------------------------------------------------------------ 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 = st.tabs(
+    ["Statistics", "Equity curves", "Allocation", "Tax detail",
+     "Correlation"])
 
 with tab_stats:
-    # one beta/alpha/bench-return column SET per benchmark
-    bench_ms = [(b["label"], b["price"].resample("ME").last()) for b in benchmarks]
+    # 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 summarize(price):
-        out = m.summary(price, None)
-        for blab, bm in bench_ms:
-            d = m.summary(price, bm)
-            sfx = "" if len(bench_ms) == 1 else f" [{blab}]"
-            out[f"beta{sfx}"] = d["beta"]
-            out[f"alpha_ann{sfx}"] = d["alpha_ann"]
-            out[f"ann_return_bench{sfx}"] = d["ann_return_bench"]
-        return out
+    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)
 
-    # one row per candidate (the whole portfolio, not its components) and
-    # per benchmark; the pre/after suffix only appears in 'both' mode
-    summaries = {}
+        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
 
-    def row(name, pre, after):
-        if both:
-            summaries[f"{name} (pre-tax)"] = summarize(pre)
-            summaries[f"{name} (after-tax)"] = summarize(after)
-        elif show_pre:
-            summaries[name] = summarize(pre)
-        else:
-            summaries[name] = summarize(after)
+        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')
 
-    for r in results:
-        row(r["name"], r["res"].equity, r["tax"].equity)
-    for b in benchmarks:
-        row(f"benchmark: {b['label']}", b["price"], b["after"])
-    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 +
@@ -342,15 +454,8 @@ with tab_equity:
     # y to the exact min/max of the visible data. No clipping, no blank space.
     from chart_widget import equity_chart_html
 
-    _win_labels = ["Max", "10Y", "5Y", "3Y", "1Y"]
-    win = st.radio("Window", _win_labels,
-                   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]))]
+    _remember(curve_mode=mode)
+    idx = rng(res.equity).index
 
     # labels carry the pre/after suffix only in 'both' mode
     palette = ["#1f77b4", "#ff7f0e", "#2ca02c", "#d62728",
@@ -364,23 +469,25 @@ with tab_equity:
     for k, r in enumerate(results):
         color = palette[k % len(palette)]
         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}
         if show_after:
             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
             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] = b["price"] / b["price"].iloc[0]
+            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] = b["after"] / b["after"].iloc[0]
+            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)
@@ -390,11 +497,12 @@ with tab_equity:
             return f"{name}: pre {pre:.2f}× / after {after:.2f}×"
         return f"{name}: {(pre if show_pre else after):.2f}×"
 
-    finals = " · ".join(final(r["name"], r["res"].equity.iloc[-1],
-                              r["tax"].equity.iloc[-1]) for r in results)
+    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']})", b["price"].iloc[-1],
-                                b["after"].iloc[-1])
+        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.")
@@ -402,7 +510,7 @@ with tab_equity:
 with tab_alloc:
     st.caption(f"Allocation for the **{results[0]['name']}** portfolio "
                "(first entry in the Symbol field).")
-    alloc = res.allocation
+    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()}")
@@ -414,13 +522,13 @@ 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 = taxres.taxes.resample("YE").sum()
+    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 = taxres.realized.resample("YE").sum()
+    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 = taxres.liq_tax.resample("YE").last()
+    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)")
@@ -429,6 +537,38 @@ with tab_tax:
     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')
-    st.caption(f"Total taxes paid to date: {taxres.taxes['total'].sum():.2%} of start · "
-               f"final after-tax value (sell everything): {taxres.equity.iloc[-1]:.2f}× "
-               f"= {taxres.equity.iloc[-1] - 1:+.1%} total return")
+    _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.")
diff --git a/data.py b/data.py
index 7d07f0b..b51c692 100644
--- a/data.py
+++ b/data.py
@@ -20,6 +20,7 @@ from __future__ import annotations
 import json
 import os
 import threading
+import time
 from dataclasses import dataclass
 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
 _BUNDLE_CACHE: dict[tuple, Bundle] = {}
 _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:
     """{filename: (mtime_ns, size)} for every csv/json file in the data root."""
     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")
 
 
-def load_bundle(root: Path = DEFAULT_ROOT, cache: Path = CACHE_DIR,
-                rebuild: bool = False) -> Bundle:
-    """Load the full dataset, using (or building) a parquet cache.
+def up_to_date(root: Path = DEFAULT_ROOT, cache: Path = CACHE_DIR) -> bool:
+    """Cheap check whether the cache matches the data dir right now."""
+    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:
     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
         _BUNDLE_CACHE.clear()
         _BUNDLE_CACHE[key] = bundle
+        _gen[(str(root), str(cache))] = _gen.get((str(root), str(cache)), 0) + 1
         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]:
     """Case-insensitive search over symbols and names.
 
diff --git a/metrics.py b/metrics.py
index 613ab67..a98d532 100644
--- a/metrics.py
+++ b/metrics.py
@@ -68,8 +68,8 @@ def summary(price: pd.Series, bench: pd.Series | None = None,
     r = daily_returns(price)
     out = {
         "total_return": total_return(price),
-        "ann_return": annualized_return(price),
-        "ann_vol": annualized_vol(r),
+        "return": annualized_return(price),
+        "vol": annualized_vol(r),
         "sharpe": sharpe(r, rf),
         "sortino": sortino(r, rf),
         "max_dd": max_drawdown(price),
@@ -78,11 +78,21 @@ def summary(price: pd.Series, bench: pd.Series | None = None,
     if bench is not None:
         b, a = beta_alpha(r, daily_returns(bench), rf)
         out["beta"] = b
-        out["alpha_ann"] = a
+        out["alpha"] = a
         out["ann_return_bench"] = annualized_return(bench)
     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:
     """{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.
     """
     df = pd.DataFrame(summaries).T
-    pct = ("total_return", "ann_return", "ann_vol", "max_dd", "alpha_ann",
-           "ann_return_bench")
+    pct = ("total_return", "return", "vol", "max_dd", "alpha", "ann_return_bench")
     two = ("sharpe", "sortino", "calmar", "beta")
     for c in df.columns:
         base = c.split(" [")[0]
diff --git a/tests/test_app.py b/tests/test_app.py
index 88a5313..aaec917 100644
--- a/tests/test_app.py
+++ b/tests/test_app.py
@@ -104,25 +104,91 @@ def _run() -> None:
 
     print("benchmarks", flush=True)
     run_app("MSFT", "V googl:0.5,amzn:0.5")
-    tbl = app().main.tabs[0].dataframe[0].value
-    check("two benchmarks, one line", not app().exception and
-          "benchmark: v" in list(tbl.index) and
-          "benchmark: googl, amzn" in list(tbl.index))
-    check("per-benchmark beta columns",
-          any(c.startswith("beta [v]") for c in tbl.columns) and
-          any(c.startswith("beta [googl, amzn]") for c in tbl.columns))
+    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")
-    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
           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")
-    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("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("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)):
@@ -163,15 +229,20 @@ def _run() -> None:
 
     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", 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())
     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("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
-    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",
-          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__":
diff --git a/tests/test_data.py b/tests/test_data.py
index ca85f19..f5232c5 100644
--- a/tests/test_data.py
+++ b/tests/test_data.py
@@ -11,6 +11,7 @@ import pathlib
 import shutil
 import sys
 import tempfile
+import time
 
 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:
-    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:
+    # 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-"))
     root, cache = tmp / "stocks", tmp / "cache"
     root.mkdir()
@@ -107,6 +111,23 @@ def main() -> int:
         after = (cache / "panel_adj.parquet").stat().st_mtime_ns
         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
         b = D.load_bundle(root=root, cache=cache, rebuild=True)
         check("forced full rebuild", list(b.adj.columns) == ["aaa", "bbb", "ddd"])