Core Functions

The main calculation and simulation functions.

CEFR Calculation

fundedness.cefr.compute_cefr(household=None, balance_sheet=None, liabilities=None, tax_model=None, planning_horizon=None, real_discount_rate=0.02, base_inflation=0.025)

Compute the Certainty-Equivalent Funded Ratio (CEFR).

CEFR = Σ(Asset × (1-τ) × λ × ρ) / PV(Liabilities)

Where

τ = tax rate λ = liquidity factor ρ = reliability factor

Parameters:

Name	Type	Description	Default
household	Household | None	Complete household model (alternative to separate components)	None
balance_sheet	BalanceSheet | None	Asset holdings (if household not provided)	None
liabilities	list[Liability] | None	Future spending obligations (if household not provided)	None
tax_model	TaxModel | None	Tax rate assumptions (defaults to TaxModel())	None
planning_horizon	int | None	Years to plan for (defaults to household horizon or 30)	None
real_discount_rate	float	Real discount rate for liability PV	0.02
base_inflation	float	Base inflation assumption	0.025

Returns:

Type	Description
CEFRResult	CEFRResult with complete breakdown

Source code in fundedness/cefr.py

def compute_cefr(
    household: Household | None = None,
    balance_sheet: BalanceSheet | None = None,
    liabilities: list[Liability] | None = None,
    tax_model: TaxModel | None = None,
    planning_horizon: int | None = None,
    real_discount_rate: float = 0.02,
    base_inflation: float = 0.025,
) -> CEFRResult:
    """Compute the Certainty-Equivalent Funded Ratio (CEFR).

    CEFR = Σ(Asset × (1-τ) × λ × ρ) / PV(Liabilities)

    Where:
        τ = tax rate
        λ = liquidity factor
        ρ = reliability factor

    Args:
        household: Complete household model (alternative to separate components)
        balance_sheet: Asset holdings (if household not provided)
        liabilities: Future spending obligations (if household not provided)
        tax_model: Tax rate assumptions (defaults to TaxModel())
        planning_horizon: Years to plan for (defaults to household horizon or 30)
        real_discount_rate: Real discount rate for liability PV
        base_inflation: Base inflation assumption

    Returns:
        CEFRResult with complete breakdown
    """
    # Extract components from household or use provided values
    if household is not None:
        balance_sheet = household.balance_sheet
        liabilities = household.liabilities
        if planning_horizon is None:
            planning_horizon = household.planning_horizon
    else:
        if balance_sheet is None:
            balance_sheet = BalanceSheet()
        if liabilities is None:
            liabilities = []

    if planning_horizon is None:
        planning_horizon = 30

    if tax_model is None:
        tax_model = TaxModel()

    # Compute asset haircuts
    asset_details = [
        compute_asset_haircuts(asset, tax_model)
        for asset in balance_sheet.assets
    ]

    # Aggregate numerator
    gross_assets = sum(d.gross_value for d in asset_details)
    total_tax_haircut = sum(d.tax_haircut for d in asset_details)
    total_liquidity_haircut = sum(d.liquidity_haircut for d in asset_details)
    total_reliability_haircut = sum(d.reliability_haircut for d in asset_details)
    net_assets = sum(d.net_value for d in asset_details)

    # Compute liability PV (denominator)
    liability_pv, _ = calculate_total_liability_pv(
        liabilities=liabilities,
        planning_horizon=planning_horizon,
        real_discount_rate=real_discount_rate,
        base_inflation=base_inflation,
    )

    # Calculate CEFR
    if liability_pv == 0:
        cefr = float("inf") if net_assets > 0 else 0.0
    else:
        cefr = net_assets / liability_pv

    return CEFRResult(
        cefr=cefr,
        gross_assets=gross_assets,
        total_tax_haircut=total_tax_haircut,
        total_liquidity_haircut=total_liquidity_haircut,
        total_reliability_haircut=total_reliability_haircut,
        net_assets=net_assets,
        liability_pv=liability_pv,
        asset_details=asset_details,
    )

Monte Carlo Simulation

fundedness.simulate.run_simulation(initial_wealth, annual_spending, config, stock_weight=0.6, spending_floor=None, inflation_rate=0.025)

Run Monte Carlo simulation of retirement portfolio.

Parameters:

