231 lines
8.5 KiB
Python
231 lines
8.5 KiB
Python
"""Monte Carlo simulator — the core of fire-planner.
|
||
|
||
Inputs:
|
||
- a `(n_paths, n_years, 3)` bootstrap of returns + CPI (`returns/`)
|
||
- a withdrawal strategy (`strategies/`)
|
||
- a glide-path function (`glide_path`)
|
||
- a tax regime (`tax/`)
|
||
- starting portfolio + spending target
|
||
|
||
Per path the simulator runs a 60-year (or whatever horizon) lifecycle:
|
||
|
||
for each year y in 0..H:
|
||
asset_alloc = glide(y)
|
||
portfolio = portfolio * (1 + alloc·stock + (1-alloc)·bond)
|
||
withdrawal = strategy.propose(state)
|
||
tax = regime.compute_tax(...)
|
||
portfolio -= (withdrawal + tax_in_addition)
|
||
if portfolio < 0: failed_path
|
||
|
||
We work entirely in REAL GBP — convert returns to real (return/inflation
|
||
factor each year). Tax brackets are assumed to inflate with CPI (a v2
|
||
follow-up on fiscal drag).
|
||
|
||
Annual savings (during the accumulation phase) get added at year start
|
||
and earn the year's return.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
from collections.abc import Callable
|
||
from dataclasses import dataclass
|
||
from decimal import Decimal
|
||
|
||
import numpy as np
|
||
import numpy.typing as npt
|
||
|
||
from fire_planner.glide_path import GlideFn
|
||
from fire_planner.strategies.base import StrategyState, WithdrawalStrategy
|
||
from fire_planner.tax.base import TaxInputs, TaxRegime
|
||
|
||
# Stock idx, bond idx, cpi idx in the bootstrap output.
|
||
STOCK = 0
|
||
BOND = 1
|
||
CPI = 2
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class SimulationResult:
|
||
"""Per-path arrays + scalar summaries.
|
||
|
||
All money in REAL GBP (today's pounds). `portfolio_real[:, 0]` is
|
||
the seed; `portfolio_real[:, k]` is end-of-year-k.
|
||
"""
|
||
portfolio_real: npt.NDArray[np.float64] # (n_paths, n_years+1)
|
||
withdrawal_real: npt.NDArray[np.float64] # (n_paths, n_years)
|
||
tax_real: npt.NDArray[np.float64] # (n_paths, n_years)
|
||
success_mask: npt.NDArray[np.bool_] # (n_paths,)
|
||
|
||
@property
|
||
def n_paths(self) -> int:
|
||
return int(self.portfolio_real.shape[0])
|
||
|
||
@property
|
||
def n_years(self) -> int:
|
||
return int(self.withdrawal_real.shape[1])
|
||
|
||
@property
|
||
def success_rate(self) -> float:
|
||
return float(np.mean(self.success_mask))
|
||
|
||
def ending_percentile(self, p: int) -> float:
|
||
return float(np.percentile(self.portfolio_real[:, -1], p))
|
||
|
||
def median_lifetime_tax(self) -> float:
|
||
return float(np.median(self.tax_real.sum(axis=1)))
|
||
|
||
def median_years_to_ruin(self) -> float | None:
|
||
"""Among failing paths, the median year-to-ruin (1-indexed).
|
||
Returns None if every path survives."""
|
||
failing = ~self.success_mask
|
||
if not failing.any():
|
||
return None
|
||
portfolios = self.portfolio_real[failing, 1:]
|
||
# First year-end where portfolio == 0 (or below)
|
||
ruin_year = np.argmax(portfolios <= 0, axis=1) + 1
|
||
return float(np.median(ruin_year))
|
||
|
||
def sequence_risk_correlation(self) -> float:
|
||
"""Pearson correlation between year-1 drawdown and total
|
||
success (1 if survived, 0 if not). Year-1 drawdown =
|
||
(initial - portfolio_after_year1) / initial.
|
||
|
||
Returns 0.0 when either variable has zero variance — e.g. all
|
||
paths share the same year-1 returns (fixed_paths fixture) or
|
||
every path succeeds.
|
||
"""
|
||
initial = self.portfolio_real[:, 0]
|
||
after_y1 = self.portfolio_real[:, 1]
|
||
drawdown = (initial - after_y1) / initial
|
||
success = self.success_mask.astype(np.float64)
|
||
if np.var(drawdown) < 1e-12 or np.var(success) < 1e-12:
|
||
return 0.0
|
||
with np.errstate(invalid="ignore"):
|
||
corr = np.corrcoef(drawdown, success)[0, 1]
|
||
return 0.0 if np.isnan(corr) else float(corr)
|
||
|
||
def fan_quantiles(self, p: int) -> npt.NDArray[np.float64]:
|
||
"""Per-year cross-path percentile of portfolio_real."""
|
||
out: npt.NDArray[np.float64] = np.percentile(self.portfolio_real, p, axis=0)
|
||
return out
|
||
|
||
|
||
# Default split of withdrawal across tax-treated buckets. The simulator
|
||
# treats withdrawals as "capital gains" by default since most accounts
|
||
# we model are taxable brokerage; ISA wraps don't tax at all and SIPP
|
||
# withdrawals are 25%-tax-free + ordinary income.
|
||
_BucketSplit = Callable[[float, int], TaxInputs]
|
||
|
||
|
||
def default_bucket_split(real_withdrawal: float, year_idx: int) -> TaxInputs:
|
||
"""Treat the entire withdrawal as long-term capital gains.
|
||
Override via `bucket_split` to reflect ISA / SIPP / divs balances."""
|
||
del year_idx
|
||
return TaxInputs(capital_gains=Decimal(str(round(real_withdrawal, 2))))
|
||
|
||
|
||
RegimeFn = Callable[[int], TaxRegime]
|
||
|
||
|
||
def constant_regime(regime: TaxRegime) -> RegimeFn:
|
||
return lambda _y: regime
|
||
|
||
|
||
def jurisdiction_schedule(
|
||
pre_departure: TaxRegime,
|
||
post_departure: TaxRegime,
|
||
leave_year: int,
|
||
) -> RegimeFn:
|
||
"""While `year < leave_year` apply `pre_departure`; from `leave_year`
|
||
onwards apply `post_departure`. Used to model "live in UK for N more
|
||
years then move to Cyprus/Bulgaria/etc."
|
||
"""
|
||
|
||
def fn(year: int) -> TaxRegime:
|
||
return pre_departure if year < leave_year else post_departure
|
||
|
||
return fn
|
||
|
||
|
||
def simulate(
|
||
paths: npt.NDArray[np.float64],
|
||
initial_portfolio: float,
|
||
spending_target: float,
|
||
glide: GlideFn,
|
||
strategy: WithdrawalStrategy,
|
||
regime: TaxRegime | RegimeFn,
|
||
horizon_years: int | None = None,
|
||
annual_savings: npt.NDArray[np.float64] | None = None,
|
||
bucket_split: _BucketSplit = default_bucket_split,
|
||
) -> SimulationResult:
|
||
"""Run the MC simulation. `paths` shape: (n_paths, n_years, 3).
|
||
|
||
`spending_target` is the year-0 real GBP draw; subsequent years are
|
||
decided by the strategy. `annual_savings`, if given, is a (n_years,)
|
||
real-GBP array — added at the start of each year while accumulating.
|
||
|
||
`regime` may be a single `TaxRegime` (constant for all years) or a
|
||
callable `(year_idx) -> TaxRegime` to model jurisdiction switches —
|
||
e.g. UK for years 0..N-1, then Cyprus from year N onward.
|
||
"""
|
||
regime_at: RegimeFn = (regime if callable(regime) else constant_regime(regime))
|
||
n_paths, n_years, _ = paths.shape
|
||
if horizon_years is None:
|
||
horizon_years = n_years
|
||
|
||
portfolio = np.full(n_paths, float(initial_portfolio), dtype=np.float64)
|
||
portfolio_history = np.zeros((n_paths, n_years + 1), dtype=np.float64)
|
||
portfolio_history[:, 0] = portfolio
|
||
withdrawal_hist = np.zeros((n_paths, n_years), dtype=np.float64)
|
||
tax_hist = np.zeros((n_paths, n_years), dtype=np.float64)
|
||
last_withdrawal = np.full(n_paths, float(spending_target), dtype=np.float64)
|
||
|
||
if annual_savings is None:
|
||
annual_savings = np.zeros(n_years, dtype=np.float64)
|
||
|
||
for y in range(n_years):
|
||
alloc = glide(y)
|
||
# Real returns for this year, all paths: shape (n_paths,)
|
||
nominal_stock = paths[:, y, STOCK]
|
||
nominal_bond = paths[:, y, BOND]
|
||
cpi = paths[:, y, CPI]
|
||
real_stock = (1 + nominal_stock) / (1 + cpi) - 1
|
||
real_bond = (1 + nominal_bond) / (1 + cpi) - 1
|
||
port_return = alloc * real_stock + (1 - alloc) * real_bond
|
||
|
||
# Add savings at year start, then apply year's return.
|
||
portfolio = (portfolio + annual_savings[y]) * (1 + port_return)
|
||
|
||
# Strategy is per-path Python — 600k iterations at 60y × 10k paths.
|
||
# Profiled: ~3 seconds for the full Trinity / GK / VPW set.
|
||
for p in range(n_paths):
|
||
state = StrategyState(
|
||
portfolio=float(portfolio[p]),
|
||
initial_portfolio=float(initial_portfolio),
|
||
initial_withdrawal=float(spending_target),
|
||
year_idx=y,
|
||
horizon_years=horizon_years,
|
||
last_withdrawal=float(last_withdrawal[p]),
|
||
)
|
||
w = strategy.propose_withdrawal(state)
|
||
# Strategies are real-GBP; clip to portfolio.
|
||
w = max(0.0, min(w, float(portfolio[p])))
|
||
tax_breakdown = regime_at(y).compute_tax(bucket_split(w, y))
|
||
t = float(tax_breakdown.total)
|
||
portfolio[p] = portfolio[p] - w
|
||
withdrawal_hist[p, y] = w
|
||
tax_hist[p, y] = t
|
||
last_withdrawal[p] = w
|
||
|
||
portfolio_history[:, y + 1] = np.clip(portfolio, a_min=0.0, a_max=None)
|
||
portfolio = portfolio_history[:, y + 1]
|
||
|
||
# Success = portfolio stayed positive through every interim year.
|
||
# Excludes the very last year-end because VPW deliberately drains
|
||
# to ~0 at the horizon boundary by construction; that's not a failure.
|
||
success_mask = portfolio_history[:, 1:-1].min(axis=1) > 0.0
|
||
return SimulationResult(
|
||
portfolio_real=portfolio_history,
|
||
withdrawal_real=withdrawal_hist,
|
||
tax_real=tax_hist,
|
||
success_mask=success_mask,
|
||
)
|