59 lines
1.9 KiB
Python
59 lines
1.9 KiB
Python
|
"""Convert life-event records into per-year cashflow adjustments.
|
||
|
|
|
||
|
|
Two event shapes the engine understands:
|
||
|
|
|
||
|
|
- **Ranged delta**: `delta_gbp_per_year != 0` applied each year in
|
||
|
|
`[year_start, year_end]` (inclusive). Use a negative delta for
|
||
|
|
expenses (childcare, sabbatical), positive for income (rental,
|
||
|
|
pension that hasn't started yet).
|
||
|
|
|
||
|
|
- **One-time amount**: `one_time_amount_gbp` applied once at
|
||
|
|
`year_start`. Inheritance, house sale proceeds, lump-sum gift.
|
||
|
|
|
||
|
|
Disabled events (`enabled=False`) are skipped. Year ranges that
|
||
|
|
extend past the simulation horizon are clipped — events beyond year
|
||
|
|
H simply don't happen in this run.
|
||
|
|
"""
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from collections.abc import Iterable
|
||
|
|
from dataclasses import dataclass
|
||
|
|
|
||
|
|
import numpy as np
|
||
|
|
import numpy.typing as npt
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass(frozen=True)
|
||
|
|
class EventInput:
|
||
|
|
"""Engine-level event shape — decoupled from the SQLAlchemy ORM and
|
||
|
|
the API Pydantic schema so callers can construct them however."""
|
||
|
|
year_start: int
|
||
|
|
year_end: int | None = None
|
||
|
|
delta_gbp_per_year: float = 0.0
|
||
|
|
one_time_amount_gbp: float | None = None
|
||
|
|
enabled: bool = True
|
||
|
|
|
||
|
|
|
||
|
|
def events_to_cashflow_array(
|
||
|
|
events: Iterable[EventInput],
|
||
|
|
horizon_years: int,
|
||
|
|
) -> npt.NDArray[np.float64]:
|
||
|
|
"""Sum a list of events into a `(horizon_years,)` real-GBP array."""
|
||
|
|
out = np.zeros(horizon_years, dtype=np.float64)
|
||
|
|
for ev in events:
|
||
|
|
if not ev.enabled:
|
||
|
|
continue
|
||
|
|
start = max(0, int(ev.year_start))
|
||
|
|
if start >= horizon_years:
|
||
|
|
continue
|
||
|
|
|
||
|
|
if ev.delta_gbp_per_year:
|
||
|
|
end = ev.year_end if ev.year_end is not None else ev.year_start
|
||
|
|
end = min(int(end), horizon_years - 1)
|
||
|
|
if end >= start:
|
||
|
|
out[start:end + 1] += float(ev.delta_gbp_per_year)
|
||
|
|
|
||
|
|
if ev.one_time_amount_gbp:
|
||
|
|
out[start] += float(ev.one_time_amount_gbp)
|
||
|
|
return out
|