fire-planner: Wave 2 chart-first — flex spending, categorised life
Some checks failed
ci/woodpecker/push/woodpecker Pipeline failed

events, interactive Visx Gantt + spending-profile chart

Charts are now the primary editor for life events. The Plan-tab body
re-orders to make charts ~80% of viewport real-estate; legacy form
sections are collapsed into a drawer.

Backend:
- alembic 0004: life_event.category enum (essential / discretionary /
  not_spending). Defaults to essential so existing rows keep their
  full spending impact.
- Simulator gains discretionary_outflows + flex_rules params. Tracks
  per-path running ATH, applies the deepest applicable cut to
  discretionary outflows when portfolio drops vs ATH (PLab-style flex
  spending). Cut amount stays in the portfolio (refund pattern).
- New flex_spending module with FlexRule + applicable_cut +
  cuts_per_year (vectorised). Sortable rules; "deepest cut wins" so
  users specify cumulative cuts at each tier.
- New /scenarios/{id}/spending-profile endpoint returning per-year
  base / essential / discretionary / flex_cut / total breakdown.
- SimulateRequest gains flex_rules + life_event.category roundtrip.
- 8 new tests; 246 total pytest pass; mypy + ruff clean.

Frontend (Visx + ECharts):
- Installed @visx/{scale,shape,group,axis,event,responsive,tooltip}
  for native SVG drag interactions.
- New <SpendingProfileChart> — Visx stacked-area of base/essential/
  discretionary with red flex-cut overlay, hover tooltip, click-to-
  scrub-year.
- New <EventGantt> — interactive Visx Gantt:
    * Click empty space → popover create at that year (default
      essential spending event)
    * Click a bar → inline edit popover (name, kind, range, £/y,
      category) with delete button
    * Drag bar middle → moves the whole event (year-resolution snap)
    * Drag bar edges → resizes year_start / year_end
    * All gestures persist via PATCH /life-events/{id}
- New <FlexRulesEditor> — list of {from_ath_pct, cut} tiers, save-on-
  change to scenario.config_json.flex_rules.
- Plan-tab redesign: NW fan dominant top with floating stat badges
  (Year/Age/NW/Δ NW/Spending/Eff. tax) over the chart; spending-
  profile chart middle; Gantt bottom; flex-rules editor; legacy form
  sections in a collapsed <details> drawer.
- Frontend typecheck + 7 vitest tests + production build all clean.
This commit is contained in:
Viktor Barzin 2026-05-10 16:49:04 +00:00
parent 9cc781a8d6
commit 64eb90c3dc
19 changed files with 2581 additions and 88 deletions

View file

