From 40431fa8be1ee6a8ec07584d7932ea479d57fcad Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Wed, 9 Sep 2026 17:47:42 -0400 Subject: [PATCH 1/2] Add opt-in annual population stock-flow accounting --- docs/stock-flow-accounting.md | 157 ++ scripts/first_estimates_birth_evidence.py | 3 + src/populace_dynamics/engine/accounting.py | 1167 +++++++++++++ tests/README-tiers.md | 6 +- .../estimates/test_birth_evidence_artifact.py | 69 +- tests/test_m6_stock_flow.py | 1537 +++++++++++++++++ tests/tier_counts.json | 4 +- 7 files changed, 2928 insertions(+), 15 deletions(-) create mode 100644 docs/stock-flow-accounting.md create mode 100644 src/populace_dynamics/engine/accounting.py create mode 100644 tests/test_m6_stock_flow.py diff --git a/docs/stock-flow-accounting.md b/docs/stock-flow-accounting.md new file mode 100644 index 00000000..37723b9c --- /dev/null +++ b/docs/stock-flow-accounting.md @@ -0,0 +1,157 @@ +# Annual population stock-flow accounting + +`populace_dynamics.engine.accounting.reconcile_period` checks whether one +annual transition's opening population, declared arrivals and departures, +and closing population reconcile. It reports person counts, weight flows, +weight changes, and arithmetic residuals without altering either frame. + +This interface is experimental and opt-in. It does not change the historical +projection engine, generate demographic events, or establish an admitted +population. Its status is `engineering-accounting-coherence-only`, and its +interface version is `stock-flow-accounting/0.1.0-experimental`. No scientific +tolerance or acceptance gate is added. + +## Input contract + +```python +reconcile_period( + opening, + closing, + *, + opening_year, + closing_year, + additions=(), + exits=(), +) +``` + +Both pandas frames require `person_id`, `year`, and `weight` columns. +Identifiers are unique within each frame; IDs and years must be integers +within the signed int64 range. Floats and booleans are rejected as IDs or +years. These are the accountant's own validation rules, which are stricter +than the historical loop's slice check. + +Weights must be real numeric, finite and nonnegative. Strings, complex +values, and booleans are rejected before conversion to binary64. Negative +weights are checked before conversion; nonzero values that underflow to zero +in binary64 are rejected. Zero-weight +rows remain visible in counts. Empty frames still require the three columns, +but empty columns may have any dtype. Each nonempty frame must carry its +stated year, and `closing_year` must equal `opening_year + 1`. + +`additions` and `exits` are sequences of `PopulationEvent` declarations. +Each event has an integer `person_id`, `kind`, closing `year`, optional +`weight`, and optional string `reason` and `source`. + +| Addition kinds | Exit kinds | +|---|---| +| `birth` | `death` | +| `scheduled_entry` | `emigration` | +| `other_entry` | `other_exit` | + +The two `other_*` kinds require a nonempty reason. At most one addition and +one exit are supported for a person in a period, with arrival before exit. +There is no finer event-timing model. An addition cannot collide with the +opening roster, and a declared departure cannot remain in the closing roster. +Unexplained changes in endpoint membership and duplicate declarations fail. +The accountant never infers that a disappearing person died or that a new +identifier represents an immigrant. + +## Counts and weights + +Counts reconcile exactly: + +```text +closing = opening + additions_total - exits_total +``` + +People who arrive and depart in the same period are counted in both flows. +These transients appear in neither endpoint frame and must have explicit +weights on both declarations. Otherwise, an omitted arrival weight uses the +closing-frame weight; an omitted departure weight uses the opening-frame +weight. These conventions are recorded in provenance through the counts of +explicit event weights. + +Weight accounting reports: + +```text +reconstructed_closing = opening + additions_total - exits_total + revaluation +weight_residual = closing - reconstructed_closing +``` + +| Revaluation component | Difference summed over the relevant people | +|---|---| +| `carried` | Closing weight minus opening weight for survivors | +| `entrant` | Closing weight minus declared arrival weight | +| `exiting` | Declared departure weight minus opening weight | +| `transient` | Declared departure weight minus declared arrival weight | + +Weights are never rebalanced. Sums use `math.fsum` over binary64 components; +subtraction and component totals still involve rounding. Nonzero arithmetic +residuals are reported without a tolerance-based verdict. An intermediate or +summary that cannot be represented with finite binary64 arithmetic raises a +`PopulationAccountingInputError`. + +## Example + +```python +import pandas as pd +from populace_dynamics.engine.accounting import PopulationEvent, reconcile_period + +opening = pd.DataFrame({ + "person_id": [1, 2, 3], "year": [2020] * 3, + "weight": [10.0, 20.0, 30.0], +}) +closing = pd.DataFrame({ + "person_id": [1, 2, 90], "year": [2021] * 3, + "weight": [11.0, 20.0, 5.0], +}) +account = reconcile_period( + opening, closing, opening_year=2020, closing_year=2021, + additions=[PopulationEvent(90, "birth", 2021, source="synthetic.birth")], + exits=[PopulationEvent(3, "death", 2021, source="synthetic.mortality")], +) +assert account.weights.closing == 36.0 +assert account.weights.revaluation.carried == 1.0 +assert account.count_residual == 0 +assert account.weight_residual == 0.0 +``` + +The weight identity is `60 + 5 - 30 + 1 = 36`. `account.to_dict()` returns +JSON-serializable counts, weights, residuals, and provenance. Person-ID tuples +are available as attributes and are omitted from this summary. Serialized +provenance is isolated from the immutable account and other serializations. + +Malformed inputs raise `PopulationAccountingInputError`. Well-formed inputs +whose declarations conflict with the frames raise +`PopulationReconciliationError`, with typed `.discrepancies` and a `.to_dict()` +representation. Validation stops at the first failing stage: frames, +declarations, reconciliation, then weights. + +## Projection-loop integration and limits + +The caller must capture declarations from its adapters or supplied schedule. +The loop activates scheduled entries before mortality in the wave ending in +`Y`, while entry frames carry `Y - 1`. Their declarations must use `Y` and +explicit weights so an entrant who dies in the same wave can be accounted for. +Birth and mortality declarations should come from the operations that perform +those transitions. An endpoint difference alone cannot establish their cause. + +Tests in `tests/test_m6_stock_flow.py` drive the real `ProjectionEngine` with +synthetic recording adapters, then reconcile adjacent output slices. They +include a birth, a death, and a scheduled entrant who dies before the wave +closes. No fitted transition law or native population is used. + +Accounting coherence does not verify event-log completeness. If both records +for a transient are omitted, the endpoint frames cannot reveal the omission; +provenance records `event_log_completeness_verified=False`. Event reasons are +also caller assertions. The accountant has no cross-period memory, so a past +ID explicitly declared as a new addition can be reused without detection. +Only `weight` is reconciled; `start_weight` and other frame columns are ignored. + +The module directly imports only the standard library, NumPy, and pandas. +Normal package import also executes the historical engine initializer and its +broader source dependencies. The historical source-identity test verifies that +this new module remains unreachable from the sealed projection roots; the +static guard covers ordinary imports and explicitly listed dynamic roots, not +arbitrary runtime imports. This page is not added to the Quarto chapter list. diff --git a/scripts/first_estimates_birth_evidence.py b/scripts/first_estimates_birth_evidence.py index 81d5bd41..fec55d44 100644 --- a/scripts/first_estimates_birth_evidence.py +++ b/scripts/first_estimates_birth_evidence.py @@ -166,6 +166,9 @@ Path("src/populace_dynamics/graph/runtime.py"), Path("src/populace_dynamics/graph/synthetic.py"), Path("src/populace_dynamics/graph/trajectory.py"), + # This opt-in accountant is unreachable from the historical projection. + # The existing engine loop, steps, and package initializer remain sealed. + Path("src/populace_dynamics/engine/accounting.py"), ) POST_REVIEW_SHARED_SOURCE_BLOBS = { Path( diff --git a/src/populace_dynamics/engine/accounting.py b/src/populace_dynamics/engine/accounting.py new file mode 100644 index 00000000..1920836d --- /dev/null +++ b/src/populace_dynamics/engine/accounting.py @@ -0,0 +1,1167 @@ +"""Experimental accounting for one annual population transition. + +``reconcile_period`` checks opening and closing person sets against the +caller's declared arrivals and departures. It does not generate transitions, +infer their causes, fit a model, or read data. Count conservation is exact; +weight stocks, flows, and separate revaluations use binary64 arithmetic with +``math.fsum`` and a reported residual. An unrepresentable summary is refused. +No scientific tolerance or acceptance gate is introduced. + +Frames require unique signed-int64 person IDs, matching integer years, and +finite nonnegative real weights. These are this interface's validation rules; +the historical projection loop has a less restrictive input check. Zero-weight +rows remain people. Every declaration uses the closing year, even when a +scheduled-entry frame carries the loop's required previous-year stamp. + +The supported lifecycle is at most one arrival followed by at most one +departure in the period. Event ordering within the year is not observed. +Declared transients appear in neither endpoint frame and need explicit +weights for both events. Omitting BOTH events is unobservable from endpoints: +accounting coherence does not establish event-log completeness or true causes. +There is no cross-period history, so declared reuse of a past ID is not caught. + +The module has only stdlib, NumPy, and pandas direct imports. A normal package +import also executes the existing engine package initializer and its broader +source dependencies. This optional module is not called by the historical +engine. See ``docs/stock-flow-accounting.md`` for the interface and limits. +""" + +from __future__ import annotations + +import math +from collections.abc import Iterable, Mapping, Sequence +from dataclasses import dataclass +from enum import Enum +from types import MappingProxyType + +import numpy as np +import pandas as pd + +__all__ = [ + "ACCOUNTING_INTERFACE_VERSION", + "ADDITION_KINDS", + "ENGINEERING_STATUS", + "ENGINEERING_STATUS_NOTE", + "EXIT_KINDS", + "PERSON_ID_COLUMN", + "REQUIRED_COLUMNS", + "WEIGHT_COLUMN", + "YEAR_COLUMN", + "AccountingDiscrepancy", + "DiscrepancyKind", + "PeriodAccount", + "PersonCounts", + "PopulationAccountingError", + "PopulationAccountingInputError", + "PopulationEvent", + "PopulationEventKind", + "PopulationReconciliationError", + "WeightRevaluation", + "WeightTotals", + "reconcile_period", +] + +ACCOUNTING_INTERFACE_VERSION = "stock-flow-accounting/0.1.0-experimental" +ENGINEERING_STATUS = "engineering-accounting-coherence-only" +ENGINEERING_STATUS_NOTE = ( + "Person sets reconcile exactly and the weight identity closes to the " + "reported arithmetic residual. This is engineering coherence only: it " + "is not scientific acceptance, not a benchmark comparison, not a gate " + "outcome, and not evidence that the population is admitted." +) +SUMMATION_METHOD = "math.fsum over binary64 components; residual reported" +INFERENCE_POLICY = ( + "none: every addition and every exit must be declared by the caller" +) + +PERSON_ID_COLUMN = "person_id" +YEAR_COLUMN = "year" +WEIGHT_COLUMN = "weight" +REQUIRED_COLUMNS = (PERSON_ID_COLUMN, YEAR_COLUMN, WEIGHT_COLUMN) +_INT64_MIN = -(2**63) +_INT64_MAX = 2**63 - 1 + + +class PopulationEventKind(str, Enum): + """The declared reason a person joins or leaves within one period.""" + + BIRTH = "birth" + SCHEDULED_ENTRY = "scheduled_entry" + OTHER_ENTRY = "other_entry" + DEATH = "death" + EMIGRATION = "emigration" + OTHER_EXIT = "other_exit" + + +ADDITION_KINDS = frozenset( + { + PopulationEventKind.BIRTH, + PopulationEventKind.SCHEDULED_ENTRY, + PopulationEventKind.OTHER_ENTRY, + } +) +EXIT_KINDS = frozenset( + { + PopulationEventKind.DEATH, + PopulationEventKind.EMIGRATION, + PopulationEventKind.OTHER_EXIT, + } +) +#: Kinds whose whole purpose is "something else happened", and which are +#: therefore only meaningful if the caller says what. +REASON_REQUIRED_KINDS = frozenset( + {PopulationEventKind.OTHER_ENTRY, PopulationEventKind.OTHER_EXIT} +) + + +class DiscrepancyKind(str, Enum): + """Ways a declared story can fail to match the two frames.""" + + UNDECLARED_ADDITION = "undeclared_addition" + UNDECLARED_EXIT = "undeclared_exit" + ADDITION_COLLIDES_WITH_OPENING = "addition_collides_with_opening" + DUPLICATE_ADDITION = "duplicate_addition" + DUPLICATE_EXIT = "duplicate_exit" + EXIT_WITHOUT_PRESENCE = "exit_without_presence" + EXIT_CONTRADICTED_BY_CLOSING = "exit_contradicted_by_closing" + ADDITION_ABSENT_AT_CLOSE = "addition_absent_at_close" + COUNT_IDENTITY_VIOLATION = "count_identity_violation" + + +_DISCREPANCY_ORDER = { + kind: index for index, kind in enumerate(DiscrepancyKind) +} + + +class PopulationAccountingError(ValueError): + """Base class for every refusal raised by this module.""" + + +class PopulationAccountingInputError(PopulationAccountingError): + """A frame or a declaration is malformed on its own terms.""" + + +class PopulationReconciliationError(PopulationAccountingError): + """Well-formed inputs whose declared story does not reconcile. + + The typed findings stay on the exception so that a caller can + distinguish an unexplained arrival from an omitted exit without + parsing a message. + """ + + def __init__( + self, + message: str, + discrepancies: Sequence[AccountingDiscrepancy], + ) -> None: + self.discrepancies: tuple[AccountingDiscrepancy, ...] = tuple( + discrepancies + ) + super().__init__(message) + + def to_dict(self) -> dict[str, object]: + """Return a JSON-safe view of the refusal.""" + return { + "interface_version": ACCOUNTING_INTERFACE_VERSION, + "error": "population_reconciliation_error", + "message": str(self), + "discrepancies": [item.to_dict() for item in self.discrepancies], + } + + +def _as_person_id(value: object, label: str) -> int: + """Coerce one identifier to ``int``, rejecting bools and floats.""" + if isinstance(value, bool) or not isinstance(value, (int, np.integer)): + raise PopulationAccountingInputError( + f"{label} must be an integer person identifier, " + f"got {value!r} of type {type(value).__name__}" + ) + return _in_integer_domain(int(value), label) + + +def _as_year(value: object, label: str) -> int: + """Coerce one calendar year to ``int``, rejecting bools and floats.""" + if isinstance(value, bool) or not isinstance(value, (int, np.integer)): + raise PopulationAccountingInputError( + f"{label} must be an integer year, got {value!r} of type " + f"{type(value).__name__}" + ) + return _in_integer_domain(int(value), label) + + +def _in_integer_domain(value: int, label: str) -> int: + """Enforce the same signed int64 domain before any array cast.""" + if not _INT64_MIN <= value <= _INT64_MAX: + raise PopulationAccountingInputError( + f"{label} must fit in signed int64, got {value!r}" + ) + return value + + +def _stable_sum(values: Iterable[float]) -> float: + """Sum binary64 components and refuse an unrepresentable summary.""" + try: + result = math.fsum(float(value) for value in values) + except (OverflowError, ValueError) as error: + raise PopulationAccountingInputError( + "weight summary is not representable with finite binary64 " + "arithmetic" + ) from error + if not math.isfinite(result): + raise PopulationAccountingInputError("weight summary is not finite") + return result + + +def _as_weight(value: object, label: str) -> float: + """Validate a real numeric weight before converting it to binary64.""" + if isinstance(value, (bool, np.bool_)) or not isinstance( + value, (int, float, np.integer, np.floating) + ): + raise PopulationAccountingInputError( + f"{label} must be a real number, got {value!r}" + ) + if value < 0: + raise PopulationAccountingInputError( + f"{label} must be finite and non-negative, got {value!r}" + ) + try: + weight = float(value) + except (OverflowError, ValueError) as error: + raise PopulationAccountingInputError( + f"{label} is not representable as a finite binary64 weight" + ) from error + if not math.isfinite(weight) or weight < 0.0: + raise PopulationAccountingInputError( + f"{label} must be finite and non-negative, got {value!r}" + ) + if weight == 0.0 and value != 0: + raise PopulationAccountingInputError( + f"{label} underflows to zero in binary64 arithmetic" + ) + return weight + + +@dataclass(frozen=True) +class PopulationEvent: + """One declared arrival or departure inside one annual period. + + Parameters + ---------- + person_id: + The identifier that joins or leaves. Must match the identifier + used in the frames. + kind: + A member of :class:`PopulationEventKind`. Plain strings are + accepted and coerced. + year: + The period's *closing* year, which labels the period. + weight: + Optional explicit weight at the moment of the event. When + omitted, an addition inherits the weight of its closing-frame + row and an exit inherits the weight of its opening-frame row. + A person who both arrives and departs within the period appears + in neither frame and must therefore declare both weights. + reason: + Free text. Required, and required to be non-empty, for + ``other_entry`` and ``other_exit`` so that "something else" + is never silent. + source: + Free-text provenance, e.g. the adapter that emitted the record. + Collected into the account's provenance block. + """ + + person_id: int + kind: PopulationEventKind + year: int + weight: float | None = None + reason: str = "" + source: str = "" + + def __post_init__(self) -> None: + object.__setattr__( + self, + "person_id", + _as_person_id(self.person_id, "PopulationEvent.person_id"), + ) + try: + kind = PopulationEventKind(self.kind) + except ValueError as error: + raise PopulationAccountingInputError( + f"unknown population event kind {self.kind!r}; expected one " + f"of {sorted(item.value for item in PopulationEventKind)}" + ) from error + object.__setattr__(self, "kind", kind) + object.__setattr__( + self, "year", _as_year(self.year, "PopulationEvent.year") + ) + if self.weight is not None: + object.__setattr__( + self, + "weight", + _as_weight(self.weight, "PopulationEvent.weight"), + ) + for field_name in ("reason", "source"): + value = getattr(self, field_name) + if not isinstance(value, str): + raise PopulationAccountingInputError( + f"PopulationEvent.{field_name} must be a string, " + f"got {value!r}" + ) + if kind in REASON_REQUIRED_KINDS and not self.reason.strip(): + raise PopulationAccountingInputError( + f"a {kind.value!r} event must state a non-empty reason; " + "this module never books an unexplained change" + ) + + @property + def is_addition(self) -> bool: + """Whether this kind adds a person to the roster.""" + return self.kind in ADDITION_KINDS + + def to_dict(self) -> dict[str, object]: + """Return a JSON-safe view of the declaration.""" + return { + "person_id": self.person_id, + "kind": self.kind.value, + "year": self.year, + "weight": self.weight, + "reason": self.reason, + "source": self.source, + } + + +@dataclass(frozen=True) +class AccountingDiscrepancy: + """One typed reason the declared story does not reconcile.""" + + kind: DiscrepancyKind + person_id: int | None + detail: str + + def to_dict(self) -> dict[str, object]: + """Return a JSON-safe view of the finding.""" + return { + "kind": self.kind.value, + "person_id": self.person_id, + "detail": self.detail, + } + + +@dataclass(frozen=True) +class PersonCounts: + """Exact integer person counts for one period. + + ``closing == opening + additions_total - exits_total`` holds + exactly, including transients, which are counted in both + ``additions_total`` and ``exits_total`` and cancel. + """ + + opening: int + closing: int + carried: int + entered: int + exited: int + transient: int + additions_total: int + exits_total: int + additions_by_kind: Mapping[str, int] + exits_by_kind: Mapping[str, int] + + def to_dict(self) -> dict[str, object]: + """Return a JSON-safe view of the counts.""" + return { + "opening": self.opening, + "closing": self.closing, + "carried": self.carried, + "entered": self.entered, + "exited": self.exited, + "transient": self.transient, + "additions_total": self.additions_total, + "exits_total": self.exits_total, + "additions_by_kind": dict(self.additions_by_kind), + "exits_by_kind": dict(self.exits_by_kind), + } + + +@dataclass(frozen=True) +class WeightRevaluation: + """Weight movement that is *not* an arrival or a departure. + + Each component is the sum of ``weight at the end of the person's + presence minus weight at the start of it`` over one presence class. + ``carried`` is the component the caller usually wants: it is the + entire change in the weight of people who were present at both ends + of the period. The other three are zero unless the caller declared + an explicit event weight that differs from the frame weight. + """ + + carried: float + entrant: float + exiting: float + transient: float + + @property + def total(self) -> float: + """Stable sum of the four binary64 components.""" + return _stable_sum( + (self.carried, self.entrant, self.exiting, self.transient) + ) + + def to_dict(self) -> dict[str, object]: + """Return a JSON-safe view of the revaluation.""" + return { + "carried": self.carried, + "entrant": self.entrant, + "exiting": self.exiting, + "transient": self.transient, + "total": self.total, + } + + +@dataclass(frozen=True) +class WeightTotals: + """Weight stocks and flows for one period. + + The identity is:: + + closing == opening + + additions_total + - exits_total + + revaluation.total + + reported against an arithmetic residual rather than a tolerance. + """ + + opening: float + closing: float + additions_total: float + exits_total: float + additions_by_kind: Mapping[str, float] + exits_by_kind: Mapping[str, float] + revaluation: WeightRevaluation + + def to_dict(self) -> dict[str, object]: + """Return a JSON-safe view of the weight totals.""" + return { + "opening": self.opening, + "closing": self.closing, + "additions_total": self.additions_total, + "exits_total": self.exits_total, + "additions_by_kind": dict(self.additions_by_kind), + "exits_by_kind": dict(self.exits_by_kind), + "revaluation": self.revaluation.to_dict(), + } + + +@dataclass(frozen=True) +class PeriodAccount: + """The reconciled stock-flow account for one annual period.""" + + opening_year: int + closing_year: int + status: str + status_note: str + counts: PersonCounts + weights: WeightTotals + count_residual: int + weight_residual: float + reconstructed_closing_weight: float + carried_person_ids: tuple[int, ...] + entered_person_ids: tuple[int, ...] + exited_person_ids: tuple[int, ...] + transient_person_ids: tuple[int, ...] + provenance: Mapping[str, object] + + @property + def added_person_ids(self) -> tuple[int, ...]: + """Every declared addition: entrants plus transients, sorted.""" + return tuple( + sorted(self.entered_person_ids + self.transient_person_ids) + ) + + @property + def departed_person_ids(self) -> tuple[int, ...]: + """Every declared exit: exiters plus transients, sorted.""" + return tuple( + sorted(self.exited_person_ids + self.transient_person_ids) + ) + + def to_dict(self) -> dict[str, object]: + """Return the serializable counts, weights and provenance. + + Person identifier tuples stay off this payload on purpose: they + are available as attributes for programmatic use, but a + serialized account is a summary, not a roster. + """ + return { + "interface_version": ACCOUNTING_INTERFACE_VERSION, + "status": self.status, + "status_note": self.status_note, + "opening_year": self.opening_year, + "closing_year": self.closing_year, + "counts": self.counts.to_dict(), + "weights": self.weights.to_dict(), + "residuals": { + "count": self.count_residual, + "weight": self.weight_residual, + "reconstructed_closing_weight": ( + self.reconstructed_closing_weight + ), + }, + "provenance": { + **self.provenance, + "declaration_sources": list( + self.provenance["declaration_sources"] + ), + }, + } + + +def _integer_column( + frame: pd.DataFrame, column: str, label: str +) -> np.ndarray: + """Return one non-null integral column as ``int64``. + + An empty column is accepted whatever its dtype: it holds no value + that could be non-integral, and the naive way to spell an empty + population, ``pd.DataFrame({"person_id": [], ...})``, yields float + columns that carry no information about the caller's intent. + """ + series = frame[column] + if len(series) == 0: + return np.empty(0, dtype=np.int64) + if series.isna().any(): + raise PopulationAccountingInputError( + f"{label} column {column!r} contains null values" + ) + if pd.api.types.is_bool_dtype(series.dtype): + raise PopulationAccountingInputError( + f"{label} column {column!r} is boolean, not integral" + ) + if pd.api.types.is_integer_dtype(series.dtype): + _in_integer_domain(int(series.min()), f"{label}.{column}") + _in_integer_domain(int(series.max()), f"{label}.{column}") + return series.to_numpy(dtype=np.int64, copy=True) + if series.dtype == object: + values = series.tolist() + for value in values: + if isinstance(value, bool) or not isinstance( + value, (int, np.integer) + ): + raise PopulationAccountingInputError( + f"{label} column {column!r} holds a non-integral value " + f"{value!r} of type {type(value).__name__}" + ) + return np.asarray( + [ + _in_integer_domain(int(value), f"{label}.{column}") + for value in values + ], + dtype=np.int64, + ) + raise PopulationAccountingInputError( + f"{label} column {column!r} must be an integer dtype, got " + f"{series.dtype!r}; float identifiers and years are rejected " + "because they cannot be compared exactly" + ) + + +def _weight_column(frame: pd.DataFrame, label: str) -> np.ndarray: + """Return the weight column as finite, non-negative ``float64``. + + As with identifiers, an empty column is accepted whatever its + dtype. + """ + series = frame[WEIGHT_COLUMN] + if len(series) == 0: + return np.empty(0, dtype=np.float64) + if series.isna().any(): + raise PopulationAccountingInputError( + f"{label} column {WEIGHT_COLUMN!r} contains null values" + ) + if pd.api.types.is_bool_dtype(series.dtype): + raise PopulationAccountingInputError( + f"{label} column {WEIGHT_COLUMN!r} is boolean, not numeric" + ) + if series.dtype == object: + return np.asarray( + [ + _as_weight(value, f"{label}.{WEIGHT_COLUMN}") + for value in series.tolist() + ], + dtype=np.float64, + ) + if not pd.api.types.is_numeric_dtype( + series.dtype + ) or pd.api.types.is_complex_dtype(series.dtype): + raise PopulationAccountingInputError( + f"{label} column {WEIGHT_COLUMN!r} must contain real numbers" + ) + if (series < 0).any(): + raise PopulationAccountingInputError( + f"{label} column {WEIGHT_COLUMN!r} contains negative weights" + ) + try: + values = series.to_numpy(dtype=np.float64, copy=True) + except (TypeError, ValueError, OverflowError) as error: + raise PopulationAccountingInputError( + f"{label} column {WEIGHT_COLUMN!r} is not numeric " + f"({series.dtype!r})" + ) from error + if not np.isfinite(values).all(): + raise PopulationAccountingInputError( + f"{label} column {WEIGHT_COLUMN!r} contains non-finite weights" + ) + if (values < 0.0).any(): + raise PopulationAccountingInputError( + f"{label} column {WEIGHT_COLUMN!r} contains negative weights" + ) + if ((values == 0.0) & series.ne(0).to_numpy(dtype=bool)).any(): + raise PopulationAccountingInputError( + f"{label} column {WEIGHT_COLUMN!r} underflows to zero in binary64" + ) + return values + + +def _read_frame( + frame: pd.DataFrame, year: int, label: str +) -> tuple[np.ndarray, np.ndarray]: + """Validate one population frame and snapshot its two columns. + + The frame is only read. Nothing is assigned, sorted in place, or + otherwise mutated, and the returned arrays are fresh. + """ + if not isinstance(frame, pd.DataFrame): + raise PopulationAccountingInputError( + f"{label} must be a pandas DataFrame, got " + f"{type(frame).__name__}" + ) + missing = [ + column for column in REQUIRED_COLUMNS if column not in frame.columns + ] + if missing: + raise PopulationAccountingInputError( + f"{label} is missing columns {missing}" + ) + if frame.columns.duplicated().any(): + raise PopulationAccountingInputError( + f"{label} has duplicate column labels" + ) + person_ids = _integer_column(frame, PERSON_ID_COLUMN, label) + unique_ids, counts = np.unique(person_ids, return_counts=True) + if unique_ids.size != person_ids.size: + repeated = unique_ids[counts > 1][:10].tolist() + raise PopulationAccountingInputError( + f"{label} contains duplicate {PERSON_ID_COLUMN} rows: {repeated}" + ) + years = _integer_column(frame, YEAR_COLUMN, label) + off_year = np.unique(years[years != year])[:10].tolist() + if off_year: + raise PopulationAccountingInputError( + f"{label} must carry year {year}; found {off_year}" + ) + weights = _weight_column(frame, label) + return person_ids, weights + + +def _index_events( + events: Sequence[PopulationEvent], + *, + expected_kinds: frozenset[PopulationEventKind], + closing_year: int, + label: str, + duplicate_kind: DiscrepancyKind, + discrepancies: list[AccountingDiscrepancy], +) -> dict[int, PopulationEvent]: + """Validate one declaration sequence and index it by person.""" + if isinstance(events, (str, bytes)) or not isinstance(events, Sequence): + raise PopulationAccountingInputError( + f"{label} must be a sequence of PopulationEvent, got " + f"{type(events).__name__}" + ) + indexed: dict[int, PopulationEvent] = {} + for position, event in enumerate(events): + if not isinstance(event, PopulationEvent): + raise PopulationAccountingInputError( + f"{label}[{position}] must be a PopulationEvent, got " + f"{type(event).__name__}" + ) + if event.kind not in expected_kinds: + raise PopulationAccountingInputError( + f"{label}[{position}] declares kind {event.kind.value!r}, " + f"which is not one of " + f"{sorted(item.value for item in expected_kinds)}; " + "arrivals and departures are declared separately" + ) + if event.year != closing_year: + raise PopulationAccountingInputError( + f"{label}[{position}] is booked to year {event.year}, but " + f"this period closes in {closing_year}; events are booked " + "to the period's closing year" + ) + if event.person_id in indexed: + discrepancies.append( + AccountingDiscrepancy( + kind=duplicate_kind, + person_id=event.person_id, + detail=( + f"person {event.person_id} is declared more than " + f"once in {label} " + f"({indexed[event.person_id].kind.value!r} then " + f"{event.kind.value!r})" + ), + ) + ) + continue + indexed[event.person_id] = event + return indexed + + +def _resolve_weight( + person_id: int, + event: PopulationEvent, + fallback: Mapping[int, float], + *, + fallback_label: str, +) -> float: + """Return the declared event weight, or its frame stand-in.""" + if event.weight is not None: + return event.weight + try: + return fallback[person_id] + except KeyError: + raise PopulationAccountingInputError( + f"person {person_id} is declared as a " + f"{event.kind.value!r} but appears in neither the opening nor " + "the closing frame, so this module cannot recover a weight for " + f"the event from the {fallback_label} frame; a person who both " + "arrives and departs within the period must declare an " + "explicit weight on both declarations" + ) from None + + +def reconcile_period( + opening: pd.DataFrame, + closing: pd.DataFrame, + *, + opening_year: int, + closing_year: int, + additions: Sequence[PopulationEvent] = (), + exits: Sequence[PopulationEvent] = (), +) -> PeriodAccount: + """Reconcile one annual period against its declared transitions. + + This is the module's only entry point. It is a pure function: it + reads the two frames and the two declaration sequences, mutates + nothing, touches no filesystem, and returns a + :class:`PeriodAccount` or raises. + + Parameters + ---------- + opening, closing: + Person-level frames carrying ``person_id``, ``year`` and + ``weight``. Identifiers must be integral and unique within + each frame; weights must be finite and non-negative. Rows with + zero weight are ordinary persons and are never dropped. + opening_year, closing_year: + The period's endpoints, stated explicitly rather than inferred, + so an empty frame is unambiguous. ``closing_year`` must be + ``opening_year + 1``. + additions, exits: + The declared arrivals and departures, each a sequence of + :class:`PopulationEvent`. Passing a departure kind in + ``additions`` (or the reverse) is an input error. + + Returns + ------- + PeriodAccount + Counts, weight stocks and flows, arithmetic residuals and + provenance. Its ``status`` is always + :data:`ENGINEERING_STATUS`: a returned account is an + engineering statement about arithmetic and identity, never a + scientific verdict. + + Raises + ------ + PopulationAccountingInputError + A frame or a declaration is malformed: a missing column, a null + or non-integral identifier, a duplicate identifier within one + frame, a non-finite or negative weight, a row carrying the + wrong year, a declaration booked to the wrong year, a + declaration of the wrong direction, or a transient whose weight + cannot be recovered. + PopulationReconciliationError + The inputs are well formed but the declared story does not + match the frames: an unexplained arrival, an omitted exit, a + duplicate or colliding declaration, or an impossible sequence. + The typed findings are on the exception's ``discrepancies``. + + Notes + ----- + Validation is staged, and the first stage to find a problem raises: + frames, then declarations, then reconciliation, then weights. A + caller with several problems at once therefore sees the earliest, + not all of them. + + The accountant sees exactly one period. It has no memory of + earlier ones and no view of what the loop *should* have scheduled; + it is not a schedule builder. A person who exits in one period and + reappears later is an unexplained arrival only if no addition is + declared. The accountant cannot detect declared reuse of a past ID, + or a transient omitted from both event sequences. + """ + opening_year = _as_year(opening_year, "opening_year") + closing_year = _as_year(closing_year, "closing_year") + if closing_year != opening_year + 1: + raise PopulationAccountingInputError( + f"closing_year must be opening_year + 1 (this is an annual " + f"accountant); got opening_year={opening_year} and " + f"closing_year={closing_year}" + ) + + opening_ids, opening_weights = _read_frame( + opening, opening_year, "opening frame" + ) + closing_ids, closing_weights = _read_frame( + closing, closing_year, "closing frame" + ) + opening_weight_by_person = dict( + zip(opening_ids.tolist(), opening_weights.tolist(), strict=True) + ) + closing_weight_by_person = dict( + zip(closing_ids.tolist(), closing_weights.tolist(), strict=True) + ) + opening_set = set(opening_weight_by_person) + closing_set = set(closing_weight_by_person) + + discrepancies: list[AccountingDiscrepancy] = [] + additions_by_person = _index_events( + additions, + expected_kinds=ADDITION_KINDS, + closing_year=closing_year, + label="additions", + duplicate_kind=DiscrepancyKind.DUPLICATE_ADDITION, + discrepancies=discrepancies, + ) + exits_by_person = _index_events( + exits, + expected_kinds=EXIT_KINDS, + closing_year=closing_year, + label="exits", + duplicate_kind=DiscrepancyKind.DUPLICATE_EXIT, + discrepancies=discrepancies, + ) + + for person_id in sorted(additions_by_person): + if person_id in opening_set: + discrepancies.append( + AccountingDiscrepancy( + kind=DiscrepancyKind.ADDITION_COLLIDES_WITH_OPENING, + person_id=person_id, + detail=( + f"person {person_id} is declared as a " + f"{additions_by_person[person_id].kind.value!r} but " + "was already present in the opening frame" + ), + ) + ) + for person_id in sorted(exits_by_person): + if person_id not in opening_set and person_id not in ( + additions_by_person + ): + discrepancies.append( + AccountingDiscrepancy( + kind=DiscrepancyKind.EXIT_WITHOUT_PRESENCE, + person_id=person_id, + detail=( + f"person {person_id} is declared as a " + f"{exits_by_person[person_id].kind.value!r} but was " + "never present: absent from the opening frame and " + "never declared as an addition" + ), + ) + ) + if person_id in closing_set: + discrepancies.append( + AccountingDiscrepancy( + kind=DiscrepancyKind.EXIT_CONTRADICTED_BY_CLOSING, + person_id=person_id, + detail=( + f"person {person_id} is declared as a " + f"{exits_by_person[person_id].kind.value!r} but is " + "still present in the closing frame" + ), + ) + ) + for person_id in sorted(closing_set - opening_set): + if person_id not in additions_by_person: + discrepancies.append( + AccountingDiscrepancy( + kind=DiscrepancyKind.UNDECLARED_ADDITION, + person_id=person_id, + detail=( + f"person {person_id} appears in the closing frame " + "with no declared addition; this module will not " + "guess whether the identifier is a birth, a " + "scheduled entry, or a defect" + ), + ) + ) + for person_id in sorted(opening_set - closing_set): + if person_id not in exits_by_person: + discrepancies.append( + AccountingDiscrepancy( + kind=DiscrepancyKind.UNDECLARED_EXIT, + person_id=person_id, + detail=( + f"person {person_id} disappears between the frames " + "with no declared exit; this module will not assume " + "the person died" + ), + ) + ) + for person_id in sorted(additions_by_person): + if ( + person_id not in closing_set + and person_id not in exits_by_person + and person_id not in opening_set + ): + discrepancies.append( + AccountingDiscrepancy( + kind=DiscrepancyKind.ADDITION_ABSENT_AT_CLOSE, + person_id=person_id, + detail=( + f"person {person_id} is declared as a " + f"{additions_by_person[person_id].kind.value!r} but " + "is absent from the closing frame and has no " + "declared exit" + ), + ) + ) + + if discrepancies: + raise PopulationReconciliationError( + _summarize(discrepancies, opening_year, closing_year), + _sorted_discrepancies(discrepancies), + ) + + carried = tuple(sorted(opening_set & closing_set)) + entered = tuple(sorted(closing_set - opening_set)) + exited = tuple(sorted(opening_set - closing_set)) + transient = tuple( + sorted(set(additions_by_person) - opening_set - closing_set) + ) + + counts = PersonCounts( + opening=len(opening_set), + closing=len(closing_set), + carried=len(carried), + entered=len(entered), + exited=len(exited), + transient=len(transient), + additions_total=len(additions_by_person), + exits_total=len(exits_by_person), + additions_by_kind=_count_by_kind(additions_by_person, ADDITION_KINDS), + exits_by_kind=_count_by_kind(exits_by_person, EXIT_KINDS), + ) + count_residual = counts.closing - ( + counts.opening + counts.additions_total - counts.exits_total + ) + if count_residual != 0: + raise PopulationReconciliationError( + "the exact person-count identity does not hold: " + f"closing {counts.closing} != opening {counts.opening} + " + f"additions {counts.additions_total} - exits " + f"{counts.exits_total}", + ( + AccountingDiscrepancy( + kind=DiscrepancyKind.COUNT_IDENTITY_VIOLATION, + person_id=None, + detail=f"count residual {count_residual}", + ), + ), + ) + + entry_weight = { + person_id: _resolve_weight( + person_id, + event, + closing_weight_by_person, + fallback_label="closing", + ) + for person_id, event in additions_by_person.items() + } + exit_weight = { + person_id: _resolve_weight( + person_id, + event, + opening_weight_by_person, + fallback_label="opening", + ) + for person_id, event in exits_by_person.items() + } + + revaluation = WeightRevaluation( + carried=_stable_sum( + closing_weight_by_person[person_id] + - opening_weight_by_person[person_id] + for person_id in carried + ), + entrant=_stable_sum( + closing_weight_by_person[person_id] - entry_weight[person_id] + for person_id in entered + ), + exiting=_stable_sum( + exit_weight[person_id] - opening_weight_by_person[person_id] + for person_id in exited + ), + transient=_stable_sum( + exit_weight[person_id] - entry_weight[person_id] + for person_id in transient + ), + ) + additions_by_kind_weight = _weight_by_kind( + additions_by_person, entry_weight, ADDITION_KINDS + ) + exits_by_kind_weight = _weight_by_kind( + exits_by_person, exit_weight, EXIT_KINDS + ) + weights = WeightTotals( + opening=_stable_sum(opening_weights.tolist()), + closing=_stable_sum(closing_weights.tolist()), + additions_total=_stable_sum(entry_weight.values()), + exits_total=_stable_sum(exit_weight.values()), + additions_by_kind=additions_by_kind_weight, + exits_by_kind=exits_by_kind_weight, + revaluation=revaluation, + ) + reconstructed = _stable_sum( + ( + weights.opening, + weights.additions_total, + -weights.exits_total, + revaluation.total, + ) + ) + weight_residual = _stable_sum((weights.closing, -reconstructed)) + + provenance = MappingProxyType( + { + "interface_version": ACCOUNTING_INTERFACE_VERSION, + "person_id_column": PERSON_ID_COLUMN, + "year_column": YEAR_COLUMN, + "weight_column": WEIGHT_COLUMN, + "summation": SUMMATION_METHOD, + "inference": INFERENCE_POLICY, + "opening_rows": int(opening_ids.size), + "closing_rows": int(closing_ids.size), + "opening_zero_weight_rows": int( + np.count_nonzero(opening_weights == 0.0) + ), + "closing_zero_weight_rows": int( + np.count_nonzero(closing_weights == 0.0) + ), + "declared_additions": len(additions_by_person), + "declared_exits": len(exits_by_person), + "declared_addition_weights": sum( + 1 + for event in additions_by_person.values() + if event.weight is not None + ), + "declared_exit_weights": sum( + 1 + for event in exits_by_person.values() + if event.weight is not None + ), + "event_log_completeness_verified": False, + "declaration_sources": tuple( + sorted( + { + event.source + for event in ( + *additions_by_person.values(), + *exits_by_person.values(), + ) + if event.source + } + ) + ), + } + ) + + return PeriodAccount( + opening_year=opening_year, + closing_year=closing_year, + status=ENGINEERING_STATUS, + status_note=ENGINEERING_STATUS_NOTE, + counts=counts, + weights=weights, + count_residual=count_residual, + weight_residual=weight_residual, + reconstructed_closing_weight=reconstructed, + carried_person_ids=carried, + entered_person_ids=entered, + exited_person_ids=exited, + transient_person_ids=transient, + provenance=provenance, + ) + + +def _count_by_kind( + indexed: Mapping[int, PopulationEvent], + kinds: frozenset[PopulationEventKind], +) -> Mapping[str, int]: + """Count declarations by kind, with every kind present as a key.""" + tally = {kind.value: 0 for kind in sorted(kinds, key=lambda k: k.value)} + for event in indexed.values(): + tally[event.kind.value] += 1 + return MappingProxyType(tally) + + +def _weight_by_kind( + indexed: Mapping[int, PopulationEvent], + resolved: Mapping[int, float], + kinds: frozenset[PopulationEventKind], +) -> Mapping[str, float]: + """Sum resolved event weights by kind, stably, with all keys present.""" + grouped: dict[str, list[float]] = { + kind.value: [] for kind in sorted(kinds, key=lambda k: k.value) + } + for person_id, event in indexed.items(): + grouped[event.kind.value].append(resolved[person_id]) + return MappingProxyType( + {kind: _stable_sum(values) for kind, values in grouped.items()} + ) + + +def _sorted_discrepancies( + discrepancies: Iterable[AccountingDiscrepancy], +) -> tuple[AccountingDiscrepancy, ...]: + """Order findings deterministically by kind then person.""" + return tuple( + sorted( + discrepancies, + key=lambda item: ( + _DISCREPANCY_ORDER[item.kind], + -1 if item.person_id is None else item.person_id, + ), + ) + ) + + +def _summarize( + discrepancies: Sequence[AccountingDiscrepancy], + opening_year: int, + closing_year: int, +) -> str: + """Build a stable one-line summary of a refusal.""" + tally: dict[str, int] = {} + for item in discrepancies: + tally[item.kind.value] = tally.get(item.kind.value, 0) + 1 + rendered = ", ".join( + f"{kind}={count}" for kind, count in sorted(tally.items()) + ) + return ( + f"population accounting for {opening_year}->{closing_year} does not " + f"reconcile: {rendered}" + ) diff --git a/tests/README-tiers.md b/tests/README-tiers.md index d19e45fe..4af88c67 100644 --- a/tests/README-tiers.md +++ b/tests/README-tiers.md @@ -38,9 +38,9 @@ pytest --collect-only -q -m oracle_policyengine | tail -1 | Tier | Tests at HEAD | |---|---:| -| `unit` | 1,624 | -| `artifact` | 2,668 | +| `unit` | 1,717 | +| `artifact` | 2,672 | | `integration_psid` | 848 | | `reproduction_legacy` | 520 | | `oracle_policyengine` | 159 | -| **Total** | **5,819** | +| **Total** | **5,916** | diff --git a/tests/estimates/test_birth_evidence_artifact.py b/tests/estimates/test_birth_evidence_artifact.py index 3641f43f..b4bfb6ba 100644 --- a/tests/estimates/test_birth_evidence_artifact.py +++ b/tests/estimates/test_birth_evidence_artifact.py @@ -95,6 +95,7 @@ def test_post_review_sources_are_outside_historical_reducer_identity(): Path("src/populace_dynamics/graph/runtime.py"), Path("src/populace_dynamics/graph/synthetic.py"), Path("src/populace_dynamics/graph/trajectory.py"), + Path("src/populace_dynamics/engine/accounting.py"), ) assert reducer.POST_REVIEW_SHARED_SOURCE_BLOBS == { Path( @@ -137,11 +138,7 @@ def _internal_imports( package_parts = module_parts if is_package else module_parts[:-1] for node in ast.walk(tree): if isinstance(node, ast.Import): - imports.update( - alias.name - for alias in node.names - if alias.name in module_paths - ) + imports.update(alias.name for alias in node.names) continue if not isinstance(node, ast.ImportFrom): continue @@ -155,13 +152,61 @@ def _internal_imports( base = ".".join(base_parts) else: base = node.module or "" - if base in module_paths: - imports.add(base) + imports.add(base) for alias in node.names: candidate = f"{base}.{alias.name}" if base else alias.name - if candidate in module_paths: - imports.add(candidate) - return imports + imports.add(candidate) + # Importing a leaf executes its parent package initializers too. Include + # those even when an imported leaf is external or not a tracked module. + internal = set() + for imported in imports: + parts = imported.split(".") + internal.update( + parent + for length in range(1, len(parts) + 1) + if (parent := ".".join(parts[:length])) in module_paths + ) + return internal + + +@pytest.mark.parametrize( + "statement", + [ + "import sample.leaf", + "from sample.leaf import function", + "from sample import leaf", + "import sample.untracked_extension", + ], +) +def test_source_reachability_includes_implicit_package_initializers( + tmp_path, statement +): + package = tmp_path / "__init__.py" + leaf = tmp_path / "leaf.py" + hidden = tmp_path / "hidden.py" + consumer = tmp_path / "consumer.py" + package.write_text("from . import hidden\n") + leaf.write_text("def function(): pass\n") + hidden.write_text("") + consumer.write_text(statement + "\n") + modules = { + "consumer": consumer, + "sample": package, + "sample.leaf": leaf, + "sample.hidden": hidden, + } + reachable = set() + pending = ["consumer"] + while pending: + name = pending.pop() + if name in reachable: + continue + reachable.add(name) + pending.extend( + _internal_imports(name, modules[name], modules) - reachable + ) + assert {"sample", "sample.hidden"}.issubset(reachable) + assert "sample.untracked_extension" not in reachable def test_psid_and_graph_exclusions_are_unreachable_from_birth_evidence(): @@ -223,6 +268,10 @@ def test_psid_and_graph_exclusions_are_unreachable_from_birth_evidence(): "opt-in graph modules became reachable from the birth-evidence " f"reducer: {sorted(graph_exclusions & reachable)}" ) + assert "populace_dynamics.engine.accounting" in module_paths + assert "populace_dynamics.engine.accounting" not in reachable + assert "populace_dynamics.engine" in reachable + assert "populace_dynamics.engine.steps" in reachable def test_reducer_accepts_explicit_unresolved_upstream_boundary(): diff --git a/tests/test_m6_stock_flow.py b/tests/test_m6_stock_flow.py new file mode 100644 index 00000000..a68fd055 --- /dev/null +++ b/tests/test_m6_stock_flow.py @@ -0,0 +1,1537 @@ +"""Tests for the experimental annual stock-flow accounting interface. + +Every input here is built in memory. The conservation tests state the +closing population and the expected identity components independently, +as hand-checked literals, rather than recomputing the module's own +formula and comparing it to itself. The projection-engine tests drive +the real :class:`populace_dynamics.engine.loop.ProjectionEngine` with +synthetic adapters that record their own event declarations, so the +accountant is exercised against frames the engine actually produced. + +No fitted model, survey extract, benefit calculation or benchmark value +is involved anywhere in this module. +""" + +from __future__ import annotations + +import json +import math +from dataclasses import FrozenInstanceError, dataclass, field +from inspect import signature + +import numpy as np +import pandas as pd +import pytest + +import populace_dynamics.engine.accounting as accounting +from populace_dynamics.engine.accounting import ( + ACCOUNTING_INTERFACE_VERSION, + ENGINEERING_STATUS, + AccountingDiscrepancy, + DiscrepancyKind, + PopulationAccountingInputError, + PopulationEvent, + PopulationEventKind, + PopulationReconciliationError, + reconcile_period, +) +from populace_dynamics.engine.loop import ( + SCHEDULED_ENTRIES_KEY, + MaritalStepResult, + PeriodModules, + ProjectionEngine, +) + +# --------------------------------------------------------------------- +# in-memory frame helpers +# --------------------------------------------------------------------- + + +def frame(year: int, rows: dict[int, float], **columns) -> pd.DataFrame: + """Build a population frame from an ``{id: weight}`` mapping.""" + person_ids = list(rows) + built = pd.DataFrame( + { + "person_id": np.asarray(person_ids, dtype=np.int64), + "year": np.full(len(person_ids), year, dtype=np.int64), + "weight": np.asarray( + [rows[key] for key in person_ids], dtype=np.float64 + ), + } + ) + for name, values in columns.items(): + built[name] = values + return built + + +def empty_frame() -> pd.DataFrame: + """Build the naive spelling of an empty population.""" + return pd.DataFrame({"person_id": [], "year": [], "weight": []}) + + +def death(person_id: int, year: int, **kwargs) -> PopulationEvent: + return PopulationEvent(person_id, "death", year, **kwargs) + + +def birth(person_id: int, year: int, **kwargs) -> PopulationEvent: + return PopulationEvent(person_id, "birth", year, **kwargs) + + +def entry(person_id: int, year: int, **kwargs) -> PopulationEvent: + return PopulationEvent(person_id, "scheduled_entry", year, **kwargs) + + +def kinds(error: PopulationReconciliationError) -> list[DiscrepancyKind]: + return [item.kind for item in error.discrepancies] + + +# --------------------------------------------------------------------- +# conservation, with independently specified expectations +# --------------------------------------------------------------------- + + +def test_hand_checked_year_reconciles_to_stated_totals(): + """A worked year whose every component is stated, not derived. + + Opening 2020 holds five people weighing 100, 200, 300, 400 and 500, + so the opening stock is 1500. Over the period person 5 dies, person + 3's weight rises from 300 to 350, and one child is born weighing + 100. The closing frame is written out independently and sums to + 1150. The identity a reader can check by hand is + ``1500 - 500 + 100 + 50 = 1150``. + """ + opening = frame(2020, {1: 100.0, 2: 200.0, 3: 300.0, 4: 400.0, 5: 500.0}) + closing = frame( + 2021, {1: 100.0, 2: 200.0, 3: 350.0, 4: 400.0, 1001: 100.0} + ) + + account = reconcile_period( + opening, + closing, + opening_year=2020, + closing_year=2021, + additions=[birth(1001, 2021)], + exits=[death(5, 2021)], + ) + + assert account.weights.opening == 1500.0 + assert account.weights.closing == 1150.0 + assert account.weights.exits_total == 500.0 + assert account.weights.additions_total == 100.0 + assert account.weights.revaluation.carried == 50.0 + assert account.weights.revaluation.total == 50.0 + assert account.reconstructed_closing_weight == 1150.0 + assert account.weight_residual == 0.0 + assert account.count_residual == 0 + assert account.counts.opening == 5 + assert account.counts.closing == 5 + assert account.counts.carried == 4 + assert account.counts.entered == 1 + assert account.counts.exited == 1 + assert account.counts.transient == 0 + assert account.carried_person_ids == (1, 2, 3, 4) + assert account.entered_person_ids == (1001,) + assert account.exited_person_ids == (5,) + + +def test_flows_split_by_declared_kind(): + """Two arrivals and two departures of different kinds stay apart. + + Opening 10 + 20 = 30. A birth of 4 and a scheduled entry of 6 + arrive; a death of 10 and an emigration of 20 depart. The closing + frame holds only the two arrivals and weighs 10. + """ + opening = frame(2030, {1: 10.0, 2: 20.0}) + closing = frame(2031, {50: 4.0, 60: 6.0}) + + account = reconcile_period( + opening, + closing, + opening_year=2030, + closing_year=2031, + additions=[birth(50, 2031), entry(60, 2031)], + exits=[ + death(1, 2031), + PopulationEvent(2, "emigration", 2031), + ], + ) + + assert account.weights.closing == 10.0 + assert dict(account.weights.additions_by_kind) == { + "birth": 4.0, + "scheduled_entry": 6.0, + "other_entry": 0.0, + } + assert dict(account.weights.exits_by_kind) == { + "death": 10.0, + "emigration": 20.0, + "other_exit": 0.0, + } + assert dict(account.counts.additions_by_kind) == { + "birth": 1, + "scheduled_entry": 1, + "other_entry": 0, + } + assert dict(account.counts.exits_by_kind) == { + "death": 1, + "emigration": 1, + "other_exit": 0, + } + assert account.weight_residual == 0.0 + + +def test_unchanged_survivors_produce_a_flat_account(): + """Nothing happens: every flow and every revaluation is zero.""" + rows = {7: 1.5, 8: 2.5, 9: 3.5} + account = reconcile_period( + frame(2040, rows), + frame(2041, rows), + opening_year=2040, + closing_year=2041, + ) + + assert account.weights.opening == 7.5 + assert account.weights.closing == 7.5 + assert account.weights.additions_total == 0.0 + assert account.weights.exits_total == 0.0 + assert account.weights.revaluation.to_dict() == { + "carried": 0.0, + "entrant": 0.0, + "exiting": 0.0, + "transient": 0.0, + "total": 0.0, + } + assert account.counts.carried == 3 + assert account.weight_residual == 0.0 + + +def test_carried_weight_change_is_its_own_component(): + """A pure revaluation moves no person and is never a flow. + + Both people survive. One weight rises by 3 and the other falls by + 1, so the stock moves from 30 to 32 with no arrival or departure. + """ + account = reconcile_period( + frame(2040, {1: 10.0, 2: 20.0}), + frame(2041, {1: 13.0, 2: 19.0}), + opening_year=2040, + closing_year=2041, + ) + + assert account.weights.opening == 30.0 + assert account.weights.closing == 32.0 + assert account.weights.additions_total == 0.0 + assert account.weights.exits_total == 0.0 + assert account.weights.revaluation.carried == 2.0 + assert account.counts.additions_total == 0 + assert account.counts.exits_total == 0 + assert account.weight_residual == 0.0 + + +def test_within_period_entry_and_exit_nets_out(): + """Someone who arrives and leaves inside the period holds no stock. + + Persons 1 and 2 carry 10 and 20 through unchanged. Person 7 + scheduled-enters weighing 5 and emigrates the same year weighing 5. + They appear in neither frame, yet both flows are booked and the + closing stock is still 30. + """ + account = reconcile_period( + frame(2050, {1: 10.0, 2: 20.0}), + frame(2051, {1: 10.0, 2: 20.0}), + opening_year=2050, + closing_year=2051, + additions=[entry(7, 2051, weight=5.0)], + exits=[PopulationEvent(7, "emigration", 2051, weight=5.0)], + ) + + assert account.counts.opening == 2 + assert account.counts.closing == 2 + assert account.counts.carried == 2 + assert account.counts.entered == 0 + assert account.counts.exited == 0 + assert account.counts.transient == 1 + assert account.counts.additions_total == 1 + assert account.counts.exits_total == 1 + assert account.transient_person_ids == (7,) + assert account.added_person_ids == (7,) + assert account.departed_person_ids == (7,) + assert account.weights.additions_total == 5.0 + assert account.weights.exits_total == 5.0 + assert account.weights.revaluation.total == 0.0 + assert account.weights.closing == 30.0 + assert account.weight_residual == 0.0 + + +def test_transient_weight_change_is_reported_not_absorbed(): + """A transient whose weight moved keeps the identity honest. + + Person 7 enters weighing 5 and leaves weighing 8. The 3 is a + transient revaluation, not a silent gap in the closing stock. + """ + account = reconcile_period( + frame(2050, {1: 10.0}), + frame(2051, {1: 10.0}), + opening_year=2050, + closing_year=2051, + additions=[entry(7, 2051, weight=5.0)], + exits=[PopulationEvent(7, "emigration", 2051, weight=8.0)], + ) + + assert account.weights.additions_total == 5.0 + assert account.weights.exits_total == 8.0 + assert account.weights.revaluation.transient == 3.0 + assert account.weights.revaluation.carried == 0.0 + assert account.weights.closing == 10.0 + assert account.weight_residual == 0.0 + + +def test_declared_exit_weight_differing_from_opening_is_a_component(): + """A departure priced away from its opening weight is visible. + + Person 2 opens at 20 but is declared to leave at 12. The 8 lands + in the ``exiting`` revaluation rather than vanishing. + """ + account = reconcile_period( + frame(2060, {1: 10.0, 2: 20.0}), + frame(2061, {1: 10.0}), + opening_year=2060, + closing_year=2061, + exits=[death(2, 2061, weight=12.0)], + ) + + assert account.weights.exits_total == 12.0 + assert account.weights.revaluation.exiting == -8.0 + assert account.weights.revaluation.carried == 0.0 + assert account.weights.closing == 10.0 + assert account.weight_residual == 0.0 + + +def test_declared_entry_weight_differing_from_closing_is_a_component(): + """An arrival repriced after entry is visible too.""" + account = reconcile_period( + frame(2060, {1: 10.0}), + frame(2061, {1: 10.0, 9: 6.0}), + opening_year=2060, + closing_year=2061, + additions=[entry(9, 2061, weight=4.0)], + ) + + assert account.weights.additions_total == 4.0 + assert account.weights.revaluation.entrant == 2.0 + assert account.weights.closing == 16.0 + assert account.weight_residual == 0.0 + + +def test_complete_extinction_reconciles(): + """Everyone dies: the closing frame is empty and the stock is zero.""" + account = reconcile_period( + frame(2070, {1: 10.0, 2: 20.0, 3: 30.0}), + empty_frame(), + opening_year=2070, + closing_year=2071, + exits=[death(person, 2071) for person in (1, 2, 3)], + ) + + assert account.counts.opening == 3 + assert account.counts.closing == 0 + assert account.counts.exited == 3 + assert account.weights.opening == 60.0 + assert account.weights.closing == 0.0 + assert account.weights.exits_total == 60.0 + assert account.weight_residual == 0.0 + + +def test_empty_to_empty_reconciles(): + """An empty population that stays empty is a valid, flat account.""" + account = reconcile_period( + empty_frame(), + empty_frame(), + opening_year=2080, + closing_year=2081, + ) + + assert account.counts.to_dict()["opening"] == 0 + assert account.counts.closing == 0 + assert account.weights.opening == 0.0 + assert account.weights.closing == 0.0 + assert account.weights.revaluation.total == 0.0 + assert account.weight_residual == 0.0 + assert account.carried_person_ids == () + + +def test_population_may_start_empty_and_be_repopulated(): + """Additions in a later period do not need an earlier stock.""" + account = reconcile_period( + empty_frame(), + frame(2081, {4: 2.0, 5: 3.0}), + opening_year=2080, + closing_year=2081, + additions=[entry(4, 2081), entry(5, 2081)], + ) + + assert account.counts.opening == 0 + assert account.counts.entered == 2 + assert account.weights.additions_total == 5.0 + assert account.weights.closing == 5.0 + assert account.weight_residual == 0.0 + + +def test_a_quiet_period_may_precede_one_with_additions(): + """Chaining periods: the closing frame becomes the next opening.""" + first_year = frame(2090, {1: 10.0, 2: 20.0}) + second_year = frame(2091, {1: 10.0, 2: 20.0}) + third_year = frame(2092, {1: 10.0, 2: 20.0, 30: 5.0}) + + quiet = reconcile_period( + first_year, + second_year, + opening_year=2090, + closing_year=2091, + ) + active = reconcile_period( + second_year, + third_year, + opening_year=2091, + closing_year=2092, + additions=[entry(30, 2092)], + ) + + assert quiet.counts.additions_total == 0 + assert active.counts.additions_total == 1 + assert quiet.weights.closing == active.weights.opening == 30.0 + assert active.weights.closing == 35.0 + assert active.weight_residual == 0.0 + + +def test_zero_weight_rows_stay_visible_as_people(): + """A zero weight is a person, not an absence. + + Three rows open, one of them weighing nothing. The zero-weight + person then dies. Counts move by one; the stock does not move. + """ + account = reconcile_period( + frame(2100, {1: 0.0, 2: 5.0, 3: 7.0}), + frame(2101, {2: 5.0, 3: 7.0}), + opening_year=2100, + closing_year=2101, + exits=[death(1, 2101)], + ) + + assert account.counts.opening == 3 + assert account.counts.closing == 2 + assert account.counts.exited == 1 + assert account.exited_person_ids == (1,) + assert account.provenance["opening_zero_weight_rows"] == 1 + assert account.provenance["closing_zero_weight_rows"] == 0 + assert account.weights.exits_total == 0.0 + assert account.weights.opening == 12.0 + assert account.weights.closing == 12.0 + assert account.weight_residual == 0.0 + + +def test_a_zero_weight_arrival_is_still_an_arrival(): + """Arriving with no weight still books a person.""" + account = reconcile_period( + frame(2100, {1: 5.0}), + frame(2101, {1: 5.0, 2: 0.0}), + opening_year=2100, + closing_year=2101, + additions=[birth(2, 2101)], + ) + + assert account.counts.entered == 1 + assert account.counts.closing == 2 + assert account.weights.additions_total == 0.0 + assert account.provenance["closing_zero_weight_rows"] == 1 + + +# --------------------------------------------------------------------- +# stable summation and residual reporting +# --------------------------------------------------------------------- + + +def test_stable_summation_preserves_small_terms_in_this_example(): + """The stock is the exactly-rounded sum, which naive addition misses. + + ``1.0 + 1e16 + 1.0`` accumulated left to right loses both ones, and + 2.0 is representable at that magnitude, so the exactly-rounded + answer is 10000000000000002.0. + """ + weights = [1.0, 1e16, 1.0] + naive = 0.0 + for value in weights: + naive += value + assert naive == 1e16 + + rows = dict(zip((1, 2, 3), weights, strict=True)) + account = reconcile_period( + frame(2110, rows), + frame(2111, rows), + opening_year=2110, + closing_year=2111, + ) + + assert account.weights.opening == 10000000000000002.0 + assert account.weights.opening != naive + assert account.weight_residual == 0.0 + + +def test_totals_do_not_depend_on_row_order(): + """Reordering rows and declarations changes no reported number.""" + rows = {1: 0.1, 2: 0.2, 3: 0.3, 4: 1e15, 5: 0.7} + closing_rows = {1: 0.1, 2: 0.2, 4: 1e15, 5: 0.7} + forward = reconcile_period( + frame(2120, rows), + frame(2121, closing_rows), + opening_year=2120, + closing_year=2121, + exits=[death(3, 2121)], + ) + reversed_rows = dict(reversed(list(rows.items()))) + reversed_closing = dict(reversed(list(closing_rows.items()))) + backward = reconcile_period( + frame(2120, reversed_rows), + frame(2121, reversed_closing), + opening_year=2120, + closing_year=2121, + exits=[death(3, 2121)], + ) + + assert forward.to_dict() == backward.to_dict() + + +def test_a_representable_residual_is_reported_and_not_gated(): + """The residual is a reported number, never a pass/fail verdict. + + Nothing in the module compares it to a tolerance, so the field is + simply present and finite on a reconciled account. + """ + account = reconcile_period( + frame(2130, {1: 0.1, 2: 0.2}), + frame(2131, {1: 0.1}), + opening_year=2130, + closing_year=2131, + exits=[death(2, 2131)], + ) + + assert isinstance(account.weight_residual, float) + assert math.isfinite(account.weight_residual) + assert account.weight_residual != 0.0 + assert account.status == ENGINEERING_STATUS + assert account.count_residual == 0 + + +def test_the_module_defines_no_tolerance_or_acceptance_knob(): + """No threshold, tolerance or gate may creep into this interface.""" + banned = ("toler", "threshold", "atol", "rtol", "epsilon", "accept") + offenders = [ + name + for name in dir(accounting) + if not name.startswith("_") + and any(token in name.lower() for token in banned) + ] + assert offenders == [] + assert not any( + any(token in name.lower() for token in banned) + for name in signature(reconcile_period).parameters + ) + + +# --------------------------------------------------------------------- +# refusal to infer +# --------------------------------------------------------------------- + + +def test_a_disappearance_is_not_assumed_to_be_a_death(): + """An undeclared disappearance is refused, not booked as mortality.""" + with pytest.raises(PopulationReconciliationError) as caught: + reconcile_period( + frame(2140, {1: 10.0, 2: 20.0}), + frame(2141, {1: 10.0}), + opening_year=2140, + closing_year=2141, + ) + + assert kinds(caught.value) == [DiscrepancyKind.UNDECLARED_EXIT] + assert caught.value.discrepancies[0].person_id == 2 + assert "will not assume the person died" in ( + caught.value.discrepancies[0].detail + ) + + +def test_a_new_identifier_is_not_assumed_to_be_an_immigrant(): + """A new identifier flagged synthetic is still refused. + + The closing frame marks the row ``synthetic_entry`` exactly as + :func:`populace_dynamics.engine.steps.materialize_maternal_births` + would. The accountant ignores the flag: only a declaration counts. + """ + closing = frame(2141, {1: 10.0, 999: 3.0}, synthetic_entry=[False, True]) + with pytest.raises(PopulationReconciliationError) as caught: + reconcile_period( + frame(2140, {1: 10.0}), + closing, + opening_year=2140, + closing_year=2141, + ) + + assert kinds(caught.value) == [DiscrepancyKind.UNDECLARED_ADDITION] + assert caught.value.discrepancies[0].person_id == 999 + + +def test_both_unexplained_directions_are_reported_together(): + """One refusal carries every finding, ordered deterministically.""" + with pytest.raises(PopulationReconciliationError) as caught: + reconcile_period( + frame(2140, {1: 10.0, 2: 20.0}), + frame(2141, {1: 10.0, 3: 30.0}), + opening_year=2140, + closing_year=2141, + ) + + assert kinds(caught.value) == [ + DiscrepancyKind.UNDECLARED_ADDITION, + DiscrepancyKind.UNDECLARED_EXIT, + ] + assert [item.person_id for item in caught.value.discrepancies] == [3, 2] + assert "undeclared_addition=1" in str(caught.value) + assert "undeclared_exit=1" in str(caught.value) + + +def test_an_explicit_other_exit_is_accepted_when_the_caller_says_why(): + """The escape hatch is declaration, never inference.""" + account = reconcile_period( + frame(2140, {1: 10.0, 2: 20.0}), + frame(2141, {1: 10.0}), + opening_year=2140, + closing_year=2141, + exits=[ + PopulationEvent( + 2, + "other_exit", + 2141, + reason="removed by the caller's own roster surgery", + ) + ], + ) + + assert account.counts.exits_by_kind["other_exit"] == 1 + assert account.weights.exits_by_kind["other_exit"] == 20.0 + assert account.weight_residual == 0.0 + + +def test_refusal_payload_is_serializable(): + """A refusal can be written down without parsing its message.""" + with pytest.raises(PopulationReconciliationError) as caught: + reconcile_period( + frame(2140, {1: 10.0}), + frame(2141, {2: 10.0}), + opening_year=2140, + closing_year=2141, + ) + + payload = json.loads(json.dumps(caught.value.to_dict())) + assert payload["error"] == "population_reconciliation_error" + assert payload["interface_version"] == ACCOUNTING_INTERFACE_VERSION + assert {item["kind"] for item in payload["discrepancies"]} == { + "undeclared_addition", + "undeclared_exit", + } + + +# --------------------------------------------------------------------- +# rejecting colliding, duplicated and impossible declarations +# --------------------------------------------------------------------- + + +def test_duplicate_addition_declarations_are_rejected(): + with pytest.raises(PopulationReconciliationError) as caught: + reconcile_period( + frame(2150, {1: 10.0}), + frame(2151, {1: 10.0, 2: 5.0}), + opening_year=2150, + closing_year=2151, + additions=[birth(2, 2151), entry(2, 2151)], + ) + + assert DiscrepancyKind.DUPLICATE_ADDITION in kinds(caught.value) + + +def test_duplicate_exit_declarations_are_rejected(): + with pytest.raises(PopulationReconciliationError) as caught: + reconcile_period( + frame(2150, {1: 10.0, 2: 5.0}), + frame(2151, {1: 10.0}), + opening_year=2150, + closing_year=2151, + exits=[ + death(2, 2151), + PopulationEvent(2, "emigration", 2151), + ], + ) + + assert DiscrepancyKind.DUPLICATE_EXIT in kinds(caught.value) + + +def test_an_addition_colliding_with_the_opening_roster_is_rejected(): + """Declaring an arrival for somebody already present is a collision.""" + with pytest.raises(PopulationReconciliationError) as caught: + reconcile_period( + frame(2150, {1: 10.0, 2: 5.0}), + frame(2151, {1: 10.0, 2: 5.0}), + opening_year=2150, + closing_year=2151, + additions=[entry(2, 2151)], + ) + + assert kinds(caught.value) == [ + DiscrepancyKind.ADDITION_COLLIDES_WITH_OPENING + ] + + +def test_an_exit_for_someone_never_present_is_rejected(): + """Leaving requires having been here: no exit before entry.""" + with pytest.raises(PopulationReconciliationError) as caught: + reconcile_period( + frame(2150, {1: 10.0}), + frame(2151, {1: 10.0}), + opening_year=2150, + closing_year=2151, + exits=[death(404, 2151)], + ) + + assert kinds(caught.value) == [DiscrepancyKind.EXIT_WITHOUT_PRESENCE] + assert caught.value.discrepancies[0].person_id == 404 + + +def test_an_exit_contradicted_by_the_closing_frame_is_rejected(): + """Declaring a death for somebody still on the roster is refused.""" + with pytest.raises(PopulationReconciliationError) as caught: + reconcile_period( + frame(2150, {1: 10.0, 2: 5.0}), + frame(2151, {1: 10.0, 2: 5.0}), + opening_year=2150, + closing_year=2151, + exits=[death(2, 2151)], + ) + + assert kinds(caught.value) == [ + DiscrepancyKind.EXIT_CONTRADICTED_BY_CLOSING + ] + + +def test_an_arrival_that_never_lands_is_rejected(): + """A declared arrival absent at the close needs a declared exit.""" + with pytest.raises(PopulationReconciliationError) as caught: + reconcile_period( + frame(2150, {1: 10.0}), + frame(2151, {1: 10.0}), + opening_year=2150, + closing_year=2151, + additions=[birth(77, 2151, weight=1.0)], + ) + + assert kinds(caught.value) == [DiscrepancyKind.ADDITION_ABSENT_AT_CLOSE] + assert caught.value.discrepancies[0].person_id == 77 + + +def test_a_transient_without_declared_weights_is_rejected(): + """A person in neither frame must price both of their own events.""" + with pytest.raises(PopulationAccountingInputError, match="explicit"): + reconcile_period( + frame(2150, {1: 10.0}), + frame(2151, {1: 10.0}), + opening_year=2150, + closing_year=2151, + additions=[entry(7, 2151)], + exits=[death(7, 2151, weight=1.0)], + ) + + +# --------------------------------------------------------------------- +# malformed declarations +# --------------------------------------------------------------------- + + +def test_an_unknown_event_kind_is_rejected(): + with pytest.raises(PopulationAccountingInputError, match="unknown"): + PopulationEvent(1, "abduction", 2020) + + +def test_an_other_kind_without_a_reason_is_rejected(): + with pytest.raises(PopulationAccountingInputError, match="reason"): + PopulationEvent(1, "other_exit", 2020) + with pytest.raises(PopulationAccountingInputError, match="reason"): + PopulationEvent(1, "other_entry", 2020, reason=" ") + + +def test_a_boolean_person_identifier_is_rejected(): + with pytest.raises(PopulationAccountingInputError, match="integer"): + PopulationEvent(True, "death", 2020) + + +def test_a_float_person_identifier_is_rejected(): + with pytest.raises(PopulationAccountingInputError, match="integer"): + PopulationEvent(1.0, "death", 2020) + + +@pytest.mark.parametrize("value", [-1.0, float("nan"), float("inf")]) +def test_an_invalid_declared_weight_is_rejected(value): + with pytest.raises(PopulationAccountingInputError, match="non-negative"): + PopulationEvent(1, "death", 2020, weight=value) + + +def test_a_non_string_reason_is_rejected(): + with pytest.raises(PopulationAccountingInputError, match="string"): + PopulationEvent(1, "death", 2020, reason=7) + + +def test_a_departure_declared_as_an_arrival_is_rejected(): + with pytest.raises(PopulationAccountingInputError, match="separately"): + reconcile_period( + frame(2160, {1: 10.0}), + frame(2161, {1: 10.0}), + opening_year=2160, + closing_year=2161, + additions=[death(1, 2161)], + ) + + +def test_an_arrival_declared_as_a_departure_is_rejected(): + with pytest.raises(PopulationAccountingInputError, match="separately"): + reconcile_period( + frame(2160, {1: 10.0}), + frame(2161, {1: 10.0}), + opening_year=2160, + closing_year=2161, + exits=[birth(1, 2161)], + ) + + +def test_an_event_booked_to_the_wrong_year_is_rejected(): + with pytest.raises(PopulationAccountingInputError, match="closing year"): + reconcile_period( + frame(2160, {1: 10.0, 2: 1.0}), + frame(2161, {1: 10.0}), + opening_year=2160, + closing_year=2161, + exits=[death(2, 2160)], + ) + + +def test_a_non_event_in_a_declaration_sequence_is_rejected(): + with pytest.raises( + PopulationAccountingInputError, match="PopulationEvent" + ): + reconcile_period( + frame(2160, {1: 10.0}), + frame(2161, {1: 10.0}), + opening_year=2160, + closing_year=2161, + exits=[{"person_id": 1, "kind": "death"}], + ) + + +def test_a_generator_of_declarations_is_rejected(): + """Declarations must be a re-readable sequence, not a one-shot stream.""" + with pytest.raises(PopulationAccountingInputError, match="sequence"): + reconcile_period( + frame(2160, {1: 10.0}), + frame(2161, {1: 10.0}), + opening_year=2160, + closing_year=2161, + exits=(event for event in ()), + ) + + +# --------------------------------------------------------------------- +# malformed frames and period coordinates +# --------------------------------------------------------------------- + + +def test_a_missing_column_is_rejected(): + opening = frame(2170, {1: 10.0}).drop(columns=["weight"]) + with pytest.raises(PopulationAccountingInputError, match="missing"): + reconcile_period( + opening, + frame(2171, {1: 10.0}), + opening_year=2170, + closing_year=2171, + ) + + +def test_a_duplicate_person_row_is_rejected(): + opening = pd.DataFrame( + { + "person_id": [1, 1], + "year": [2170, 2170], + "weight": [10.0, 10.0], + } + ) + with pytest.raises(PopulationAccountingInputError, match="duplicate"): + reconcile_period( + opening, + frame(2171, {1: 10.0}), + opening_year=2170, + closing_year=2171, + ) + + +def test_a_float_identifier_column_is_rejected(): + opening = frame(2170, {1: 10.0}) + opening["person_id"] = opening["person_id"].astype(np.float64) + with pytest.raises(PopulationAccountingInputError, match="integer dtype"): + reconcile_period( + opening, + frame(2171, {1: 10.0}), + opening_year=2170, + closing_year=2171, + ) + + +def test_a_null_identifier_is_rejected(): + opening = frame(2170, {1: 10.0}) + opening["person_id"] = pd.array([pd.NA], dtype="Int64") + with pytest.raises(PopulationAccountingInputError, match="null"): + reconcile_period( + opening, + frame(2171, {1: 10.0}), + opening_year=2170, + closing_year=2171, + ) + + +@pytest.mark.parametrize( + ("value", "message"), + [ + (float("nan"), "null"), + (float("inf"), "non-finite"), + (-0.5, "negative"), + ], +) +def test_an_invalid_frame_weight_is_rejected(value, message): + closing = frame(2171, {1: value}) + with pytest.raises(PopulationAccountingInputError, match=message): + reconcile_period( + frame(2170, {1: 10.0}), + closing, + opening_year=2170, + closing_year=2171, + ) + + +def test_a_row_carrying_the_wrong_year_is_rejected(): + closing = frame(2171, {1: 10.0, 2: 5.0}) + closing.loc[1, "year"] = 2172 + with pytest.raises(PopulationAccountingInputError, match="must carry"): + reconcile_period( + frame(2170, {1: 10.0, 2: 5.0}), + closing, + opening_year=2170, + closing_year=2171, + ) + + +def test_a_non_annual_period_is_rejected(): + with pytest.raises(PopulationAccountingInputError, match="annual"): + reconcile_period( + frame(2170, {1: 10.0}), + frame(2172, {1: 10.0}), + opening_year=2170, + closing_year=2172, + ) + + +def test_a_non_frame_input_is_rejected(): + with pytest.raises(PopulationAccountingInputError, match="DataFrame"): + reconcile_period( + {"person_id": [1]}, + frame(2171, {1: 10.0}), + opening_year=2170, + closing_year=2171, + ) + + +def test_a_float_year_coordinate_is_rejected(): + with pytest.raises(PopulationAccountingInputError, match="integer year"): + reconcile_period( + frame(2170, {1: 10.0}), + frame(2171, {1: 10.0}), + opening_year=2170.0, + closing_year=2171, + ) + + +# --------------------------------------------------------------------- +# purity, immutability and the serializable payload +# --------------------------------------------------------------------- + + +def test_input_frames_are_not_mutated(): + """The accountant reads; it never writes back. + + The rows are supplied deliberately out of identifier order and with + a non-default index, so an in-place sort, a reindex or an added + bookkeeping column would all show up here. + """ + opening = frame(2180, {2: 20.0, 1: 10.0}, note=["b", "a"]) + closing = frame(2181, {3: 1.0, 2: 25.0}, note=["c", "b"]) + opening.index = pd.Index([11, 10], name="row") + closing.index = pd.Index([13, 12], name="row") + opening_before = opening.copy(deep=True) + closing_before = closing.copy(deep=True) + + reconcile_period( + opening, + closing, + opening_year=2180, + closing_year=2181, + additions=[birth(3, 2181)], + exits=[death(1, 2181)], + ) + + pd.testing.assert_frame_equal(opening, opening_before) + pd.testing.assert_frame_equal(closing, closing_before) + assert list(opening["person_id"]) == [2, 1] + assert list(closing["person_id"]) == [3, 2] + assert list(opening.columns) == ["person_id", "year", "weight", "note"] + assert list(closing.columns) == ["person_id", "year", "weight", "note"] + assert list(opening.index) == [11, 10] + assert list(closing.index) == [13, 12] + + +def test_the_account_is_immutable(): + account = reconcile_period( + frame(2180, {1: 10.0}), + frame(2181, {1: 10.0}), + opening_year=2180, + closing_year=2181, + ) + + with pytest.raises(FrozenInstanceError): + account.weight_residual = 1.0 + with pytest.raises(TypeError): + account.provenance["opening_rows"] = 99 + with pytest.raises(TypeError): + account.counts.additions_by_kind["birth"] = 99 + + +def test_the_account_payload_is_json_serializable_and_flat(): + account = reconcile_period( + frame(2180, {1: 10.0, 2: 20.0}), + frame(2181, {1: 10.0, 3: 4.0}), + opening_year=2180, + closing_year=2181, + additions=[birth(3, 2181, source="synthetic.fertility")], + exits=[death(2, 2181, source="synthetic.mortality")], + ) + + payload = account.to_dict() + round_tripped = json.loads(json.dumps(payload)) + assert round_tripped == payload + + def leaves(value): + if isinstance(value, dict): + for item in value.values(): + yield from leaves(item) + elif isinstance(value, list): + for item in value: + yield from leaves(item) + else: + yield value + + assert all( + isinstance(leaf, (int, float, str, bool)) or leaf is None + for leaf in leaves(payload) + ) + assert payload["provenance"]["declaration_sources"] == [ + "synthetic.fertility", + "synthetic.mortality", + ] + assert payload["provenance"]["inference"].startswith("none:") + + +def test_the_status_is_explicitly_engineering_only(): + account = reconcile_period( + frame(2180, {1: 10.0}), + frame(2181, {1: 10.0}), + opening_year=2180, + closing_year=2181, + ) + + assert account.status == ENGINEERING_STATUS + assert account.status == "engineering-accounting-coherence-only" + note = account.status_note.lower() + assert "not scientific acceptance" in note + assert "not a benchmark comparison" in note + assert "not a gate outcome" in note + + +def test_a_discrepancy_record_is_serializable(): + record = AccountingDiscrepancy( + kind=DiscrepancyKind.UNDECLARED_EXIT, + person_id=5, + detail="example", + ) + assert json.loads(json.dumps(record.to_dict())) == { + "kind": "undeclared_exit", + "person_id": 5, + "detail": "example", + } + + +def test_an_event_record_is_serializable(): + event = birth(3, 2181, source="synthetic.fertility") + assert json.loads(json.dumps(event.to_dict())) == { + "person_id": 3, + "kind": "birth", + "year": 2181, + "weight": None, + "reason": "", + "source": "synthetic.fertility", + } + assert event.is_addition is True + assert death(3, 2181).is_addition is False + assert set(PopulationEventKind) == ( + accounting.ADDITION_KINDS | accounting.EXIT_KINDS + ) + + +# --------------------------------------------------------------------- +# the real ProjectionEngine, driven by recording synthetic adapters +# --------------------------------------------------------------------- + + +@dataclass +class EventLog: + """Declarations captured from the synthetic adapters themselves.""" + + additions: list[PopulationEvent] = field(default_factory=list) + exits: list[PopulationEvent] = field(default_factory=list) + + def for_year(self, year: int) -> tuple[list, list]: + return ( + [item for item in self.additions if item.year == year], + [item for item in self.exits if item.year == year], + ) + + +def _recording_modules( + log: EventLog, + deaths_by_year: dict[int, tuple[int, ...]], + births_by_year: dict[int, tuple[int, ...]], +) -> PeriodModules: + """Build eight adapters that record every presence change they make. + + These are deliberately trivial: no fitted component, no draw, no + demography. Their only job is to move people in and out of the + roster through the engine's real seams and write down what they + did. + """ + + def mortality(current, context, rng): + del rng + doomed = set(deaths_by_year.get(context.year, ())) + leaving = current["person_id"].isin(doomed) + for row in current.loc[leaving].to_dict("records"): + log.exits.append( + PopulationEvent( + person_id=int(row["person_id"]), + kind=PopulationEventKind.DEATH, + year=context.year, + weight=float(row["weight"]), + source="synthetic.mortality", + ) + ) + return current.loc[~leaving].reset_index(drop=True) + + def aging(current, context, rng): + del rng + out = current.copy() + out["year"] = context.year + out["age"] = out["age"].to_numpy(dtype=np.int64) + 1 + return out + + def marital_core(current, context, rng): + del current, context, rng + return MaritalStepResult( + sim_years=pd.DataFrame(), births=pd.DataFrame() + ) + + def fertility(current, context, marital, rng): + del marital, rng + parents = births_by_year.get(context.year, ()) + if not parents: + return current + weight_of = dict( + zip( + current["person_id"].tolist(), + current["weight"].tolist(), + strict=True, + ) + ) + child_ids = context.synthetic_id_allocator.allocate(len(parents)) + children = pd.DataFrame( + { + "person_id": child_ids, + "year": np.full(len(parents), context.year, dtype=np.int64), + "age": np.zeros(len(parents), dtype=np.int64), + "weight": np.asarray( + [weight_of[parent] for parent in parents], + dtype=np.float64, + ), + } + ) + for child_id in child_ids.tolist(): + log.additions.append( + PopulationEvent( + person_id=int(child_id), + kind=PopulationEventKind.BIRTH, + year=context.year, + source="synthetic.fertility", + ) + ) + return pd.concat([current, children], ignore_index=True) + + def unchanged(current, context, rng): + del context, rng + return current + + def unchanged_reader(current, context, marital, rng): + del context, marital, rng + return current + + return PeriodModules( + mortality=mortality, + aging=aging, + marital_core=marital_core, + fertility=fertility, + disability=unchanged, + earnings=unchanged, + claiming=unchanged, + household_composition=unchanged_reader, + ) + + +def _run_projection() -> tuple[object, EventLog]: + """Project 2020-2023 with a birth, a death and a transient entrant. + + The 2022 scheduled entrant is registered on the metadata seam the + loop reads, joins the roster before mortality, and dies in the same + wave -- so they appear in no projected slice at all. + """ + log = EventLog() + initial = frame(2020, {1: 10.0, 2: 20.0, 3: 30.0}, age=[40, 41, 42]) + entrants_2022 = frame(2021, {100: 7.0}, age=[50]) + schedule = {2022: entrants_2022} + for year, entrant_frame in schedule.items(): + for row in entrant_frame.to_dict("records"): + log.additions.append( + PopulationEvent( + person_id=int(row["person_id"]), + kind=PopulationEventKind.SCHEDULED_ENTRY, + year=year, + weight=float(row["weight"]), + source="loop.m6_scheduled_entries_by_year", + ) + ) + engine = ProjectionEngine( + _recording_modules( + log, + deaths_by_year={2021: (3,), 2022: (100,)}, + births_by_year={2021: (1,)}, + ) + ) + result = engine.project( + initial, + end_year=2023, + draw_index=0, + metadata={SCHEDULED_ENTRIES_KEY: schedule}, + ) + return result, log + + +def test_projection_slices_reconcile_year_by_year(): + """Every wave of a real projection balances against its records. + + The projection is small enough to state outright: 2021 loses person + 3 (weight 30) and gains one child carrying the mother's weight 10, + so the stock goes 60 -> 40. 2022 admits and then buries person 100 + (weight 7), leaving the stock at 40. 2023 does nothing. + """ + result, log = _run_projection() + assert [int(slice_.iloc[0]["year"]) for slice_ in result.slices] == [ + 2020, + 2021, + 2022, + 2023, + ] + + accounts = [] + for index in range(len(result.slices) - 1): + opening_year = 2020 + index + additions, exits = log.for_year(opening_year + 1) + accounts.append( + reconcile_period( + result.slices[index], + result.slices[index + 1], + opening_year=opening_year, + closing_year=opening_year + 1, + additions=additions, + exits=exits, + ) + ) + + first, second, third = accounts + + assert first.counts.opening == 3 + assert first.counts.closing == 3 + assert first.counts.entered == 1 + assert first.counts.exited == 1 + assert first.counts.transient == 0 + assert first.counts.additions_by_kind["birth"] == 1 + assert first.counts.exits_by_kind["death"] == 1 + assert first.weights.opening == 60.0 + assert first.weights.closing == 40.0 + assert first.weights.additions_total == 10.0 + assert first.weights.exits_total == 30.0 + assert first.weights.revaluation.total == 0.0 + assert first.exited_person_ids == (3,) + + assert second.counts.opening == 3 + assert second.counts.closing == 3 + assert second.counts.carried == 3 + assert second.counts.transient == 1 + assert second.transient_person_ids == (100,) + assert second.counts.additions_by_kind["scheduled_entry"] == 1 + assert second.counts.exits_by_kind["death"] == 1 + assert second.weights.opening == 40.0 + assert second.weights.closing == 40.0 + assert second.weights.additions_total == 7.0 + assert second.weights.exits_total == 7.0 + assert second.weights.revaluation.total == 0.0 + + assert third.counts.carried == 3 + assert third.counts.additions_total == 0 + assert third.counts.exits_total == 0 + assert third.weights.closing == 40.0 + + assert [account.count_residual for account in accounts] == [0, 0, 0] + assert [account.weight_residual for account in accounts] == [ + 0.0, + 0.0, + 0.0, + ] + assert {account.status for account in accounts} == {ENGINEERING_STATUS} + + +def test_the_transient_entrant_never_appears_in_a_projected_slice(): + """The within-period arrival is invisible to the frames alone. + + Only the captured declarations show that person 100 was ever in the + population. Endpoint frames cannot expose omission of both events; + the accountant does not verify event-log completeness. + """ + result, log = _run_projection() + for slice_ in result.slices: + assert 100 not in set(slice_["person_id"].tolist()) + assert [item.person_id for item in log.additions if item.year == 2022] == [ + 100 + ] + assert [item.person_id for item in log.exits if item.year == 2022] == [100] + + +def test_dropping_one_captured_death_makes_the_projection_refuse(): + """Omitting a real event is caught against real engine output.""" + result, log = _run_projection() + additions, exits = log.for_year(2021) + withheld = [item for item in exits if item.person_id != 3] + assert len(withheld) == len(exits) - 1 + + with pytest.raises(PopulationReconciliationError) as caught: + reconcile_period( + result.slices[0], + result.slices[1], + opening_year=2020, + closing_year=2021, + additions=additions, + exits=withheld, + ) + + assert kinds(caught.value) == [DiscrepancyKind.UNDECLARED_EXIT] + assert caught.value.discrepancies[0].person_id == 3 + + +def test_dropping_one_captured_birth_makes_the_projection_refuse(): + """A synthetic identifier is never quietly accepted as a birth.""" + result, log = _run_projection() + _, exits = log.for_year(2021) + + with pytest.raises(PopulationReconciliationError) as caught: + reconcile_period( + result.slices[0], + result.slices[1], + opening_year=2020, + closing_year=2021, + additions=[], + exits=exits, + ) + + assert kinds(caught.value) == [DiscrepancyKind.UNDECLARED_ADDITION] + + +# Root regression cases to append after builder releases source ownership. + + +@pytest.mark.parametrize("column", ["person_id", "year"]) +@pytest.mark.parametrize( + "dtype,value", + [(np.uint64, 2**64 - 1), (object, 2**63), (object, -(2**63) - 1)], +) +def test_frame_integer_domain_cannot_wrap(column, dtype, value): + opening = frame(2020, {1: 1.0}) + opening[column] = pd.Series([value], dtype=dtype) + with pytest.raises(PopulationAccountingInputError): + reconcile_period( + opening, + frame(2021, {-1: 1.0}), + opening_year=2020, + closing_year=2021, + ) + + +@pytest.mark.parametrize("field_name", ["person_id", "year"]) +@pytest.mark.parametrize("value", [2**63, -(2**63) - 1, np.uint64(2**64 - 1)]) +def test_event_integer_domain_matches_frames(field_name, value): + kwargs = {"person_id": 1, "kind": "death", "year": 2021} + kwargs[field_name] = value + with pytest.raises(PopulationAccountingInputError): + PopulationEvent(**kwargs) + + +@pytest.mark.parametrize( + "value,dtype", + [ + ("2.0", None), + (2 + 3j, None), + (True, object), + (np.bool_(True), object), + (1 + 0j, object), + (10**400, object), + ], +) +def test_frame_weights_refuse_lossy_or_non_numeric_values(value, dtype): + opening = frame(2020, {1: 1.0}) + opening["weight"] = pd.Series([value], dtype=dtype) + with pytest.raises(PopulationAccountingInputError): + reconcile_period( + opening, + frame(2021, {1: 1.0}), + opening_year=2020, + closing_year=2021, + ) + + +def test_unrepresentable_weight_total_has_typed_refusal(): + with pytest.raises( + PopulationAccountingInputError, match="represent|overflow|finite" + ): + reconcile_period( + frame(2020, {1: 1e308, 2: 1e308}), + frame(2021, {1: 1e308, 2: 1e308}), + opening_year=2020, + closing_year=2021, + ) + + +def test_unrepresentable_event_weight_has_typed_refusal(): + with pytest.raises(PopulationAccountingInputError): + PopulationEvent(1, "death", 2021, weight=10**400) + + +def test_serialized_provenance_cannot_mutate_original_or_other_payload(): + account = reconcile_period( + empty_frame(), + frame(2021, {1: 1.0}), + opening_year=2020, + closing_year=2021, + additions=[birth(1, 2021, source="synthetic.birth")], + ) + first, second = account.to_dict(), account.to_dict() + first["provenance"]["declaration_sources"].append("invented") + assert second["provenance"]["declaration_sources"] == ["synthetic.birth"] + assert account.to_dict()["provenance"]["declaration_sources"] == [ + "synthetic.birth" + ] + assert account.provenance["declaration_sources"] == ("synthetic.birth",) + + +def test_omitting_both_transient_events_is_not_observable_at_endpoints(): + result, log = _run_projection() + additions, exits = log.for_year(2022) + assert additions and exits + account = reconcile_period( + result.slices[1], + result.slices[2], + opening_year=2021, + closing_year=2022, + ) + assert account.counts.transient == 0 + assert account.count_residual == 0 + assert account.provenance["event_log_completeness_verified"] is False + + +@pytest.mark.parametrize("person_id", [-(2**63), 2**63 - 1]) +def test_signed_int64_boundary_ids_remain_exact(person_id): + account = reconcile_period( + frame(2020, {person_id: 1.0}), + empty_frame(), + opening_year=2020, + closing_year=2021, + exits=[death(person_id, 2021)], + ) + assert account.exited_person_ids == (person_id,) + assert account.weights.exits_total == 1.0 + + +def test_nullable_integer_columns_and_real_object_weights_are_supported(): + opening = pd.DataFrame( + { + "person_id": pd.Series([1], dtype="Int64"), + "year": pd.Series([2020], dtype="Int64"), + "weight": pd.Series([np.float64(2.5)], dtype=object), + } + ) + account = reconcile_period( + opening, + frame(2021, {1: 2.5}), + opening_year=2020, + closing_year=2021, + ) + assert account.carried_person_ids == (1,) + assert account.weights.opening == 2.5 + + +@pytest.mark.skipif( + np.finfo(np.longdouble).minexp >= np.finfo(np.float64).minexp, + reason="platform longdouble has no wider exponent range than binary64", +) +@pytest.mark.parametrize("sign", [-1, 1]) +@pytest.mark.parametrize("via", ["frame", "event"]) +def test_extended_weight_cannot_lose_sign_or_mass_on_conversion(sign, via): + weight = sign * np.nextafter(np.longdouble(0), np.longdouble(1)) + assert weight != 0 + assert float(weight) == 0.0 + with pytest.raises(PopulationAccountingInputError): + if via == "event": + death(1, 2021, weight=weight) + else: + opening = frame(2020, {1: 1.0}) + opening["weight"] = np.array([weight], dtype=np.longdouble) + reconcile_period( + opening, + frame(2021, {1: 0.0}), + opening_year=2020, + closing_year=2021, + ) diff --git a/tests/tier_counts.json b/tests/tier_counts.json index c1ba14b2..c383a179 100644 --- a/tests/tier_counts.json +++ b/tests/tier_counts.json @@ -1,8 +1,8 @@ { "schema_version": 1, "counts": { - "unit": 1624, - "artifact": 2668, + "unit": 1717, + "artifact": 2672, "integration_psid": 848, "reproduction_legacy": 520, "oracle_policyengine": 159 From 651fcbfde009c29351911db4ff01d64b107156ec Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Wed, 9 Sep 2026 18:50:01 -0400 Subject: [PATCH 2/2] Account for annual mortality trajectory stock and flows --- .github/workflows/tests.yml | 2 +- docs/trajectory-accounting.md | 89 +++ scripts/first_estimates_birth_evidence.py | 1 + .../graph/trajectory_accounting.py | 284 ++++++++ tests/README-tiers.md | 4 +- .../estimates/test_birth_evidence_artifact.py | 2 + tests/test_graph_trajectory_accounting.py | 641 ++++++++++++++++++ tests/tier_counts.json | 2 +- 8 files changed, 1021 insertions(+), 4 deletions(-) create mode 100644 docs/trajectory-accounting.md create mode 100644 src/populace_dynamics/graph/trajectory_accounting.py create mode 100644 tests/test_graph_trajectory_accounting.py diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 9d308c2d..ccd2d0e8 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -52,7 +52,7 @@ jobs: - name: Require typed graph capabilities run: python -c "from populace_dynamics.graph._compat import require_graph; require_graph()" - name: Run graph and existing mortality regressions - run: pytest -q tests/test_graph_mortality.py tests/test_graph_mortality_trajectory.py tests/test_m6_engine_refit.py tests/test_m6_engine_steps.py + run: pytest -q tests/test_graph_mortality.py tests/test_graph_mortality_trajectory.py tests/test_graph_trajectory_accounting.py tests/test_m6_engine_refit.py tests/test_m6_engine_steps.py # Fan-in jobs keeping the branch-protection context names # ("pytest (3.11)" / "pytest (3.13)") stable across the shard split. diff --git a/docs/trajectory-accounting.md b/docs/trajectory-accounting.md new file mode 100644 index 00000000..16dc8ded --- /dev/null +++ b/docs/trajectory-accounting.md @@ -0,0 +1,89 @@ +# Accounting for annual mortality transitions + +The optional `populace_dynamics.graph.trajectory_accounting` module adds one +accounting node per year to the existing synthetic mortality/ageing graph. +It executes the real Microcosm DAG with the same reviewed graph/Frame pin. +It does not change fitted laws, draws, population frames, weights, scientific +gates, or the historical projection engine. + +```python +from populace_dynamics.graph.trajectory_accounting import ( + run_accounted_mortality_trajectory, +) + +result = run_accounted_mortality_trajectory( + training="synthetic-training.json", + rates="synthetic-rates.json", + initial="synthetic-initial.json", + holdouts={2015: "synthetic-annual-2015.json"}, + end_year=2015, + output_dir="synthetic-accounting-output", +) +``` + +These paths must contain the exact synthetic contracts documented in +[the population graph and mortality trajectory](population-graph.md). +The optional module +requires Python 3.13 or later and the reviewed graph dependencies. Importing +the ordinary `populace_dynamics.graph` package remains lazy and unchanged. +`build_accounted_trajectory_graph` also exposes the declaration and kernel +registry for callers using the executor directly. + +## Declared events and frozen snapshots + +Each `account_YEAR` node reads only the corresponding typed mortality +transition and frozen population snapshot. It runs against the graph's +separate training population version without reading that population's +columns. It has no holdout, model, source, or RNG input. Accounting is not a +prerequisite of any subsequent mortality transition. + +The adapter validates the transition's person and observation bindings and +the snapshot's year, row, period, and weight structure. It copies the opening +and closing rows, preserving the supported columns and weight positions. +Only transition records declaring `survives=false` produce death events. +Missing or additional endpoint people cannot supply their own event causes. +No births, migration, or other entry/exit events are assumed. + +Completed transitions call the +[annual stock-flow accountant](stock-flow-accounting.md). Count conservation +is exact; weight residuals and revaluation components are reported without +an acceptance threshold. A changed survivor weight can therefore produce a +complete account while the original mortality evaluation independently +fails its unchanged-weight check. Accounting completion establishes neither +scientific acceptance nor completeness or truth of the declared event log. +The adapter cannot detect an arbitrary same-length rearrangement of supplied +weights without independent binding evidence; it accounts for the supplied, +content-addressed snapshot. + +After complete extinction, subsequent completed empty periods receive explicit +zero-to-zero accounts even though the population contains no new period +groups. A failed or blocked mortality transition instead produces +`account=null` and `accounting_status=not_evaluated`, retaining its diagnostic +and last completed year. It never turns a stale snapshot into deaths. +Malformed inputs or reconciliation refusals produce a separate failed +accounting artifact; original mortality and evaluation receipts remain intact. + +The runner writes `accounting-report.json` and the actual `manifest.json`. +Its `AccountedTrajectoryRun` contains those accounting summaries and the +manifest. It does not rewrite the original runner's `report.json`, model, or +trajectory exports. The manifest retains original application/evaluation +receipts and artifact references; an accounting status is not their rollup. +Source hashes cover the accountant and the reused snapshot/transition helpers. + +## Household and location contract still required + +The current initial source accepts exactly `person_id`, `age`, `sex`, and +`weight`. Its snapshot accepts only the existing person-period identity, +age, sex, period, and weight structure. Additional atomic-location or +household-link columns are refused, not silently discarded. This integration +does not yet transport household location or membership. + +A separate extension must preserve household atomic-location columns **and +household/member links** in both source and snapshot contracts. Microcosm +assigns the household anchor once **before support clones are created**; +clones and Dynamics inherit it. Larger geographies must derive from that +anchor through the **same versioned mapping**. Location may change only +through a separate declared mobility or migration event. Accounting must not +allocate locations, create independent geography assignments, or infer a move +from a roster difference. That extension needs its own schema, lineage, +membership, mapping-version, and declared-event tests. diff --git a/scripts/first_estimates_birth_evidence.py b/scripts/first_estimates_birth_evidence.py index fec55d44..9de25745 100644 --- a/scripts/first_estimates_birth_evidence.py +++ b/scripts/first_estimates_birth_evidence.py @@ -166,6 +166,7 @@ Path("src/populace_dynamics/graph/runtime.py"), Path("src/populace_dynamics/graph/synthetic.py"), Path("src/populace_dynamics/graph/trajectory.py"), + Path("src/populace_dynamics/graph/trajectory_accounting.py"), # This opt-in accountant is unreachable from the historical projection. # The existing engine loop, steps, and package initializer remain sealed. Path("src/populace_dynamics/engine/accounting.py"), diff --git a/src/populace_dynamics/graph/trajectory_accounting.py b/src/populace_dynamics/graph/trajectory_accounting.py new file mode 100644 index 00000000..82b3f654 --- /dev/null +++ b/src/populace_dynamics/graph/trajectory_accounting.py @@ -0,0 +1,284 @@ +"""Optional accounting of the existing annual mortality graph's artifacts. + +No population or transition is changed. Death declarations come from the +typed mortality transition, never from the difference between two rosters. +Import this optional module only with the reviewed graph dependencies. +""" + +from __future__ import annotations + +from dataclasses import dataclass, replace +from pathlib import Path + +import pandas as pd +from microcosm.graph.decl import ( + ArtifactInput, + ArtifactOutput, + ArtifactType, + Node, + compile_graph, +) +from microcosm.graph.executor import run_graph +from microcosm.graph.kernel import KernelResult, source_hash +from microcosm.graph.store import ContentStore + +from populace_dynamics.engine import accounting + +from . import trajectory +from ._compat import require_graph +from .model import json_bytes, parse_json + +ACCOUNT_TYPE = ArtifactType("populace-dynamics.mortality-stock-flow", 1) +ACCOUNT_KERNEL = "dynamics.trajectory.stock-flow@1" + + +def _validate_raw_snapshot(payload): + """Apply the accountant's scalar domain before pandas can coerce values.""" + raw = parse_json(payload) + rt = trajectory.rt + if ( + not isinstance(raw, dict) + or not isinstance(raw.get("observations"), list) + or not isinstance(raw.get("periods"), list) + or not isinstance(raw.get("weights"), list) + or len(raw["observations"]) != len(raw["weights"]) + ): + raise ValueError("invalid accounting snapshot row/weight structure") + accounting._as_year(raw.get("year"), "snapshot.year") + for row in raw["observations"]: + if not isinstance(row, dict) or set(row) != { + rt.OID, + rt.PID, + rt.PERIOD_ID, + "age", + "sex", + }: + raise ValueError("invalid annual mortality snapshot row binding") + for column in (rt.OID, rt.PID, rt.PERIOD_ID, "age"): + accounting._as_person_id(row[column], f"snapshot.{column}") + for row in raw["periods"]: + if not isinstance(row, dict) or set(row) != {"period_id", "period"}: + raise ValueError("invalid accounting snapshot period structure") + for column in ("period_id", "period"): + accounting._as_year(row[column], f"snapshot.{column}") + for value in raw["weights"]: + accounting._as_weight(value, "snapshot.weight") + + +def _period_frames(context): + """Copy endpoint rows, retaining columns and positional weight binding.""" + rt = trajectory.rt + observations = context.tables[rt.OBS] + periods = rt._periods(context, observations) + calendar = context.tables["period"].period + if ( + not calendar.is_unique + or (calendar > context.params["year"]).any() + or (calendar < context.params["boundary_year"]).any() + or pd.DataFrame({"person_id": observations[rt.PID], "year": periods}) + .duplicated() + .any() + ): + raise ValueError("invalid accounting snapshot person-period history") + frames = [] + for year in (context.params["year"] - 1, context.params["year"]): + mask = periods == year + frame = observations.loc[mask].copy() + frame["person_id"] = frame[rt.PID].to_numpy(copy=True) + frame["year"] = periods.loc[mask].to_numpy(copy=True) + frame["weight"] = context.weights[rt.OBS].values[mask.to_numpy()] + frames.append(frame) + return frames + + +def _account(context): + year = context.params["year"] + report = { + "format": ACCOUNT_TYPE.name, + "schema_version": 1, + "scope": "synthetic_engineering", + "from_year": year - 1, + "year": year, + "application_status": None, + "completed_year": None, + "accounting_status": "failed", + "account": None, + "diagnostic": None, + "transition_node": f"apply_{year}", + "snapshot_node": f"snapshot_{year}", + } + try: + outcome = trajectory._decode_transition( + context.artifacts["transition"].payload, year + ) + if outcome["completed_year"] < context.params["boundary_year"]: + raise ValueError("transition predates the model boundary") + # JSON exponent overflow can produce infinity despite parse_constant. + # Refuse it before copying arbitrary diagnostic fields into a report. + json_bytes(outcome["diagnostic"]) + report.update( + application_status=outcome["status"], + completed_year=outcome["completed_year"], + ) + if outcome["status"] != "complete": + # Do not reconcile a stale snapshot or parse it after failure. + report.update( + accounting_status="not_evaluated", + diagnostic=outcome["diagnostic"], + ) + else: + _validate_raw_snapshot(context.artifacts["snapshot"].payload) + frozen = trajectory._evaluation_context(context) + _, outcome = trajectory._transition(frozen) + opening, closing = _period_frames(frozen) + exits = tuple( + accounting.PopulationEvent( + person_id=row["person_id"], + kind=accounting.PopulationEventKind.DEATH, + year=year, + source=f"apply_{year}:mortality-transition@1", + ) + for row in outcome["records"] + if not row["survives"] + ) + account = accounting.reconcile_period( + opening, + closing, + opening_year=year - 1, + closing_year=year, + exits=exits, + ) + report.update( + accounting_status="complete", account=account.to_dict() + ) + except Exception as error: + # Keep an accounting refusal inspectable without changing mortality + # gates or stopping another year's already declared transition. + report["diagnostic"] = { + "exception_type": type(error).__name__, + "message": str(error), + } + if isinstance(error, accounting.PopulationReconciliationError): + report["diagnostic"]["reconciliation"] = error.to_dict() + return KernelResult( + artifacts={"account": json_bytes(report)}, + receipt={ + "accounting_status": report["accounting_status"], + "application_status": report["application_status"], + "completed_year": report["completed_year"], + }, + ) + + +class _AccountingKernel(trajectory.rt._Kernel): + def implementation_hash(self): + return source_hash( + self.function, + accounting, + trajectory, + trajectory.rt, + trajectory.rt.model_module, + dependencies=self.capabilities.dependencies, + ) + + +def build_accounted_trajectory_graph(**kwargs): + """Append isolated accounting nodes to the unchanged trajectory graph.""" + require_graph() + graph, registry = trajectory.build_trajectory_graph(**kwargs) + registry.register(_AccountingKernel(ACCOUNT_KERNEL, _account)) + boundary = kwargs.get("boundary_year", 2014) + nodes = tuple( + Node( + f"account_{year}", + ACCOUNT_KERNEL, + # EXPAND consumes ordinary members of its base. This separate + # population keeps accounting outside all later transition keys. + population="training", + params={"year": year, "boundary_year": boundary}, + artifact_inputs=( + ArtifactInput( + "transition", + f"apply_{year}", + "transition", + trajectory.TRANSITION_TYPE, + ), + ArtifactInput( + "snapshot", + f"snapshot_{year}", + "snapshot", + trajectory.SNAPSHOT_TYPE, + ), + ), + artifact_outputs=(ArtifactOutput("account", ACCOUNT_TYPE),), + ) + for year in range(boundary + 1, kwargs["end_year"] + 1) + ) + return replace(graph, nodes=(*graph.nodes, *nodes)), registry + + +@dataclass(frozen=True) +class AccountedTrajectoryRun: + """Accounting summaries and the actual graph's unmodified receipts.""" + + manifest: object + report: dict + + +def run_accounted_mortality_trajectory( + *, training, rates, initial, holdouts, output_dir, **kwargs +): + """Run once and export accounting separately from mortality evaluations. + + Graph coordinates are those of ``build_trajectory_graph``. Inputs retain + its exact synthetic schemas; arbitrary household/location columns are + unsupported and refused by the original source reader. + """ + graph, registry = build_accounted_trajectory_graph(**kwargs) + boundary = kwargs.get("boundary_year", 2014) + years = range(boundary + 1, kwargs["end_year"] + 1) + if ( + not isinstance(holdouts, dict) + or any(type(year) is not int for year in holdouts) + or set(holdouts) != set(years) + ): + raise ValueError("holdouts must supply exactly one source per year") + sources = { + "training": Path(training).resolve(), + "rates": Path(rates).resolve(), + "initial": Path(initial).resolve(), + **{ + f"holdout_{year}": Path(holdouts[year]).resolve() for year in years + }, + } + output = Path(output_dir).resolve() + output.mkdir(parents=True, exist_ok=True) + store = ContentStore(output / "store") + manifest = run_graph( + compile_graph(graph), sources=sources, store=store, kernels=registry + ) + periods = { + str(year): parse_json( + store.load_bytes( + manifest.nodes[f"account_{year}"].opaque_artifacts["account"] + ) + ) + for year in years + } + statuses = {period["accounting_status"] for period in periods.values()} + report = { + "scope": "synthetic_engineering", + "boundary_year": boundary, + "end_year": kwargs["end_year"], + "accounting_status": ( + "failed" + if "failed" in statuses + else ( + "not_evaluated" if "not_evaluated" in statuses else "complete" + ) + ), + "periods": periods, + } + (output / "manifest.json").write_text(manifest.to_json(), encoding="utf-8") + (output / "accounting-report.json").write_bytes(json_bytes(report)) + return AccountedTrajectoryRun(manifest, report) diff --git a/tests/README-tiers.md b/tests/README-tiers.md index 4af88c67..7434c89d 100644 --- a/tests/README-tiers.md +++ b/tests/README-tiers.md @@ -38,9 +38,9 @@ pytest --collect-only -q -m oracle_policyengine | tail -1 | Tier | Tests at HEAD | |---|---:| -| `unit` | 1,717 | +| `unit` | 1,757 | | `artifact` | 2,672 | | `integration_psid` | 848 | | `reproduction_legacy` | 520 | | `oracle_policyengine` | 159 | -| **Total** | **5,916** | +| **Total** | **5,956** | diff --git a/tests/estimates/test_birth_evidence_artifact.py b/tests/estimates/test_birth_evidence_artifact.py index b4bfb6ba..f398cd24 100644 --- a/tests/estimates/test_birth_evidence_artifact.py +++ b/tests/estimates/test_birth_evidence_artifact.py @@ -95,6 +95,7 @@ def test_post_review_sources_are_outside_historical_reducer_identity(): Path("src/populace_dynamics/graph/runtime.py"), Path("src/populace_dynamics/graph/synthetic.py"), Path("src/populace_dynamics/graph/trajectory.py"), + Path("src/populace_dynamics/graph/trajectory_accounting.py"), Path("src/populace_dynamics/engine/accounting.py"), ) assert reducer.POST_REVIEW_SHARED_SOURCE_BLOBS == { @@ -229,6 +230,7 @@ def test_psid_and_graph_exclusions_are_unreachable_from_birth_evidence(): or name.startswith("populace_dynamics.graph.") } assert graph_exclusions + assert "populace_dynamics.graph.trajectory_accounting" in graph_exclusions module_by_path = { path.resolve(): module_name for module_name, path in module_paths.items() diff --git a/tests/test_graph_trajectory_accounting.py b/tests/test_graph_trajectory_accounting.py new file mode 100644 index 00000000..47d4e9ab --- /dev/null +++ b/tests/test_graph_trajectory_accounting.py @@ -0,0 +1,641 @@ +"""Invented-source checks of accounting on the actual optional graph.""" + +import copy +import json +import math +from dataclasses import replace +from pathlib import Path +from types import SimpleNamespace + +import pandas as pd +import pytest + +from populace_dynamics.graph.synthetic import write_synthetic_inputs + + +@pytest.fixture +def runtime(): + from populace_dynamics.graph._compat import require_graph + + try: + require_graph() + except ImportError as error: + pytest.skip(str(error)) + from populace_dynamics.graph import trajectory_accounting + + return trajectory_accounting + + +def _read(path): + return json.loads(path.read_text()) + + +def _write(path, value): + path.write_text(json.dumps(value)) + + +@pytest.fixture +def inputs(tmp_path): + sources = write_synthetic_inputs(tmp_path / "inputs") + sources.pop("holdout") + holdouts = {} + for year in range(2015, 2019): + path = tmp_path / "inputs" / f"annual-{year}.json" + _write( + path, + { + "scope": "synthetic_engineering", + "year": year, + "expected_death_rate": 0.2, + "fixture_max_abs_death_rate_gap": 0.25, + }, + ) + holdouts[year] = path + return {**sources, "holdouts": holdouts} + + +def _sources(inputs, end_year): + return { + **inputs, + "holdouts": { + year: path + for year, path in inputs["holdouts"].items() + if year <= end_year + }, + } + + +def _run(runtime, inputs, tmp_path, *, end_year=2017, **kwargs): + return runtime.run_accounted_mortality_trajectory( + **_sources(inputs, end_year), + end_year=end_year, + output_dir=tmp_path / "output", + **kwargs, + ) + + +def _regime(inputs, *, all_die): + training = _read(inputs["training"]) + for row in training: + row["death"] = 1.0 if all_die else 0.0 + row["exposure"] = 1e-9 if all_die else 1.0 + _write(inputs["training"], training) + + +def _artifact(runtime, result, tmp_path, node, name): + store = runtime.ContentStore(tmp_path / "output" / "store") + return runtime.parse_json( + store.load_bytes(result.manifest.nodes[node].opaque_artifacts[name]) + ) + + +def _context(runtime, result, tmp_path, year=2015): + return SimpleNamespace( + params={"year": year, "boundary_year": 2014}, + sources={}, + artifacts={ + name: SimpleNamespace( + payload=runtime.json_bytes( + _artifact(runtime, result, tmp_path, node, name) + ) + ) + for name, node in ( + ("transition", f"apply_{year}"), + ("snapshot", f"snapshot_{year}"), + ) + }, + ) + + +def test_actual_graph_preserves_original_keys_populations_and_evaluations( + runtime, inputs, tmp_path +): + original = runtime.trajectory.run_mortality_trajectory( + **_sources(inputs, 2017), + end_year=2017, + output_dir=tmp_path / "output", + ) + result = _run(runtime, inputs, tmp_path) + for name, node in original.manifest.nodes.items(): + actual = result.manifest.nodes[name] + assert actual.hit, name + assert actual.key == node.key, name + assert actual.receipt == node.receipt, name + assert result.report["accounting_status"] == "complete" + for year in range(2015, 2018): + name = f"advance_{year}" + old = original.manifest.population(name) + new = result.manifest.population(name) + pd.testing.assert_frame_equal( + old.table("person_period"), new.table("person_period") + ) + assert not result.manifest.nodes[f"account_{year}"].hit + period = result.report["periods"][str(year)] + assert period["application_status"] == "complete" + assert period["completed_year"] == year + account = period["account"] + assert account["status"] == runtime.accounting.ENGINEERING_STATUS + transition = _artifact( + runtime, result, tmp_path, f"apply_{year}", "transition" + ) + deaths = [row for row in transition["records"] if not row["survives"]] + assert account["counts"]["exits_total"] == len(deaths) + assert account["counts"]["exits_by_kind"]["death"] == len(deaths) + assert account["counts"]["additions_total"] == 0 + weights = account["weights"] + assert weights["revaluation"]["total"] == 0 + assert weights["closing"] == math.fsum( + (weights["opening"], -weights["exits_total"]) + ) + assert account["residuals"]["count"] == 0 + assert account["residuals"]["weight"] == 0 + assert _read(tmp_path / "output" / "accounting-report.json") == ( + result.report + ) + assert _read(tmp_path / "output" / "manifest.json") == json.loads( + result.manifest.to_json() + ) + # Existing outputs from the original runner were not overwritten. + assert _read(tmp_path / "output" / "report.json") == original.report + + +def test_cold_warm_and_horizon_extension(runtime, inputs, tmp_path): + cold = _run(runtime, inputs, tmp_path, end_year=2015) + warm = _run(runtime, inputs, tmp_path, end_year=2015) + assert not any(node.hit for node in cold.manifest.nodes.values()) + assert all(node.hit for node in warm.manifest.nodes.values()) + assert cold.report == warm.report + extended = _run(runtime, inputs, tmp_path, end_year=2018) + for name, node in warm.manifest.nodes.items(): + assert extended.manifest.nodes[name].hit, name + assert extended.manifest.nodes[name].key == node.key, name + assert extended.report["periods"]["2015"] == cold.report["periods"]["2015"] + assert not extended.manifest.nodes["account_2018"].hit + + +@pytest.mark.parametrize("malformed", [False, True]) +def test_holdout_changes_only_its_evaluation( + runtime, inputs, tmp_path, malformed +): + original = _run(runtime, inputs, tmp_path) + path = inputs["holdouts"][2016] + if malformed: + path.write_text("{invalid fixture JSON") + else: + holdout = _read(path) + holdout["expected_death_rate"] = 1.0 + holdout["fixture_max_abs_death_rate_gap"] = 0.0 + _write(path, holdout) + changed = _run(runtime, inputs, tmp_path) + for name, node in changed.manifest.nodes.items(): + if name == "evaluate_2016": + assert not node.hit + assert node.receipt["outcome"] == "fail" + else: + assert node.hit, name + assert node.key == original.manifest.nodes[name].key, name + assert changed.report == original.report + + +@pytest.mark.parametrize("all_die", [False, True]) +def test_extinction_and_zero_mortality(runtime, inputs, tmp_path, all_die): + _regime(inputs, all_die=all_die) + result = _run(runtime, inputs, tmp_path) + for year in range(2015, 2018): + period = result.report["periods"][str(year)] + assert period["accounting_status"] == "complete" + account = period["account"] + counts = account["counts"] + expected_opening = 0 if all_die and year > 2015 else 20 + assert counts["opening"] == expected_opening + assert counts["closing"] == (0 if all_die else 20) + assert counts["exits_total"] == (expected_opening if all_die else 0) + assert account["opening_year"] == year - 1 + assert account["closing_year"] == year + assert account["residuals"]["count"] == 0 + if all_die: + population = result.manifest.population("advance_2017") + assert population.table("period").period.tolist() == [2014] + + +def test_failed_and_blocked_application_never_reconciles_stale_snapshots( + runtime, inputs, tmp_path, monkeypatch +): + _regime(inputs, all_die=False) + initial = _read(inputs["initial"]) + initial[0]["age"] = 120 + _write(inputs["initial"], initial) + original = runtime.accounting.reconcile_period + calls = [] + + def tracked(*args, **kwargs): + calls.append(kwargs["closing_year"]) + return original(*args, **kwargs) + + monkeypatch.setattr(runtime.accounting, "reconcile_period", tracked) + cold = _run(runtime, inputs, tmp_path) + warm = _run(runtime, inputs, tmp_path) + assert calls == [2015] + assert cold.report == warm.report + assert all(node.hit for node in warm.manifest.nodes.values()) + assert cold.report["accounting_status"] == "not_evaluated" + for year, status in ((2016, "failed"), (2017, "blocked")): + period = cold.report["periods"][str(year)] + assert period["application_status"] == status + assert period["completed_year"] == 2015 + assert period["accounting_status"] == "not_evaluated" + assert period["account"] is None + transition = _artifact( + runtime, cold, tmp_path, f"apply_{year}", "transition" + ) + assert period["diagnostic"] == transition["diagnostic"] + + +def test_accounting_error_does_not_change_mortality_or_later_application( + runtime, inputs, tmp_path, monkeypatch +): + def refuse(*args, **kwargs): + raise runtime.accounting.PopulationAccountingInputError( + "invented accounting refusal" + ) + + monkeypatch.setattr(runtime.accounting, "reconcile_period", refuse) + result = _run(runtime, inputs, tmp_path) + assert result.report["accounting_status"] == "failed" + for year in range(2015, 2018): + period = result.report["periods"][str(year)] + assert period["account"] is None + assert period["application_status"] == "complete" + assert period["completed_year"] == year + assert period["diagnostic"]["message"] == "invented accounting refusal" + assert ( + result.manifest.nodes[f"apply_{year}"].receipt["outcome"] == "pass" + ) + assert ( + result.manifest.nodes[f"evaluate_{year}"].receipt["outcome"] + == "pass" + ) + + +@pytest.mark.parametrize( + "mutation", + ["person", "observation", "boolean", "probability", "missing_record"], +) +def test_transition_records_must_bind_to_snapshot( + runtime, inputs, tmp_path, mutation +): + result = _run(runtime, inputs, tmp_path, end_year=2015) + context = _context(runtime, result, tmp_path) + value = runtime.parse_json(context.artifacts["transition"].payload) + row = value["records"][0] + if mutation == "person": + row["person_id"] += 9999 + elif mutation == "observation": + row["observation_id"] += 9999 + elif mutation == "boolean": + row["survives"] = 1 + elif mutation == "probability": + row["death_probability"] = 1.1 + else: + value["records"].pop() + context.artifacts["transition"].payload = runtime.json_bytes(value) + report = runtime.parse_json(runtime._account(context).artifacts["account"]) + assert report["accounting_status"] == "failed" + assert report["account"] is None + assert report["diagnostic"]["exception_type"] == "ValueError" + + +@pytest.mark.parametrize("mutation", ["omit_survivor", "retain_death"]) +def test_endpoint_differences_do_not_infer_deaths( + runtime, inputs, tmp_path, mutation +): + result = _run(runtime, inputs, tmp_path, end_year=2015) + context = _context(runtime, result, tmp_path) + snapshot = runtime.parse_json(context.artifacts["snapshot"].payload) + transition = runtime.parse_json(context.artifacts["transition"].payload) + rows = snapshot["observations"] + if mutation == "omit_survivor": + index = next( + i + for i, row in enumerate(rows) + if row["person_period_period_id"] == 2015 + ) + rows.pop(index) + snapshot["weights"].pop(index) + expected = "undeclared_exit" + else: + dead = next( + row for row in transition["records"] if not row["survives"] + ) + index = next( + i + for i, row in enumerate(rows) + if row["person_period_id"] == dead["observation_id"] + ) + child = { + **rows[index], + "person_period_id": 9999, + "person_period_period_id": 2015, + } + rows.append(child) + snapshot["weights"].append(snapshot["weights"][index]) + expected = "exit_contradicted_by_closing" + context.artifacts["snapshot"].payload = runtime.json_bytes(snapshot) + report = runtime.parse_json(runtime._account(context).artifacts["account"]) + assert report["accounting_status"] == "failed" + assert report["account"] is None + discrepancies = report["diagnostic"]["reconciliation"]["discrepancies"] + assert expected in {item["kind"] for item in discrepancies} + + +def test_accounting_copies_rows_without_mutating_snapshot_or_columns( + runtime, inputs, tmp_path +): + result = _run(runtime, inputs, tmp_path, end_year=2015) + context = _context(runtime, result, tmp_path) + before = copy.deepcopy(context) + frozen = runtime.trajectory._evaluation_context(context) + opening, closing = runtime._period_frames(frozen) + for frame in (opening, closing): + assert {"age", "sex", "person_period_id"} <= set(frame.columns) + opening.loc[:, "age"] = -99 + assert (frozen.tables["person_period"].age >= 0).all() + runtime._account(context) + assert ( + context.artifacts["snapshot"].payload + == before.artifacts["snapshot"].payload + ) + assert ( + context.artifacts["transition"].payload + == before.artifacts["transition"].payload + ) + + +@pytest.mark.parametrize("column", ["atomic_location_id", "household_id"]) +def test_initial_reader_refuses_unsupported_location_and_household_columns( + runtime, inputs, tmp_path, column +): + initial = _read(inputs["initial"]) + for row in initial: + row[column] = "invented-anchor" + _write(inputs["initial"], initial) + with pytest.raises(Exception, match="unsupported fields"): + _run(runtime, inputs, tmp_path) + assert _read(inputs["initial"]) == initial + + +def test_snapshot_refuses_unsupported_location_column( + runtime, inputs, tmp_path +): + result = _run(runtime, inputs, tmp_path, end_year=2015) + context = _context(runtime, result, tmp_path) + snapshot = runtime.parse_json(context.artifacts["snapshot"].payload) + for row in snapshot["observations"]: + row["atomic_location_id"] = "invented-anchor" + context.artifacts["snapshot"].payload = runtime.json_bytes(snapshot) + report = runtime.parse_json(runtime._account(context).artifacts["account"]) + assert report["accounting_status"] == "failed" + assert report["account"] is None + assert "snapshot row binding" in report["diagnostic"]["message"] + + +@pytest.mark.parametrize( + "mutation", + ["period_alias", "duplicate_history", "weight_length", "future_period"], +) +def test_snapshot_binding_and_period_history_fail_closed( + runtime, inputs, tmp_path, mutation +): + result = _run(runtime, inputs, tmp_path, end_year=2015) + context = _context(runtime, result, tmp_path) + snapshot = runtime.parse_json(context.artifacts["snapshot"].payload) + if mutation == "period_alias": + snapshot["periods"].append({"period_id": 9999, "period": 2015}) + elif mutation == "duplicate_history": + snapshot["observations"].append( + {**snapshot["observations"][0], "person_period_id": 9999} + ) + snapshot["weights"].append(snapshot["weights"][0]) + elif mutation == "weight_length": + snapshot["weights"].pop() + else: + snapshot["periods"].append({"period_id": 2016, "period": 2016}) + context.artifacts["snapshot"].payload = runtime.json_bytes(snapshot) + report = runtime.parse_json(runtime._account(context).artifacts["account"]) + assert report["accounting_status"] == "failed" + assert report["account"] is None + assert report["diagnostic"]["exception_type"] == "ValueError" + + +def test_changed_snapshot_survivor_weight_is_reported_separately_from_evaluation( + runtime, inputs, tmp_path, monkeypatch +): + original = runtime.trajectory._snapshot + + def changed_weight(context): + result = original(context) + snapshot = runtime.parse_json(result.artifacts["snapshot"]) + if context.params["year"] == 2016: + index = next( + i + for i, row in enumerate(snapshot["observations"]) + if row["person_period_period_id"] == 2016 + ) + snapshot["weights"][index] += 3.0 + return replace( + result, artifacts={"snapshot": runtime.json_bytes(snapshot)} + ) + + monkeypatch.setattr(runtime.trajectory, "_snapshot", changed_weight) + result = _run(runtime, inputs, tmp_path) + account = result.report["periods"]["2016"]["account"] + assert account["weights"]["revaluation"]["carried"] == 3.0 + assert account["residuals"]["weight"] == 0.0 + assert result.report["accounting_status"] == "complete" + assert result.manifest.nodes["evaluate_2016"].receipt["outcome"] == "fail" + assert result.manifest.nodes["apply_2017"].receipt["outcome"] == "pass" + assert result.manifest.nodes["evaluate_2017"].receipt["outcome"] == "pass" + + +def test_incomplete_application_does_not_parse_snapshot(runtime): + outcome = { + "format": runtime.trajectory.TRANSITION_TYPE.name, + "schema_version": 1, + "from_year": 2015, + "year": 2016, + "status": "failed", + "completed_year": 2015, + "records": [], + "diagnostic": {"message": "synthetic prior refusal"}, + } + context = SimpleNamespace( + params={"year": 2016, "boundary_year": 2014}, + artifacts={ + "transition": SimpleNamespace(payload=runtime.json_bytes(outcome)), + "snapshot": SimpleNamespace(payload=b"{invalid snapshot JSON"), + }, + ) + report = runtime.parse_json(runtime._account(context).artifacts["account"]) + assert report["accounting_status"] == "not_evaluated" + assert report["account"] is None + assert report["diagnostic"] == outcome["diagnostic"] + + +@pytest.mark.parametrize( + "mutation", ["json", "completed_year", "unknown_status"] +) +def test_malformed_transition_does_not_invent_application_completion( + runtime, mutation +): + value = { + "format": runtime.trajectory.TRANSITION_TYPE.name, + "schema_version": 1, + "from_year": 2015, + "year": 2016, + "status": "failed", + "completed_year": 2015, + "records": [], + "diagnostic": {"message": "synthetic prior refusal"}, + } + if mutation == "completed_year": + value["completed_year"] = 2013 + elif mutation == "unknown_status": + value["status"] = "unknown" + context = SimpleNamespace( + params={"year": 2016, "boundary_year": 2014}, + artifacts={ + "transition": SimpleNamespace( + payload=( + b"{invalid" + if mutation == "json" + else runtime.json_bytes(value) + ) + ) + }, + ) + report = runtime.parse_json(runtime._account(context).artifacts["account"]) + assert report["accounting_status"] == "failed" + assert report["account"] is None + assert report["application_status"] is None + assert report["completed_year"] is None + + +def test_runner_requires_exact_annual_holdouts(runtime, inputs, tmp_path): + inputs["holdouts"].pop(2016) + with pytest.raises(ValueError, match="exactly one source per year"): + _run(runtime, inputs, tmp_path) + + +@pytest.mark.parametrize("value", ["1.0", True]) +def test_raw_snapshot_weights_are_checked_before_float_conversion( + runtime, inputs, tmp_path, value +): + result = _run(runtime, inputs, tmp_path, end_year=2015) + context = _context(runtime, result, tmp_path) + snapshot = runtime.parse_json(context.artifacts["snapshot"].payload) + snapshot["weights"][0] = value + context.artifacts["snapshot"].payload = runtime.json_bytes(snapshot) + report = runtime.parse_json(runtime._account(context).artifacts["account"]) + assert report["accounting_status"] == "failed" + assert report["account"] is None + assert ( + report["diagnostic"]["exception_type"] + == "PopulationAccountingInputError" + ) + assert "real number" in report["diagnostic"]["message"] + + +@pytest.mark.parametrize("identifier", ["person", "observation"]) +def test_oversized_raw_identifier_cannot_wrap_into_transition_binding( + runtime, inputs, tmp_path, identifier +): + result = _run(runtime, inputs, tmp_path, end_year=2015) + context = _context(runtime, result, tmp_path) + snapshot = runtime.parse_json(context.artifacts["snapshot"].payload) + transition = runtime.parse_json(context.artifacts["transition"].payload) + if identifier == "person": + column, field = "person_period_person_id", "person_id" + else: + column, field = "person_period_id", "observation_id" + original_id = transition["records"][0][field] + for row in snapshot["observations"]: + if row[column] == original_id: + row[column] = 2**64 - 1 + transition["records"][0][field] = -1 + context.artifacts["snapshot"].payload = runtime.json_bytes(snapshot) + context.artifacts["transition"].payload = runtime.json_bytes(transition) + report = runtime.parse_json(runtime._account(context).artifacts["account"]) + assert report["accounting_status"] == "failed" + assert report["account"] is None + assert ( + report["diagnostic"]["exception_type"] + == "PopulationAccountingInputError" + ) + assert "signed int64" in report["diagnostic"]["message"] + + +@pytest.mark.parametrize("status", ["failed", "blocked"]) +def test_overflowing_diagnostic_becomes_a_serializable_accounting_refusal( + runtime, status +): + payload = ( + '{"format":"populace-dynamics.mortality-transition",' + '"schema_version":1,"from_year":2015,"year":2016,' + f'"status":"{status}","completed_year":2015,"records":[], ' + '"diagnostic":{"message":"synthetic refusal","nested":[1e400]}}' + ).encode() + context = SimpleNamespace( + params={"year": 2016, "boundary_year": 2014}, + artifacts={"transition": SimpleNamespace(payload=payload)}, + ) + report = runtime.parse_json(runtime._account(context).artifacts["account"]) + assert report["accounting_status"] == "failed" + assert report["account"] is None + assert report["application_status"] is None + assert report["completed_year"] is None + assert report["diagnostic"]["exception_type"] == "ValueError" + json.dumps(report, allow_nan=False) + + +@pytest.mark.parametrize( + "module", ["accounting", "trajectory", "runtime", "model"] +) +def test_accounting_hash_includes_reused_source_modules( + runtime, monkeypatch, module +): + _, registry = runtime.build_accounted_trajectory_graph(end_year=2015) + kernel = registry.get(runtime.ACCOUNT_KERNEL) + modules = { + "accounting": runtime.accounting, + "trajectory": runtime.trajectory, + "runtime": runtime.trajectory.rt, + "model": runtime.trajectory.rt.model_module, + } + target = Path(modules[module].__file__).resolve() + before = kernel.implementation_hash() + original = Path.read_bytes + + def changed(path): + payload = original(path) + return ( + payload + b"\n# synthetic source change\n" + if path.resolve() == target + else payload + ) + + monkeypatch.setattr(Path, "read_bytes", changed) + assert kernel.implementation_hash() != before + + +def test_account_nodes_have_only_artifact_dependencies(runtime): + graph, _ = runtime.build_accounted_trajectory_graph(end_year=2017) + for node in graph.nodes: + if not node.id.startswith("account_"): + continue + assert node.population == "training" + assert node.inputs == () + assert node.sources == () + assert {item.name for item in node.artifact_inputs} == { + "transition", + "snapshot", + } diff --git a/tests/tier_counts.json b/tests/tier_counts.json index c383a179..7748446a 100644 --- a/tests/tier_counts.json +++ b/tests/tier_counts.json @@ -1,7 +1,7 @@ { "schema_version": 1, "counts": { - "unit": 1717, + "unit": 1757, "artifact": 2672, "integration_psid": 848, "reproduction_legacy": 520,