fire-planner: ProjectionLab parity Wave 1 — tabbed shell, year stats, goals,

income streams, Sankey cashflow, progress overlay, settings sub-pages

Wave 1 (9 features across 4 streams):

Stream A — dashboard skeleton
  1.A.1 ScenarioShell with top tabs (Plan/Cash Flow/Tax Analytics/Compare/
        Reports/Estate/Settings) + left Sidebar with Plans switcher.
  1.A.2 GET /scenarios/{id}/year-stats?year=N returning per-year metrics
        (NW, Δ NW, taxable income, taxes, eff. rate, spending, contribs,
        investment growth). YearScrubber + YearStatsPanel render the
        right-hand sidebar; URL ?year= preserves selection.
  1.A.3 FanChart gains optional `milestones` prop (lib/milestone.ts maps
        life_event.kind → emoji) + selectedYear marker line.

Stream B — goals + progress
  1.B.1 New goals_eval module: target_nw_by_year / never_run_out /
        target_real_income probability evaluation. Wired into POST
        /simulate (exact, per-path) and GET /scenarios/{id}/projection
        (approximated from persisted fan via percentile interpolation).
        GoalsSection renders pass/fail badges.
  1.B.2 GET /scenarios/{id}/progress overlays AccountSnapshot totals on
        the projection fan; ProgressPage shows variance side-panel.

Stream C — income + cashflow
  1.C.1 New IncomeStream model + alembic 0003 + CRUD endpoints. Engine
        aggregates streams into per-year inflows + taxable arrays;
        income tax routes through the jurisdiction tax engine.
        IncomeStreamsSection on Plan tab.
  1.C.2 GET /scenarios/{id}/cashflow?year=N returns sources/sinks for
        an ECharts Sankey (sums conserve). CashflowTab body.

Stream D — settings
  1.D.1 SettingsTab + sub-nav (Milestones/Rates/Dividends/Bonds/Tax/
        Metrics/Other/Notes); placeholder cards for unbuilt sub-pages.
  1.D.2 LifeEventsSection relocated to /scenarios/:id/settings.
  1.D.3 RatesSettings (Fixed/Historical/Advanced segmented + per-asset
        cards). SimulateRequest gains rates_mode, inflation_pct,
        stocks/bonds growth + dividend, stocks_allocation. New
        build_fixed_paths() in simulator. Real-return arithmetic
        verified against (1+g+d)/(1+i)−1 ≈ 5.4%.
  1.D.4 NotesSettings — markdown textarea, save-on-blur, stored in
        scenario.config_json.notes.

Backend: 238 pytest pass (+19 new), mypy + ruff clean.
Frontend: typecheck + 7 unit tests + production build clean.

Roadmap for Wave 2-N is documented in the implementation plan.

fire_planner/api/cashflow.py

