From 69e493e567794a4cb621e2c54f080d4f21d33076 Mon Sep 17 00:00:00 2001 From: Juha Kangas <42040080+valuecodes@users.noreply.github.com> Date: Tue, 21 Jul 2026 17:49:29 +0300 Subject: [PATCH] feat: add moving-average trend-timing backtest --- README.md | 22 +- app/pages/3_Trend_Timing.py | 289 +++++++++++++++++++++++++ src/portfolio_research_lab/__init__.py | 11 +- src/portfolio_research_lab/models.py | 65 ++++++ src/portfolio_research_lab/timing.py | 229 ++++++++++++++++++++ tests/test_timing.py | 178 +++++++++++++++ 6 files changed, 789 insertions(+), 5 deletions(-) create mode 100644 app/pages/3_Trend_Timing.py create mode 100644 src/portfolio_research_lab/timing.py create mode 100644 tests/test_timing.py diff --git a/README.md b/README.md index 4d445ca..8feb40b 100644 --- a/README.md +++ b/README.md @@ -26,6 +26,12 @@ The initial release is a small but working foundation: (validated with Pydantic). - Run a **buy-and-hold** or **periodically rebalanced** (monthly / quarterly / annually) simulation with transparent accounting. +- Backtest a tactical **cash-deployment** strategy (deploy a reserve into stocks + as the market falls, refill at new highs) and search its parameters with a + walk-forward **optimizer**. +- Backtest a **trend-timing** strategy — hold stocks above a moving average + (simple or exponential), step aside to cash below it — with a hysteresis band, + optional per-switch transaction cost, and a one-bar signal lag (no look-ahead). - Calculate **portfolio value, total return, CAGR, annualised volatility and maximum drawdown**. - Display an interactive **equity curve** and **drawdown** chart (Plotly). @@ -143,14 +149,21 @@ request (see `.github/workflows/ci.yml`). ``` market-simulation-lab/ ├── app/ -│ └── Home.py # Streamlit UI (presentation layer only) +│ ├── Home.py # fixed-weight backtest UI +│ └── pages/ +│ ├── 1_Cash_Deploy.py # tactical cash-deploy UI +│ ├── 2_Optimize.py # walk-forward optimizer UI +│ └── 3_Trend_Timing.py # moving-average trend-timing UI ├── src/ │ └── portfolio_research_lab/ │ ├── __init__.py # public API │ ├── models.py # Pydantic configuration models │ ├── data.py # CSV / rate loading + cleaning │ ├── strategies.py # strategy definitions (buy-and-hold) -│ ├── simulator.py # portfolio accounting / engine +│ ├── simulator.py # fixed-weight portfolio engine +│ ├── cash_deploy.py # tactical cash-deployment engine +│ ├── timing.py # moving-average trend-timing engine +│ ├── optimizer.py # Optuna walk-forward parameter search │ └── metrics.py # returns, CAGR, volatility, drawdown ├── tests/ # pytest unit tests ├── data/ # local price CSVs (git-ignored) @@ -163,8 +176,9 @@ market-simulation-lab/ To keep the foundation small and understandable, this project deliberately avoids databases, authentication, cloud services, machine learning and -premature optimization. Transaction costs, taxes and leverage are **not** -modelled yet — the strategy interface leaves room to add them later. +premature optimization. Taxes and leverage are **not** modelled; transaction +costs are modelled only where a strategy trades on a signal (the trend-timing +per-switch cost) — the other engines assume frictionless rebalancing. ## License diff --git a/app/pages/3_Trend_Timing.py b/app/pages/3_Trend_Timing.py new file mode 100644 index 0000000..0830e6e --- /dev/null +++ b/app/pages/3_Trend_Timing.py @@ -0,0 +1,289 @@ +"""Trend Timing — a moving-average crossover backtest page. + +Holds 100% stocks while the S&P is above its moving average and moves 100% to +cash (money-market at the fed funds rate) when it falls below, re-entering when +price climbs back above. Compares the result against a 100%-stocks buy-and-hold +and a static 60/40 split so the user can see whether trend timing actually helped +— the classic "does market timing beat buy-and-hold?" question. + +Like ``1_Cash_Deploy.py`` this is a thin presentation layer: all accounting lives +in ``portfolio_research_lab`` (the timing 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.data import load_stocks_cash # noqa: E402 +from portfolio_research_lab.models import StrategyConfig, TimingConfig # noqa: E402 +from portfolio_research_lab.simulator import run_simulation # noqa: E402 +from portfolio_research_lab.timing import TimingResult, run_ma_timing # 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)" +MA_KIND_LABELS: dict[str, str] = {"Simple (SMA)": "simple", "Exponential (EMA)": "exponential"} + +st.set_page_config( + page_title="Trend Timing · 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 _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: TimingResult, 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 _signal_chart(prices: pd.DataFrame, result: TimingResult) -> go.Figure: + """Stock price and its moving average, with the switch points marked.""" + fig = go.Figure() + fig.add_trace( + go.Scatter(x=prices.index, y=prices[STOCK], name="S&P 500", mode="lines", line={"width": 1}) + ) + fig.add_trace( + go.Scatter( + x=result.moving_average.index, + y=result.moving_average, + name="Moving average", + mode="lines", + line={"width": 1, "dash": "dot"}, + ) + ) + events = result.events + for kind, symbol, color in ( + ("to_stocks", "triangle-up", "green"), + ("to_cash", "triangle-down", "red"), + ): + hits = events[events["type"] == kind] + if not hits.empty: + fig.add_trace( + go.Scatter( + x=hits.index, + y=hits["price"], + name="Enter stocks" if kind == "to_stocks" else "Exit to cash", + mode="markers", + marker={"symbol": symbol, "size": 9, "color": color}, + ) + ) + fig.update_layout(margin={"t": 20, "b": 20}, yaxis_title="Price", 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("📈 Trend Timing") + st.caption( + "Hold stocks while the price is above its moving average, step aside to " + "cash when it falls below. 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." + ) + st.warning( + "**The signal is lagged one day (no look-ahead):** each day's position is " + "decided from the *previous* close and traded at today's close. The " + "200-day average is the conventional choice, not one fitted to this " + "history — searching windows for the best backtest would be curve-fitting." + ) + + try: + prices = _load_stocks_cash() + except (ValueError, FileNotFoundError) as exc: + st.error(f"Could not load price data: {exc}") + st.stop() + + # --- Signal ----------------------------------------------------------- + st.sidebar.header("1 · Signal") + ma_window = st.sidebar.slider("Moving-average window (trading days)", 20, 300, 200, step=10) + ma_kind_label = st.sidebar.selectbox("Average type", list(MA_KIND_LABELS), index=0) + ma_kind = MA_KIND_LABELS[ma_kind_label] + band_pct = st.sidebar.slider("Hysteresis band (%)", 0, 10, 0, step=1) / 100.0 + if band_pct > 0: + st.sidebar.caption( + f"Exit below -{band_pct:.0%} of the average, re-enter above +{band_pct:.0%}; " + "hold inside the band." + ) + + # --- Costs & capital -------------------------------------------------- + st.sidebar.header("2 · Costs & capital") + initial_capital = st.sidebar.number_input( + "Initial capital", min_value=100.0, value=10_000.0, step=1_000.0 + ) + cost_bps = st.sidebar.slider("Transaction cost per switch (bps)", 0, 100, 0, step=5) + st.sidebar.caption("10 bps = 0.10% of the traded amount, charged on each switch.") + + # --- Window ----------------------------------------------------------- + st.sidebar.header("3 · 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.caption( + "The average is recomputed on the selected window, so the strategy starts " + "fully invested for the first " + f"{ma_window - 1} trading days (warm-up)." + ) + + # --- Compare ---------------------------------------------------------- + st.sidebar.header("4 · Compare") + show_stocks = st.sidebar.checkbox("100% stocks", value=True) + show_static = st.sidebar.checkbox("Static 60/40 (annual)", value=True) + + # --- Run -------------------------------------------------------------- + try: + config = TimingConfig( + name="Trend Timing", + initial_capital=initial_capital, + ma_window=ma_window, + ma_kind=ma_kind, + band_pct=band_pct, + cost_bps=float(cost_bps), + stock_symbol=STOCK, + cash_symbol=CASH, + ) + result = run_ma_timing(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: + baselines["Static 60/40"] = _baseline_equity( + prices, + {STOCK: 0.6, CASH: 0.4}, + initial_capital, + rebalance="annually", + name="Static 60/40", + ) + + # --- Headline --------------------------------------------------------- + st.subheader("Results") + c1, c2 = st.columns(2) + c1.metric("Time in market", f"{result.time_in_market:.1%}") + c2.metric("Switches", f"{result.n_switches:,}") + + rows = [ + _metrics_row( + config.name, + result.metrics(), + 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), + metrics.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("Price vs moving average") + st.caption( + "Green ▲ = entered stocks, red ▼ = exited to cash (executed one day after the signal)." + ) + st.plotly_chart(_signal_chart(prices, result), width="stretch") + + st.subheader("Drawdown") + st.plotly_chart(_drawdown_chart(result.drawdown()), width="stretch") + + with st.expander(f"Switch events ({len(result.events)})"): + if result.events.empty: + st.write("No switches in this window — the strategy stayed fully invested.") + else: + display = result.events.copy() + display["price"] = display["price"].map(lambda x: f"{x:,.2f}") + display["ma"] = display["ma"].map(lambda x: f"{x:,.2f}") + display["cost"] = display["cost"].map(lambda x: f"{x:,.2f}") + display["value"] = display["value"].map(lambda x: f"{x:,.0f}") + st.dataframe(display, width="stretch") + + +if __name__ == "__main__": + main() diff --git a/src/portfolio_research_lab/__init__.py b/src/portfolio_research_lab/__init__.py index 3a1dba0..3c70efc 100644 --- a/src/portfolio_research_lab/__init__.py +++ b/src/portfolio_research_lab/__init__.py @@ -17,7 +17,12 @@ parse_price_csv, rate_to_index, ) -from portfolio_research_lab.models import CashDeployConfig, DeployRule, StrategyConfig +from portfolio_research_lab.models import ( + CashDeployConfig, + DeployRule, + StrategyConfig, + TimingConfig, +) from portfolio_research_lab.optimizer import ( ObjectiveKind, OptimizationResult, @@ -28,6 +33,7 @@ ) from portfolio_research_lab.simulator import SimulationResult, run_simulation from portfolio_research_lab.strategies import BuyAndHold, Strategy +from portfolio_research_lab.timing import TimingResult, run_ma_timing __all__ = [ "BuyAndHold", @@ -40,6 +46,8 @@ "SimulationResult", "Strategy", "StrategyConfig", + "TimingConfig", + "TimingResult", "WalkForwardResult", "infer_periods_per_year", "load_price_data", @@ -48,6 +56,7 @@ "parse_price_csv", "rate_to_index", "run_cash_deploy", + "run_ma_timing", "run_simulation", "walk_forward", ] diff --git a/src/portfolio_research_lab/models.py b/src/portfolio_research_lab/models.py index 7db1f75..143c932 100644 --- a/src/portfolio_research_lab/models.py +++ b/src/portfolio_research_lab/models.py @@ -21,6 +21,9 @@ # (weights are set once and left to drift). REBALANCE_FREQUENCIES = frozenset({"monthly", "quarterly", "annually"}) +# Moving-average kinds the trend-timing engine understands. +MA_KINDS = frozenset({"simple", "exponential"}) + class StrategyConfig(BaseModel): """Reusable description of a portfolio strategy. @@ -230,6 +233,68 @@ def _symbols_must_differ(self) -> CashDeployConfig: return self +class TimingConfig(BaseModel): + """Configuration for a moving-average trend-timing backtest. + + Hold 100% stocks while the stock price is above its ``ma_window`` moving + average, and 100% cash (a money-market account) when it falls below. The + signal is evaluated on the previous close and acted on the next day (the + engine lags it one bar to avoid look-ahead). + + Attributes + ---------- + name: + Human-readable label shown in the UI and charts. + initial_capital: + Starting capital, in the currency of the price data. + ma_window: + Moving-average lookback in trading days (e.g. ``200``). Must be > 1. + ma_kind: + ``"simple"`` (equal-weighted rolling mean) or ``"exponential"`` (EWM with + ``span = ma_window``). Both stay fully invested until the average is + defined (the first ``ma_window - 1`` steps). + band_pct: + Hysteresis buffer as a positive fraction in ``[0, 1)``. Exit to cash only + when ``price < ma * (1 - band_pct)`` and re-enter only when + ``price > ma * (1 + band_pct)``; inside the band the position is retained, + which damps whipsaws around the crossover. + cost_bps: + Transaction cost charged on each switch, in basis points of the traded + notional (``10`` = 0.10%). Bounded below 10,000 bps so a switch can never + wipe out or invert a leg. Defaults to 0 for parity with the other engines. + trading_days_per_year: + Periods used to annualise metrics (252 for daily data). + stock_symbol / cash_symbol: + Column names of the stock and cash legs in the price frame. + """ + + model_config = {"extra": "forbid"} + + name: str = Field(default="Trend Timing", min_length=1) + initial_capital: float = Field(default=10_000.0, gt=0, allow_inf_nan=False) + ma_window: int = Field(default=200, gt=1) + ma_kind: str = Field(default="simple") + band_pct: float = Field(default=0.0, ge=0.0, lt=1.0, allow_inf_nan=False) + cost_bps: float = Field(default=0.0, ge=0.0, lt=10_000.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) + + @field_validator("ma_kind") + @classmethod + def _known_ma_kind(cls, value: str) -> str: + if value not in MA_KINDS: + allowed = ", ".join(sorted(MA_KINDS)) + raise ValueError(f"ma_kind must be one of {{{allowed}}}, got {value!r}") + return value + + @model_validator(mode="after") + def _symbols_must_differ(self) -> TimingConfig: + 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. Thresholds and usages are # expressed as fractions (10% -> 0.10). PRESET_RULES: dict[str, DeployRule] = { diff --git a/src/portfolio_research_lab/timing.py b/src/portfolio_research_lab/timing.py new file mode 100644 index 0000000..8a5ccec --- /dev/null +++ b/src/portfolio_research_lab/timing.py @@ -0,0 +1,229 @@ +"""The moving-average trend-timing engine. + +Holds 100% stocks while the stock price is above its moving average and moves +100% to cash (a money-market account) when it falls below. Like +:mod:`~portfolio_research_lab.cash_deploy` this is *path dependent* — the position +held on any day depends on the prior signal and the running in/out state — so it +is a dedicated explicit daily loop rather than a fixed-weight strategy plugged +into the shared simulator. + +The signal is lagged one bar to avoid look-ahead: the position held *during* day +``i`` (and therefore earning day ``i``'s return) is decided from the previous +close (``price[i-1]`` versus ``ma[i-1]``); any resulting switch executes at day +``i``'s close. A hysteresis ``band`` keeps the position unchanged while the price +sits within ``±band`` of the average, damping whipsaws. + +The engine consumes the same two-column ``[stock, cash]`` price frame the app +builds (see :func:`portfolio_research_lab.data.load_stocks_cash`). Cash held out +of the market accrues at the money-market rate, taken as the ratio of consecutive +cash-index levels — identical to the cash-deploy engine. +""" + +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 TimingConfig + +# Columns of the switch-event log emitted by a run. +_EVENT_COLUMNS = ("type", "price", "ma", "cost", "value") + + +@dataclass(slots=True) +class TimingResult: + """Output of a moving-average trend-timing run. + + Attributes + ---------- + config: + The configuration that produced this run. + equity: + Total portfolio value over time (stock leg + cash). + stock_value: + Market value of the stock leg over time (0 while out of the market). + cash: + Cash balance over time (0 while invested). + position: + Exposure held *during* each day, as 1.0 (in stocks) or 0.0 (in cash). + Recorded before any same-day switch, so ``position[i]`` is the exposure + that earned day ``i``'s return. + moving_average: + The moving average of the stock price (NaN during the warm-up window). + events: + One row per switch, indexed by date, with columns ``type`` + (``"to_stocks"``/``"to_cash"``), ``price`` (execution close), ``ma`` (the + prior-day average that triggered the switch), ``cost`` (cash lost to the + transaction cost) and ``value`` (portfolio value just after the switch). + """ + + config: TimingConfig + equity: pd.Series + stock_value: pd.Series + cash: pd.Series + position: pd.Series + moving_average: 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) + + @property + def n_switches(self) -> int: + """Number of in/out switches over the run.""" + return len(self.events) + + @property + def time_in_market(self) -> float: + """Fraction of return intervals spent invested in stocks. + + Averaged over the ``n-1`` return intervals (the first observation has no + preceding return), so it reflects realised exposure rather than a raw row + count. + """ + if len(self.position) < 2: + return float(self.position.iloc[0]) if len(self.position) else 0.0 + return float(self.position.iloc[1:].mean()) + + +def _moving_average(stock: np.ndarray, config: TimingConfig) -> np.ndarray: + """Moving average of the stock price, NaN for the first ``ma_window-1`` steps. + + Both kinds share the same warm-up mask so the two modes initialise + identically: the strategy stays fully invested until the average is defined. + """ + series = pd.Series(stock) + window = config.ma_window + if config.ma_kind == "exponential": + ma = series.ewm(span=window, adjust=False).mean() + # ``ewm`` is defined from the first row; mask the warm-up so it matches the + # simple average's ``min_periods=window`` behaviour. + ma.iloc[: window - 1] = np.nan + else: + ma = series.rolling(window, min_periods=window).mean() + return ma.to_numpy(dtype=float) + + +def run_ma_timing(prices: pd.DataFrame, config: TimingConfig) -> TimingResult: + """Run a moving-average trend-timing simulation. + + Parameters + ---------- + prices: + Wide-format price frame containing (at least) ``config.stock_symbol`` and + ``config.cash_symbol`` columns, indexed by a sorted date index. The cash + column is a money-market growth index; its per-step ratio drives interest + on cash held out of the market. + config: + The trend-timing configuration to simulate. + """ + if prices.empty: + raise ValueError("cannot simulate on empty price data") + if len(prices) < 2: + raise ValueError("need at least two price rows to simulate") + if not prices.index.is_monotonic_increasing: + raise ValueError("price data must be sorted by a monotonically increasing index") + + 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).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. Mirrors the cash-deploy engine. + 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 + + ma = _moving_average(stock, config) + band = config.band_pct + cost_rate = config.cost_bps / 10_000.0 + + # --- State --- + invested = True # start fully invested; the MA is undefined during warm-up + units = config.initial_capital / stock[0] + cash = 0.0 + + n = len(index) + equity_out = np.empty(n, dtype=float) + cash_out = np.empty(n, dtype=float) + stock_out = np.empty(n, dtype=float) + position_out = np.empty(n, dtype=float) + events: list[tuple[pd.Timestamp, str, float, float, float, float]] = [] + + for i in range(n): + price = stock[i] + if i > 0: + cash *= factor[i] # accrue money-market yield on any idle cash + + # The exposure that earns day i's return is the state entering the day, + # before any switch executed today (which only affects tomorrow's return). + position_out[i] = 1.0 if invested else 0.0 + + # Decide from the PREVIOUS close (no look-ahead); execute at today's close. + # Transitions are state-dependent with a hysteresis band: exit only below + # the lower boundary, enter only above the upper one, otherwise hold — so a + # flat price at the average (band 0) never oscillates. + if i >= 1 and np.isfinite(ma[i - 1]): + prev_price = stock[i - 1] + prev_ma = ma[i - 1] + if invested and prev_price < prev_ma * (1.0 - band): + notional = units * price # stocks -> cash + cost = cost_rate * notional + cash = notional - cost + units = 0.0 + invested = False + events.append((index[i], "to_cash", price, prev_ma, cost, cash)) + elif not invested and prev_price > prev_ma * (1.0 + band): + notional = cash # cash -> stocks + cost = cost_rate * notional + units = (cash - cost) / price + cash = 0.0 + invested = True + events.append((index[i], "to_stocks", price, prev_ma, cost, units * price)) + + 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 TimingResult( + config=config, + equity=pd.Series(equity_out, index=index, name="equity"), + stock_value=pd.Series(stock_out, index=index, name=config.stock_symbol), + cash=pd.Series(cash_out, index=index, name=config.cash_symbol), + position=pd.Series(position_out, index=index, name="position"), + moving_average=pd.Series(ma, index=index, name="moving_average"), + events=events_frame, + ) diff --git a/tests/test_timing.py b/tests/test_timing.py new file mode 100644 index 0000000..405612e --- /dev/null +++ b/tests/test_timing.py @@ -0,0 +1,178 @@ +"""Tests for the moving-average trend-timing engine.""" + +from __future__ import annotations + +from collections.abc import Callable + +import pandas as pd +import pytest +from pydantic import ValidationError + +from portfolio_research_lab.models import TimingConfig +from portfolio_research_lab.timing import run_ma_timing + +STOCK = "S&P 500" +CASH = "Cash (Fed Funds)" + + +def _frame(stock: list[float], cash: list[float]) -> pd.DataFrame: + """Build a two-column [stock, cash] price frame on business days.""" + dates = pd.bdate_range("2020-01-01", periods=len(stock), name="date") + return pd.DataFrame( + {STOCK: [float(x) for x in stock], CASH: [float(x) for x in cash]}, index=dates + ) + + +def _config( + *, + initial_capital: float = 1_000.0, + ma_window: int = 2, + ma_kind: str = "simple", + band_pct: float = 0.0, + cost_bps: float = 0.0, +) -> TimingConfig: + """A timing config with a 2-day window (MA defined from the second row).""" + return TimingConfig( + initial_capital=initial_capital, + ma_window=ma_window, + ma_kind=ma_kind, + band_pct=band_pct, + cost_bps=cost_bps, + stock_symbol=STOCK, + cash_symbol=CASH, + ) + + +def test_starts_fully_invested(): + result = run_ma_timing(_frame([100, 110, 120, 130], [100] * 4), _config()) + assert result.equity.iloc[0] == pytest.approx(1_000.0) + assert result.stock_value.iloc[0] == pytest.approx(1_000.0) + assert result.cash.iloc[0] == pytest.approx(0.0) + assert result.position.iloc[0] == 1.0 + + +def test_uptrend_never_exits(): + # Price always above its rising MA => never a reason to step aside. + result = run_ma_timing(_frame([100, 110, 120, 130], [100] * 4), _config()) + assert result.n_switches == 0 + assert (result.position == 1.0).all() + # Behaves exactly like buy-and-hold: 10 units * 130. + assert result.equity.iloc[-1] == pytest.approx(1_300.0) + + +def test_exit_is_lagged_one_bar_then_cash_accrues(): + # Price first prints below its MA at i=2 (90 < 95), but the exit is decided + # from the *previous* close, so it executes at i=3 — proving no look-ahead. + prices = _frame([100, 100, 90, 90, 90], [100, 100, 100, 110, 121]) + result = run_ma_timing(prices, _config()) + + events = result.events + assert list(events["type"]) == ["to_cash"] + day3 = prices.index[3] + assert events.loc[day3, "type"] == "to_cash" + assert events.loc[day3, "price"] == pytest.approx(90.0) + # 10 units sold at 90 => 900 cash, no cost. + assert result.cash.iloc[3] == pytest.approx(900.0) + assert result.stock_value.iloc[3] == pytest.approx(0.0) + # Position entering day 3 was still invested (switch recorded before the day's + # return); the earlier below-MA bar (i=2) did not yet trade. + assert list(result.position) == [1.0, 1.0, 1.0, 1.0, 0.0] + # Out of the market on day 4, the reserve accrues the +10% cash step. + assert result.cash.iloc[4] == pytest.approx(990.0) + assert result.equity.iloc[4] == pytest.approx(990.0) + + +def test_exit_then_reentry(): + # Drop below MA (exit), then climb back above it (re-enter): two switches. + prices = _frame([100, 90, 90, 100, 110], [100] * 5) + result = run_ma_timing(prices, _config()) + + assert list(result.events["type"]) == ["to_cash", "to_stocks"] + # Exit at i=2 (price[1]=90 < ma[1]=95), selling 10 units at 90 => 900 cash. + assert result.cash.iloc[2] == pytest.approx(900.0) + # Re-enter at i=4 (price[3]=100 > ma[3]=95), buying at 110 with 900 cash. + assert result.stock_value.iloc[4] == pytest.approx(900.0) + assert result.cash.iloc[4] == pytest.approx(0.0) + + +def test_band_and_equality_retain_state(): + # A 10% band: a shallow dip to 95 never breaches ma*(1-0.10), so no exit. + banded = run_ma_timing(_frame([100, 100, 95, 95], [100] * 4), _config(band_pct=0.10)) + assert banded.n_switches == 0 + # At band 0, price exactly equal to its MA must also retain the position. + flat = run_ma_timing(_frame([100, 100, 100], [100] * 3), _config(band_pct=0.0)) + assert flat.n_switches == 0 + + +def test_transaction_cost_charged_on_both_directions(): + prices = _frame([100, 90, 90, 100, 110], [100] * 5) + result = run_ma_timing(prices, _config(cost_bps=100.0)) # 1% per switch + + # Exit: 10 units * 90 = 900 notional, 1% cost = 9 => 891 cash. + assert list(result.events["cost"]) == pytest.approx([9.0, 8.91]) + assert result.equity.iloc[2] == pytest.approx(891.0) + # Re-enter: 891 cash notional, 1% cost = 8.91 => 882.09 invested. + assert result.equity.iloc[4] == pytest.approx(882.09) + + +def test_time_in_market_and_switch_count(): + prices = _frame([100, 90, 90, 100, 110], [100] * 5) + result = run_ma_timing(prices, _config()) + # Positions entering each day: invested, invested, invested, cash, cash. + assert list(result.position) == [1.0, 1.0, 1.0, 0.0, 0.0] + # Averaged over the 4 return intervals (drops the first observation): 2/4. + assert result.time_in_market == pytest.approx(0.5) + assert result.n_switches == 2 + + +def test_warmup_masks_both_kinds_identically(): + prices = _frame([100, 101, 102, 103, 104], [100] * 5) + for kind in ("simple", "exponential"): + result = run_ma_timing(prices, _config(ma_window=3, ma_kind=kind)) + ma = result.moving_average + assert ma.iloc[:2].isna().all() # first ma_window - 1 rows are NaN + assert ma.iloc[2:].notna().all() + + +def test_events_schema(): + result = run_ma_timing(_frame([100, 90, 90, 100, 110], [100] * 5), _config()) + assert list(result.events.columns) == ["type", "price", "ma", "cost", "value"] + + +@pytest.mark.parametrize( + "make", + [ + lambda: TimingConfig(cost_bps=10_000.0), + lambda: TimingConfig(band_pct=1.0), + lambda: TimingConfig(cost_bps=float("inf")), + lambda: TimingConfig(ma_window=1), + lambda: TimingConfig(ma_kind="triangular"), + ], +) +def test_invalid_config_rejected(make: Callable[[], TimingConfig]): + with pytest.raises(ValidationError): + make() + + +def test_empty_prices_raise(): + with pytest.raises(ValueError, match="empty price data"): + run_ma_timing(pd.DataFrame(), _config()) + + +def test_single_row_raises(): + with pytest.raises(ValueError, match="at least two"): + run_ma_timing(_frame([100], [100]), _config()) + + +def test_non_monotonic_index_raises(): + dates = pd.DatetimeIndex(pd.to_datetime(["2020-01-03", "2020-01-02"]), name="date") + prices = pd.DataFrame({STOCK: [100.0, 101.0], CASH: [100.0, 100.0]}, index=dates) + with pytest.raises(ValueError, match="monoton"): + run_ma_timing(prices, _config()) + + +def test_missing_columns_raise(): + dates = pd.bdate_range("2020-01-01", periods=3, name="date") + prices = pd.DataFrame({STOCK: [100.0, 100.0, 100.0]}, index=dates) + with pytest.raises(ValueError, match="missing required columns"): + run_ma_timing(prices, _config())