From 5fd9b47b8277fa0bbed6189bb9f5d37149035607 Mon Sep 17 00:00:00 2001 From: Juha Kangas <42040080+valuecodes@users.noreply.github.com> Date: Tue, 21 Jul 2026 17:46:46 +0300 Subject: [PATCH 1/3] feat: add point-in-time optimal cash reserve page --- app/pages/3_Reserve_Over_Time.py | 310 ++++++++++++++++++++++++ src/portfolio_research_lab/__init__.py | 6 + src/portfolio_research_lab/optimizer.py | 203 +++++++++++++++- tests/test_optimize.py | 144 +++++++++++ 4 files changed, 662 insertions(+), 1 deletion(-) create mode 100644 app/pages/3_Reserve_Over_Time.py diff --git a/app/pages/3_Reserve_Over_Time.py b/app/pages/3_Reserve_Over_Time.py new file mode 100644 index 0000000..8649121 --- /dev/null +++ b/app/pages/3_Reserve_Over_Time.py @@ -0,0 +1,310 @@ +"""Optimal Cash Reserve Over Time β€” how the best reserve % drifts through history. + +For a fixed deploy rule and refill rate, this sweeps the cash-reserve target at a +series of expanding-window snapshots: at each date it uses only the price history +*up to that point* (no look-ahead) and records the reserve that maximized the +chosen objective. It answers "standing here in history, what reserve would have +looked best on the past so far?" β€” an **in-sample fit at each date**, not an +out-of-sample claim (that is the Optimizer page's walk-forward job). + +Like the other pages it is a thin presentation layer; all accounting and search +logic live in ``portfolio_research_lab``. +""" + +from __future__ import annotations + +import sys +from pathlib import Path + +# Allow running directly (`streamlit run app/Home.py`) without installing the +# package, by adding the src/ layout to the import path. +PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent +SRC = PROJECT_ROOT / "src" +if SRC.exists() and str(SRC) not in sys.path: + sys.path.insert(0, str(SRC)) + +import pandas as pd # noqa: E402 +import plotly.graph_objects as go # noqa: E402 +import streamlit as st # noqa: E402 + +from portfolio_research_lab import metrics, optimizer # noqa: E402 +from portfolio_research_lab.data import load_stocks_cash # noqa: E402 +from portfolio_research_lab.models import PRESET_RULES, CashDeployConfig # noqa: E402 +from portfolio_research_lab.optimizer import ( # noqa: E402 + ObjectiveKind, + ReserveSweepResult, + SearchSpace, + index_equity, +) + +SP500_DATA = PROJECT_ROOT / "data" / "sp500daily.csv" +FED_FUNDS_DATA = PROJECT_ROOT / "data" / "fed-funds-rate.csv" +STOCK = "S&P 500" +CASH = "Cash (Fed Funds)" + +_OBJECTIVE_LABELS: dict[str, ObjectiveKind] = { + "Excess CAGR vs index": ObjectiveKind.EXCESS_CAGR, + "Excess CAGR, drawdown-capped": ObjectiveKind.EXCESS_CAGR_DD_CAPPED, + "Sharpe (vs cash)": ObjectiveKind.SHARPE_VS_CASH, + "CAGR": ObjectiveKind.CAGR, +} + +st.set_page_config( + page_title="Reserve Over Time Β· Portfolio Research Lab", page_icon="πŸ“ˆ", layout="wide" +) + + +@st.cache_data(show_spinner=False) +def _load_stocks_cash() -> pd.DataFrame: + return load_stocks_cash(SP500_DATA, FED_FUNDS_DATA, stock_name=STOCK, cash_name=CASH) + + +def _reserve_grid(steps: int) -> tuple[float, ...]: + """Evenly spaced reserve grid over the optimizer's reserve range.""" + lo, hi = SearchSpace().reserve_range + if steps < 2: + return (lo,) + return tuple(lo + (hi - lo) * i / (steps - 1) for i in range(steps)) + + +def _reserve_line_chart(result: ReserveSweepResult, drawdown: pd.Series) -> go.Figure: + """Optimal reserve over time, with the index drawdown on a secondary axis.""" + dates = [p.as_of for p in result.points] + reserves = [p.optimal_reserve for p in result.points] + fig = go.Figure() + fig.add_trace( + go.Scatter( + x=drawdown.index, + y=drawdown, + name="Index drawdown", + mode="lines", + line={"width": 1, "color": "rgba(200,80,80,0.45)"}, + yaxis="y2", + ) + ) + fig.add_trace( + go.Scatter( + x=dates, + y=reserves, + name="Optimal reserve", + mode="lines+markers", + line={"width": 2}, + ) + ) + fig.update_layout( + margin={"t": 20, "b": 20}, + yaxis={"title": "Optimal cash reserve", "tickformat": ".0%"}, + yaxis2={ + "title": "Index drawdown", + "overlaying": "y", + "side": "right", + "tickformat": ".0%", + "showgrid": False, + }, + hovermode="x unified", + legend={"orientation": "h", "y": 1.1}, + ) + return fig + + +def _objective_heatmap(result: ReserveSweepResult) -> go.Figure: + """The full objective surface: reserve (y) by snapshot date (x), coloured by objective.""" + dates = [p.as_of for p in result.points] + reserves = [f"{r:.0%}" for r in result.reserve_grid] + # z[i][j] = objective at reserve i, snapshot j (columns aligned with dates). + z = [ + [p.objective_by_reserve[i] for p in result.points] for i in range(len(result.reserve_grid)) + ] + fig = go.Figure( + go.Heatmap( + x=dates, + y=reserves, + z=z, + colorscale="Viridis", + colorbar={"title": "Objective"}, + ) + ) + fig.add_trace( + go.Scatter( + x=dates, + y=[f"{p.optimal_reserve:.0%}" for p in result.points], + name="Optimal", + mode="lines+markers", + line={"color": "white", "width": 2}, + marker={"size": 5, "color": "white"}, + ) + ) + fig.update_layout( + margin={"t": 20, "b": 20}, + xaxis_title="Snapshot date", + yaxis_title="Cash reserve", + legend={"orientation": "h", "y": 1.1}, + ) + return fig + + +def _snapshot_table(result: ReserveSweepResult) -> pd.DataFrame: + rows = [ + { + "As of": f"{p.as_of:%Y-%m-%d}", + "History (rows)": p.n_rows, + "Optimal reserve": f"{p.optimal_reserve:.1%}", + "In-sample excess CAGR": f"{p.excess_cagr:+.2%}", + } + for p in result.points + ] + return pd.DataFrame(rows).set_index("As of") + + +def main() -> None: + st.title("πŸ“ˆ Optimal Cash Reserve Over Time") + st.caption( + "For a fixed deploy rule and refill rate, sweep the cash-reserve target at a " + "series of expanding-window snapshots and track the reserve that looked best " + "on the history up to each date. Research and education only β€” not financial advice." + ) + st.warning( + "**Each point is an in-sample fit.** At every date the reserve is the one that " + "maximized the objective *on the past so far* β€” no future data is used (no " + "look-ahead), but this is still curve-fitting to history, not an out-of-sample " + "result. For honest out-of-sample numbers use the **Optimizer** page's walk-forward." + ) + st.info( + "**S&P 500 price return only** β€” dividends are not reinvested, which understates " + "stock returns and mildly favours holding cash, so a higher 'optimal' reserve here " + "is partly an artefact of the price-return data." + ) + + try: + prices = _load_stocks_cash() + except (ValueError, FileNotFoundError) as exc: + st.error(f"Could not load price data: {exc}") + st.stop() + + # --- Controls --------------------------------------------------------- + st.sidebar.header("1 Β· Objective") + objective_label = st.sidebar.selectbox("Maximize", list(_OBJECTIVE_LABELS), index=0) + objective_kind = _OBJECTIVE_LABELS[objective_label] + dd_cap = 0.35 + if objective_kind is ObjectiveKind.EXCESS_CAGR_DD_CAPPED: + dd_cap = st.sidebar.slider("Max drawdown cap (%)", 10, 80, 35, step=5) / 100.0 + + st.sidebar.header("2 Β· Fixed strategy") + st.sidebar.caption("The deploy rule and refill are held fixed; only the reserve is swept.") + rule_name = st.sidebar.selectbox( + "Deploy rule", list(PRESET_RULES), index=list(PRESET_RULES).index("Recommended") + ) + rule = PRESET_RULES[rule_name] + refill = st.sidebar.slider("Refill rate (% / year)", 0, 100, 25, step=5) / 100.0 + + st.sidebar.header("3 Β· Grid & window") + n_points = st.sidebar.slider("Time snapshots", 4, 40, 24) + reserve_steps = st.sidebar.slider("Reserve resolution", 5, 41, 21, step=2) + warmup_years = st.sidebar.slider("Warm-up (years)", 1, 15, 5) + + base = CashDeployConfig( + name="Reserve sweep", + rule=rule, + stock_symbol=STOCK, + cash_symbol=CASH, + ) + min_rows = warmup_years * base.trading_days_per_year + + years = prices.index.map(lambda ts: ts.year) + start_year = st.sidebar.slider( + "Start year", int(years.min()), int(years.max()), int(years.min()) + ) + prices = prices[years >= start_year] + + grid = _reserve_grid(reserve_steps) + total_backtests = n_points * len(grid) + st.sidebar.caption( + f"β‰ˆ {total_backtests:,} backtests " + f"(~{total_backtests * 0.04:.0f}s at ~40ms each). Deterministic β€” no random seed." + ) + + run = st.sidebar.button("Run sweep", type="primary") + + if run: + progress = st.progress(0.0) + status = st.empty() + last_update = {"n": 0} + + def on_progress(done: int, total: int, snapshot_best: float) -> None: + if done - last_update["n"] >= len(grid) or done >= total: + last_update["n"] = done + progress.progress(min(done / total, 1.0)) + snapshot = -(-done // len(grid)) # ceil division β†’ 1-based snapshot index + status.write( + f"Snapshot {snapshot:,}/{n_points:,} Β· " + f"best objective in this snapshot: {snapshot_best:.4f}" + ) + + try: + with st.spinner("Sweeping…"): + result = optimizer.optimal_reserve_over_time( + prices, + base=base, + rule=rule, + refill_rate_per_year=refill, + objective_kind=objective_kind, + reserve_grid=grid, + n_points=n_points, + min_rows=min_rows, + dd_cap=dd_cap, + on_progress=on_progress, + ) + except ValueError as exc: + progress.empty() + status.empty() + st.warning(f"Could not run the sweep: {exc}") + st.stop() + + progress.progress(1.0) + status.write(f"Done β€” evaluated {total_backtests:,} backtests.") + st.session_state["reserve_result"] = result + st.session_state["reserve_prices"] = prices + + result = st.session_state.get("reserve_result") + if result is None: + st.info("Set the controls in the sidebar and press **Run sweep**.") + return + + prices = st.session_state["reserve_prices"] + latest = result.points[-1] + + # --- Headline --------------------------------------------------------- + st.subheader("Latest snapshot") + c1, c2 = st.columns(2) + c1.metric(f"Optimal reserve β€” as of {latest.as_of:%Y-%m-%d}", f"{latest.optimal_reserve:.1%}") + c2.metric("In-sample excess CAGR vs index", f"{latest.excess_cagr:+.2%}") + st.caption( + f"Fitted on {latest.n_rows:,} rows of history through " + f"**{latest.as_of:%Y-%m-%d}** β€” the last observation in the data, not " + f"necessarily the present day. Deploy rule: **{result.rule.name}**, " + f"refill **{result.refill_rate_per_year:.0%}/yr**." + ) + + # --- Trajectory ------------------------------------------------------- + st.subheader("Optimal reserve over time") + st.caption( + "How the best in-sample reserve drifts as history accumulates, against the " + "index drawdown for context β€” does dry powder look better after crashes?" + ) + idx_drawdown = metrics.drawdown_series(index_equity(prices, base)) + st.plotly_chart(_reserve_line_chart(result, idx_drawdown), width="stretch") + + # --- Objective surface ------------------------------------------------ + st.subheader("Objective surface") + st.caption( + "The objective at every reserve and snapshot. A flat column means the reserve " + "barely mattered at that date; a sharp ridge means it mattered a lot." + ) + st.plotly_chart(_objective_heatmap(result), width="stretch") + + st.subheader("Snapshots") + st.dataframe(_snapshot_table(result), width="stretch") + + +if __name__ == "__main__": + main() diff --git a/src/portfolio_research_lab/__init__.py b/src/portfolio_research_lab/__init__.py index 3a1dba0..1325db2 100644 --- a/src/portfolio_research_lab/__init__.py +++ b/src/portfolio_research_lab/__init__.py @@ -21,8 +21,11 @@ from portfolio_research_lab.optimizer import ( ObjectiveKind, OptimizationResult, + ReserveSweepPoint, + ReserveSweepResult, SearchSpace, WalkForwardResult, + optimal_reserve_over_time, optimize, walk_forward, ) @@ -36,6 +39,8 @@ "DeployRule", "ObjectiveKind", "OptimizationResult", + "ReserveSweepPoint", + "ReserveSweepResult", "SearchSpace", "SimulationResult", "Strategy", @@ -44,6 +49,7 @@ "infer_periods_per_year", "load_price_data", "load_rate_series", + "optimal_reserve_over_time", "optimize", "parse_price_csv", "rate_to_index", diff --git a/src/portfolio_research_lab/optimizer.py b/src/portfolio_research_lab/optimizer.py index d6e347c..09e12eb 100644 --- a/src/portfolio_research_lab/optimizer.py +++ b/src/portfolio_research_lab/optimizer.py @@ -17,7 +17,8 @@ from __future__ import annotations -from collections.abc import Callable +import math +from collections.abc import Callable, Sequence from dataclasses import dataclass, field from enum import Enum @@ -432,3 +433,203 @@ def fold_progress(done: int, _total: int, best: float) -> None: mean_test_excess = sum(f.test_excess_cagr for f in folds) / len(folds) return WalkForwardResult(folds=folds, mean_test_excess_cagr=mean_test_excess, final=final) + + +# --- Point-in-time reserve sweep --------------------------------------------- +# +# A cheaper, deterministic complement to the Optuna search: instead of one static +# config fitted over the whole window, ask "at each point in history, what cash +# reserve would have looked best on the past so far?". The deploy rule and refill +# rate are held fixed and only ``reserve_pct`` is swept, over expanding windows. +# There is no sampler, so results are exact and reproducible without a seed. + + +@dataclass(frozen=True, slots=True) +class ReserveSweepPoint: + """Optimal cash reserve at one expanding-window snapshot. + + ``as_of`` is the last date of the window this snapshot was fitted on; the fit + used only rows up to and including it, so there is no look-ahead. + ``objective_by_reserve`` is aligned element-for-element with the sweep's + ``reserve_grid``, and is the full objective surface behind ``optimal_reserve``. + """ + + as_of: pd.Timestamp + n_rows: int + optimal_reserve: float + best_objective: float + excess_cagr: float + objective_by_reserve: tuple[float, ...] + + +@dataclass(slots=True) +class ReserveSweepResult: + """Point-in-time optimal cash reserve over an expanding window. + + Each :class:`ReserveSweepPoint` fixes the deploy ``rule`` and + ``refill_rate_per_year`` and sweeps ``reserve_grid`` on the data available up + to its ``as_of`` date, recording the reserve that maximized ``objective_kind``. + This is an *in-sample* fit at each date β€” honest about look-ahead (only past + rows are used) but **not** out-of-sample validated; use :func:`walk_forward` + for the latter. + """ + + reserve_grid: tuple[float, ...] + points: list[ReserveSweepPoint] + rule: DeployRule + refill_rate_per_year: float + objective_kind: ObjectiveKind + + +def default_reserve_grid(steps: int = 21) -> tuple[float, ...]: + """An evenly spaced reserve grid over :class:`SearchSpace`'s reserve range.""" + lo, hi = SearchSpace().reserve_range + if steps < 2: + return (lo,) + return tuple(lo + (hi - lo) * i / (steps - 1) for i in range(steps)) + + +def _argmax_finite(scores: Sequence[float]) -> int | None: + """Index of the maximum finite score; the *lowest* index wins ties. + + Returns ``None`` when no score is finite. With an ascending reserve grid this + means the smallest reserve wins on a tie β€” a deliberate, stable choice. + """ + best_idx: int | None = None + best = float("-inf") + for i, score in enumerate(scores): + if math.isfinite(score) and score > best: + best = score + best_idx = i + return best_idx + + +def _snapshot_cuts(n: int, n_points: int, min_rows: int) -> list[int]: + """Row-count cutoffs for the default expanding-window anchors. + + Returns ``n_points`` strictly increasing cut lengths in ``[min_rows, n]``; the + window for a cut is ``prices.iloc[:cut]``. Raises if that many distinct integer + cuts cannot be produced (``n_points`` too large for the data and warm-up). + """ + if n_points == 1: + return [n] + span = n - min_rows + cuts = [min_rows + round(span * i / (n_points - 1)) for i in range(n_points)] + if len(set(cuts)) != n_points: + raise ValueError( + f"n_points={n_points} is too large for {n} rows with min_rows={min_rows}; " + "reduce the number of points or the warm-up" + ) + return cuts + + +def optimal_reserve_over_time( + prices: pd.DataFrame, + *, + base: CashDeployConfig, + rule: DeployRule, + refill_rate_per_year: float, + objective_kind: ObjectiveKind = ObjectiveKind.EXCESS_CAGR, + reserve_grid: Sequence[float] | None = None, + as_of_dates: Sequence[pd.Timestamp] | None = None, + n_points: int = 24, + min_rows: int = 252, + dd_cap: float = 0.35, + on_progress: ProgressCallback | None = None, +) -> ReserveSweepResult: + """Sweep the cash reserve at each expanding-window snapshot, holding the rest fixed. + + At each snapshot date ``T`` the deploy ``rule`` and ``refill_rate_per_year`` + are held fixed and only ``reserve_pct`` is varied across ``reserve_grid``; the + reserve maximizing ``objective_kind`` on the rows up to ``T`` is recorded. On + ties the *lowest* reserve wins (the grid is sorted ascending). Every backtest + runs on data ``<= T`` only, so there is no look-ahead β€” though it remains an + in-sample fit at each date. + + Snapshots come from ``as_of_dates`` when given (each windowed as + ``prices.loc[:T]``), which makes them stable as new data arrives; otherwise + ``n_points`` expanding windows are anchored by row count from ``min_rows`` to + the full length. ``on_progress(done, total, snapshot_best)`` reports progress, + where ``snapshot_best`` is the best objective seen so far *within the snapshot + currently being swept* (a global best across windows of different lengths and + benchmarks would not be comparable). + """ + if min_rows < 2: + raise ValueError("min_rows must be at least 2") + if not math.isfinite(refill_rate_per_year) or refill_rate_per_year < 0: + raise ValueError("refill_rate_per_year must be a finite, non-negative number") + if not math.isfinite(dd_cap): + raise ValueError("dd_cap must be finite") + + grid_values = default_reserve_grid() if reserve_grid is None else tuple(reserve_grid) + if not grid_values: + raise ValueError("reserve_grid must be non-empty") + if any(not math.isfinite(r) or not 0.0 <= r <= 1.0 for r in grid_values): + raise ValueError("every reserve in reserve_grid must be finite and within [0, 1]") + if len(set(grid_values)) != len(grid_values): + raise ValueError("reserve_grid must not contain duplicate values") + # Sort ascending so the lowest-index tie-break is the lowest-reserve tie-break. + grid = tuple(sorted(grid_values)) + + n = len(prices) + if as_of_dates is None: + if n_points < 1: + raise ValueError("n_points must be at least 1") + if n <= min_rows: + raise ValueError(f"need more than min_rows={min_rows} rows to sweep, got {n}") + windows = [prices.iloc[:cut] for cut in _snapshot_cuts(n, n_points, min_rows)] + else: + windows = [prices.loc[:t] for t in as_of_dates] + for as_of, window in zip(as_of_dates, windows, strict=True): + if len(window) < 2: + raise ValueError(f"snapshot {as_of} has fewer than two rows of history") + + total = len(windows) * len(grid) + done = 0 + points: list[ReserveSweepPoint] = [] + for window in windows: + ctx = _build_context(window, base, dd_cap) + objective = make_objective(objective_kind, ctx) + scores: list[float] = [] + metrics_by_reserve: list[dict[str, float]] = [] + snapshot_best = float("-inf") + for reserve in grid: + config = base.model_copy( + update={ + "reserve_pct": reserve, + "refill_rate_per_year": refill_rate_per_year, + "rule": rule, + } + ) + result = run_cash_deploy(window, config) + m = result.metrics() + score = objective(result, m) + scores.append(score) + metrics_by_reserve.append(m) + if math.isfinite(score) and score > snapshot_best: + snapshot_best = score + done += 1 + if on_progress is not None: + on_progress(done, total, snapshot_best) + + best_idx = _argmax_finite(scores) + if best_idx is None: + raise RuntimeError(f"no finite objective at snapshot ending {window.index[-1]}") + points.append( + ReserveSweepPoint( + as_of=window.index[-1], + n_rows=len(window), + optimal_reserve=grid[best_idx], + best_objective=scores[best_idx], + excess_cagr=metrics_by_reserve[best_idx]["cagr"] - ctx.benchmark_cagr, + objective_by_reserve=tuple(scores), + ) + ) + + return ReserveSweepResult( + reserve_grid=grid, + points=points, + rule=rule, + refill_rate_per_year=refill_rate_per_year, + objective_kind=objective_kind, + ) diff --git a/tests/test_optimize.py b/tests/test_optimize.py index 1912b3f..8d170f7 100644 --- a/tests/test_optimize.py +++ b/tests/test_optimize.py @@ -228,3 +228,147 @@ def test_walk_forward_rejects_too_many_folds(): ) with pytest.raises(ValueError, match="not enough data"): opt.walk_forward(prices, base=_base(), n_folds=5, n_trials=3) + + +# --- optimal_reserve_over_time ------------------------------------------------ + + +def test_optimal_reserve_over_time_shape(recovery_prices: pd.DataFrame): + grid = (0.1, 0.2, 0.3, 0.4) + result = opt.optimal_reserve_over_time( + recovery_prices, + base=_base(), + rule=_base().rule, + refill_rate_per_year=0.25, + reserve_grid=grid, + n_points=5, + min_rows=50, + ) + assert result.reserve_grid == grid # already ascending, preserved + assert len(result.points) == 5 + # Snapshots are strictly increasing in time (distinct expanding windows). + for earlier, later in pairwise(result.points): + assert earlier.as_of < later.as_of + assert earlier.n_rows < later.n_rows + for point in result.points: + assert point.optimal_reserve in grid + assert len(point.objective_by_reserve) == len(grid) + # The recorded winner is the argmax of the stored surface (lowest on ties). + best = max(point.objective_by_reserve) + winners = [grid[i] for i, s in enumerate(point.objective_by_reserve) if s == best] + assert point.optimal_reserve == min(winners) + assert point.best_objective == pytest.approx(best) + + +def test_optimal_reserve_sorts_grid_and_breaks_ties_low(recovery_prices: pd.DataFrame): + # Pass an unsorted grid; it must come back ascending, and a tie must resolve to + # the lowest reserve. CAGR over a flat-cash series is identical for two reserves + # only if they behave identically, so instead assert the sort + argmax contract. + result = opt.optimal_reserve_over_time( + recovery_prices, + base=_base(), + rule=_base().rule, + refill_rate_per_year=0.25, + reserve_grid=(0.4, 0.1, 0.25), + n_points=3, + min_rows=50, + ) + assert result.reserve_grid == (0.1, 0.25, 0.4) + + +def test_optimal_reserve_is_deterministic(recovery_prices: pd.DataFrame): + def run() -> opt.ReserveSweepResult: + return opt.optimal_reserve_over_time( + recovery_prices, + base=_base(), + rule=_base().rule, + refill_rate_per_year=0.25, + reserve_grid=(0.1, 0.2, 0.3), + n_points=4, + min_rows=50, + ) + + a, b = run(), run() + assert [p.optimal_reserve for p in a.points] == [p.optimal_reserve for p in b.points] + assert [p.objective_by_reserve for p in a.points] == [p.objective_by_reserve for p in b.points] + + +def test_optimal_reserve_matches_bruteforce(recovery_prices: pd.DataFrame): + grid = (0.05, 0.15, 0.30, 0.45, 0.60) + base = _base() + rule = base.rule + result = opt.optimal_reserve_over_time( + recovery_prices, + base=base, + rule=rule, + refill_rate_per_year=0.25, + reserve_grid=grid, + n_points=3, + min_rows=50, + ) + # Recompute the final snapshot's whole objective surface independently. + last = result.points[-1] + window = recovery_prices.iloc[: last.n_rows] + ctx = _context(window) + objective = make_objective(ObjectiveKind.EXCESS_CAGR, ctx) + expected = [] + for reserve in grid: + cfg = base.model_copy( + update={"reserve_pct": reserve, "refill_rate_per_year": 0.25, "rule": rule} + ) + res = run_cash_deploy(window, cfg) + expected.append(objective(res, res.metrics())) + assert list(last.objective_by_reserve) == pytest.approx(expected) + assert last.optimal_reserve == grid[expected.index(max(expected))] + + +def test_optimal_reserve_no_lookahead(recovery_prices: pd.DataFrame): + # A snapshot fixed by calendar date must not change when future rows are added: + # windowing is prices.loc[:T], so rows after T are irrelevant. + as_of = recovery_prices.index[199] + full = opt.optimal_reserve_over_time( + recovery_prices, + base=_base(), + rule=_base().rule, + refill_rate_per_year=0.25, + reserve_grid=(0.1, 0.3, 0.5), + as_of_dates=[as_of], + ) + truncated = opt.optimal_reserve_over_time( + recovery_prices.iloc[:230], + base=_base(), + rule=_base().rule, + refill_rate_per_year=0.25, + reserve_grid=(0.1, 0.3, 0.5), + as_of_dates=[as_of], + ) + a, b = full.points[0], truncated.points[0] + assert a.as_of == b.as_of == as_of + assert a.n_rows == b.n_rows == 200 + assert a.optimal_reserve == b.optimal_reserve + assert a.objective_by_reserve == b.objective_by_reserve + + +def test_optimal_reserve_rejects_insufficient_data(recovery_prices: pd.DataFrame): + with pytest.raises(ValueError, match="more than min_rows"): + opt.optimal_reserve_over_time( + recovery_prices.iloc[:100], + base=_base(), + rule=_base().rule, + refill_rate_per_year=0.25, + n_points=5, + min_rows=100, + ) + + +def test_optimal_reserve_rejects_bad_grid(recovery_prices: pd.DataFrame): + with pytest.raises(ValueError, match="within"): + opt.optimal_reserve_over_time( + recovery_prices, + base=_base(), + rule=_base().rule, + refill_rate_per_year=0.25, + reserve_grid=(0.1, 1.5), + n_points=3, + min_rows=50, + ) From 244983414ee6e2a891a5e412b2fd1ab9c7fe5284 Mon Sep 17 00:00:00 2001 From: Juha Kangas <42040080+valuecodes@users.noreply.github.com> Date: Tue, 21 Jul 2026 17:53:08 +0300 Subject: [PATCH 2/3] fix: address review findings --- app/pages/3_Reserve_Over_Time.py | 9 +++++--- src/portfolio_research_lab/optimizer.py | 8 ++++++- tests/test_optimize.py | 29 +++++++++++++++++++++++++ 3 files changed, 42 insertions(+), 4 deletions(-) diff --git a/app/pages/3_Reserve_Over_Time.py b/app/pages/3_Reserve_Over_Time.py index 8649121..6d860cb 100644 --- a/app/pages/3_Reserve_Over_Time.py +++ b/app/pages/3_Reserve_Over_Time.py @@ -110,7 +110,10 @@ def _reserve_line_chart(result: ReserveSweepResult, drawdown: pd.Series) -> go.F def _objective_heatmap(result: ReserveSweepResult) -> go.Figure: """The full objective surface: reserve (y) by snapshot date (x), coloured by objective.""" dates = [p.as_of for p in result.points] - reserves = [f"{r:.0%}" for r in result.reserve_grid] + # Numeric reserves for the y-axis (formatted via tickformat): using rounded + # percent strings would collapse distinct grid values into one row at higher + # reserve resolutions. + reserves = list(result.reserve_grid) # z[i][j] = objective at reserve i, snapshot j (columns aligned with dates). z = [ [p.objective_by_reserve[i] for p in result.points] for i in range(len(result.reserve_grid)) @@ -127,7 +130,7 @@ def _objective_heatmap(result: ReserveSweepResult) -> go.Figure: fig.add_trace( go.Scatter( x=dates, - y=[f"{p.optimal_reserve:.0%}" for p in result.points], + y=[p.optimal_reserve for p in result.points], name="Optimal", mode="lines+markers", line={"color": "white", "width": 2}, @@ -137,7 +140,7 @@ def _objective_heatmap(result: ReserveSweepResult) -> go.Figure: fig.update_layout( margin={"t": 20, "b": 20}, xaxis_title="Snapshot date", - yaxis_title="Cash reserve", + yaxis={"title": "Cash reserve", "tickformat": ".0%"}, legend={"orientation": "h", "y": 1.1}, ) return fig diff --git a/src/portfolio_research_lab/optimizer.py b/src/portfolio_research_lab/optimizer.py index 09e12eb..2a0af2e 100644 --- a/src/portfolio_research_lab/optimizer.py +++ b/src/portfolio_research_lab/optimizer.py @@ -575,10 +575,16 @@ def optimal_reserve_over_time( if as_of_dates is None: if n_points < 1: raise ValueError("n_points must be at least 1") - if n <= min_rows: + if n < 2: + raise ValueError(f"need at least two rows to sweep, got {n}") + # Several distinct snapshots need history beyond the warm-up; a single + # snapshot is just the full-history fit, so min_rows does not gate it. + if n_points > 1 and n <= min_rows: raise ValueError(f"need more than min_rows={min_rows} rows to sweep, got {n}") windows = [prices.iloc[:cut] for cut in _snapshot_cuts(n, n_points, min_rows)] else: + if len(as_of_dates) == 0: + raise ValueError("as_of_dates must be non-empty") windows = [prices.loc[:t] for t in as_of_dates] for as_of, window in zip(as_of_dates, windows, strict=True): if len(window) < 2: diff --git a/tests/test_optimize.py b/tests/test_optimize.py index 8d170f7..90fe8b1 100644 --- a/tests/test_optimize.py +++ b/tests/test_optimize.py @@ -361,6 +361,35 @@ def test_optimal_reserve_rejects_insufficient_data(recovery_prices: pd.DataFrame ) +def test_optimal_reserve_rejects_empty_as_of_dates(recovery_prices: pd.DataFrame): + # An empty date sequence would otherwise yield a point-less result, deferring + # the failure to consumers that read points[-1]. + with pytest.raises(ValueError, match="non-empty"): + opt.optimal_reserve_over_time( + recovery_prices, + base=_base(), + rule=_base().rule, + refill_rate_per_year=0.25, + as_of_dates=[], + ) + + +def test_optimal_reserve_single_snapshot_ignores_warmup(recovery_prices: pd.DataFrame): + # A single snapshot is the full-history fit; min_rows must not gate it even + # when it equals the row count. + result = opt.optimal_reserve_over_time( + recovery_prices, + base=_base(), + rule=_base().rule, + refill_rate_per_year=0.25, + reserve_grid=(0.1, 0.3), + n_points=1, + min_rows=len(recovery_prices), + ) + assert len(result.points) == 1 + assert result.points[0].n_rows == len(recovery_prices) + + def test_optimal_reserve_rejects_bad_grid(recovery_prices: pd.DataFrame): with pytest.raises(ValueError, match="within"): opt.optimal_reserve_over_time( From 92c84b74abbf41c9a51dbe9ee2c661e5e5251e3e Mon Sep 17 00:00:00 2001 From: Juha Kangas <42040080+valuecodes@users.noreply.github.com> Date: Tue, 21 Jul 2026 18:22:26 +0300 Subject: [PATCH 3/3] fix: address PR review feedback --- app/pages/3_Reserve_Over_Time.py | 2 ++ src/portfolio_research_lab/optimizer.py | 7 +++++-- tests/test_optimize.py | 25 +++++++++++++++++++++++++ 3 files changed, 32 insertions(+), 2 deletions(-) diff --git a/app/pages/3_Reserve_Over_Time.py b/app/pages/3_Reserve_Over_Time.py index 6d860cb..f771e3a 100644 --- a/app/pages/3_Reserve_Over_Time.py +++ b/app/pages/3_Reserve_Over_Time.py @@ -213,6 +213,8 @@ def main() -> None: ) min_rows = warmup_years * base.trading_days_per_year + # `.map` rather than the vectorized `.index.year`: ty types `.index` as the + # generic Index, which has no `.year`; this mirrors 2_Optimize.py. years = prices.index.map(lambda ts: ts.year) start_year = st.sidebar.slider( "Start year", int(years.min()), int(years.max()), int(years.min()) diff --git a/src/portfolio_research_lab/optimizer.py b/src/portfolio_research_lab/optimizer.py index 2a0af2e..9b033c2 100644 --- a/src/portfolio_research_lab/optimizer.py +++ b/src/portfolio_research_lab/optimizer.py @@ -21,6 +21,7 @@ from collections.abc import Callable, Sequence from dataclasses import dataclass, field from enum import Enum +from itertools import pairwise import optuna import pandas as pd @@ -558,8 +559,8 @@ def optimal_reserve_over_time( raise ValueError("min_rows must be at least 2") if not math.isfinite(refill_rate_per_year) or refill_rate_per_year < 0: raise ValueError("refill_rate_per_year must be a finite, non-negative number") - if not math.isfinite(dd_cap): - raise ValueError("dd_cap must be finite") + if not math.isfinite(dd_cap) or not 0.0 <= dd_cap <= 1.0: + raise ValueError("dd_cap must be a fraction within [0, 1]") grid_values = default_reserve_grid() if reserve_grid is None else tuple(reserve_grid) if not grid_values: @@ -585,6 +586,8 @@ def optimal_reserve_over_time( else: if len(as_of_dates) == 0: raise ValueError("as_of_dates must be non-empty") + if any(b <= a for a, b in pairwise(as_of_dates)): + raise ValueError("as_of_dates must be strictly increasing") windows = [prices.loc[:t] for t in as_of_dates] for as_of, window in zip(as_of_dates, windows, strict=True): if len(window) < 2: diff --git a/tests/test_optimize.py b/tests/test_optimize.py index 90fe8b1..15f403b 100644 --- a/tests/test_optimize.py +++ b/tests/test_optimize.py @@ -401,3 +401,28 @@ def test_optimal_reserve_rejects_bad_grid(recovery_prices: pd.DataFrame): n_points=3, min_rows=50, ) + + +def test_optimal_reserve_rejects_bad_dd_cap(recovery_prices: pd.DataFrame): + with pytest.raises(ValueError, match="dd_cap"): + opt.optimal_reserve_over_time( + recovery_prices, + base=_base(), + rule=_base().rule, + refill_rate_per_year=0.25, + n_points=3, + min_rows=50, + dd_cap=1.5, + ) + + +def test_optimal_reserve_rejects_unsorted_as_of_dates(recovery_prices: pd.DataFrame): + dates = [recovery_prices.index[200], recovery_prices.index[100]] + with pytest.raises(ValueError, match="strictly increasing"): + opt.optimal_reserve_over_time( + recovery_prices, + base=_base(), + rule=_base().rule, + refill_rate_per_year=0.25, + as_of_dates=dates, + )