From 6d623883ceac2005d73cda20e90457854232adfc Mon Sep 17 00:00:00 2001 From: Juha Kangas <42040080+valuecodes@users.noreply.github.com> Date: Mon, 13 Jul 2026 22:00:41 +0300 Subject: [PATCH 1/3] feat: add cash-deploy backtest with drawdown-triggered deployment MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add a tactical cash-deployment strategy: hold a cash reserve at the fed funds rate, deploy it into stocks in tranches as the S&P draws down from its all-time high, and drip it back toward the target reserve at new highs. Includes named deploy rules (Käyttäjän sääntö, Kasvuoptimi, Riskioptimi, Suositus), a dedicated stateful engine, a Streamlit page, and comparison baselines (100% stocks, static split) via the existing simulator. Co-Authored-By: Claude Opus 4.8 --- app/Home.py | 13 +- app/pages/1_Cash_Deploy.py | 317 ++++++++++++++++++++++ src/portfolio_research_lab/cash_deploy.py | 183 +++++++++++++ src/portfolio_research_lab/data.py | 29 ++ src/portfolio_research_lab/metrics.py | 24 ++ src/portfolio_research_lab/models.py | 137 ++++++++++ tests/conftest.py | 19 ++ tests/test_cash_deploy.py | 191 +++++++++++++ tests/test_data.py | 47 +++- tests/test_metrics.py | 24 ++ tests/test_models.py | 79 +++++- 11 files changed, 1052 insertions(+), 11 deletions(-) create mode 100644 app/pages/1_Cash_Deploy.py create mode 100644 src/portfolio_research_lab/cash_deploy.py create mode 100644 tests/test_cash_deploy.py diff --git a/app/Home.py b/app/Home.py index df7eeda..82ae751 100644 --- a/app/Home.py +++ b/app/Home.py @@ -28,9 +28,8 @@ from portfolio_research_lab.data import ( # noqa: E402 load_price_data, - load_rate_series, + load_stocks_cash, parse_price_csv, - rate_to_index, ) from portfolio_research_lab.models import StrategyConfig # noqa: E402 from portfolio_research_lab.simulator import run_simulation # noqa: E402 @@ -60,13 +59,9 @@ def _load_sp500() -> pd.DataFrame: @st.cache_data(show_spinner=False) def _load_stocks_cash() -> pd.DataFrame: # Two-asset frame: S&P 500 alongside a synthetic cash account that compounds - # the daily federal funds rate. The cash index is built on the fed funds' - # native (calendar) dates so weekend accrual is baked into each level, then - # sampled onto the S&P trading days by the join. dropna trims the S&P-only - # years before 1954 (when the fed funds series begins). - stocks = _load_sp500() - cash = rate_to_index(load_rate_series(FED_FUNDS_DATA)).rename("Cash (Fed Funds)") - return stocks.join(cash, how="left").ffill().dropna(how="any") + # the daily federal funds rate. Built by the shared engine loader + # (data.load_stocks_cash), which the Cash Deploy page reuses too. + return load_stocks_cash(SP500_DATA, FED_FUNDS_DATA) def _load_uploaded(file) -> pd.DataFrame: diff --git a/app/pages/1_Cash_Deploy.py b/app/pages/1_Cash_Deploy.py new file mode 100644 index 0000000..25de3b5 --- /dev/null +++ b/app/pages/1_Cash_Deploy.py @@ -0,0 +1,317 @@ +"""Cash Deploy — a tactical cash-deployment backtest page. + +Holds part of the portfolio in a cash reserve (money-market at the fed funds +rate), deploys it into stocks in tranches as the S&P draws down from its +all-time high, and drips it back to target once the market recovers. Compares +the result against a 100%-stocks buy-and-hold and a static stock/cash split so +the user can see whether tactical timing actually helped. + +Like ``Home.py`` this is a thin presentation layer: all accounting lives in +``portfolio_research_lab`` (the cash-deploy engine and the shared simulator used +for the baselines). +""" + +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 # noqa: E402 +from portfolio_research_lab.cash_deploy import CashDeployResult, run_cash_deploy # noqa: E402 +from portfolio_research_lab.data import load_stocks_cash # noqa: E402 +from portfolio_research_lab.models import ( # noqa: E402 + PRESET_RULES, + CashDeployConfig, + DeployRule, + StrategyConfig, +) +from portfolio_research_lab.simulator import run_simulation # noqa: E402 + +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)" +CUSTOM_RULE = "Custom…" + +st.set_page_config(page_title="Cash Deploy · 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 _metrics_row(label: str, m: dict[str, float], sharpe: float) -> dict[str, str]: + return { + "Strategy": label, + "Total return": f"{m['total_return']:.2%}", + "CAGR": f"{m['cagr']:.2%}", + "Volatility (ann.)": f"{m['annualized_volatility']:.2%}", + "Max drawdown": f"{m['max_drawdown']:.2%}", + "Sharpe (vs cash)": f"{sharpe:.2f}", + } + + +def _sharpe_vs_cash(equity: pd.Series, cash_level: pd.Series, periods_per_year: int) -> float: + # Risk-free = the money-market leg's own periodic return, aligned to the + # equity curve's return dates inside sharpe_ratio. + returns = metrics.periodic_returns(equity) + risk_free = cash_level.pct_change() + return metrics.sharpe_ratio(returns, periods_per_year, risk_free) + + +def _sidebar_rule() -> DeployRule | None: + """Collect a deploy rule from the sidebar (a preset or a custom one).""" + choice = st.sidebar.selectbox("Deploy rule", [*PRESET_RULES, CUSTOM_RULE], index=3) + if choice != CUSTOM_RULE: + rule = PRESET_RULES[choice] + th_str = " / ".join(f"{t:.0%}" for t in rule.thresholds) + us_str = " / ".join(f"{u:.0%}" for u in rule.usages) + st.sidebar.caption(f"Drawdown: {th_str}\n\nReserve used: {us_str}") + return rule + + st.sidebar.caption( + "One row per tranche: the drawdown that triggers it and the % of the " + "reserve (locked at the drawdown's start) to deploy." + ) + n = st.sidebar.number_input("Number of tranches", min_value=1, max_value=8, value=5) + thresholds: list[float] = [] + usages: list[float] = [] + for i in range(int(n)): + cols = st.sidebar.columns(2) + th = cols[0].number_input( + f"Drawdown {i + 1} (%)", + min_value=1, + max_value=99, + value=min(10 * (i + 1), 99), + key=f"th{i}", + ) + us = cols[1].number_input( + f"Reserve {i + 1} (%)", min_value=1, max_value=100, value=20, key=f"us{i}" + ) + thresholds.append(th / 100.0) + usages.append(us / 100.0) + try: + return DeployRule(name="Custom", thresholds=tuple(thresholds), usages=tuple(usages)) + except ValueError as exc: + st.sidebar.error(f"Invalid rule: {exc}") + return None + + +def _baseline_equity( + prices: pd.DataFrame, + allocations: dict[str, float], + capital: float, + *, + rebalance: str | None, + name: str, +) -> pd.Series: + config = StrategyConfig.from_weights( + allocations, name=name, initial_capital=capital, rebalance_frequency=rebalance + ) + return run_simulation(prices, config).equity + + +def _equity_chart(result: CashDeployResult, baselines: dict[str, pd.Series]) -> go.Figure: + fig = go.Figure() + for label, series in baselines.items(): + fig.add_trace( + go.Scatter(x=series.index, y=series, name=label, mode="lines", line={"width": 1}) + ) + fig.add_trace( + go.Scatter(x=result.equity.index, y=result.equity, name=result.config.name, mode="lines") + ) + fig.update_layout( + margin={"t": 20, "b": 20}, yaxis_title="Portfolio value", hovermode="x unified" + ) + return fig + + +def _composition_chart(result: CashDeployResult) -> go.Figure: + fig = go.Figure() + fig.add_trace( + go.Scatter( + x=result.stock_value.index, + y=result.stock_value, + name="Stocks", + mode="lines", + stackgroup="alloc", + line={"width": 0.5}, + ) + ) + fig.add_trace( + go.Scatter( + x=result.cash.index, + y=result.cash, + name="Cash reserve", + mode="lines", + stackgroup="alloc", + line={"width": 0.5}, + ) + ) + fig.update_layout(margin={"t": 20, "b": 20}, yaxis_title="Value", hovermode="x unified") + return fig + + +def _drawdown_chart(dd: pd.Series) -> go.Figure: + fig = go.Figure() + fig.add_trace(go.Scatter(x=dd.index, y=dd, name="Drawdown", mode="lines", fill="tozeroy")) + fig.update_layout( + margin={"t": 20, "b": 20}, + yaxis_title="Drawdown", + yaxis_tickformat=".0%", + hovermode="x unified", + ) + return fig + + +def main() -> None: + st.title("💵 Cash Deploy") + st.caption( + "Hold a cash reserve, deploy it into stocks as the market falls, and " + "refill it as the market recovers. For research and education only — " + "not financial advice." + ) + st.info( + "**S&P 500 price return only** — dividends are not reinvested, which " + "understates stock returns by roughly the dividend yield and mildly " + "favours holding cash. Read the comparisons with that in mind." + ) + + try: + prices = _load_stocks_cash() + except (ValueError, FileNotFoundError) as exc: + st.error(f"Could not load price data: {exc}") + st.stop() + + # --- Reserve & capital ------------------------------------------------ + st.sidebar.header("1 · Reserve") + reserve_pct = st.sidebar.slider("Cash reserve (%)", 0, 100, 30, step=5) / 100.0 + st.sidebar.caption(f"Start: {1 - reserve_pct:.0%} stocks · {reserve_pct:.0%} cash") + initial_capital = st.sidebar.number_input( + "Initial capital", min_value=100.0, value=10_000.0, step=1_000.0 + ) + + # --- Deploy rule ------------------------------------------------------ + st.sidebar.header("2 · Deploy rule") + rule = _sidebar_rule() + if rule is None: + st.warning("Fix the deploy rule in the sidebar to run the backtest.") + st.stop() + if abs(rule.usage_sum - 1.0) > 1e-9: + tail = ( + "Some reserve is never deployed." + if rule.usage_sum < 1 + else "Deployment is capped by available cash." + ) + st.sidebar.warning(f"Reserve usages sum to {rule.usage_sum:.0%}, not 100%. {tail}") + + # --- Refill & window -------------------------------------------------- + st.sidebar.header("3 · Refill") + refill_rate = st.sidebar.slider("Refill rate (% of target / year)", 0, 100, 25, step=5) / 100.0 + st.sidebar.caption("Applied only while at a new all-time high.") + + st.sidebar.header("4 · Window") + # `.map` sidesteps the missing DatetimeIndex.year type stub while staying a + # plain int Index we can slice on. + 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] + if len(prices) < 2: + st.warning("Not enough data in the selected window.") + st.stop() + + st.sidebar.header("5 · Compare") + show_stocks = st.sidebar.checkbox("100% stocks", value=True) + show_static = st.sidebar.checkbox( + f"Static {1 - reserve_pct:.0%}/{reserve_pct:.0%} (annual)", value=True + ) + + # --- Run -------------------------------------------------------------- + try: + config = CashDeployConfig( + name="Cash Deploy", + initial_capital=initial_capital, + reserve_pct=reserve_pct, + rule=rule, + refill_rate_per_year=refill_rate, + stock_symbol=STOCK, + cash_symbol=CASH, + ) + result = run_cash_deploy(prices, config) + except ValueError as exc: + st.error(f"Simulation error: {exc}") + st.stop() + + periods_per_year = config.trading_days_per_year + cash_level = prices[CASH] + + baselines: dict[str, pd.Series] = {} + if show_stocks: + baselines["100% stocks"] = _baseline_equity( + prices, {STOCK: 1.0}, initial_capital, rebalance=None, name="100% stocks" + ) + if show_static and 0.0 < reserve_pct < 1.0: + baselines[f"Static {1 - reserve_pct:.0%}/{reserve_pct:.0%}"] = _baseline_equity( + prices, + {STOCK: 1 - reserve_pct, CASH: reserve_pct}, + initial_capital, + rebalance="annually", + name="Static split", + ) + + # --- Metrics ---------------------------------------------------------- + st.subheader("Results") + rows = [ + _metrics_row( + config.name, + result.metrics(), + _sharpe_vs_cash(result.equity, cash_level, periods_per_year), + ) + ] + for label, series in baselines.items(): + rows.append( + _metrics_row( + label, + metrics.summarize(series, periods_per_year), + _sharpe_vs_cash(series, cash_level, periods_per_year), + ) + ) + st.dataframe(pd.DataFrame(rows).set_index("Strategy"), width="stretch") + + st.subheader("Equity curve") + st.plotly_chart(_equity_chart(result, baselines), width="stretch") + + st.subheader("Reserve deployment") + st.caption("Stacked value of the stock leg and the cash reserve over time.") + st.plotly_chart(_composition_chart(result), width="stretch") + + st.subheader("Drawdown") + st.plotly_chart(_drawdown_chart(result.drawdown()), width="stretch") + + with st.expander(f"Deploy / refill events ({len(result.events)})"): + if result.events.empty: + st.write("No deployments or refills in this window.") + else: + display = result.events.copy() + display["drawdown"] = display["drawdown"].map(lambda x: f"{x:.1%}") + display["amount"] = display["amount"].map(lambda x: f"{x:,.0f}") + display["cash_after"] = display["cash_after"].map(lambda x: f"{x:,.0f}") + st.dataframe(display, width="stretch") + + +if __name__ == "__main__": + main() diff --git a/src/portfolio_research_lab/cash_deploy.py b/src/portfolio_research_lab/cash_deploy.py new file mode 100644 index 0000000..20b911a --- /dev/null +++ b/src/portfolio_research_lab/cash_deploy.py @@ -0,0 +1,183 @@ +"""The tactical cash-deployment engine. + +Unlike the fixed-weight :mod:`~portfolio_research_lab.simulator` engine, this +simulation is *path dependent*: it holds a cash reserve and deploys it into +stocks as the market draws down from its running peak, then refills the reserve +by drip-selling stocks once a new all-time high is reached. That state (running +peak, which tranches have fired this episode, the locked tranche base) cannot be +expressed as a per-period weight vector, so this is a dedicated explicit daily +loop rather than a strategy plugged into the shared simulator. + +The engine consumes the same two-column ``[stock, cash]`` price frame the app +builds (see :func:`portfolio_research_lab.data.load_stocks_cash`). The cash leg +is a money-market growth index; cash held in the reserve accrues at the same +per-step rate, taken as the ratio of consecutive cash-index levels. +""" + +from __future__ import annotations + +from dataclasses import dataclass + +import numpy as np +import pandas as pd + +from portfolio_research_lab import metrics +from portfolio_research_lab.models import CashDeployConfig + +# Columns of the event log emitted by a run. +_EVENT_COLUMNS = ("type", "drawdown", "amount", "cash_after") + + +@dataclass(slots=True) +class CashDeployResult: + """Output of a cash-deploy simulation run. + + Attributes + ---------- + config: + The configuration that produced this run. + equity: + Total portfolio value over time (cash + stock leg). + cash: + Cash reserve balance over time. + stock_value: + Market value of the stock leg over time. + events: + One row per deploy/refill action, indexed by date, with columns + ``type`` (``"deploy"``/``"refill"``), ``drawdown`` (the S&P drawdown at + the time), ``amount`` (cash moved) and ``cash_after`` (reserve balance + after the action). + """ + + config: CashDeployConfig + equity: pd.Series + cash: pd.Series + stock_value: pd.Series + events: pd.DataFrame + + def metrics(self) -> dict[str, float]: + """Headline metrics for the strategy equity curve.""" + return metrics.summarize(self.equity, self.config.trading_days_per_year) + + def drawdown(self) -> pd.Series: + """Drawdown series of the strategy equity curve.""" + return metrics.drawdown_series(self.equity) + + +def run_cash_deploy(prices: pd.DataFrame, config: CashDeployConfig) -> CashDeployResult: + """Run a tactical cash-deployment simulation. + + Parameters + ---------- + prices: + Wide-format price frame containing (at least) ``config.stock_symbol`` and + ``config.cash_symbol`` columns. The cash column is a money-market growth + index; its per-step ratio drives interest on the held reserve. + config: + The cash-deploy configuration to simulate. + """ + if prices.empty: + raise ValueError("cannot simulate on empty price data") + + missing = [s for s in (config.stock_symbol, config.cash_symbol) if s not in prices.columns] + if missing: + raise ValueError(f"price data is missing required columns: {missing}") + + index = prices.index + stock = prices[config.stock_symbol].to_numpy(dtype=float) + cash_level = prices[config.cash_symbol].to_numpy(dtype=float) + + if stock[0] <= 0.0 or not np.isfinite(stock[0]): + raise ValueError(f"the first stock price must be positive, got {stock[0]}") + + # Per-step cash growth factor = ratio of consecutive money-market levels; the + # first step has no prior level so it is 1.0. A non-positive/non-finite prior + # level would corrupt the ratio, so guard it (falls back to no growth). + factor = np.ones(len(stock), dtype=float) + if len(stock) > 1: + prev = cash_level[:-1] + ratio = np.divide( + cash_level[1:], + prev, + out=np.ones(len(stock) - 1, dtype=float), + where=(prev > 0.0), + ) + ratio[~np.isfinite(ratio)] = 1.0 + factor[1:] = ratio + + thresholds = config.rule.thresholds + usages = config.rule.usages + r = config.reserve_pct + refill_daily = config.refill_rate_per_year / config.trading_days_per_year + + # --- State --- + cash = config.initial_capital * r + units = config.initial_capital * (1.0 - r) / stock[0] + peak = stock[0] + deployed = [False] * len(thresholds) # which tranches have fired this episode + base: float | None = None # reserve locked at the first dip of an episode + + n = len(index) + equity_out = np.empty(n, dtype=float) + cash_out = np.empty(n, dtype=float) + stock_out = np.empty(n, dtype=float) + events: list[tuple[pd.Timestamp, str, float, float, float]] = [] + + for i in range(n): + price = stock[i] + if i > 0: + cash *= factor[i] # accrue the reserve's money-market yield + + if price >= peak: + # At or making a new all-time high: the drawdown episode is over. + peak = price + deployed = [False] * len(thresholds) + base = None + # Refill the reserve toward its target by drip-selling stocks. The + # target is a share of the *current* portfolio, so this also trims + # winners at new highs to hold the reserve at `reserve_pct` — not + # only after a deployment. It is therefore a slow rebalance toward + # the target allocation, accelerated on the buy side by drawdown + # deployment. Gating on price >= peak is the literal "after back to + # ATH" reading; on a long plateau just below the peak the refill + # stalls. To relax that, widen to a band like `price >= peak*(1-b)`. + stock_value = units * price + target_cash = r * (cash + stock_value) + if cash < target_cash: + move = min(refill_daily * target_cash, target_cash - cash, stock_value) + if move > 0.0: + units -= move / price + cash += move + events.append((index[i], "refill", 0.0, move, cash)) + else: + drawdown = price / peak - 1.0 # negative + if base is None: + base = cash # lock the tranche base at the episode's first dip + for k, threshold in enumerate(thresholds): + # Every threshold newly crossed this step fires, so a single + # gap-down day can deploy several tranches at once. + if not deployed[k] and -drawdown >= threshold: + amount = min(base * usages[k], cash) + if amount > 0.0: + units += amount / price + cash -= amount + events.append((index[i], "deploy", drawdown, amount, cash)) + deployed[k] = True + + equity_out[i] = cash + units * price + cash_out[i] = cash + stock_out[i] = units * price + + events_frame = pd.DataFrame( + [e[1:] for e in events], + index=pd.DatetimeIndex([e[0] for e in events], name=index.name), + columns=list(_EVENT_COLUMNS), + ) + + return CashDeployResult( + config=config, + equity=pd.Series(equity_out, index=index, name="equity"), + cash=pd.Series(cash_out, index=index, name=config.cash_symbol), + stock_value=pd.Series(stock_out, index=index, name=config.stock_symbol), + events=events_frame, + ) diff --git a/src/portfolio_research_lab/data.py b/src/portfolio_research_lab/data.py index b828be3..e5869fa 100644 --- a/src/portfolio_research_lab/data.py +++ b/src/portfolio_research_lab/data.py @@ -158,6 +158,35 @@ def rate_to_index( return base * factor.cumprod() +def load_stocks_cash( + sp500_path: str | Path, + fed_funds_path: str | Path, + *, + stock_name: str = "S&P 500", + cash_name: str = "Cash (Fed Funds)", +) -> pd.DataFrame: + """Build the two-asset stocks + cash price frame used across the app. + + Loads daily S&P 500 closes and turns the federal funds rate into a + compounding money-market index via :func:`rate_to_index`, then joins them on + the S&P trading days. The cash index is built on the fed funds' native + (calendar) dates first, so weekend interest accrual is baked into each level + before the join samples it onto trading days. The final ``dropna`` trims the + S&P-only years before the fed funds series begins (1954). + + Returns a DataFrame with exactly two float columns, ``stock_name`` and + ``cash_name``, indexed by a sorted :class:`~pandas.DatetimeIndex`. + """ + if stock_name == cash_name: + raise ValueError("stock_name and cash_name must be different") + stocks = load_price_data(sp500_path).rename(columns={"close": stock_name}) + cash = rate_to_index(load_rate_series(fed_funds_path)).rename(cash_name) + frame = stocks.join(cash, how="left").ffill().dropna(how="any") + if frame.empty: + raise ValueError("stocks + cash frame is empty after aligning the two series") + return frame + + def infer_periods_per_year(index: pd.DatetimeIndex) -> int: """Guess how many observations occur per year from the index spacing. diff --git a/src/portfolio_research_lab/metrics.py b/src/portfolio_research_lab/metrics.py index ed6811e..7851fcc 100644 --- a/src/portfolio_research_lab/metrics.py +++ b/src/portfolio_research_lab/metrics.py @@ -55,6 +55,30 @@ def annualized_volatility( return float(returns.std(ddof=1) * np.sqrt(periods_per_year)) +def sharpe_ratio( + returns: pd.Series, + periods_per_year: int = TRADING_DAYS_PER_YEAR, + risk_free: pd.Series | float = 0.0, +) -> float: + """Annualised Sharpe ratio of periodic returns. + + Excess return over ``risk_free`` (a per-period rate, either a constant or a + series aligned to ``returns``) divided by the annualised volatility of those + excess returns. Returns 0.0 when there is too little data or no variation. + """ + if len(returns) < 2: + return 0.0 + if isinstance(risk_free, pd.Series): + rf: pd.Series | float = risk_free.reindex(returns.index).fillna(0.0) + else: + rf = risk_free + excess = returns - rf + vol = annualized_volatility(excess, periods_per_year) + if vol == 0.0: + return 0.0 + return float(excess.mean() * periods_per_year / vol) + + def drawdown_series(equity: pd.Series) -> pd.Series: """Drawdown at each point: ``value / running_peak - 1`` (values <= 0).""" running_peak = equity.cummax() diff --git a/src/portfolio_research_lab/models.py b/src/portfolio_research_lab/models.py index 441aa91..57c3387 100644 --- a/src/portfolio_research_lab/models.py +++ b/src/portfolio_research_lab/models.py @@ -9,6 +9,7 @@ import math from collections.abc import Mapping +from itertools import pairwise from pydantic import BaseModel, Field, field_validator, model_validator @@ -117,3 +118,139 @@ def from_weights( trading_days_per_year=trading_days_per_year, rebalance_frequency=rebalance_frequency, ) + + +class DeployRule(BaseModel): + """A tactical cash-deployment rule for the cash-deploy backtest. + + A rule splits a cash reserve into tranches keyed to market-drawdown depth: + when the market draws down past each ``thresholds[k]`` (a positive fraction, + so ``0.10`` means a 10% fall from the running peak), the fraction + ``usages[k]`` of the reserve is deployed into stocks. + + Attributes + ---------- + name: + Human-readable label shown in the UI. + thresholds: + Drawdown magnitudes as positive fractions, strictly increasing, each in + ``(0, 1)``. E.g. ``(0.10, 0.20, 0.30, 0.40, 0.50)``. + usages: + Fraction of the reserve to deploy at each matching threshold. Same length + as ``thresholds`` and each positive. Usages summing to 1.0 mean the whole + reserve is deployed by the deepest threshold, but this is *not* enforced: + a sum below 1.0 leaves a permanent cash residue, and a sum above 1.0 is + capped by the cash actually available at run time. + """ + + model_config = {"extra": "forbid"} + + name: str = Field(min_length=1) + thresholds: tuple[float, ...] = Field(min_length=1) + usages: tuple[float, ...] = Field(min_length=1) + + @field_validator("thresholds") + @classmethod + def _thresholds_increasing_in_unit_interval(cls, value: tuple[float, ...]) -> tuple[float, ...]: + for threshold in value: + if not math.isfinite(threshold) or not 0.0 < threshold < 1.0: + raise ValueError(f"each threshold must be a number in (0, 1), got {threshold}") + if any(b <= a for a, b in pairwise(value)): + raise ValueError(f"thresholds must be strictly increasing, got {value}") + return value + + @field_validator("usages") + @classmethod + def _usages_must_be_positive(cls, value: tuple[float, ...]) -> tuple[float, ...]: + for usage in value: + if not math.isfinite(usage) or usage <= 0: + raise ValueError(f"each reserve usage must be a positive number, got {usage}") + return value + + @model_validator(mode="after") + def _thresholds_and_usages_align(self) -> DeployRule: + if len(self.thresholds) != len(self.usages): + raise ValueError( + f"thresholds and usages must have equal length, got " + f"{len(self.thresholds)} and {len(self.usages)}" + ) + return self + + @property + def usage_sum(self) -> float: + """Total reserve fraction deployed if every threshold is reached.""" + return sum(self.usages) + + +class CashDeployConfig(BaseModel): + """Configuration for a tactical cash-deployment backtest. + + Start with ``reserve_pct`` of capital in cash (a money-market account) and + the rest in stocks. As the market draws down, deploy the reserve into stocks + per ``rule``; once the market recovers to a new high, refill the reserve + toward its target by drip-selling stocks at ``refill_rate_per_year`` of the + target per year. + + Attributes + ---------- + name: + Human-readable label shown in the UI and charts. + initial_capital: + Starting capital, in the currency of the price data. + reserve_pct: + Cash fraction at inception, in ``[0, 1]``. Stocks get ``1 - reserve_pct``. + rule: + The :class:`DeployRule` that drives deployment. + refill_rate_per_year: + Fraction of the *target* reserve moved back from stocks to cash per year, + while at a new all-time high. ``0.25`` refills a fully-drained reserve in + roughly four years. + trading_days_per_year: + Periods used to annualise the refill drip and metrics (252 for daily). + stock_symbol / cash_symbol: + Column names of the stock and cash legs in the price frame passed to the + engine. + """ + + model_config = {"extra": "forbid"} + + name: str = Field(default="Cash Deploy", min_length=1) + initial_capital: float = Field(default=10_000.0, gt=0, allow_inf_nan=False) + reserve_pct: float = Field(default=0.30, ge=0.0, le=1.0) + rule: DeployRule + refill_rate_per_year: float = Field(default=0.25, ge=0.0, allow_inf_nan=False) + trading_days_per_year: int = Field(default=252, gt=0) + stock_symbol: str = Field(default="S&P 500", min_length=1) + cash_symbol: str = Field(default="Cash (Fed Funds)", min_length=1) + + @model_validator(mode="after") + def _symbols_must_differ(self) -> CashDeployConfig: + if self.stock_symbol == self.cash_symbol: + raise ValueError("stock_symbol and cash_symbol must be different") + return self + + +# The named deploy rules from the research brief, keyed by their (Finnish) name. +# Thresholds and usages are expressed as fractions (10% -> 0.10). +PRESET_RULES: dict[str, DeployRule] = { + "Käyttäjän sääntö": DeployRule( + name="Käyttäjän sääntö", + thresholds=(0.10, 0.20, 0.30, 0.40, 0.50), + usages=(0.15, 0.30, 0.30, 0.20, 0.05), + ), + "Kasvuoptimi": DeployRule( + name="Kasvuoptimi", + thresholds=(0.10, 0.15, 0.20, 0.30, 0.40), + usages=(0.30, 0.25, 0.20, 0.15, 0.10), + ), + "Riskioptimi": DeployRule( + name="Riskioptimi", + thresholds=(0.15, 0.20, 0.30, 0.40, 0.50), + usages=(0.10, 0.15, 0.20, 0.25, 0.30), + ), + "Suositus": DeployRule( + name="Suositus", + thresholds=(0.15, 0.20, 0.30, 0.40, 0.50), + usages=(0.30, 0.25, 0.20, 0.15, 0.10), + ), +} diff --git a/tests/conftest.py b/tests/conftest.py index 4a9b9a7..08bbdde 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -25,3 +25,22 @@ def two_asset_prices(flat_dates: pd.DatetimeIndex) -> pd.DataFrame: }, index=flat_dates, ) + + +@pytest.fixture +def deploy_prices(flat_dates: pd.DatetimeIndex) -> pd.DataFrame: + """A hand-checkable stocks + cash panel for the cash-deploy engine. + + Six business days. ``S&P 500`` sits at its opening high, gaps down 25% (in + one step, so a two-tranche rule with thresholds at 10% and 20% fires both + tranches at once), holds, then recovers to the old high and makes a new one. + ``Cash (Fed Funds)`` is a flat index (per-step growth factor 1.0), so the + reserve earns no interest and the arithmetic stays exact. + """ + return pd.DataFrame( + { + "S&P 500": [100.0, 100.0, 75.0, 75.0, 100.0, 125.0], + "Cash (Fed Funds)": [100.0, 100.0, 100.0, 100.0, 100.0, 100.0], + }, + index=flat_dates, + ) diff --git a/tests/test_cash_deploy.py b/tests/test_cash_deploy.py new file mode 100644 index 0000000..ae3eba6 --- /dev/null +++ b/tests/test_cash_deploy.py @@ -0,0 +1,191 @@ +"""Tests for the tactical cash-deployment engine.""" + +from __future__ import annotations + +import pandas as pd +import pytest + +from portfolio_research_lab.cash_deploy import run_cash_deploy +from portfolio_research_lab.models import CashDeployConfig, DeployRule + + +def _config( + *, + initial_capital: float = 1_000.0, + reserve_pct: float = 0.5, + rule: DeployRule | None = None, + refill_rate_per_year: float = 0.25, + trading_days_per_year: int = 252, +) -> CashDeployConfig: + """A cash-deploy config with a simple two-tranche rule (40% then 60%).""" + if rule is None: + rule = DeployRule(name="two", thresholds=(0.10, 0.20), usages=(0.4, 0.6)) + return CashDeployConfig( + initial_capital=initial_capital, + reserve_pct=reserve_pct, + rule=rule, + refill_rate_per_year=refill_rate_per_year, + trading_days_per_year=trading_days_per_year, + ) + + +def test_initial_split(deploy_prices: pd.DataFrame): + result = run_cash_deploy(deploy_prices, _config()) + # 50% of 1,000 into cash, 50% into stocks at the opening price of 100. + assert result.equity.iloc[0] == pytest.approx(1_000.0) + assert result.cash.iloc[0] == pytest.approx(500.0) + assert result.stock_value.iloc[0] == pytest.approx(500.0) + + +def test_no_drawdown_never_deploys(): + # A stock that only ever rises never triggers a deploy. (It may still refill + # toward the growing target reserve — deployment is the drawdown-only side.) + dates = pd.bdate_range("2020-01-01", periods=4, name="date") + prices = pd.DataFrame( + { + "S&P 500": [100.0, 110.0, 120.0, 130.0], + "Cash (Fed Funds)": [100.0, 100.0, 100.0, 100.0], + }, + index=dates, + ) + result = run_cash_deploy(prices, _config()) + assert (result.events["type"] == "deploy").sum() == 0 + assert result.equity.iloc[-1] > result.equity.iloc[0] + + +def test_gap_down_fires_both_tranches_same_day(deploy_prices: pd.DataFrame): + result = run_cash_deploy(deploy_prices, _config()) + day2 = deploy_prices.index[2] # price 75 => -25% in one step + + fired = result.events.loc[[day2]] + assert list(fired["type"]) == ["deploy", "deploy"] + # Tranche base is the reserve at the episode start = 500. 40% then 60%. + assert list(fired["amount"]) == pytest.approx([200.0, 300.0]) + # Reserve fully drained; both tranches summed to 100% of the base. + assert result.cash.loc[day2] == pytest.approx(0.0) + # Deploy is a transfer at the current price, so equity is unchanged by it: + # 5 units * 75 + 500 cash = 875 either way, all now in stocks. + assert result.equity.loc[day2] == pytest.approx(875.0) + assert result.stock_value.loc[day2] == pytest.approx(875.0) + + +def test_deploy_capped_by_available_cash(): + # Usages summing above 1.0: the second tranche can only spend what's left. + dates = pd.bdate_range("2020-01-01", periods=3, name="date") + prices = pd.DataFrame( + {"S&P 500": [100.0, 100.0, 75.0], "Cash (Fed Funds)": [100.0, 100.0, 100.0]}, + index=dates, + ) + rule = DeployRule(name="greedy", thresholds=(0.10, 0.20), usages=(0.7, 0.7)) + result = run_cash_deploy(prices, _config(rule=rule)) + # base 500: first tranche 350, second capped at remaining 150 (not 350). + assert list(result.events["amount"]) == pytest.approx([350.0, 150.0]) + assert result.cash.iloc[-1] == pytest.approx(0.0) + + +def test_tranche_base_locked_at_episode_start(): + # The reserve grows via cash yield mid-episode, but tranche sizing uses the + # reserve locked at the episode's first dip, not the grown balance. + dates = pd.bdate_range("2020-01-01", periods=4, name="date") + prices = pd.DataFrame( + { + "S&P 500": [100.0, 100.0, 95.0, 89.0], # -5% (no deploy) then -11% + "Cash (Fed Funds)": [100.0, 100.0, 110.0, 121.0], # +10%/step + }, + index=dates, + ) + rule = DeployRule(name="one", thresholds=(0.10,), usages=(0.5,)) + result = run_cash_deploy(prices, _config(rule=rule)) + # First dip is day2 (price 95): cash has already grown to 550 -> base = 550. + # Deploy fires at day3 (-11%); amount = 0.5 * 550 = 275, NOT 0.5 * 605. + assert list(result.events["amount"]) == pytest.approx([275.0]) + + +def test_no_refill_below_all_time_high(): + # A partial recovery that never regains the prior peak must not refill. + dates = pd.bdate_range("2020-01-01", periods=4, name="date") + prices = pd.DataFrame( + {"S&P 500": [100.0, 80.0, 80.0, 90.0], "Cash (Fed Funds)": [100.0] * 4}, + index=dates, + ) + result = run_cash_deploy(prices, _config()) + assert (result.events["type"] == "refill").sum() == 0 + + +def test_refill_drips_toward_target_at_new_high(): + # tdy=2 with a 100%/yr refill => a 50%-of-target drip per ATH day, giving + # hand-checkable refill steps. A -50% drop deploys the whole reserve into + # clean units (5 + 500/50 = 15). + dates = pd.bdate_range("2020-01-01", periods=5, name="date") + prices = pd.DataFrame( + { + "S&P 500": [100.0, 50.0, 50.0, 100.0, 100.0], + "Cash (Fed Funds)": [100.0] * 5, + }, + index=dates, + ) + config = _config(refill_rate_per_year=1.0, trading_days_per_year=2) + result = run_cash_deploy(prices, config) + + day1 = dates[1] # -50%: both tranches fire, cash -> 0, units 15 + assert result.cash.loc[day1] == pytest.approx(0.0) + + day3 = dates[3] # first new high: equity = 15 * 100 = 1500, target = 750, + # drip = 50% of target = 375. + refills = result.events[result.events["type"] == "refill"] + assert refills.loc[day3, "amount"] == pytest.approx(375.0) + assert result.cash.loc[day3] == pytest.approx(375.0) + # Refill is a transfer, so equity is unchanged by it. + assert result.equity.loc[day3] == pytest.approx(1500.0) + + day4 = dates[4] # second drip reaches the target exactly. + assert result.cash.loc[day4] == pytest.approx(750.0) + # Cash never overshoots its target. + target_series = config.reserve_pct * result.equity + assert (result.cash <= target_series + 1e-6).all() + + +def test_cash_accrues_yield_when_idle(): + # No drawdown, but the cash index grows: the reserve should compound with it. + dates = pd.bdate_range("2020-01-01", periods=3, name="date") + prices = pd.DataFrame( + { + "S&P 500": [100.0, 100.0, 100.0], + "Cash (Fed Funds)": [100.0, 101.0, 102.01], # +1%/step + }, + index=dates, + ) + result = run_cash_deploy(prices, _config()) + # 500 cash compounding at +1%/step, stock leg flat at 500. + assert result.cash.iloc[-1] == pytest.approx(500.0 * 1.01 * 1.01) + assert result.equity.iloc[-1] == pytest.approx(500.0 * 1.0201 + 500.0) + + +def test_events_schema(deploy_prices: pd.DataFrame): + result = run_cash_deploy(deploy_prices, _config()) + assert list(result.events.columns) == ["type", "drawdown", "amount", "cash_after"] + + +def test_empty_prices_raise(): + with pytest.raises(ValueError, match="empty price data"): + run_cash_deploy(pd.DataFrame(), _config()) + + +def test_missing_columns_raise(): + dates = pd.bdate_range("2020-01-01", periods=3, name="date") + prices = pd.DataFrame({"S&P 500": [100.0, 100.0, 100.0]}, index=dates) + with pytest.raises(ValueError, match="missing required columns"): + run_cash_deploy(prices, _config()) + + +def test_all_cash_reserve_never_deploys_beyond_cash(): + # reserve_pct = 1.0 => everything starts in cash; deploys draw it down but + # equity stays continuous and non-negative. + dates = pd.bdate_range("2020-01-01", periods=3, name="date") + prices = pd.DataFrame( + {"S&P 500": [100.0, 100.0, 80.0], "Cash (Fed Funds)": [100.0] * 3}, + index=dates, + ) + result = run_cash_deploy(prices, _config(reserve_pct=1.0)) + assert (result.cash >= -1e-9).all() + assert result.equity.iloc[0] == pytest.approx(1_000.0) diff --git a/tests/test_data.py b/tests/test_data.py index 7243581..a6c29d6 100644 --- a/tests/test_data.py +++ b/tests/test_data.py @@ -6,7 +6,12 @@ import pytest from portfolio_research_lab import data -from portfolio_research_lab.data import load_rate_series, parse_price_csv, rate_to_index +from portfolio_research_lab.data import ( + load_rate_series, + load_stocks_cash, + parse_price_csv, + rate_to_index, +) _VALID_CSV = "date,STOCKS,BONDS\n2020-01-01,100.0,100.0\n2020-01-02,101.0,100.5\n" @@ -139,3 +144,43 @@ def test_parse_price_csv_rejects_overlong_symbol(): csv = f"date,{long_name}\n2020-01-01,100.0\n2020-01-02,101.0\n" with pytest.raises(ValueError, match="asset name exceeds"): parse_price_csv(csv) + + +# --- load_stocks_cash ---------------------------------------------------- + + +def _write_stocks_cash(tmp_path): + # S&P starts a year before the fed funds series so the join must trim it. + sp = tmp_path / "sp.csv" + sp.write_text( + "date,close\n" + "2019-01-01,50.0\n" # pre-rate-history: dropped by the inner join + "2020-01-01,100.0\n" + "2020-01-02,101.0\n" + ) + fed = tmp_path / "fed.csv" + fed.write_text('"Date","Value"\n"01/01/2020",3.6\n"01/02/2020",3.6\n') + return sp, fed + + +def test_load_stocks_cash_columns_and_trim(tmp_path): + sp, fed = _write_stocks_cash(tmp_path) + frame = load_stocks_cash(sp, fed) + assert list(frame.columns) == ["S&P 500", "Cash (Fed Funds)"] + # The 2019 S&P row has no cash level and is trimmed. + assert frame.index.min() == pd.Timestamp("2020-01-01") + assert len(frame) == 2 + # Cash is a growth index pinned to its base on the first shared day. + assert (frame["Cash (Fed Funds)"].diff().dropna() > 0).all() + + +def test_load_stocks_cash_custom_names(tmp_path): + sp, fed = _write_stocks_cash(tmp_path) + frame = load_stocks_cash(sp, fed, stock_name="Stocks", cash_name="Money Market") + assert list(frame.columns) == ["Stocks", "Money Market"] + + +def test_load_stocks_cash_rejects_equal_names(tmp_path): + sp, fed = _write_stocks_cash(tmp_path) + with pytest.raises(ValueError, match="must be different"): + load_stocks_cash(sp, fed, stock_name="X", cash_name="X") diff --git a/tests/test_metrics.py b/tests/test_metrics.py index 26bafad..bb0ea9d 100644 --- a/tests/test_metrics.py +++ b/tests/test_metrics.py @@ -79,3 +79,27 @@ def test_summarize_keys(): "annualized_volatility", "max_drawdown", } + + +# --- Sharpe ratio -------------------------------------------------------- + + +def test_sharpe_zero_when_too_few_points(): + assert metrics.sharpe_ratio(pd.Series([0.01])) == 0.0 + + +def test_sharpe_zero_when_no_variation(): + # Constant returns => zero volatility => Sharpe defined as 0.0. + assert metrics.sharpe_ratio(pd.Series([0.01, 0.01, 0.01])) == 0.0 + + +def test_sharpe_positive_for_steady_gains_above_risk_free(): + returns = pd.Series([0.01, 0.02, 0.015, 0.005]) + assert metrics.sharpe_ratio(returns, periods_per_year=252) > 0.0 + + +def test_sharpe_subtracts_risk_free_series(): + returns = pd.Series([0.02, 0.02, 0.02, 0.02], index=pd.RangeIndex(4)) + # A risk-free series equal to the returns leaves zero excess => Sharpe 0. + rf = pd.Series([0.02, 0.02, 0.02, 0.02], index=pd.RangeIndex(4)) + assert metrics.sharpe_ratio(returns, risk_free=rf) == 0.0 diff --git a/tests/test_models.py b/tests/test_models.py index 860a895..e9d7d0b 100644 --- a/tests/test_models.py +++ b/tests/test_models.py @@ -7,7 +7,12 @@ import pytest from pydantic import ValidationError -from portfolio_research_lab.models import StrategyConfig +from portfolio_research_lab.models import ( + PRESET_RULES, + CashDeployConfig, + DeployRule, + StrategyConfig, +) def test_valid_config(): @@ -69,3 +74,75 @@ def test_valid_rebalance_frequencies(freq: str): def test_unknown_rebalance_frequency_rejected(): with pytest.raises(ValidationError, match="rebalance_frequency must be one of"): StrategyConfig(allocations={"A": 1.0}, rebalance_frequency="daily") + + +# --- DeployRule ---------------------------------------------------------- + + +def test_deploy_rule_valid(): + rule = DeployRule(name="r", thresholds=(0.1, 0.2, 0.3), usages=(0.5, 0.3, 0.2)) + assert rule.usage_sum == pytest.approx(1.0) + + +def test_deploy_rule_thresholds_must_increase(): + with pytest.raises(ValidationError, match="strictly increasing"): + DeployRule(name="r", thresholds=(0.2, 0.1), usages=(0.5, 0.5)) + + +def test_deploy_rule_thresholds_must_be_in_unit_interval(): + with pytest.raises(ValidationError, match=r"in \(0, 1\)"): + DeployRule(name="r", thresholds=(0.1, 1.5), usages=(0.5, 0.5)) + + +@pytest.mark.parametrize("bad", [math.nan, math.inf]) +def test_deploy_rule_rejects_non_finite_threshold(bad: float): + with pytest.raises(ValidationError): + DeployRule(name="r", thresholds=(0.1, bad), usages=(0.5, 0.5)) + + +def test_deploy_rule_usages_must_be_positive(): + with pytest.raises(ValidationError, match="must be a positive number"): + DeployRule(name="r", thresholds=(0.1, 0.2), usages=(0.5, -0.1)) + + +def test_deploy_rule_lengths_must_match(): + with pytest.raises(ValidationError, match="equal length"): + DeployRule(name="r", thresholds=(0.1, 0.2), usages=(1.0,)) + + +# --- CashDeployConfig ---------------------------------------------------- + +_RULE = DeployRule(name="r", thresholds=(0.1, 0.2), usages=(0.5, 0.5)) + + +def test_cash_deploy_config_defaults(): + config = CashDeployConfig(rule=_RULE) + assert config.reserve_pct == pytest.approx(0.30) + assert config.refill_rate_per_year == pytest.approx(0.25) + assert config.trading_days_per_year == 252 + + +@pytest.mark.parametrize("reserve", [-0.1, 1.1]) +def test_cash_deploy_config_reserve_pct_bounds(reserve: float): + with pytest.raises(ValidationError): + CashDeployConfig(rule=_RULE, reserve_pct=reserve) + + +def test_cash_deploy_config_rejects_equal_symbols(): + with pytest.raises(ValidationError, match="must be different"): + CashDeployConfig(rule=_RULE, stock_symbol="X", cash_symbol="X") + + +def test_cash_deploy_config_rejects_non_positive_capital(): + with pytest.raises(ValidationError): + CashDeployConfig(rule=_RULE, initial_capital=0) + + +# --- Presets ------------------------------------------------------------- + + +def test_preset_rules_present_and_valid(): + assert set(PRESET_RULES) == {"Käyttäjän sääntö", "Kasvuoptimi", "Riskioptimi", "Suositus"} + for rule in PRESET_RULES.values(): + assert len(rule.thresholds) == len(rule.usages) == 5 + assert rule.usage_sum == pytest.approx(1.0) From 7b3a5e93ac82d689bd1690fd1f9f96ef3c572eb8 Mon Sep 17 00:00:00 2001 From: Juha Kangas <42040080+valuecodes@users.noreply.github.com> Date: Mon, 13 Jul 2026 22:23:17 +0300 Subject: [PATCH 2/3] refactor: rename deploy-rule presets to English Rename the four preset deploy rules from Finnish to English: User rule, Growth optimum, Risk optimum, Recommended. Co-Authored-By: Claude Opus 4.8 --- src/portfolio_research_lab/models.py | 20 ++++++++++---------- tests/test_models.py | 2 +- 2 files changed, 11 insertions(+), 11 deletions(-) diff --git a/src/portfolio_research_lab/models.py b/src/portfolio_research_lab/models.py index 57c3387..7db1f75 100644 --- a/src/portfolio_research_lab/models.py +++ b/src/portfolio_research_lab/models.py @@ -230,26 +230,26 @@ def _symbols_must_differ(self) -> CashDeployConfig: return self -# The named deploy rules from the research brief, keyed by their (Finnish) name. -# Thresholds and usages are expressed as fractions (10% -> 0.10). +# The named deploy rules from the research brief. Thresholds and usages are +# expressed as fractions (10% -> 0.10). PRESET_RULES: dict[str, DeployRule] = { - "Käyttäjän sääntö": DeployRule( - name="Käyttäjän sääntö", + "User rule": DeployRule( + name="User rule", thresholds=(0.10, 0.20, 0.30, 0.40, 0.50), usages=(0.15, 0.30, 0.30, 0.20, 0.05), ), - "Kasvuoptimi": DeployRule( - name="Kasvuoptimi", + "Growth optimum": DeployRule( + name="Growth optimum", thresholds=(0.10, 0.15, 0.20, 0.30, 0.40), usages=(0.30, 0.25, 0.20, 0.15, 0.10), ), - "Riskioptimi": DeployRule( - name="Riskioptimi", + "Risk optimum": DeployRule( + name="Risk optimum", thresholds=(0.15, 0.20, 0.30, 0.40, 0.50), usages=(0.10, 0.15, 0.20, 0.25, 0.30), ), - "Suositus": DeployRule( - name="Suositus", + "Recommended": DeployRule( + name="Recommended", thresholds=(0.15, 0.20, 0.30, 0.40, 0.50), usages=(0.30, 0.25, 0.20, 0.15, 0.10), ), diff --git a/tests/test_models.py b/tests/test_models.py index e9d7d0b..b9d0eae 100644 --- a/tests/test_models.py +++ b/tests/test_models.py @@ -142,7 +142,7 @@ def test_cash_deploy_config_rejects_non_positive_capital(): def test_preset_rules_present_and_valid(): - assert set(PRESET_RULES) == {"Käyttäjän sääntö", "Kasvuoptimi", "Riskioptimi", "Suositus"} + assert set(PRESET_RULES) == {"User rule", "Growth optimum", "Risk optimum", "Recommended"} for rule in PRESET_RULES.values(): assert len(rule.thresholds) == len(rule.usages) == 5 assert rule.usage_sum == pytest.approx(1.0) From c4b18d106c67dd37c01f89d35043f7e903a84b4c Mon Sep 17 00:00:00 2001 From: Juha Kangas <42040080+valuecodes@users.noreply.github.com> Date: Mon, 13 Jul 2026 22:25:21 +0300 Subject: [PATCH 3/3] fix: improve error validation Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- src/portfolio_research_lab/cash_deploy.py | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/src/portfolio_research_lab/cash_deploy.py b/src/portfolio_research_lab/cash_deploy.py index 20b911a..66b1927 100644 --- a/src/portfolio_research_lab/cash_deploy.py +++ b/src/portfolio_research_lab/cash_deploy.py @@ -87,9 +87,13 @@ def run_cash_deploy(prices: pd.DataFrame, config: CashDeployConfig) -> CashDeplo stock = prices[config.stock_symbol].to_numpy(dtype=float) cash_level = prices[config.cash_symbol].to_numpy(dtype=float) - if stock[0] <= 0.0 or not np.isfinite(stock[0]): - raise ValueError(f"the first stock price must be positive, got {stock[0]}") + if (stock <= 0.0).any() or not np.isfinite(stock).all(): + bad = stock[~np.isfinite(stock) | (stock <= 0.0)][0] + raise ValueError(f"stock prices must be finite and positive, got {bad}") + if (cash_level <= 0.0).any() or not np.isfinite(cash_level).all(): + bad = cash_level[~np.isfinite(cash_level) | (cash_level <= 0.0)][0] + raise ValueError(f"cash levels must be finite and positive, got {bad}") # Per-step cash growth factor = ratio of consecutive money-market levels; the # first step has no prior level so it is 1.0. A non-positive/non-finite prior # level would corrupt the ratio, so guard it (falls back to no growth).