@@ -0,0 +1,130 @@
+"""Cashflow Sankey data for one year — ProjectionLab parity Wave 1.C.2.
+
+Returns a ``{sources, sinks}`` map of named tributaries / drains. Sums
+conserve at the Sankey level: total inflow == total outflow, with
+"net change in NW" absorbing the residual so the diagram balances.
+"""
+from __future__ import annotations
+
+from decimal import Decimal
+
+from fastapi import APIRouter, Depends, HTTPException, Query
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from fire_planner.api.dependencies import get_session
+from fire_planner.api.schemas import CashflowResponse
+from fire_planner.db import IncomeStream, LifeEvent, McRun, ProjectionYearly, Scenario
+
+router = APIRouter(prefix="/scenarios", tags=["cashflow"])
+
+
+def _income_inflow_at(streams: list[IncomeStream], year_idx: int) -> dict[str, Decimal]:
+    """Per-stream-kind inflow at ``year_idx``."""
+    out: dict[str, Decimal] = {}
+    for s in streams:
+        if not s.enabled:
+            continue
+        if year_idx < s.start_year:
+            continue
+        if s.end_year is not None and year_idx > s.end_year:
+            continue
+        years_active = year_idx - s.start_year
+        growth = (Decimal("1") + Decimal(str(s.growth_pct)))**years_active
+        amount = Decimal(str(s.amount_gbp_per_year)) * growth
+        key = f"income:{s.kind}"
+        out[key] = out.get(key, Decimal("0")) + amount
+    return out
+
+
+def _life_event_inflow_at(events: list[LifeEvent], year_idx: int) -> dict[str, Decimal]:
+    out: dict[str, Decimal] = {}
+    for ev in events:
+        if not ev.enabled:
+            continue
+        if year_idx < ev.year_start:
+            continue
+        end = ev.year_end if ev.year_end is not None else ev.year_start
+        if year_idx > end:
+            continue
+        delta = Decimal(str(ev.delta_gbp_per_year or 0))
+        if ev.year_start == year_idx and ev.one_time_amount_gbp is not None:
+            delta += Decimal(str(ev.one_time_amount_gbp))
+        if delta == 0:
+            continue
+        key = f"event:{ev.name}"
+        if delta > 0:
+            out[key] = out.get(key, Decimal("0")) + delta
+    return out
+
+
+@router.get("/{scenario_id}/cashflow", response_model=CashflowResponse)
+async def get_cashflow(
+    scenario_id: int,
+    year: int = Query(ge=0, le=100, description="Year offset from scenario start (0 = now)"),
+    session: AsyncSession = Depends(get_session),
+) -> CashflowResponse:
+    scen = await session.get(Scenario, scenario_id)
+    if scen is None:
+        raise HTTPException(status_code=404, detail="Scenario not found")
+    run = (await session.execute(
+        select(McRun).where(McRun.scenario_id == scenario_id).order_by(
+            McRun.run_at.desc()).limit(1))).scalar_one_or_none()
+    if run is None:
+        raise HTTPException(status_code=404, detail="No MC runs persisted for this scenario yet")
+
+    yearly_rows = (await session.execute(
+        select(ProjectionYearly).where(ProjectionYearly.mc_run_id == run.id).order_by(
+            ProjectionYearly.year_idx))).scalars().all()
+    by_year = {r.year_idx: r for r in yearly_rows}
+    if year not in by_year:
+        raise HTTPException(status_code=404, detail=f"No projection row for year {year}")
+
+    streams = list((await session.execute(
+        select(IncomeStream).where(
+            IncomeStream.scenario_id == scenario_id))).scalars().all())
+    events = list((await session.execute(
+        select(LifeEvent).where(LifeEvent.scenario_id == scenario_id))).scalars().all())
+
+    row = by_year[year]
+    prev = by_year.get(year - 1) if year > 0 else None
+    nw = Decimal(str(row.p50_portfolio_gbp))
+    prev_nw = (Decimal(str(prev.p50_portfolio_gbp)) if prev is not None else Decimal(str(
+        scen.nw_seed_gbp)))
+    nw_change = nw - prev_nw
+
+    sources: dict[str, Decimal] = {}
+    sources.update(_income_inflow_at(streams, year))
+    sources.update(_life_event_inflow_at(events, year))
+    if scen.savings_per_year_gbp > 0:
+        sources["savings"] = Decimal(str(scen.savings_per_year_gbp))
+
+    spending = Decimal(str(row.p50_withdrawal_gbp))
+    taxes = Decimal(str(row.p50_tax_gbp))
+    sinks: dict[str, Decimal] = {
+        "spending": spending,
+        "taxes": taxes,
+    }
+
+    sources_total = sum(sources.values(), Decimal("0"))
+    sinks_total = sum(sinks.values(), Decimal("0"))
+    investment_returns = nw_change + sinks_total - sources_total
+    if investment_returns > 0:
+        sources["investment_returns"] = investment_returns
+        sources_total += investment_returns
+    elif investment_returns < 0:
+        sinks["portfolio_drawdown"] = -investment_returns
+        sinks_total += -investment_returns
+
+    delta = sources_total - sinks_total
+    if delta > 0:
+        sinks["net_change_in_nw"] = delta
+    elif delta < 0:
+        sources["net_change_in_nw"] = -delta
+
+    return CashflowResponse(
+        scenario_id=scenario_id,
+        year=year,
+        sources=sources,
+        sinks=sinks,
+    )

fire_planner/api/income_streams.py

