70101c836c
All checks were successful
ci/woodpecker/push/woodpecker Pipeline was successful
The Plan-tab editors (interactive Gantt for life events, flex spending
rules) are now available in What-If too — with local state instead of
DB persistence so users can tweak before committing to a scenario.
Architecture refactor:
- EventGantt is now a controlled component. The `scenarioId` prop +
internal useMutation/useQueryClient hooks went away; the component
takes a `persister: { create, patch, delete }` prop and delegates
every mutation through it. Plan tab wires it to lifeEventsApi +
cache invalidation; What-If wires it to React local state with
negative ids for new events.
- FlexRulesEditor is similarly controlled. Takes `rules + onChange`
instead of a `scenario` object. Plan tab wraps it with the PATCH
/scenarios/:id mutation; What-If wraps it with setFlexRules.
Backend:
- New stateless POST /scenarios/spending-profile-preview endpoint
takes base_spending_gbp + horizon + life_events + flex_rules in the
body and returns the same SpendingProfileResponse shape as the
read-only /scenarios/{id}/spending-profile endpoint. Used by
What-If to render the stacked-area chart against unsaved events.
- SpendingProfileResponse.scenario_id loosened to int | None to
support the preview variant.
Frontend:
- What-If page gains `events` + `flexRules` local state, an
EventGantt + FlexRulesEditor wired through them, and a Visx
spending-profile chart fed by /spending-profile-preview.
- Sim auto-refresh: a 600ms debounced effect re-fires /simulate
whenever the form / events / flex rules change. Manual "Run
simulation" button stays as an immediate trigger.
- "Save as scenario" still works — preserves the scenario params but
not yet the life events / flex rules (a Wave-3 follow-up could
POST them after the scenario is created).
247 pytest pass; mypy + ruff + frontend typecheck/test/build all clean.
526 lines
18 KiB
Python
526 lines
18 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 GoalProbability(BaseModel):
|
||
goal_id: int | None
|
||
name: str
|
||
kind: str
|
||
probability: Decimal
|
||
threshold: Decimal
|
||
passed: bool
|
||
|
||
|
||
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]
|
||
goals_probability: list[GoalProbability] = Field(default_factory=list)
|
||
|
||
|
||
# ── 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 the headline figure used as the "Annual spending"
|
||
default in the WhatIf form. It is **inflation-adjusted to today's
|
||
£** (each month's nominal pence revalued forward by
|
||
`inflation_pct` compounded monthly), matching the simulator's
|
||
real-£ convention.
|
||
|
||
`nominal_total_gbp` is the same window without inflation
|
||
adjustment — for transparency / comparison.
|
||
|
||
`raw_total_gbp` is the nominal sum *including* groups that were
|
||
excluded (e.g. investment transfers) — useful when you want to
|
||
see your full cash outflow.
|
||
"""
|
||
months: int
|
||
window_start: str # "YYYY-MM" (oldest month included)
|
||
window_end: str # "YYYY-MM" (newest)
|
||
excluded_groups: list[str]
|
||
inflation_pct: Decimal # annual rate applied
|
||
total_gbp: Decimal # inflation-adjusted, after exclusions
|
||
nominal_total_gbp: Decimal # not adjusted, after exclusions
|
||
raw_total_gbp: Decimal # nominal, before exclusions
|
||
by_group_total_gbp: dict[str, Decimal] # nominal 12-mo group sums (incl. excluded)
|
||
monthly: list[SpendingMonth]
|
||
|
||
|
||
# ── life events ──────────────────────────────────────────────────────
|
||
|
||
|
||
_CATEGORY_PATTERN = "^(essential|discretionary|not_spending)$"
|
||
|
||
|
||
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
|
||
category: str = "essential"
|
||
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
|
||
category: str = Field(default="essential", pattern=_CATEGORY_PATTERN)
|
||
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
|
||
category: str | None = Field(default=None, pattern=_CATEGORY_PATTERN)
|
||
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
|
||
|
||
|
||
# ── income streams ───────────────────────────────────────────────────
|
||
|
||
|
||
class IncomeStreamOut(_Base):
|
||
id: int
|
||
scenario_id: int
|
||
kind: str
|
||
name: str
|
||
start_year: int
|
||
end_year: int | None
|
||
amount_gbp_per_year: Decimal
|
||
growth_pct: Decimal
|
||
tax_treatment: str
|
||
enabled: bool
|
||
payload: dict[str, Any] | None
|
||
created_at: datetime
|
||
|
||
|
||
class IncomeStreamCreate(BaseModel):
|
||
kind: str
|
||
name: str = Field(min_length=1, max_length=200)
|
||
start_year: int = Field(ge=0, le=100, default=0)
|
||
end_year: int | None = Field(default=None, ge=0, le=100)
|
||
amount_gbp_per_year: Decimal = Field(ge=0, default=Decimal("0"))
|
||
growth_pct: Decimal = Field(default=Decimal("0"), ge=-1, le=1)
|
||
tax_treatment: str = Field(default="income", pattern="^(income|dividend|cgt|tax_free)$")
|
||
enabled: bool = True
|
||
payload: dict[str, Any] | None = None
|
||
|
||
|
||
class IncomeStreamPatch(BaseModel):
|
||
kind: str | None = None
|
||
name: str | None = None
|
||
start_year: int | None = None
|
||
end_year: int | None = None
|
||
amount_gbp_per_year: Decimal | None = None
|
||
growth_pct: Decimal | None = None
|
||
tax_treatment: str | None = None
|
||
enabled: bool | None = None
|
||
payload: dict[str, Any] | None = None
|
||
|
||
|
||
class IncomeStreamInput(BaseModel):
|
||
"""Engine-level income stream — same shape minus the row metadata."""
|
||
kind: str = "salary"
|
||
start_year: int = Field(ge=0, le=100, default=0)
|
||
end_year: int | None = Field(default=None, ge=0, le=100)
|
||
amount_gbp_per_year: Decimal = Field(ge=0, default=Decimal("0"))
|
||
growth_pct: Decimal = Field(default=Decimal("0"))
|
||
tax_treatment: str = Field(default="income", pattern="^(income|dividend|cgt|tax_free)$")
|
||
enabled: bool = True
|
||
|
||
|
||
# ── year stats (per-year scrubber sidebar) ───────────────────────────
|
||
|
||
|
||
class YearStats(BaseModel):
|
||
"""Per-year metrics for the right-hand stats sidebar.
|
||
|
||
Most fields derive from the latest persisted MC run + scenario
|
||
config. Liquid NW / Expenses / Savings rate / Portfolio allocations
|
||
are stubbed null until the relevant Wave 2 features land
|
||
(real-estate split, expense streams, allocation editor).
|
||
"""
|
||
year_idx: int
|
||
calendar_year: int
|
||
age: int | None
|
||
net_worth_p50: Decimal
|
||
change_in_nw: Decimal
|
||
taxable_income: Decimal
|
||
taxes: Decimal
|
||
effective_tax_rate: Decimal
|
||
spending: Decimal
|
||
contributions: Decimal
|
||
investment_growth: Decimal
|
||
liquid_nw: Decimal | None = None
|
||
expenses: Decimal | None = None
|
||
savings_rate: Decimal | None = None
|
||
portfolio_allocations: dict[str, Decimal] | None = None
|
||
|
||
|
||
# ── progress overlay (actuals vs projection) ─────────────────────────
|
||
|
||
|
||
class ProgressActualPoint(BaseModel):
|
||
snapshot_date: date
|
||
total_gbp: Decimal
|
||
|
||
|
||
class ProgressProjectedPoint(BaseModel):
|
||
year_idx: int
|
||
p10_portfolio_gbp: Decimal
|
||
p50_portfolio_gbp: Decimal
|
||
p90_portfolio_gbp: Decimal
|
||
|
||
|
||
class ProgressVariancePoint(BaseModel):
|
||
year_idx: int
|
||
actual_avg_gbp: Decimal
|
||
projected_p50_gbp: Decimal
|
||
delta_gbp: Decimal
|
||
|
||
|
||
class ProgressResponse(BaseModel):
|
||
"""`actual` is daily NW totals from `account_snapshot`. `projected`
|
||
is the latest persisted fan in `projection_yearly`. `alignment_anchor`
|
||
is the date corresponding to year_idx=0 — earliest snapshot or scenario
|
||
`created_at` when no snapshots exist yet."""
|
||
scenario_id: int
|
||
alignment_anchor: date
|
||
actual: list[ProgressActualPoint]
|
||
projected: list[ProgressProjectedPoint]
|
||
variance: list[ProgressVariancePoint]
|
||
|
||
|
||
# ── cashflow Sankey ──────────────────────────────────────────────────
|
||
|
||
|
||
class CashflowResponse(BaseModel):
|
||
"""Per-year sources/sinks for a Sankey diagram. Sums conserve at the
|
||
Sankey level: total inflow == total outflow, with savings + ending-NW
|
||
delta absorbing any leftover."""
|
||
scenario_id: int
|
||
year: int
|
||
sources: dict[str, Decimal]
|
||
sinks: dict[str, Decimal]
|
||
|
||
|
||
# ── 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
|
||
category: str = Field(default="essential", pattern=_CATEGORY_PATTERN)
|
||
enabled: bool = True
|
||
|
||
|
||
class FlexRule(BaseModel):
|
||
"""ProjectionLab-style flex-spending rule.
|
||
|
||
When the portfolio falls ``from_ath_pct`` below its running all-time-high,
|
||
cut discretionary spending by ``cut_discretionary_pct``. Multiple rules
|
||
stack via "worst applicable threshold wins" — at -30% from ATH a rule
|
||
keyed at -10% AND a rule keyed at -25% both apply, but only the deeper
|
||
cut takes effect (so users specify *cumulative* cuts, not per-tier).
|
||
|
||
`from_ath_pct` is the absolute drop magnitude as a positive fraction:
|
||
0.30 means "the portfolio is 30% below its ATH". This matches the way
|
||
PLab labels its sliders ("if down 30%, cut 60%").
|
||
"""
|
||
from_ath_pct: Decimal = Field(ge=0, le=1)
|
||
cut_discretionary_pct: Decimal = Field(ge=0, le=1)
|
||
|
||
|
||
# ── spending profile ────────────────────────────────────────────────
|
||
|
||
|
||
class SpendingProfilePoint(BaseModel):
|
||
year_idx: int
|
||
base_gbp: Decimal
|
||
essential_gbp: Decimal
|
||
discretionary_gbp: Decimal
|
||
not_spending_gbp: Decimal
|
||
flex_cut_gbp: Decimal
|
||
total_gbp: Decimal
|
||
|
||
|
||
class SpendingProfileResponse(BaseModel):
|
||
scenario_id: int | None = None
|
||
horizon_years: int
|
||
points: list[SpendingProfilePoint]
|
||
|
||
|
||
class SpendingProfilePreviewRequest(BaseModel):
|
||
"""Stateless spending-profile preview — used by the What-If page where
|
||
the user is editing in-memory life events that aren't persisted yet.
|
||
|
||
No scenario_id needed; the caller supplies the baseline spending,
|
||
horizon, and any flex rules. Flex-cut estimation against a portfolio
|
||
fan is skipped (caller already has a /simulate response to pair
|
||
with this if they want flex visualisation).
|
||
"""
|
||
base_spending_gbp: Decimal = Field(ge=0)
|
||
horizon_years: int = Field(ge=1, le=100)
|
||
life_events: list[LifeEventInput] = Field(default_factory=list)
|
||
flex_rules: list[FlexRule] = Field(default_factory=list)
|
||
|
||
|
||
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
|
||
income_streams: list[IncomeStreamInput] = Field(default_factory=list)
|
||
goals: list[GoalCreate] = Field(default_factory=list)
|
||
flex_rules: list[FlexRule] = Field(default_factory=list)
|
||
# Rates settings (Wave 1.D.3). When `rates_mode='fixed'`, the engine
|
||
# synthesises a deterministic real-return path from the per-asset
|
||
# growth + dividend + inflation rates below, weighted by the static
|
||
# 100% glide. `historical` falls back to Shiller bootstrap (legacy
|
||
# `returns_mode` honoured when rates_mode is not set). `advanced` is
|
||
# a Wave 2 stub.
|
||
rates_mode: str | None = Field(default=None, pattern="^(fixed|historical|advanced)$")
|
||
inflation_pct: Decimal = Field(default=Decimal("0.03"))
|
||
stocks_growth_pct: Decimal = Field(default=Decimal("0.06"))
|
||
stocks_dividend_pct: Decimal = Field(default=Decimal("0.025"))
|
||
bonds_growth_pct: Decimal = Field(default=Decimal("0.015"))
|
||
bonds_dividend_pct: Decimal = Field(default=Decimal("0.035"))
|
||
stocks_allocation: Decimal = Field(default=Decimal("1.0"), ge=0, le=1)
|
||
# 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]
|
||
goals_probability: list[GoalProbability] = Field(default_factory=list)
|
||
|
||
|
||
class CompareRequest(BaseModel):
|
||
scenarios: list[SimulateRequest] = Field(min_length=2, max_length=5)
|
||
|
||
|
||
class CompareResult(BaseModel):
|
||
results: list[SimulateResult]
|