Name	Type	Description	Default
initial_wealth	float	Starting portfolio value	required
annual_spending	float | ndarray	Annual spending (constant or array by year)	required
config	SimulationConfig	Simulation configuration	required
stock_weight	float | ndarray	Allocation to stocks (constant or array by year)	0.6
spending_floor	float | None	Minimum acceptable spending (for floor breach tracking)	None
inflation_rate	float	Annual inflation rate for real spending	0.025

Returns:

Type	Description
SimulationResult	SimulationResult with all paths and metrics

Source code in fundedness/simulate.py

def run_simulation(
    initial_wealth: float,
    annual_spending: float | np.ndarray,
    config: SimulationConfig,
    stock_weight: float | np.ndarray = 0.6,
    spending_floor: float | None = None,
    inflation_rate: float = 0.025,
) -> SimulationResult:
    """Run Monte Carlo simulation of retirement portfolio.

    Args:
        initial_wealth: Starting portfolio value
        annual_spending: Annual spending (constant or array by year)
        config: Simulation configuration
        stock_weight: Allocation to stocks (constant or array by year)
        spending_floor: Minimum acceptable spending (for floor breach tracking)
        inflation_rate: Annual inflation rate for real spending

    Returns:
        SimulationResult with all paths and metrics
    """
    n_sim = config.n_simulations
    n_years = config.n_years
    seed = config.random_seed

    # Handle spending as array
    if isinstance(annual_spending, (int, float)):
        spending_schedule = np.full(n_years, annual_spending)
    else:
        spending_schedule = np.array(annual_spending)[:n_years]
        if len(spending_schedule) < n_years:
            # Extend with last value
            spending_schedule = np.pad(
                spending_schedule,
                (0, n_years - len(spending_schedule)),
                mode="edge",
            )

    # Handle stock weight as array
    if isinstance(stock_weight, (int, float)):
        stock_weights = np.full(n_years, stock_weight)
    else:
        stock_weights = np.array(stock_weight)[:n_years]
        if len(stock_weights) < n_years:
            stock_weights = np.pad(
                stock_weights,
                (0, n_years - len(stock_weights)),
                mode="edge",
            )

    # Generate returns for each year's allocation
    # For simplicity, use average allocation for return generation
    avg_stock_weight = np.mean(stock_weights)
    returns = generate_returns(
        n_simulations=n_sim,
        n_years=n_years,
        market_model=config.market_model,
        stock_weight=avg_stock_weight,
        random_seed=seed,
    )

    # Initialize paths
    wealth_paths = np.zeros((n_sim, n_years + 1))
    wealth_paths[:, 0] = initial_wealth

    spending_paths = np.zeros((n_sim, n_years)) if config.track_spending else None

    time_to_ruin = np.full(n_sim, np.inf)
    time_to_floor_breach = np.full(n_sim, np.inf) if spending_floor else None

    # Simulate year by year
    for year in range(n_years):
        # Current wealth
        current_wealth = wealth_paths[:, year]

        # Spending (adjusted for inflation)
        real_spending = spending_schedule[year]
        nominal_spending = real_spending * (1 + inflation_rate) ** year

        # Actual spending (can't spend more than we have)
        actual_spending = np.minimum(nominal_spending, np.maximum(current_wealth, 0))

        if spending_paths is not None:
            spending_paths[:, year] = actual_spending

        # Track floor breach
        if time_to_floor_breach is not None and spending_floor:
            floor_breach_mask = (actual_spending < spending_floor * (1 + inflation_rate) ** year)
            floor_breach_mask &= np.isinf(time_to_floor_breach)
            time_to_floor_breach[floor_breach_mask] = year

        # Wealth after spending
        wealth_after_spending = current_wealth - actual_spending

        # Apply returns (only on positive wealth)
        wealth_with_returns = wealth_after_spending * (1 + returns[:, year])
        wealth_with_returns = np.maximum(wealth_with_returns, 0)  # Can't go negative

        wealth_paths[:, year + 1] = wealth_with_returns

        # Track ruin (wealth hits zero)
        ruin_mask = (wealth_paths[:, year + 1] <= 0) & np.isinf(time_to_ruin)
        time_to_ruin[ruin_mask] = year + 1

    # Calculate percentiles
    wealth_percentiles = {}
    spending_percentiles = {}

    for p in config.percentiles:
        key = f"P{p}"
        wealth_percentiles[key] = np.percentile(wealth_paths[:, 1:], p, axis=0)
        if spending_paths is not None:
            spending_percentiles[key] = np.percentile(spending_paths, p, axis=0)

    # Aggregate metrics
    terminal_wealth = wealth_paths[:, -1]
    success_rate = np.mean(np.isinf(time_to_ruin))
    floor_breach_rate = 0.0
    if time_to_floor_breach is not None:
        floor_breach_rate = np.mean(~np.isinf(time_to_floor_breach))

    return SimulationResult(
        wealth_paths=wealth_paths[:, 1:],  # Exclude initial wealth
        spending_paths=spending_paths,
        time_to_ruin=time_to_ruin,
        time_to_floor_breach=time_to_floor_breach,
        wealth_percentiles=wealth_percentiles,
        spending_percentiles=spending_percentiles,
        success_rate=success_rate,
        floor_breach_rate=floor_breach_rate,
        median_terminal_wealth=np.median(terminal_wealth),
        mean_terminal_wealth=np.mean(terminal_wealth),
        n_simulations=n_sim,
        n_years=n_years,
        random_seed=seed,
    )