@@ -0,0 +1,98 @@
+"""Income-stream CRUD nested under a scenario.
+
+Streams are typed (salary / dividend / rental / pension / social_security
+/ rsu / other) so the simulator can route the after-tax cash through the
+jurisdiction tax engine using `tax_treatment` as the bucket.
+"""
+from __future__ import annotations
+
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy import delete, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from fire_planner.api.dependencies import get_session
+from fire_planner.api.schemas import (
+    IncomeStreamCreate,
+    IncomeStreamOut,
+    IncomeStreamPatch,
+)
+from fire_planner.db import IncomeStream, Scenario
+
+router = APIRouter(tags=["income-streams"])
+
+
+@router.get(
+    "/scenarios/{scenario_id}/income-streams",
+    response_model=list[IncomeStreamOut],
+)
+async def list_streams(
+    scenario_id: int,
+    session: AsyncSession = Depends(get_session),
+) -> list[IncomeStream]:
+    scen = await session.get(Scenario, scenario_id)
+    if scen is None:
+        raise HTTPException(status_code=404, detail="Scenario not found")
+    rows = (await session.execute(
+        select(IncomeStream).where(IncomeStream.scenario_id == scenario_id).order_by(
+            IncomeStream.start_year, IncomeStream.id))).scalars().all()
+    return list(rows)
+
+
+@router.post(
+    "/scenarios/{scenario_id}/income-streams",
+    response_model=IncomeStreamOut,
+    status_code=201,
+)
+async def create_stream(
+    scenario_id: int,
+    payload: IncomeStreamCreate,
+    session: AsyncSession = Depends(get_session),
+) -> IncomeStream:
+    scen = await session.get(Scenario, scenario_id)
+    if scen is None:
+        raise HTTPException(status_code=404, detail="Scenario not found")
+    if payload.end_year is not None and payload.end_year < payload.start_year:
+        raise HTTPException(status_code=400, detail="end_year < start_year")
+    stream = IncomeStream(scenario_id=scenario_id, **payload.model_dump())
+    session.add(stream)
+    await session.commit()
+    await session.refresh(stream)
+    return stream
+
+
+@router.patch(
+    "/income-streams/{stream_id}",
+    response_model=IncomeStreamOut,
+)
+async def patch_stream(
+    stream_id: int,
+    payload: IncomeStreamPatch,
+    session: AsyncSession = Depends(get_session),
+) -> IncomeStream:
+    stream = await session.get(IncomeStream, stream_id)
+    if stream is None:
+        raise HTTPException(status_code=404, detail="Income stream not found")
+    updates = payload.model_dump(exclude_unset=True)
+    for k, v in updates.items():
+        setattr(stream, k, v)
+    if stream.end_year is not None and stream.end_year < stream.start_year:
+        raise HTTPException(status_code=400, detail="end_year < start_year")
+    await session.commit()
+    await session.refresh(stream)
+    return stream
+
+
+@router.delete(
+    "/income-streams/{stream_id}",
+    status_code=204,
+    response_model=None,
+)
+async def delete_stream(
+    stream_id: int,
+    session: AsyncSession = Depends(get_session),
+) -> None:
+    stream = await session.get(IncomeStream, stream_id)
+    if stream is None:
+        raise HTTPException(status_code=404, detail="Income stream not found")
+    await session.execute(delete(IncomeStream).where(IncomeStream.id == stream_id))
+    await session.commit()

fire_planner/api/progress.py

