3bfa46ad4f
All checks were successful
ci/woodpecker/push/woodpecker Pipeline was successful
Adds a thin read-only client for the actualbudget HTTP API
(`fire_planner/actualbudget.py`) and a `GET /spending/annual` endpoint
that returns trailing-N-month spending broken out by category group.
Default exclusions ("Investments and Savings", "Budget Reset") strip
out wealth transfers so the headline number reflects actual
consumption — for Viktor's data, ~£41k/yr instead of the raw £210k
total. Caller can pass `?exclude=...` to override.
Frontend uses the headline `total_gbp` to autofill the Annual spending
input (same pattern as nw_seed from networth), with a small
provenance line below the input showing the window + which groups
were excluded.
Auth: 3 new env vars (ACTUALBUDGET_API_URL/KEY/SYNC_ID) sourced from
Vault `secret/fire-planner` via the existing ExternalSecret —
infra/stacks/fire-planner applied separately. Backend silently keeps
the hardcoded default if the upstream is unreachable.
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
303 lines
10 KiB
Python
303 lines
10 KiB
Python
"""Pydantic response/request schemas for the HTTP API.
|
||
|
||
Mirror the SQLAlchemy ORM but keep them de-coupled — the API surface is a
|
||
contract for the frontend; we don't want migrations to silently change
|
||
JSON shape.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
from datetime import date, datetime
|
||
from decimal import Decimal
|
||
from typing import Any
|
||
|
||
from pydantic import BaseModel, ConfigDict, Field
|
||
|
||
|
||
class _Base(BaseModel):
|
||
model_config = ConfigDict(from_attributes=True)
|
||
|
||
|
||
# ── scenarios ────────────────────────────────────────────────────────
|
||
|
||
|
||
class ScenarioOut(_Base):
|
||
id: int
|
||
external_id: str
|
||
kind: str
|
||
name: str | None
|
||
description: str | None
|
||
parent_scenario_id: int | None
|
||
jurisdiction: str
|
||
strategy: str
|
||
leave_uk_year: int
|
||
glide_path: str
|
||
spending_gbp: Decimal
|
||
horizon_years: int
|
||
nw_seed_gbp: Decimal
|
||
savings_per_year_gbp: Decimal
|
||
config_json: dict[str, Any]
|
||
created_at: datetime
|
||
|
||
|
||
class ScenarioCreate(BaseModel):
|
||
"""Body for POST /scenarios — user-defined scenario."""
|
||
name: str = Field(min_length=1, max_length=200)
|
||
description: str | None = None
|
||
parent_scenario_id: int | None = None
|
||
jurisdiction: str
|
||
strategy: str
|
||
leave_uk_year: int = Field(ge=0, le=60)
|
||
glide_path: str
|
||
spending_gbp: Decimal = Field(gt=0)
|
||
horizon_years: int = Field(ge=5, le=100, default=60)
|
||
nw_seed_gbp: Decimal = Field(ge=0)
|
||
savings_per_year_gbp: Decimal = Field(ge=0, default=Decimal("0"))
|
||
config_json: dict[str, Any] = Field(default_factory=dict)
|
||
|
||
|
||
class ScenarioPatch(BaseModel):
|
||
"""Body for PATCH /scenarios/{id} — all fields optional."""
|
||
name: str | None = None
|
||
description: str | None = None
|
||
jurisdiction: str | None = None
|
||
strategy: str | None = None
|
||
leave_uk_year: int | None = None
|
||
glide_path: str | None = None
|
||
spending_gbp: Decimal | None = None
|
||
horizon_years: int | None = None
|
||
nw_seed_gbp: Decimal | None = None
|
||
savings_per_year_gbp: Decimal | None = None
|
||
config_json: dict[str, Any] | None = None
|
||
|
||
|
||
# ── projections ──────────────────────────────────────────────────────
|
||
|
||
|
||
class ProjectionPoint(_Base):
|
||
year_idx: int
|
||
p10_portfolio_gbp: Decimal
|
||
p25_portfolio_gbp: Decimal
|
||
p50_portfolio_gbp: Decimal
|
||
p75_portfolio_gbp: Decimal
|
||
p90_portfolio_gbp: Decimal
|
||
p50_withdrawal_gbp: Decimal
|
||
p50_tax_gbp: Decimal
|
||
survival_rate: Decimal
|
||
|
||
|
||
class ScenarioProjection(BaseModel):
|
||
"""Latest MC run + per-year fan-chart series for a scenario."""
|
||
scenario_id: int
|
||
external_id: str
|
||
mc_run_id: int
|
||
run_at: datetime
|
||
n_paths: int
|
||
success_rate: Decimal
|
||
p10_ending_gbp: Decimal
|
||
p50_ending_gbp: Decimal
|
||
p90_ending_gbp: Decimal
|
||
median_lifetime_tax_gbp: Decimal
|
||
median_years_to_ruin: Decimal | None
|
||
yearly: list[ProjectionPoint]
|
||
|
||
|
||
# ── net worth ────────────────────────────────────────────────────────
|
||
|
||
|
||
class AccountSnapshotOut(_Base):
|
||
account_id: str
|
||
account_name: str
|
||
account_type: str
|
||
currency: str
|
||
snapshot_date: date
|
||
market_value: Decimal
|
||
market_value_gbp: Decimal
|
||
cost_basis_gbp: Decimal | None
|
||
|
||
|
||
class NetWorthCurrent(BaseModel):
|
||
"""Snapshot at one point in time (latest by default)."""
|
||
snapshot_date: date
|
||
total_gbp: Decimal
|
||
accounts: list[AccountSnapshotOut]
|
||
|
||
|
||
class NetWorthHistoryPoint(BaseModel):
|
||
snapshot_date: date
|
||
total_gbp: Decimal
|
||
by_account: dict[str, Decimal]
|
||
|
||
|
||
class NetWorthHistory(BaseModel):
|
||
"""Per-day NW totals + per-account breakdown for a stacked area chart."""
|
||
points: list[NetWorthHistoryPoint]
|
||
|
||
|
||
# ── annual spending (from actualbudget) ──────────────────────────────
|
||
|
||
|
||
class SpendingMonth(BaseModel):
|
||
"""One month's outflows (positive £) by category group, after
|
||
income groups have been dropped upstream."""
|
||
month: str # "YYYY-MM"
|
||
by_group: dict[str, Decimal]
|
||
total_gbp: Decimal
|
||
|
||
|
||
class AnnualSpending(BaseModel):
|
||
"""Aggregated trailing-N-month spending pulled from actualbudget.
|
||
|
||
`total_gbp` is what we suggest as the "Annual spending" default in
|
||
the WhatIf form: the sum of all category groups across the trailing
|
||
window, *minus* whichever groups the caller asked to exclude
|
||
(default: investments/savings transfers, which aren't consumption).
|
||
"""
|
||
months: int
|
||
window_start: str # "YYYY-MM" (oldest month included)
|
||
window_end: str # "YYYY-MM" (newest)
|
||
excluded_groups: list[str]
|
||
total_gbp: Decimal # the headline number
|
||
raw_total_gbp: Decimal # before exclusions, for transparency
|
||
by_group_total_gbp: dict[str, Decimal] # 12-mo group sums (incl. excluded)
|
||
monthly: list[SpendingMonth]
|
||
|
||
|
||
# ── life events ──────────────────────────────────────────────────────
|
||
|
||
|
||
class LifeEventOut(_Base):
|
||
id: int
|
||
scenario_id: int
|
||
kind: str
|
||
name: str
|
||
year_start: int
|
||
year_end: int | None
|
||
delta_gbp_per_year: Decimal
|
||
one_time_amount_gbp: Decimal | None
|
||
enabled: bool
|
||
payload: dict[str, Any] | None
|
||
created_at: datetime
|
||
|
||
|
||
class LifeEventCreate(BaseModel):
|
||
kind: str
|
||
name: str = Field(min_length=1, max_length=200)
|
||
year_start: int = Field(ge=0, le=100)
|
||
year_end: int | None = Field(default=None, ge=0, le=100)
|
||
delta_gbp_per_year: Decimal = Decimal("0")
|
||
one_time_amount_gbp: Decimal | None = None
|
||
enabled: bool = True
|
||
payload: dict[str, Any] | None = None
|
||
|
||
|
||
class LifeEventPatch(BaseModel):
|
||
kind: str | None = None
|
||
name: str | None = None
|
||
year_start: int | None = None
|
||
year_end: int | None = None
|
||
delta_gbp_per_year: Decimal | None = None
|
||
one_time_amount_gbp: Decimal | None = None
|
||
enabled: bool | None = None
|
||
payload: dict[str, Any] | None = None
|
||
|
||
|
||
# ── goals ────────────────────────────────────────────────────────────
|
||
|
||
|
||
class GoalOut(_Base):
|
||
id: int
|
||
scenario_id: int
|
||
kind: str
|
||
name: str
|
||
target_amount_gbp: Decimal | None
|
||
target_year: int | None
|
||
comparator: str
|
||
success_threshold: Decimal
|
||
enabled: bool
|
||
payload: dict[str, Any] | None
|
||
created_at: datetime
|
||
|
||
|
||
class GoalCreate(BaseModel):
|
||
kind: str
|
||
name: str = Field(min_length=1, max_length=200)
|
||
target_amount_gbp: Decimal | None = None
|
||
target_year: int | None = Field(default=None, ge=0, le=100)
|
||
comparator: str = ">="
|
||
success_threshold: Decimal = Field(default=Decimal("0.95"), ge=0, le=1)
|
||
enabled: bool = True
|
||
payload: dict[str, Any] | None = None
|
||
|
||
|
||
# ── simulate / compare ───────────────────────────────────────────────
|
||
|
||
|
||
class LifeEventInput(BaseModel):
|
||
"""Engine-level event shape — same as the DB row's relevant fields."""
|
||
year_start: int = Field(ge=0, le=100)
|
||
year_end: int | None = Field(default=None, ge=0, le=100)
|
||
delta_gbp_per_year: Decimal = Decimal("0")
|
||
one_time_amount_gbp: Decimal | None = None
|
||
enabled: bool = True
|
||
|
||
|
||
class SimulateRequest(BaseModel):
|
||
"""Sync, non-persisted simulate. Used by the React UI for what-if.
|
||
|
||
Allocation is hardcoded to 100% stocks at the engine layer
|
||
(`api/simulate.py::_project`). The UI removed the glide-path knob
|
||
in 2026-05; persisted Cartesian scenarios still carry their own
|
||
`glide_path` string on the `scenario` table.
|
||
"""
|
||
jurisdiction: str
|
||
strategy: str
|
||
leave_uk_year: int = Field(ge=0, le=60)
|
||
spending_gbp: Decimal = Field(gt=0)
|
||
nw_seed_gbp: Decimal = Field(ge=0)
|
||
savings_per_year_gbp: Decimal = Decimal("0")
|
||
horizon_years: int = Field(ge=5, le=100, default=60)
|
||
floor_gbp: Decimal | None = None
|
||
n_paths: int = Field(ge=100, le=50_000, default=5_000)
|
||
seed: int = 42
|
||
life_events: list[LifeEventInput] = Field(default_factory=list)
|
||
# Returns model — controls how `paths` (n_paths × n_years × 3) is built:
|
||
# "shiller" — block-bootstrap of Shiller 1871+ historical returns
|
||
# (or the synthetic Shiller-calibrated stream when the
|
||
# CSV isn't mounted). The default; broadest regime
|
||
# coverage including 1929/1973/2000/2008.
|
||
# "manual" — every year of every path = `manual_real_return_pct`.
|
||
# Deterministic, no fan, useful for sanity checks.
|
||
# "wealthfolio" — block-bootstrap of the user's actual blended real
|
||
# returns derived from wealthfolio_sync. Reflects the
|
||
# recent regime only (~6 years). Glide path is moot.
|
||
returns_mode: str = Field(default="shiller", pattern="^(shiller|manual|wealthfolio)$")
|
||
manual_real_return_pct: Decimal | None = None
|
||
# Custom spending-plan parameters — only consulted when strategy="custom".
|
||
# All real-£ / real-fraction. annual_real_adjust_pct = 0 means constant
|
||
# real spending (Trinity-shape). Non-zero scales last year's withdrawal
|
||
# multiplicatively each year (e.g. -0.005 for slow-down with age,
|
||
# +0.02 for healthcare creep). Guardrail cuts spending by
|
||
# `guardrail_cut_pct` whenever the portfolio falls below
|
||
# `guardrail_threshold_pct` of its starting value; null disables.
|
||
annual_real_adjust_pct: Decimal = Decimal("0")
|
||
guardrail_threshold_pct: Decimal | None = None
|
||
guardrail_cut_pct: Decimal = Decimal("0.10")
|
||
|
||
|
||
class SimulateResult(BaseModel):
|
||
success_rate: Decimal
|
||
p10_ending_gbp: Decimal
|
||
p50_ending_gbp: Decimal
|
||
p90_ending_gbp: Decimal
|
||
median_lifetime_tax_gbp: Decimal
|
||
median_years_to_ruin: Decimal | None
|
||
elapsed_seconds: Decimal
|
||
yearly: list[ProjectionPoint]
|
||
|
||
|
||
class CompareRequest(BaseModel):
|
||
scenarios: list[SimulateRequest] = Field(min_length=2, max_length=5)
|
||
|
||
|
||
class CompareResult(BaseModel):
|
||
results: list[SimulateResult]
|