Strategy Comparison

fundedness.withdrawals.comparison.compare_strategies(policies, initial_wealth, config, stock_weight=0.6, starting_age=65, spending_floor=None)

Compare multiple withdrawal strategies using the same random draws.

Parameters:

Name	Type	Description	Default
policies	list[WithdrawalPolicy]	List of withdrawal policies to compare	required
initial_wealth	float	Starting portfolio value	required
config	SimulationConfig	Simulation configuration	required
stock_weight	float	Asset allocation to stocks	0.6
starting_age	int	Starting age for age-based strategies	65
spending_floor	float | None	Minimum acceptable spending	None

Returns:

Type	Description
StrategyComparisonResult	StrategyComparisonResult with all results and metrics

Source code in fundedness/withdrawals/comparison.py

def compare_strategies(
    policies: list[WithdrawalPolicy],
    initial_wealth: float,
    config: SimulationConfig,
    stock_weight: float = 0.6,
    starting_age: int = 65,
    spending_floor: float | None = None,
) -> StrategyComparisonResult:
    """Compare multiple withdrawal strategies using the same random draws.

    Args:
        policies: List of withdrawal policies to compare
        initial_wealth: Starting portfolio value
        config: Simulation configuration
        stock_weight: Asset allocation to stocks
        starting_age: Starting age for age-based strategies
        spending_floor: Minimum acceptable spending

    Returns:
        StrategyComparisonResult with all results and metrics
    """
    strategy_names = [p.name for p in policies]
    results = {}
    metrics = {}

    # Use same seed for all strategies for fair comparison
    base_seed = config.random_seed or 42

    for i, policy in enumerate(policies):
        # Use same seed for reproducibility
        config_copy = SimulationConfig(
            n_simulations=config.n_simulations,
            n_years=config.n_years,
            random_seed=base_seed,
            market_model=config.market_model,
            tax_model=config.tax_model,
            utility_model=config.utility_model,
            percentiles=config.percentiles,
        )

        result = run_strategy_simulation(
            policy=policy,
            initial_wealth=initial_wealth,
            config=config_copy,
            stock_weight=stock_weight,
            starting_age=starting_age,
            spending_floor=spending_floor,
        )

        results[policy.name] = result

        # Calculate additional metrics
        spending_paths = result.spending_paths
        if spending_paths is not None:
            # Spending volatility (coefficient of variation of spending changes)
            spending_changes = np.diff(spending_paths, axis=1) / spending_paths[:, :-1]
            spending_volatility = np.nanstd(spending_changes)

            # Median initial spending
            median_initial_spending = np.median(spending_paths[:, 0])

            # Average spending
            avg_spending = np.mean(spending_paths)
        else:
            spending_volatility = 0
            median_initial_spending = 0
            avg_spending = 0

        metrics[policy.name] = {
            "success_rate": result.success_rate,
            "floor_breach_rate": result.floor_breach_rate,
            "median_terminal_wealth": result.median_terminal_wealth,
            "mean_terminal_wealth": result.mean_terminal_wealth,
            "median_initial_spending": median_initial_spending,
            "average_spending": avg_spending,
            "spending_volatility": spending_volatility,
        }

    return StrategyComparisonResult(
        strategy_names=strategy_names,
        results=results,
        metrics=metrics,
    )