@@ -0,0 +1,104 @@
+"""Progress overlay — actual NW from `account_snapshot` vs the persisted
+projection fan.
+
+Used by the left-sidebar "Progress" page. The alignment anchor is the
+earliest available snapshot date (year 0 of the projection); subsequent
+snapshots are bucketed into the projection's calendar-year axis so the
+overlay lines up.
+"""
+from __future__ import annotations
+
+from collections import defaultdict
+from datetime import date
+from decimal import Decimal
+
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from fire_planner.api.dependencies import get_session
+from fire_planner.api.schemas import (
+    ProgressActualPoint,
+    ProgressProjectedPoint,
+    ProgressResponse,
+    ProgressVariancePoint,
+)
+from fire_planner.db import AccountSnapshot, McRun, ProjectionYearly, Scenario
+
+router = APIRouter(prefix="/scenarios", tags=["progress"])
+
+
+@router.get("/{scenario_id}/progress", response_model=ProgressResponse)
+async def get_progress(
+    scenario_id: int,
+    session: AsyncSession = Depends(get_session),
+) -> ProgressResponse:
+    scen = await session.get(Scenario, scenario_id)
+    if scen is None:
+        raise HTTPException(status_code=404, detail="Scenario not found")
+
+    actuals_rows = (await session.execute(
+        select(
+            AccountSnapshot.snapshot_date,
+            AccountSnapshot.market_value_gbp,
+        ))).all()
+
+    by_date: dict[date, Decimal] = defaultdict(lambda: Decimal("0"))
+    for snap_date, value in actuals_rows:
+        by_date[snap_date] += Decimal(str(value))
+
+    actual = [
+        ProgressActualPoint(snapshot_date=d, total_gbp=by_date[d]) for d in sorted(by_date.keys())
+    ]
+    if actual:
+        anchor = actual[0].snapshot_date
+    else:
+        anchor = scen.created_at.date() if scen.created_at is not None else date.today()
+
+    run = (await session.execute(
+        select(McRun).where(McRun.scenario_id == scenario_id).order_by(
+            McRun.run_at.desc()).limit(1))).scalar_one_or_none()
+    projected: list[ProgressProjectedPoint] = []
+    yearly_rows: list[ProjectionYearly] = []
+    if run is not None:
+        rows = (await session.execute(
+            select(ProjectionYearly).where(ProjectionYearly.mc_run_id == run.id).order_by(
+                ProjectionYearly.year_idx))).scalars().all()
+        yearly_rows = list(rows)
+        projected = [
+            ProgressProjectedPoint(
+                year_idx=r.year_idx,
+                p10_portfolio_gbp=r.p10_portfolio_gbp,
+                p50_portfolio_gbp=r.p50_portfolio_gbp,
+                p90_portfolio_gbp=r.p90_portfolio_gbp,
+            ) for r in yearly_rows
+        ]
+
+    variance: list[ProgressVariancePoint] = []
+    if yearly_rows and actual:
+        actuals_by_year: dict[int, list[Decimal]] = defaultdict(list)
+        for pt in actual:
+            year_idx = (pt.snapshot_date.year - anchor.year)
+            if year_idx >= 0:
+                actuals_by_year[year_idx].append(pt.total_gbp)
+        for r in yearly_rows:
+            samples = actuals_by_year.get(r.year_idx)
+            if not samples:
+                continue
+            avg = sum(samples, Decimal("0")) / Decimal(len(samples))
+            projected_p50 = Decimal(str(r.p50_portfolio_gbp))
+            variance.append(
+                ProgressVariancePoint(
+                    year_idx=r.year_idx,
+                    actual_avg_gbp=avg,
+                    projected_p50_gbp=projected_p50,
+                    delta_gbp=avg - projected_p50,
+                ))
+
+    return ProgressResponse(
+        scenario_id=scenario_id,
+        alignment_anchor=anchor,
+        actual=actual,
+        projected=projected,
+        variance=variance,
+    )

fire_planner/api/scenarios.py

@@ -11,6 +11,7 @@ Mixed surface:
 from __future__ import annotations
 
 import uuid
+from decimal import Decimal
 
 from fastapi import APIRouter, Depends, HTTPException
 from sqlalchemy import delete, select
@@ -18,17 +19,109 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from fire_planner.api.dependencies import get_session
 from fire_planner.api.schemas import (
+    GoalProbability,
     ProjectionPoint,
     ScenarioCreate,
     ScenarioOut,
     ScenarioPatch,
     ScenarioProjection,
 )
-from fire_planner.db import McRun, ProjectionYearly, Scenario
+from fire_planner.db import McRun, ProjectionYearly, RetirementGoal, Scenario
 
 router = APIRouter(prefix="/scenarios", tags=["scenarios"])
 
 