@ -0,0 +1,167 @@
"""Tests for the spending-profile endpoint."""
from __future__ import annotations
from collections.abc import AsyncIterator
from datetime import UTC, datetime
from decimal import Decimal
import pytest_asyncio
from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker
from fire_planner.api.dependencies import get_session
from fire_planner.app import app
from fire_planner.db import LifeEvent, McRun, ProjectionYearly, Scenario
@pytest_asyncio.fixture
async def client(engine: AsyncEngine, session: AsyncSession) -> AsyncIterator[AsyncClient]:
factory = async_sessionmaker(engine, expire_on_commit=False)
async def _override() -> AsyncIterator[AsyncSession]:
async with factory() as s:
yield s
app.dependency_overrides[get_session] = _override
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as ac:
yield ac
app.dependency_overrides.clear()
async def _seed(session: AsyncSession,
flex_rules: list[dict] | None = None) -> int:
config: dict = {}
if flex_rules:
config["flex_rules"] = flex_rules
scen = Scenario(
external_id="user-sp",
kind="user",
name="SP test",
jurisdiction="uk",
strategy="trinity",
leave_uk_year=0,
glide_path="static",
spending_gbp=Decimal("60000"),
horizon_years=5,
nw_seed_gbp=Decimal("1000000"),
savings_per_year_gbp=Decimal("0"),
config_json=config,
)
session.add(scen)
await session.commit()
await session.refresh(scen)
# One persistent essential life event (kid at home), one
# discretionary (travel), one income inflow.
session.add_all([
LifeEvent(
scenario_id=scen.id,
kind="kid_at_home",
name="Kid 1",
year_start=0,
year_end=4,
delta_gbp_per_year=Decimal("-15000"),
category="essential",
enabled=True,
),
LifeEvent(
scenario_id=scen.id,
kind="travel",
name="Travel",
year_start=0,
year_end=4,
delta_gbp_per_year=Decimal("-10000"),
category="discretionary",
enabled=True,
),
LifeEvent(
scenario_id=scen.id,
kind="rental",
name="Rental",
year_start=0,
year_end=4,
delta_gbp_per_year=Decimal("8000"),
category="essential",
enabled=True,
),
])
await session.commit()
return scen.id
async def test_spending_profile_with_no_run(
client: AsyncClient,
session: AsyncSession,
) -> None:
sid = await _seed(session)
resp = await client.get(f"/scenarios/{sid}/spending-profile")
assert resp.status_code == 200, resp.text
body = resp.json()
assert body["horizon_years"] == 5
assert len(body["points"]) == 5
p0 = body["points"][0]
# base = 60000 - 8000 inflow = 52000
assert Decimal(p0["base_gbp"]) == Decimal("52000")
assert Decimal(p0["essential_gbp"]) == Decimal("15000")
assert Decimal(p0["discretionary_gbp"]) == Decimal("10000")
# No projection yet → no flex cut.
assert Decimal(p0["flex_cut_gbp"]) == Decimal("0")
# total = 52000 + 15000 + 10000 = 77000
assert Decimal(p0["total_gbp"]) == Decimal("77000")
async def test_spending_profile_with_flex_rules(
client: AsyncClient,
session: AsyncSession,
) -> None:
flex = [{"from_ath_pct": 0.20, "cut_discretionary_pct": 0.50}]
sid = await _seed(session, flex_rules=flex)
# Persist a fan that drops to 70% of seed (i.e. 30% drawdown vs ATH).
run = McRun(
scenario_id=sid,
run_at=datetime.now(UTC),
n_paths=10,
seed=1,
success_rate=Decimal("1"),
p10_ending_gbp=Decimal("0"),
p50_ending_gbp=Decimal("0"),
p90_ending_gbp=Decimal("0"),
median_lifetime_tax_gbp=Decimal("0"),
median_years_to_ruin=None,
elapsed_seconds=Decimal("0"),
)
session.add(run)
await session.commit()
await session.refresh(run)
rows = [
ProjectionYearly(
mc_run_id=run.id,
year_idx=y,
p10_portfolio_gbp=Decimal("0"),
p25_portfolio_gbp=Decimal("0"),
# year 0 = 1M (ATH); year 1 = 700k (down 30% — flex fires);
# years 2-4 = 800k (still down 20% from ATH 1M).
p50_portfolio_gbp=Decimal(
str([1_000_000, 700_000, 800_000, 800_000, 800_000][y])),
p75_portfolio_gbp=Decimal("0"),
p90_portfolio_gbp=Decimal("0"),
p50_withdrawal_gbp=Decimal("0"),
p50_tax_gbp=Decimal("0"),
survival_rate=Decimal("1"),
) for y in range(5)
]
session.add_all(rows)
await session.commit()
resp = await client.get(f"/scenarios/{sid}/spending-profile")
assert resp.status_code == 200
pts = resp.json()["points"]
# Year 0: portfolio == ATH → no cut.
assert Decimal(pts[0]["flex_cut_gbp"]) == Decimal("0")
# Year 1: drawdown 30% → 50% cut on £10k discretionary = £5k.
assert Decimal(pts[1]["flex_cut_gbp"]) == Decimal("5000.00")
# Year 1 total = 52000 + 15000 + 10000 - 5000 = 72000
assert Decimal(pts[1]["total_gbp"]) == Decimal("72000.00")

View file