+def _approx_prob_above(yearly_row: ProjectionYearly, target_amount: float) -> float:
+    """Approximate the fraction of paths whose portfolio at the row's
+    year is at or above ``target_amount``, using linear interpolation
+    across the persisted p10/p25/p50/p75/p90 cells.
+
+    Exact only when target_amount lands on a stored quantile, otherwise
+    a piecewise-linear estimate. Good enough for Wave 1 — the live
+    /simulate endpoint computes the exact path-by-path probability.
+    """
+    from itertools import pairwise
+
+    cells = [
+        (10, float(yearly_row.p10_portfolio_gbp)),
+        (25, float(yearly_row.p25_portfolio_gbp)),
+        (50, float(yearly_row.p50_portfolio_gbp)),
+        (75, float(yearly_row.p75_portfolio_gbp)),
+        (90, float(yearly_row.p90_portfolio_gbp)),
+    ]
+    cells.sort(key=lambda kv: kv[1])
+    if target_amount <= cells[0][1]:
+        return 1.0
+    if target_amount >= cells[-1][1]:
+        return 0.0
+    for (lo_pct, lo_v), (hi_pct, hi_v) in pairwise(cells):
+        if lo_v <= target_amount <= hi_v:
+            span = hi_v - lo_v
+            if span == 0:
+                return max(0.0, 1.0 - lo_pct / 100.0)
+            t = (target_amount - lo_v) / span
+            pct = lo_pct + t * (hi_pct - lo_pct)
+            return max(0.0, min(1.0, 1.0 - pct / 100.0))
+    return 0.0
+
+
+def _evaluate_goal_against_fan(
+    goal: RetirementGoal,
+    yearly_rows: list[ProjectionYearly],
+    horizon_years: int,
+) -> GoalProbability | None:
+    if not goal.enabled:
+        return None
+    threshold = float(goal.success_threshold)
+
+    def _result(prob: float) -> GoalProbability:
+        return GoalProbability(
+            goal_id=goal.id,
+            name=goal.name,
+            kind=goal.kind,
+            probability=Decimal(str(round(prob, 4))),
+            threshold=Decimal(str(round(threshold, 4))),
+            passed=prob >= threshold,
+        )
+
+    by_year = {row.year_idx: row for row in yearly_rows}
+    if not by_year:
+        return _result(0.0)
+    last_year = max(by_year)
+
+    if goal.kind == "target_nw_by_year":
+        if goal.target_year is None or goal.target_amount_gbp is None:
+            return _result(0.0)
+        y = max(0, min(int(goal.target_year), last_year))
+        row = by_year.get(y)
+        if row is None:
+            return _result(0.0)
+        prob = _approx_prob_above(row, float(goal.target_amount_gbp))
+        return _result(prob)
+
+    if goal.kind == "never_run_out":
+        end = int(goal.target_year) if goal.target_year is not None else horizon_years
+        end = max(0, min(end, last_year))
+        row = by_year.get(end)
+        if row is None:
+            return _result(0.0)
+        return _result(float(row.survival_rate))
+
+    if goal.kind == "target_real_income":
+        if goal.target_amount_gbp is None:
+            return _result(0.0)
+        target = float(goal.target_amount_gbp)
+        start_y = int(goal.target_year) if goal.target_year is not None else 0
+        window = [r for r in yearly_rows if r.year_idx >= start_y]
+        if not window:
+            return _result(0.0)
+        median_wd = sorted(float(r.p50_withdrawal_gbp) for r in window)
+        mid = median_wd[len(median_wd) // 2]
+        return _result(1.0 if mid >= target else 0.0)
+
+    return _result(0.0)
+
+
 @router.get("", response_model=list[ScenarioOut])
 async def list_scenarios(
     kind: str | None = None,
@@ -149,9 +242,17 @@ async def get_scenario_projection(
             McRun.run_at.desc()).limit(1))).scalar_one_or_none()
     if run is None:
         raise HTTPException(status_code=404, detail="No MC runs persisted for this scenario yet")
-    yearly_rows = (await session.execute(
+    yearly_rows = list((await session.execute(
         select(ProjectionYearly).where(ProjectionYearly.mc_run_id == run.id).order_by(
-            ProjectionYearly.year_idx))).scalars().all()
+            ProjectionYearly.year_idx))).scalars().all())
+    goals_rows = list((await session.execute(
+        select(RetirementGoal).where(
+            RetirementGoal.scenario_id == scenario_id))).scalars().all())
+    goals_probability: list[GoalProbability] = []
+    for goal in goals_rows:
+        evaluation = _evaluate_goal_against_fan(goal, yearly_rows, scen.horizon_years)
+        if evaluation is not None:
+            goals_probability.append(evaluation)
     return ScenarioProjection(
         scenario_id=scen.id,
         external_id=scen.external_id,
@@ -165,4 +266,5 @@ async def get_scenario_projection(
         median_lifetime_tax_gbp=run.median_lifetime_tax_gbp,
         median_years_to_ruin=run.median_years_to_ruin,
         yearly=[ProjectionPoint.model_validate(y) for y in yearly_rows],
+        goals_probability=goals_probability,
     )

fire_planner/api/schemas.py

@@ -85,6 +85,15 @@ class ProjectionPoint(_Base):
     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
@@ -99,6 +108,7 @@ class ScenarioProjection(BaseModel):
     median_lifetime_tax_gbp: Decimal
     median_years_to_ruin: Decimal | None
     yearly: list[ProjectionPoint]
+    goals_probability: list[GoalProbability] = Field(default_factory=list)
 
 
 # ── net worth ────────────────────────────────────────────────────────
@@ -239,6 +249,134 @@ class GoalCreate(BaseModel):
     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 ───────────────────────────────────────────────
 
 
@@ -282,6 +420,21 @@ class SimulateRequest(BaseModel):
     #                   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)
+    # 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
@@ -303,6 +456,7 @@ class SimulateResult(BaseModel):
     median_years_to_ruin: Decimal | None
     elapsed_seconds: Decimal
     yearly: list[ProjectionPoint]
+    goals_probability: list[GoalProbability] = Field(default_factory=list)
 
 
 class CompareRequest(BaseModel):

fire_planner/api/simulate.py

@@ -21,11 +21,14 @@ from sqlalchemy.ext.asyncio import async_sessionmaker
 from fire_planner.api.schemas import (
     CompareRequest,
     CompareResult,
+    GoalProbability,
     ProjectionPoint,
     SimulateRequest,
     SimulateResult,
 )
 from fire_planner.glide_path import static
+from fire_planner.goals_eval import evaluate_goals
+from fire_planner.income_streams import IncomeStreamInput, streams_to_arrays
 from fire_planner.ingest.wealthfolio_pg import create_wf_sync_engine_from_env
 from fire_planner.life_events import EventInput, events_to_cashflow_array
 from fire_planner.returns.bootstrap import block_bootstrap
@@ -35,7 +38,7 @@ from fire_planner.returns.wealthfolio_returns import (
     constant_real_return_paths,
 )
 from fire_planner.scenarios import build_regime_schedule, build_strategy
-from fire_planner.simulator import SimulationResult, simulate
+from fire_planner.simulator import SimulationResult, build_fixed_paths, simulate
 
 router = APIRouter(tags=["simulate"])
 
@@ -64,6 +67,16 @@ async def _wealthfolio_paths(seed: int, n_paths: int, n_years: int) -> np.ndarra
 
 
 async def _build_paths(req: SimulateRequest) -> np.ndarray:
+    if req.rates_mode == "fixed":
+        return build_fixed_paths(
+            n_paths=req.n_paths,
+            n_years=req.horizon_years,
+            inflation_pct=float(req.inflation_pct),
+            stocks_growth_pct=float(req.stocks_growth_pct),
+            stocks_dividend_pct=float(req.stocks_dividend_pct),
+            bonds_growth_pct=float(req.bonds_growth_pct),
+            bonds_dividend_pct=float(req.bonds_dividend_pct),
+        )
     if req.returns_mode == "manual":
         if req.manual_real_return_pct is None:
             raise HTTPException(
@@ -105,6 +118,22 @@ def _project(req: SimulateRequest, paths: np.ndarray) -> tuple[SimulationResult,
         ]
         cashflow_adjustments = events_to_cashflow_array(engine_events, req.horizon_years)
 
+    income_inflows = None
+    income_taxable = None
+    if req.income_streams:
+        engine_streams = [
+            IncomeStreamInput(
+                kind=s.kind,
+                start_year=s.start_year,
+                end_year=s.end_year,
+                amount_gbp_per_year=float(s.amount_gbp_per_year),
+                growth_pct=float(s.growth_pct),
+                tax_treatment=s.tax_treatment,
+                enabled=s.enabled,
+            ) for s in req.income_streams
+        ]
+        income_inflows, income_taxable = streams_to_arrays(engine_streams, req.horizon_years)
+
     strategy = build_strategy(
         req.strategy,
         floor=floor,
@@ -114,23 +143,31 @@ def _project(req: SimulateRequest, paths: np.ndarray) -> tuple[SimulationResult,
         guardrail_cut_pct=float(req.guardrail_cut_pct),
     )
 
+    glide_alloc = float(req.stocks_allocation) if req.rates_mode == "fixed" else 1.0
+
     started = time.perf_counter()
     result = simulate(
         paths=paths,
         initial_portfolio=float(req.nw_seed_gbp),
         spending_target=float(req.spending_gbp),
-        glide=static(1.0),
+        glide=static(glide_alloc),
         strategy=strategy,
         regime=build_regime_schedule(req.jurisdiction, req.leave_uk_year),
         horizon_years=req.horizon_years,
         annual_savings=annual_savings,
         cashflow_adjustments=cashflow_adjustments,
+        income_inflows=income_inflows,
+        income_taxable=income_taxable,
     )
     elapsed = time.perf_counter() - started
     return result, elapsed
 
 
-def _to_response(result: SimulationResult, elapsed: float) -> SimulateResult:
+def _to_response(
+    result: SimulationResult,
+    elapsed: float,
+    req: SimulateRequest | None = None,
+) -> SimulateResult:
     # portfolio_real has n_years+1 columns (year 0 = seed, year k = end-of-year k).
     # withdrawal_real / tax_real have n_years columns (year k = withdrawn in year k+1).
     # Yearly point k describes "end of year k+1": portfolio after withdrawal & growth.
@@ -156,6 +193,19 @@ def _to_response(result: SimulationResult, elapsed: float) -> SimulateResult:
         ) for y in range(n_years)
     ]
     median_ytr = result.median_years_to_ruin()
+    goals_probability: list[GoalProbability] = []
+    if req is not None and req.goals:
+        evaluations = evaluate_goals(result, req.goals, req.horizon_years)
+        goals_probability = [
+            GoalProbability(
+                goal_id=None,
+                name=ev.name,
+                kind=ev.kind,
+                probability=Decimal(str(round(ev.probability, 4))),
+                threshold=Decimal(str(round(ev.threshold, 4))),
+                passed=ev.passed,
+            ) for ev in evaluations
+        ]
     return SimulateResult(
         success_rate=Decimal(str(round(float(result.success_rate), 4))),
         p10_ending_gbp=Decimal(str(round(float(result.ending_percentile(10)), 2))),
@@ -166,6 +216,7 @@ def _to_response(result: SimulationResult, elapsed: float) -> SimulateResult:
                               if median_ytr is not None else None),
         elapsed_seconds=Decimal(str(round(elapsed, 3))),
         yearly=yearly,
+        goals_probability=goals_probability,
     )
 
 