@ -0,0 +1,59 @@
"""Tests for the flex-spending engine."""
from __future__ import annotations
import numpy as np
import pytest
from fire_planner.flex_spending import FlexRule, applicable_cut, cuts_per_year
def test_applicable_cut_picks_deepest_rule() -> None:
rules = [
FlexRule(from_ath_pct=0.10, cut_discretionary_pct=0.20),
FlexRule(from_ath_pct=0.30, cut_discretionary_pct=0.60),
FlexRule(from_ath_pct=0.50, cut_discretionary_pct=0.90),
]
# No drawdown — no cut.
assert applicable_cut(0.0, rules) == 0.0
# 9% drop — below first threshold.
assert applicable_cut(0.09, rules) == 0.0
# 15% drop — only first rule fires.
assert applicable_cut(0.15, rules) == pytest.approx(0.20)
# 35% drop — first + second; deepest cut wins (0.60, not 0.80).
assert applicable_cut(0.35, rules) == pytest.approx(0.60)
# 60% drop — all three; 0.90 wins.
assert applicable_cut(0.60, rules) == pytest.approx(0.90)
def test_applicable_cut_empty_rules() -> None:
assert applicable_cut(0.5, []) == 0.0
def test_cuts_per_year_handles_running_ath() -> None:
# Single path. Year 0 seed=1000, year 1 = 1200 (new ATH), year 2 = 800
# (-33% from ATH 1200), year 3 = 900 (still -25% from ATH 1200), year
# 4 = 1300 (new ATH).
portfolio = np.array([[1000, 1200, 800, 900, 1300]], dtype=np.float64)
rules = [
FlexRule(from_ath_pct=0.10, cut_discretionary_pct=0.20),
FlexRule(from_ath_pct=0.30, cut_discretionary_pct=0.60),
]
cuts = cuts_per_year(portfolio, rules)
# cuts[:, y] uses portfolio[:, y] (start-of-year decision based on
# the prior year's close).
# y=0: portfolio=1000 == ATH → 0
# y=1: portfolio=1200 == ATH → 0
# y=2: drawdown = 1 - 800/1200 = 0.333 → 0.60
# y=3: drawdown = 1 - 900/1200 = 0.25 → 0.20
assert cuts.shape == (1, 4)
assert cuts[0, 0] == pytest.approx(0.0)
assert cuts[0, 1] == pytest.approx(0.0)
assert cuts[0, 2] == pytest.approx(0.60)
assert cuts[0, 3] == pytest.approx(0.20)
def test_cuts_per_year_no_rules_returns_zeros() -> None:
portfolio = np.array([[1000, 800, 600]], dtype=np.float64)
cuts = cuts_per_year(portfolio, [])
assert cuts.shape == (1, 2)
assert (cuts == 0).all()

View file

@ -0,0 +1,70 @@
"""End-to-end test that flex-spending rules survive £ in the portfolio."""
from __future__ import annotations
import numpy as np
from fire_planner.flex_spending import FlexRule
from fire_planner.glide_path import static
from fire_planner.simulator import simulate
from fire_planner.strategies.trinity import TrinityStrategy
from fire_planner.tax.uae import UaeTaxRegime
def _flat_paths(n_paths: int, n_years: int, real_return: float = 0.0) -> np.ndarray:
"""Returns paths cube where real return == 0% — easy to reason about."""
paths = np.zeros((n_paths, n_years, 3), dtype=np.float64)
paths[:, :, 0] = real_return # nominal stocks
paths[:, :, 1] = real_return # nominal bonds
paths[:, :, 2] = 0.0 # cpi
return paths
def test_flex_rule_saves_money_at_drawdown() -> None:
"""A scenario that drops below ATH triggers a discretionary cut and
ends up richer than the same scenario with no flex rules."""
paths = _flat_paths(n_paths=10, n_years=5, real_return=-0.05)
initial = 1_000_000.0
common = dict(
paths=paths,
initial_portfolio=initial,
spending_target=10_000.0,
glide=static(1.0),
strategy=TrinityStrategy(),
regime=UaeTaxRegime(),
horizon_years=5,
cashflow_adjustments=np.full(5, -20_000.0, dtype=np.float64),
discretionary_outflows=np.full(5, 20_000.0, dtype=np.float64),
)
no_flex = simulate(**common)
with_flex = simulate(
**common,
flex_rules=[FlexRule(from_ath_pct=0.05, cut_discretionary_pct=0.50)],
)
no_flex_end = float(np.median(no_flex.portfolio_real[:, -1]))
with_flex_end = float(np.median(with_flex.portfolio_real[:, -1]))
assert with_flex_end > no_flex_end
assert no_flex_end > 0 # didn't ruin — meaningful comparison
def test_flex_rule_no_op_without_drawdown() -> None:
"""Strong-positive returns, never below ATH → flex rules do nothing."""
paths = _flat_paths(n_paths=10, n_years=5, real_return=0.10)
common = dict(
paths=paths,
initial_portfolio=1_000_000.0,
spending_target=40_000.0,
glide=static(1.0),
strategy=TrinityStrategy(),
regime=UaeTaxRegime(),
horizon_years=5,
cashflow_adjustments=np.full(5, -10_000.0, dtype=np.float64),
discretionary_outflows=np.full(5, 10_000.0, dtype=np.float64),
)
no_flex = simulate(**common)
with_flex = simulate(
**common,
flex_rules=[FlexRule(from_ath_pct=0.10, cut_discretionary_pct=0.50)],
)
assert np.allclose(no_flex.portfolio_real, with_flex.portfolio_real)