@@ -177,7 +228,7 @@ async def simulate_one(req: SimulateRequest) -> SimulateResult:
         result, elapsed = await asyncio.to_thread(_project, req, paths)
     except KeyError as e:
         raise HTTPException(status_code=400, detail=f"Unknown name: {e}") from None
-    return _to_response(result, elapsed)
+    return _to_response(result, elapsed, req)
 
 
 @router.post("/compare", response_model=CompareResult)
@@ -186,7 +237,7 @@ async def compare_scenarios(req: CompareRequest) -> CompareResult:
     async def one(s: SimulateRequest) -> SimulateResult:
         paths = await _build_paths(s)
         result, elapsed = await asyncio.to_thread(_project, s, paths)
-        return _to_response(result, elapsed)
+        return _to_response(result, elapsed, s)
 
     try:
         results = await asyncio.gather(*(one(s) for s in req.scenarios))

fire_planner/api/year_stats.py

@@ -0,0 +1,106 @@
+"""Per-year stats for the scrubbable right-hand sidebar (Plan tab).
+
+All cells derive from the latest persisted MC run plus scenario config.
+Future-Wave fields (liquid_nw, expenses, savings_rate, portfolio
+allocations) are returned as null until the relevant feature lands.
+"""
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from decimal import Decimal
+
+from fastapi import APIRouter, Depends, HTTPException, Query
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from fire_planner.api.dependencies import get_session
+from fire_planner.api.schemas import YearStats
+from fire_planner.db import IncomeStream, McRun, ProjectionYearly, Scenario
+
+router = APIRouter(prefix="/scenarios", tags=["year-stats"])
+
+
+def _income_at(streams: list[IncomeStream], year_idx: int) -> tuple[Decimal, Decimal]:
+    """Return ``(total_inflow, taxable_portion)`` at ``year_idx``."""
+    inflow = Decimal("0")
+    taxable = Decimal("0")
+    for s in streams:
+        if not s.enabled:
+            continue
+        if year_idx < s.start_year:
+            continue
+        if s.end_year is not None and year_idx > s.end_year:
+            continue
+        years_active = year_idx - s.start_year
+        growth = (Decimal("1") + Decimal(str(s.growth_pct)))**years_active
+        amount = Decimal(str(s.amount_gbp_per_year)) * growth
+        inflow += amount
+        if s.tax_treatment == "income":
+            taxable += amount
+    return inflow, taxable
+
+
+@router.get("/{scenario_id}/year-stats", response_model=YearStats)
+async def get_year_stats(
+    scenario_id: int,
+    year: int = Query(ge=0, le=100, description="Year offset from scenario start (0 = now)"),
+    session: AsyncSession = Depends(get_session),
+) -> YearStats:
+    scen = await session.get(Scenario, scenario_id)
+    if scen is None:
+        raise HTTPException(status_code=404, detail="Scenario not found")
+    run = (await session.execute(
+        select(McRun).where(McRun.scenario_id == scenario_id).order_by(
+            McRun.run_at.desc()).limit(1))).scalar_one_or_none()
+    if run is None:
+        raise HTTPException(status_code=404, detail="No MC runs persisted for this scenario yet")
+
+    yearly_rows = (await session.execute(
+        select(ProjectionYearly).where(ProjectionYearly.mc_run_id == run.id).order_by(
+            ProjectionYearly.year_idx))).scalars().all()
+    by_year = {r.year_idx: r for r in yearly_rows}
+    if year not in by_year:
+        raise HTTPException(status_code=404, detail=f"No projection row for year {year}")
+
+    streams = list((await session.execute(
+        select(IncomeStream).where(
+            IncomeStream.scenario_id == scenario_id))).scalars().all())
+
+    row = by_year[year]
+    prev = by_year.get(year - 1) if year > 0 else None
+
+    nw = Decimal(str(row.p50_portfolio_gbp))
+    prev_nw = (Decimal(str(prev.p50_portfolio_gbp)) if prev is not None else Decimal(str(
+        scen.nw_seed_gbp)))
+    change = nw - prev_nw
+
+    income_inflow, taxable = _income_at(streams, year)
+    contributions_from_savings = Decimal(str(scen.savings_per_year_gbp))
+    contributions = contributions_from_savings + income_inflow
+
+    spending = Decimal(str(row.p50_withdrawal_gbp))
+    taxes = Decimal(str(row.p50_tax_gbp))
+    eff_tax_rate = (taxes / taxable if taxable > 0 else
+                    (taxes / spending if spending > 0 else Decimal("0")))
+
+    investment_growth = change - contributions + spending + taxes
+
+    payload = scen.config_json or {}
+    birth_year = payload.get("birth_year") if isinstance(payload, dict) else None
+    base_year = datetime.now(UTC).year
+    calendar_year = base_year + year
+    age = (calendar_year - int(birth_year)) if isinstance(birth_year, int) else None
+
+    return YearStats(
+        year_idx=year,
+        calendar_year=calendar_year,
+        age=age,
+        net_worth_p50=nw,
+        change_in_nw=change,
+        taxable_income=taxable,
+        taxes=taxes,
+        effective_tax_rate=eff_tax_rate.quantize(Decimal("0.0001")),
+        spending=spending,
+        contributions=contributions,
+        investment_growth=investment_growth,
+    )