From 7426f729fc91cf2a65085c3c42a26f7c4e105423 Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Wed, 9 Sep 2026 14:25:44 -0400 Subject: [PATCH] Recover experimental entrant schedules with explicit support boundaries --- docs/entrant-seam.md | 85 +++ scripts/first_estimates_birth_evidence.py | 3 + .../engine/entrant_domains.py | 380 ++++++++++++ .../engine/entrant_schedule.py | 427 ++++++++++++++ tests/README-tiers.md | 4 +- .../estimates/test_birth_evidence_artifact.py | 9 + tests/test_entrant_domains.py | 381 ++++++++++++ tests/test_entrant_schedule.py | 550 ++++++++++++++++++ tests/tier_counts.json | 2 +- 9 files changed, 1838 insertions(+), 3 deletions(-) create mode 100644 docs/entrant-seam.md create mode 100644 src/populace_dynamics/engine/entrant_domains.py create mode 100644 src/populace_dynamics/engine/entrant_schedule.py create mode 100644 tests/test_entrant_domains.py create mode 100644 tests/test_entrant_schedule.py diff --git a/docs/entrant-seam.md b/docs/entrant-seam.md new file mode 100644 index 00000000..6ec8af94 --- /dev/null +++ b/docs/entrant-seam.md @@ -0,0 +1,85 @@ +# Experimental entrant schedule and support restrictions + +This source slice recovers the two engine modules and synthetic tests from +local commit `61bbf1c7e25a7e55033c134bbc2e846022b8850b`. It accepts an explicitly +supplied donor frame and annual controls in thousands. It does not include the +original native donor/control readers, snapshots, build script, or run artifact. +The original frame-reader round-trip test is deferred with that source path. + +`build_entrant_schedule` returns frames for the existing +`metadata[SCHEDULED_ENTRIES_KEY]` interface, plus arithmetic alignment and +provenance records. Positive-weight donors are reweighted for each positive +control. Zero controls remain in alignment but produce no frame and consume +no person IDs; zero-weight donors do not become demographic actors. The input +boundary rejects noninteger ages/activation keys and ambiguous boolean flags. +Controls are validated before cohort allocation begins. + +The schedule retains the historical gross-positive-inflow convention. That +control convention is not a count of observed border arrivals, a net migration +law, or a complete population stock model. The recorded donor composition and +control provenance describe caller-supplied inputs; this interface does not +admit their sources or fit/calibrate a population. + +## Existing loop timing + +A cohort scheduled for year Y carries `year=Y-1` and `age=entry_age`. +The existing loop activates it before mortality, then increments age. Its +activation-year slice therefore carries `entry_age+1`. This is the existing +engine convention, not a newly established arrival-exposure assumption. +The caller must reserve real IDs in the shared allocator. The loop rejects +duplicate scheduled IDs and builds stable person ordinals across cohorts. + +## Explicit support boundaries + +`EntrantClaimingAdapter(step)` calls a supplied claiming adapter only on +incumbents. It rejects existing entrant claim ages, claim years, claimed state, +or disability-conversion events before executing the incumbent step. Such +observations need a separately admitted claim-history path. The wrapper does +not erase possible observed entitlement or declare those people ineligible. +Supported exclusion inputs retain missing plans/years and a structural false +claiming flag. With only entrants, the incumbent step is not called. + +Explicit synthetic rows require a known `entry_kind`; losing or misspelling +that marker is an error. Legacy closed panels without synthetic markers retain +their incumbent interpretation. Birth and realized-opener markers remain +distinct from immigrant cohorts. Provenance counts reject missing/unknown kinds. + +The historical birth materializer leaves the new provenance column missing. +For this experimental path, `materialize_births_with_provenance` delegates to +that actual materializer and labels only the children it just allocated. It +validates existing rows first; it cannot retrospectively relabel unidentified +synthetic rows. It accepts supplied birth records and does not establish a +fitted fertility adapter or enforce the fertility risk-set restriction. + +`suppress_entrant_benefit_outputs` marks specified unsupported outputs missing. +It does not calculate benefits. Consumers must report the unsupported population +separately; missing benefits must not become zero benefits in an average or be +used to label a partial-population result national. + +The fertility/disability ID helpers and earnings-domain assertion express +support restrictions. `exclusion_report` is explicitly `inventory_only` with +`execution_verified=False`; it does not claim those restrictions were applied +in a projection. The actual claiming wrapper enforces its own narrower contract. +The historical fertility function treats an empty `holdout_ids` set as all +roster IDs and ignores that argument in its precomputed-birth branch. An ID-set +complement alone therefore cannot enforce the entrant fertility restriction. + +## Validation and remaining work + +Synthetic tests exercise the real scheduled-entry loop, cohort extinction and +later activation, mortality-before-ageing order, positive/zero cohort weights, +claiming exclusion, missing benefit outputs, and incumbent random consumption. +The mortality and other fitted transitions in those integration fixtures are +explicitly synthetic. Placeholder benefit values test output suppression only; +no policy calculation, Axiom runtime, or national score is claimed. + +The historical registered assembly, engine steps, and scientific gates remain +unchanged. Exact historical source exclusions are justified by assertions that +these two experimental modules remain unreachable from the birth-evidence +reducer and its registered input roots. + +An entrant-supported Social Security score still requires admitted donor and +control sources, migration-universe/exposure decisions, covered-work histories, +insured-status and benefit support, and coherent family, disability, earnings, +and other post-entry transitions. This recovery addresses an engineering +boundary; it does not establish DynaSim parity. diff --git a/scripts/first_estimates_birth_evidence.py b/scripts/first_estimates_birth_evidence.py index efae0662..f70fb7cc 100644 --- a/scripts/first_estimates_birth_evidence.py +++ b/scripts/first_estimates_birth_evidence.py @@ -156,6 +156,9 @@ Path("src/populace_dynamics/estimates/anchor_context_registry.py"), Path("src/populace_dynamics/estimates/anchor_context_rehearsal.py"), Path("src/populace_dynamics/estimates/anchor_context_report.py"), + # Experimental entrants are outside the registered projection call graph. + Path("src/populace_dynamics/engine/entrant_schedule.py"), + Path("src/populace_dynamics/engine/entrant_domains.py"), ) POST_REVIEW_SHARED_SOURCE_BLOBS = { Path( diff --git a/src/populace_dynamics/engine/entrant_domains.py b/src/populace_dynamics/engine/entrant_domains.py new file mode 100644 index 00000000..0be9cd7b --- /dev/null +++ b/src/populace_dynamics/engine/entrant_domains.py @@ -0,0 +1,380 @@ +"""Experimental support restrictions for explicitly scheduled entrants. + +Recovered from local entrant work at 61bbf1c7e25a7e55033c134bbc2e846022b8850b. +The existing fitted 2014 earnings state and observed disability panel do not +supply histories for newly allocated IDs. The historical claiming adapter also +has no insured-status gate. This module therefore identifies unsupported rows +and offers an explicit claiming wrapper; it does not supply entrant behavior, +entitlement, or an admitted population. Fertility/disability ID inventories +and demographic scope declarations do not demonstrate step execution. + +A source-only recovery accepts caller-supplied synthetic frames. No native +donor, control release, fitted model, or scientific gate is invoked here. +""" + +from __future__ import annotations + +from collections.abc import Iterable, Mapping +from dataclasses import dataclass +from typing import Any + +import numpy as np +import pandas as pd + +from populace_dynamics.engine.earnings_domain import EARNINGS_DOMAIN_COLUMN +from populace_dynamics.engine.entrant_schedule import ( + ENTRY_KIND_BIRTH, + ENTRY_KIND_COLUMN, + ENTRY_KIND_IMMIGRANT, + ENTRY_KIND_INCUMBENT, + ENTRY_KIND_REALIZED_OPENER, +) +from populace_dynamics.engine.steps import materialize_maternal_births + +__all__ = [ + "EXCLUDED_DOMAINS", + "EntrantClaimingAdapter", + "EntrantExclusionReport", + "entrant_mask", + "materialize_births_with_provenance", + "assert_entrants_out_of_earnings_domain", + "excluded_claiming_ids", + "excluded_disability_ids", + "excluded_fertility_ids", + "exclusion_report", + "suppress_entrant_benefit_outputs", +] + +#: The four domains an entrant is excluded from, and why each one is unfitted. +EXCLUDED_DOMAINS: Mapping[str, str] = { + "fertility_risk": ( + "steps.py:451-506 materializes births only for on-roster mothers and " + "initializes parity at zero; no entrant parity or birth-history seed " + "exists, so leaving entrants at risk would assert every entrant " + "arrived childless" + ), + "claiming_eligibility": ( + "steps.py:390-450 draws a claim age for everyone aged 50+ with no " + "insured-status, quarters-of-coverage, AIME or PIA test; reported " + "year of entry does not identify FIRST entry, so prior US covered " + "earnings are unknown/censored rather than zero and insured status " + "cannot be established from either survey" + ), + "disability_panel": ( + "disability.py:56-60 filters a PSID-built DisabilityPanel to " + "holdout_ids; entrants are not in the panel, and ASEC/CPS disability " + "items are not realized PSID M4 status" + ), + "earnings_domain": ( + "earnings_domain.py:150-208 keys membership on the generator's fitted " + "2014 state and forward_earnings.py:1421-1436 raises without it; the " + "section 2.8.3a certificate was never fitted on entrants and does not " + "transfer to them" + ), +} + + +def entrant_mask( + frame: pd.DataFrame, + *, + entry_kinds: Iterable[str] = (ENTRY_KIND_IMMIGRANT,), +) -> np.ndarray: + """Boolean membership: is each row a scheduled entrant? + + Reads the explicit :data:`~populace_dynamics.engine.entrant_schedule.ENTRY_KIND_COLUMN` + provenance column. A frame without that column is a closed-panel frame + and every row is an incumbent -- that is the honest reading, and it keeps + this predicate safe to call on a roster that has never seen a schedule. + A missing value is also treated as incumbent unless the row is explicitly + synthetic. Synthetic rows require a known entry kind, so losing that + provenance cannot silently remove their exclusion. + """ + known = { + ENTRY_KIND_BIRTH, + ENTRY_KIND_INCUMBENT, + ENTRY_KIND_IMMIGRANT, + ENTRY_KIND_REALIZED_OPENER, + } + if isinstance(entry_kinds, (str, bytes)): + raise ValueError( + "entry_kinds must be a collection, not a scalar string" + ) + kinds = set(entry_kinds) + if not kinds: + raise ValueError("entrant_mask needs at least one entry kind") + if not kinds.issubset(known): + raise ValueError("entry_kinds contains unknown selectors") + values = frame.get( + ENTRY_KIND_COLUMN, pd.Series(pd.NA, index=frame.index) + ).to_numpy() + synthetic = frame.get( + "synthetic_entry", pd.Series(False, index=frame.index) + ).to_numpy() + for value, synthetic_value in zip(values, synthetic, strict=True): + if not pd.isna(synthetic_value) and not isinstance( + synthetic_value, (bool, np.bool_) + ): + raise ValueError( + "synthetic_entry must contain booleans or missing" + ) + is_synthetic = not pd.isna(synthetic_value) and synthetic_value + if pd.isna(value): + if is_synthetic: + raise ValueError( + "synthetic entrants require explicit entry_kind" + ) + elif not isinstance(value, str) or value not in known: + raise ValueError(f"unknown entry_kind {value!r}") + return np.asarray( + [(not pd.isna(value)) and str(value) in kinds for value in values], + dtype=bool, + ) + + +def materialize_births_with_provenance( + frame: pd.DataFrame, + births: pd.DataFrame, + context: Any, + rng: np.random.Generator, +) -> pd.DataFrame: + """Label children at the actual historical birth-materialization boundary. + + Existing synthetic rows must already have valid provenance. Only rows + allocated by this call receive the maternal-birth marker. This helper + handles supplied birth records; it does not fit/simulate fertility or + enforce the entrant fertility risk restriction. + """ + entrant_mask(frame) + out = materialize_maternal_births(frame, births, context, rng) + added = ~out["person_id"].isin(frame["person_id"]) + if ENTRY_KIND_COLUMN not in out: + out[ENTRY_KIND_COLUMN] = pd.Series( + pd.NA, index=out.index, dtype="object" + ) + out.loc[added, ENTRY_KIND_COLUMN] = ENTRY_KIND_BIRTH + return out + + +def _entrant_ids(frame: pd.DataFrame, **kwargs: Any) -> set[int]: + mask = entrant_mask(frame, **kwargs) + return {int(value) for value in frame.loc[mask, "person_id"]} + + +def excluded_fertility_ids(frame: pd.DataFrame, **kwargs: Any) -> set[int]: + """Person IDs to remove from the fertility risk set. + + This is an inventory, not a fertility adapter. Historical + ``apply_fertility`` treats an empty ``holdout_ids`` as all roster IDs and + its precomputed-birth path does not apply that argument. Passing this + set's complement therefore does not establish entrant exclusion. + """ + return _entrant_ids(frame, **kwargs) + + +def excluded_claiming_ids(frame: pd.DataFrame, **kwargs: Any) -> set[int]: + """Person IDs whose claiming draw must be suppressed.""" + return _entrant_ids(frame, **kwargs) + + +def excluded_disability_ids(frame: pd.DataFrame, **kwargs: Any) -> set[int]: + """Person IDs to keep out of the M4 disability panel's holdout set.""" + return _entrant_ids(frame, **kwargs) + + +def assert_entrants_out_of_earnings_domain(frame: pd.DataFrame) -> int: + """Fail loudly if any entrant is marked inside the fitted earnings domain. + + Returns the number of entrants checked. There is no "exclude" step to + perform here -- ``earnings_domain.membership`` already excludes them by + construction, because a synthetic ID cannot be in the fitted 2014 state + maps. This is the assertion that the construction was not circumvented, + which is the failure mode ``validate_domain`` exists to catch. + """ + mask = entrant_mask(frame) + if not mask.any(): + return 0 + if EARNINGS_DOMAIN_COLUMN not in frame.columns: + return int(mask.sum()) + marked = frame.loc[mask, EARNINGS_DOMAIN_COLUMN] + offending = marked.fillna(False).astype(bool).to_numpy() + if offending.any(): + bad = frame.loc[mask].loc[offending, "person_id"].tolist()[:10] + raise ValueError( + "scheduled entrants are marked inside the fitted earnings " + f"domain: {bad}; the section 2.8.3a certificate does not transfer " + "to a population it was never fitted on" + ) + return int(mask.sum()) + + +def suppress_entrant_benefit_outputs( + frame: pd.DataFrame, *, columns: Iterable[str] = ("aime", "pia", "benefit") +) -> pd.DataFrame: + """Blank entrant benefit outputs to missing, never to zero. + + A zero AIME is a *measurement*: it says this person had no covered + earnings. For an entrant it would be a fabrication, because reported year + of entry does not identify first entry, so prior US covered earnings are + censored rather than absent. Missing is the only honest value until a + stock-to-arrival and insured-status bridge exists. + """ + out = frame.copy() + mask = entrant_mask(out) + if not mask.any(): + return out + for column in columns: + if column in out.columns: + out.loc[mask, column] = pd.NA + return out + + +@dataclass(frozen=True) +class EntrantExclusionReport: + """What was excluded, from where, and how many -- for the run artifact.""" + + n_rows: int + n_entrants: int + excluded: dict[str, dict[str, Any]] + + def as_dict(self) -> dict[str, Any]: + return { + "n_rows": self.n_rows, + "n_entrants": self.n_entrants, + "excluded_domains": self.excluded, + "gated": False, + "status": "inventory_only", + "interpretation": ( + "an excluded entrant is OUTSIDE the estimand for that domain, " + "not a person modelled as having no children, no disability " + "and no claim; suppressed counts are reported instead of " + "zeros for exactly that reason" + ), + "execution_verified": False, + "intended_demographic_domains": [ + "mortality (age/sex draw at the entry age)", + "aging (deterministic advance)", + ], + "not_certified": ( + "entrant benefit levels, insured status, prior US covered " + "earnings, legal status, population stocks and every " + "post-entry transition remain out of scope" + ), + } + + +def exclusion_report(frame: pd.DataFrame) -> EntrantExclusionReport: + """Measure the exclusions on one roster frame.""" + mask = entrant_mask(frame) + n_entrants = int(mask.sum()) + age = ( + frame["age"].to_numpy(dtype=np.float64) + if "age" in frame.columns + else np.full(len(frame), np.nan) + ) + missing_plan = ( + frame.get("claim_age", pd.Series(pd.NA, index=frame.index)) + .isna() + .to_numpy() + ) + claim_exposed = int((mask & (age >= 50) & missing_plan).sum()) + female = ( + frame["sex"].astype(str).to_numpy() == "female" + if "sex" in frame.columns + else np.zeros(len(frame), dtype=bool) + ) + fertile_exposed = int((mask & female & (age >= 15) & (age <= 49)).sum()) + + excluded: dict[str, dict[str, Any]] = {} + for domain, reason in EXCLUDED_DOMAINS.items(): + record: dict[str, Any] = { + "n_excluded": n_entrants, + "reason": reason, + } + if domain == "claiming_eligibility": + record["n_would_have_drawn_a_claim_age"] = claim_exposed + record["counterfactual"] = ( + "without this exclusion apply_claiming would draw a claim age " + f"for {claim_exposed} entrant rows aged 50+ without a plan; " + "this inventory does not verify execution or entitlement" + ) + if domain == "fertility_risk": + record["n_would_have_been_at_risk"] = fertile_exposed + if domain == "earnings_domain": + record["n_checked_out_of_domain"] = ( + assert_entrants_out_of_earnings_domain(frame) + ) + excluded[domain] = record + return EntrantExclusionReport( + n_rows=int(len(frame)), n_entrants=n_entrants, excluded=excluded + ) + + +@dataclass(frozen=True) +class EntrantClaimingAdapter: + """Run a claiming step on incumbents only, leaving entrants unclaimed. + + The direct analogue of :class:`~populace_dynamics.engine.earnings_domain.EarningsDomainAdapter`: + it keeps a support restriction outside the historical step rather than + editing the historical step; entrant rows never reach that step. + + Existing entrant plans, claim years, claimed state, or conversion events + are rejected before the incumbent step runs. They require a separately + admitted claim-history path; this wrapper does not erase them or rule on + observed entitlement. For admitted inputs, those three fields come back as + missing / ``False`` / missing. ``claimed = False`` is not a behavioural + claim that entrants never retire -- it is the roster's structural default + for a person outside the claiming estimand, and + :func:`exclusion_report` publishes how many rows it applied to so the + suppression is never mistaken for a measured zero. + """ + + step: Any + entry_kinds: tuple[str, ...] = (ENTRY_KIND_IMMIGRANT,) + + def __call__( + self, + frame: pd.DataFrame, + context: Any, + rng: np.random.Generator, + ) -> pd.DataFrame: + mask = entrant_mask(frame, entry_kinds=self.entry_kinds) + if not mask.any(): + return self.step(frame, context, rng) + incumbents = frame.loc[~mask].copy() + entrants = frame.loc[mask].copy() + # Preserve possible observed entitlement by rejecting unsupported + # state, rather than overwriting it with an exclusion default. + for column in ("claim_age", "claim_year"): + if column in entrants and entrants[column].notna().any(): + raise ValueError( + f"excluded entrants have existing {column}; an admitted " + "claim-history path is required" + ) + for column in ("claimed", "di_converted"): + if column in entrants: + observed = entrants[column].dropna() + if any( + not isinstance(value, (bool, np.bool_)) or value + for value in observed + ): + raise ValueError( + f"excluded entrants have unsupported {column} state" + ) + advanced = ( + self.step(incumbents, context, rng) + if not incumbents.empty + else incumbents + ) + for column, default in ( + ("claim_age", pd.NA), + ("claimed", False), + ("claim_year", pd.NA), + ): + entrants[column] = pd.array( + [default] * len(entrants), + dtype="bool" if column == "claimed" else "Int64", + ) + out = pd.concat([advanced, entrants], ignore_index=True, sort=False) + return out.sort_values("person_id", kind="stable").reset_index( + drop=True + ) diff --git a/src/populace_dynamics/engine/entrant_schedule.py b/src/populace_dynamics/engine/entrant_schedule.py new file mode 100644 index 00000000..5bfe38c8 --- /dev/null +++ b/src/populace_dynamics/engine/entrant_schedule.py @@ -0,0 +1,427 @@ +"""Build entrant cohorts for the loop's scheduled-entries seam. + +REPORT-ONLY. Nothing this module produces enters a fitted law, and no gate +scores it. It converts a sized control (Trustees Table V.A2 gross inflow) and +an explicitly supplied demographic donor into +frames the existing seam already accepts. + +**The seam is not new.** ``engine/loop.py`` has activated scheduled entries +since the M6 openers: :data:`~populace_dynamics.engine.loop.SCHEDULED_ENTRIES_KEY` +at ``loop.py:27``, validation at ``loop.py:219-247``, activation at +``loop.py:262-281``. This module only produces its input, so three properties +are inherited rather than designed: + +1. **The frame coordinate is the year BEFORE activation.** ``loop.py:236-239`` + rejects any other year. Following the documented opener convention at + ``m6_population.py:328-334`` -- "the frame coordinate is the reference year + immediately before the anchor interview, while age is the realized + collection-wave age" -- an entrant row carries ``year = activation_year - 1`` + and ``age = entry_age``. The loop's first step is therefore a **mortality + draw at the entrant's entry age**, before any aging; the aging step then + advances them to ``entry_age + 1`` in the activation year. That entry-year + exposure convention is fixed by the seam, not chosen here. +2. **IDs come from the projection-wide allocator.** ``loop.py:41-63`` raises + on any overlap with ``reserved_real_ids``, which is what stops an entrant + from silently inheriting a fitted person's ``u_w``. +3. **RNG streams are stable per person** (``loop.py:285-292, 351-353``), so + reproducibility is free. + +**Sizing.** Cohorts are sized to V.A2's *gross positive inflow* (LPR inflow + +temporary-or-unlawfully-present inflow), never to its total net change. Net +change is a residual whose age/sex/family composition has no literal +interpretation, and adjustment of status is a reclassification between two +stocks rather than a new person. Native control and donor readers from the original branch are not included +in this isolated source slice; all controls and rows are supplied by callers. + +**Method.** The donor pool is reweighted, not resampled: every positive-weight donor row +appears once per positive-inflow activation year with its weight scaled by a single factor so +the cohort's weighted total equals the control. That is deterministic, +consumes no RNG, and reproduces the donor composition exactly, so any residual +against the control is arithmetic rather than sampling noise. + +**Provenance.** Every emitted row carries ``entry_kind``, so births, realized +openers and immigrant cohorts stop being inferred from ID arithmetic. That +inference is what ``harness/m6_runner.py:1026-1032`` currently does when it +writes ``"immigrant_cohorts": 0`` under the comment "Every synthetic ID +allocated by this closed-panel engine is a step-4 materialized maternal +birth" -- true today, and false the moment a schedule is supplied. +""" + +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from dataclasses import dataclass +from numbers import Integral +from typing import Any + +import numpy as np +import pandas as pd + +from populace_dynamics.engine.loop import SyntheticPersonIdAllocator + +__all__ = [ + "ENTRY_KIND_COLUMN", + "ENTRY_KIND_IMMIGRANT", + "ENTRY_KIND_BIRTH", + "ENTRY_KIND_REALIZED_OPENER", + "ENTRY_KIND_INCUMBENT", + "EntrantSchedule", + "build_entrant_schedule", + "entrant_provenance_counters", +] + +#: The provenance column every open-addition row should carry. +ENTRY_KIND_COLUMN = "entry_kind" +ENTRY_KIND_IMMIGRANT = "immigrant_cohort" +ENTRY_KIND_BIRTH = "maternal_birth" +ENTRY_KIND_REALIZED_OPENER = "realized_opener" +ENTRY_KIND_INCUMBENT = "incumbent" + +#: V.A2 is published in thousands of persons. +_THOUSANDS = 1000.0 + +#: Columns this module owns on an entrant row. Anything else present on the +#: roster is emitted as NA, which is the same disposition the birth path uses +#: (``steps.py:485-487`` builds children on ``frame.columns`` and fills only +#: what a newborn genuinely has). +_OWNED_COLUMNS = ( + "person_id", + "year", + "age", + "sex", + "birth_year", + "weight", + "start_weight", + "synthetic_entry", + ENTRY_KIND_COLUMN, + "entry_year", + "entry_age", + "donor_person_id", + "donor_source_year", + "donor_peinusyr", + "donor_prcitshp", + "donor_penatvty", + "foreign_born", +) + + +@dataclass(frozen=True) +class EntrantSchedule: + """Scheduled entrant frames plus the audit record of how they were sized.""" + + frames: dict[int, pd.DataFrame] + alignment: dict[int, dict[str, float]] + provenance: dict[str, Any] + + def as_metadata(self) -> dict[int, pd.DataFrame]: + """The value for ``metadata[SCHEDULED_ENTRIES_KEY]``.""" + return {year: frame.copy() for year, frame in self.frames.items()} + + def total_rows(self) -> int: + return int(sum(len(frame) for frame in self.frames.values())) + + def total_weight(self) -> float: + return float( + sum(frame["weight"].sum() for frame in self.frames.values()) + ) + + +def _validate_donor(donor: pd.DataFrame) -> None: + required = { + "person_id", + "weight", + "entry_age", + "is_female", + "source_year", + "peinusyr", + "prcitshp", + "penatvty", + # Carried, never asserted: the recent-arrival band's universe is + # everyone not born in the fifty states, so it contains natives too + # (see nativity_frame.recent_arrival_donor). + "foreign_born", + } + missing = required - set(donor.columns) + if missing: + raise ValueError(f"entrant donor is missing columns {sorted(missing)}") + if donor.empty: + raise ValueError("entrant donor pool is empty") + weight = donor["weight"].to_numpy(dtype=np.float64) + if not np.isfinite(weight).all() or (weight < 0).any(): + raise ValueError("entrant donor weights must be finite and >= 0") + if not np.isfinite(weight.sum()) or weight.sum() <= 0: + raise ValueError( + "entrant donor pool needs a finite positive total weight" + ) + age = donor["entry_age"].to_numpy(dtype=np.float64) + if ( + not np.isfinite(age).all() + or (age < 0).any() + or (age != np.floor(age)).any() + or (age >= np.iinfo(np.int64).max).any() + ): + raise ValueError( + "entrant donor entry_age must be nonnegative integers" + ) + for column in ("is_female", "foreign_born"): + if not all( + isinstance(value, (bool, np.bool_)) for value in donor[column] + ): + raise ValueError(f"entrant donor {column} must contain booleans") + + +def build_entrant_schedule( + donor: pd.DataFrame, + inflow_thousands_by_year: Mapping[int, float], + *, + allocator: SyntheticPersonIdAllocator, + roster_columns: Sequence[str] | None = None, + entry_kind: str = ENTRY_KIND_IMMIGRANT, + control_provenance: Mapping[str, Any] | None = None, + donor_provenance: Mapping[str, Any] | None = None, +) -> EntrantSchedule: + """Materialize one entrant frame per activation year. + + ``donor`` supplies the explicit demographic/provenance columns; no native + donor reader or source admission is included in this experimental slice. + ``inflow_thousands_by_year`` supplies scenario controls in thousands using + the historical gross-positive-inflow convention. ``allocator`` is the + projection-wide + :class:`~populace_dynamics.engine.loop.SyntheticPersonIdAllocator`; passing + the projection's own allocator is what guarantees entrant IDs never + collide with a fitted person-keyed support. + """ + _validate_donor(donor) + if entry_kind != ENTRY_KIND_IMMIGRANT: + raise ValueError("immigrant schedules require immigrant_cohort kind") + if any( + isinstance(year, (bool, np.bool_)) or not isinstance(year, Integral) + for year in inflow_thousands_by_year + ): + raise ValueError("activation years must be integers") + years = sorted(inflow_thousands_by_year) + if not years: + raise ValueError("no activation years requested") + controls = {} + for year in years: + inflow = float(inflow_thousands_by_year[year]) + if ( + not np.isfinite(inflow) + or inflow < 0 + or not np.isfinite(inflow * _THOUSANDS) + ): + raise ValueError( + f"activation year {year} has a non-finite or negative control " + f"inflow {inflow!r}" + ) + controls[year] = inflow + + # Zero-weight donors never become demographic actors or consume IDs. + # Keep the float representation validated above; an integer sum can + # overflow before scaling even when every individual weight is valid. + donor = donor.copy() + donor["weight"] = donor["weight"].to_numpy(dtype=np.float64) + donor = donor.loc[donor["weight"] > 0].copy() + + donor_weight_total = float(donor["weight"].sum()) + donor_weight = donor["weight"].to_numpy(dtype=np.float64) + donor_proportion = donor_weight / donor_weight_total + for inflow in controls.values(): + if inflow == 0: + continue + scale = inflow * _THOUSANDS / donor_weight_total + with np.errstate(over="ignore", under="ignore"): + extrema = donor_proportion[ + [donor_weight.argmin(), donor_weight.argmax()] + ] * (inflow * _THOUSANDS) + if ( + not np.isfinite(scale) + or scale <= 0 + or not np.isfinite(extrema).all() + or (extrema <= 0).any() + ): + raise ValueError( + "control scaling must yield finite positive donor weights" + ) + entry_age = donor["entry_age"].to_numpy(dtype=np.int64) + sex = np.where(donor["is_female"].to_numpy(dtype=bool), "female", "male") + + frames: dict[int, pd.DataFrame] = {} + alignment: dict[int, dict[str, float]] = {} + for year in years: + inflow = controls[year] + target_weight = inflow * _THOUSANDS + scale = target_weight / donor_weight_total + if target_weight == 0: + alignment[year] = { + "control_inflow_thousands": inflow, + "target_weighted_persons": 0.0, + "scheduled_weighted_persons": 0.0, + "residual_persons": 0.0, + "relative_residual": 0.0, + "n_rows": 0, + "donor_weight_scale": 0.0, + } + # The existing loop rejects empty scheduled frames. Omitting the + # executable frame also preserves its ID/RNG state for this year. + continue + # Normalize before scaling: a positive subnormal common scale can + # lose material precision even though the target is representable. + weight = donor_proportion * target_weight + person_id = allocator.allocate(len(donor)) + frame_year = year - 1 + row = pd.DataFrame( + { + "person_id": person_id, + "year": np.full(len(donor), frame_year, dtype=np.int64), + "age": entry_age, + "sex": sex, + "birth_year": frame_year - entry_age, + "weight": weight, + "start_weight": weight, + "synthetic_entry": np.ones(len(donor), dtype=bool), + ENTRY_KIND_COLUMN: np.full( + len(donor), entry_kind, dtype=object + ), + "entry_year": np.full(len(donor), year, dtype=np.int64), + "entry_age": entry_age, + "donor_person_id": donor["person_id"].to_numpy(), + "donor_source_year": donor["source_year"].to_numpy(), + "donor_peinusyr": donor["peinusyr"].to_numpy(), + "donor_prcitshp": donor["prcitshp"].to_numpy(), + "donor_penatvty": donor["penatvty"].to_numpy(), + "foreign_born": donor["foreign_born"].to_numpy(dtype=bool), + } + ) + if roster_columns is not None: + for column in roster_columns: + if column not in row.columns: + row[column] = pd.NA + row = row[ + list(roster_columns) + + [c for c in row.columns if c not in set(roster_columns)] + ] + row = row.sort_values("person_id", kind="stable").reset_index( + drop=True + ) + frames[year] = row + realized = float(row["weight"].sum()) + alignment[year] = { + "control_inflow_thousands": inflow, + "target_weighted_persons": target_weight, + "scheduled_weighted_persons": realized, + "residual_persons": realized - target_weight, + "relative_residual": ( + (realized - target_weight) / target_weight + if target_weight + else 0.0 + ), + "n_rows": int(len(row)), + "donor_weight_scale": scale, + } + + provenance: dict[str, Any] = { + "method": "donor_reweighted_to_control", + "method_detail": ( + "every positive-weight donor appears once per positive-inflow " + "activation year with its weight " + "scaled by a single factor, so the cohort's weighted total equals " + "the control and its composition equals the donor exactly; no RNG " + "is consumed and no row is resampled" + ), + "sizing_basis": "trustees_va2_gross_positive_inflow", + "sizing_excludes": [ + "outflow (the engine has no emigration law)", + "adjustment of status (a reclassification, not a new person)", + "total net change (a residual with no literal composition)", + ], + "sizing_basis_disclosure": ( + "a stock-accounting inflow proxy, NOT a count of physical " + "arrivals: V.A2's temporary-or-unlawfully-present inflow counts " + "only those who remain to year-end, so the gross total understates " + "border arrivals and the cohort must not be read as one " + "(2026 OASDI Trustees Report Table V.A2; the same qualification " + "PR #218 section 0 states for this control)" + ), + "frame_coordinate": ( + "year = activation_year - 1, age = entry_age; the loop's first " + "step is a mortality draw at the entry age, then aging advances " + "to entry_age + 1 in the activation year (loop.py:236-239, " + "262-281; convention per m6_population.py:328-334)" + ), + "entry_kind": entry_kind, + "activation_years": years, + "scheduled_activation_years": sorted(frames), + "zero_inflow_years": [year for year in years if controls[year] == 0], + "id_allocation": ( + "projection-wide SyntheticPersonIdAllocator; loop.py:52-61 raises " + "on any overlap with reserved_real_ids" + ), + "gated": False, + "report_only": True, + } + if control_provenance is not None: + provenance["control"] = dict(control_provenance) + if donor_provenance is not None: + provenance["donor"] = dict(donor_provenance) + return EntrantSchedule( + frames=frames, alignment=alignment, provenance=provenance + ) + + +def entrant_provenance_counters( + frames: Mapping[int, pd.DataFrame], +) -> dict[str, Any]: + """Counts by ``entry_kind`` and by year, for the run artifact. + + This is the replacement for inferring an entrant's kind from ID + arithmetic. ``harness/m6_runner.py:1026-1032`` and ``:1210-1216`` publish + a hardcoded ``"immigrant_cohorts": 0`` justified by a comment that holds + only while no schedule exists; a counter keyed on an explicit column + survives the schedule existing. + """ + by_kind: dict[str, int] = {} + by_year: dict[int, dict[str, Any]] = {} + weighted_by_kind: dict[str, float] = {} + for year, frame in sorted(frames.items()): + if ENTRY_KIND_COLUMN not in frame.columns: + raise ValueError( + f"scheduled entries {year} carry no {ENTRY_KIND_COLUMN!r} " + "column; entrant provenance cannot be counted" + ) + if ( + not frame[ENTRY_KIND_COLUMN] + .isin( + { + ENTRY_KIND_BIRTH, + ENTRY_KIND_INCUMBENT, + ENTRY_KIND_IMMIGRANT, + ENTRY_KIND_REALIZED_OPENER, + } + ) + .all() + ): + raise ValueError( + "scheduled provenance has missing or unknown entry_kind" + ) + counts = frame[ENTRY_KIND_COLUMN].value_counts().to_dict() + by_year[int(year)] = { + "n_rows": int(len(frame)), + "weighted_persons": float(frame["weight"].sum()), + "by_entry_kind": {str(k): int(v) for k, v in counts.items()}, + } + for kind, count in counts.items(): + by_kind[str(kind)] = by_kind.get(str(kind), 0) + int(count) + mask = frame[ENTRY_KIND_COLUMN] == kind + weighted_by_kind[str(kind)] = weighted_by_kind.get( + str(kind), 0.0 + ) + float(frame.loc[mask, "weight"].sum()) + return { + "n_rows_by_entry_kind": by_kind, + "weighted_persons_by_entry_kind": weighted_by_kind, + "by_activation_year": by_year, + "immigrant_cohorts": by_kind.get(ENTRY_KIND_IMMIGRANT, 0), + "counter_basis": ( + f"explicit {ENTRY_KIND_COLUMN!r} column, not ID arithmetic" + ), + } diff --git a/tests/README-tiers.md b/tests/README-tiers.md index a5ef9187..6cbfa5c1 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,563 | +| `unit` | 1,641 | | `artifact` | 2,668 | | `integration_psid` | 848 | | `reproduction_legacy` | 520 | | `oracle_policyengine` | 159 | -| **Total** | **5,758** | +| **Total** | **5,836** | diff --git a/tests/estimates/test_birth_evidence_artifact.py b/tests/estimates/test_birth_evidence_artifact.py index d4e838a1..70e78c47 100644 --- a/tests/estimates/test_birth_evidence_artifact.py +++ b/tests/estimates/test_birth_evidence_artifact.py @@ -88,6 +88,8 @@ def test_post_review_sources_are_outside_historical_reducer_identity(): Path("src/populace_dynamics/estimates/anchor_context_registry.py"), Path("src/populace_dynamics/estimates/anchor_context_rehearsal.py"), Path("src/populace_dynamics/estimates/anchor_context_report.py"), + Path("src/populace_dynamics/engine/entrant_schedule.py"), + Path("src/populace_dynamics/engine/entrant_domains.py"), ) assert reducer.POST_REVIEW_SHARED_SOURCE_BLOBS == { Path( @@ -205,6 +207,13 @@ def test_psid_identity_exclusions_are_unreachable_from_birth_evidence(): "historically excluded PSID modules became reachable from the " f"birth-evidence reducer: {sorted(psid_exclusions & reachable)}" ) + entrant_modules = { + "populace_dynamics.engine.entrant_schedule", + "populace_dynamics.engine.entrant_domains", + } + assert entrant_modules.issubset(module_paths) + assert entrant_modules.isdisjoint(reachable) + assert "populace_dynamics.engine.steps" in reachable def test_reducer_accepts_explicit_unresolved_upstream_boundary(): diff --git a/tests/test_entrant_domains.py b/tests/test_entrant_domains.py new file mode 100644 index 00000000..3896b7a0 --- /dev/null +++ b/tests/test_entrant_domains.py @@ -0,0 +1,381 @@ +"""Tests for the entrant exclusion adapters. + +The load-bearing test is +``test_without_the_adapter_every_entrant_over_50_draws_a_claim_age``: it runs +the historical ``apply_claiming`` step directly on entrant rows and shows the +unconditional draw happening, which is the failure the adapter exists to +prevent. +""" + +from __future__ import annotations + +import numpy as np +import pandas as pd +import pytest + +from populace_dynamics.engine import entrant_domains as edm +from populace_dynamics.engine.earnings_domain import EARNINGS_DOMAIN_COLUMN +from populace_dynamics.engine.entrant_schedule import ( + ENTRY_KIND_BIRTH, + ENTRY_KIND_COLUMN, + ENTRY_KIND_IMMIGRANT, +) +from populace_dynamics.engine.loop import ( + PeriodContext, + SyntheticPersonIdAllocator, +) +from populace_dynamics.engine.rng import ProjectionRNGRegistry +from populace_dynamics.engine.steps import ( + ClaimingSchedule, + apply_claiming, + materialize_maternal_births, +) + + +def _roster() -> pd.DataFrame: + """Four incumbents and four entrants, spanning the claiming threshold.""" + return pd.DataFrame( + { + "person_id": [1, 2, 3, 4, 101, 102, 103, 104], + "year": [2026] * 8, + "age": [30, 55, 62, 70, 30, 55, 62, 70], + "sex": ["female", "male", "female", "male"] * 2, + "weight": [1.0] * 8, + ENTRY_KIND_COLUMN: [None] * 4 + [ENTRY_KIND_IMMIGRANT] * 4, + } + ) + + +def _claiming_schedule() -> ClaimingSchedule: + return ClaimingSchedule( + pmf={ + ("female", 2026): {62: 0.5, 67: 0.5}, + ("male", 2026): {62: 0.5, 67: 0.5}, + } + ) + + +def _context() -> PeriodContext: + return PeriodContext(period_index=1, year=2026, draw_index=0, metadata={}) + + +# -------------------------------------------------------------------------- +# Membership +# -------------------------------------------------------------------------- +def test_entrant_mask_reads_the_provenance_column(): + mask = edm.entrant_mask(_roster()) + assert mask.tolist() == [False] * 4 + [True] * 4 + + +def test_a_frame_without_the_column_has_no_entrants(): + """Safe to call on a closed-panel roster that never saw a schedule.""" + frame = _roster().drop(columns=[ENTRY_KIND_COLUMN]) + assert not edm.entrant_mask(frame).any() + + +def test_a_missing_value_is_an_incumbent(): + frame = _roster() + frame.loc[4, ENTRY_KIND_COLUMN] = pd.NA + assert edm.entrant_mask(frame).tolist() == [False] * 5 + [True] * 3 + + +def test_births_are_not_immigrant_entrants_by_default(): + frame = _roster() + frame.loc[4, ENTRY_KIND_COLUMN] = ENTRY_KIND_BIRTH + assert edm.entrant_mask(frame).sum() == 3 + assert ( + edm.entrant_mask( + frame, entry_kinds=(ENTRY_KIND_IMMIGRANT, ENTRY_KIND_BIRTH) + ).sum() + == 4 + ) + + +def test_the_three_exclusion_id_sets_are_the_entrant_ids(): + frame = _roster() + expected = {101, 102, 103, 104} + assert edm.excluded_fertility_ids(frame) == expected + assert edm.excluded_claiming_ids(frame) == expected + assert edm.excluded_disability_ids(frame) == expected + + +# -------------------------------------------------------------------------- +# Claiming: the hazard, and the adapter that removes it +# -------------------------------------------------------------------------- +def test_without_the_adapter_every_entrant_over_50_draws_a_claim_age(): + """The unconditional draw at ``steps.py:411-413``, demonstrated. + + This establishes behavioral claiming-age assignment only. It does not + calculate benefits or establish a direction for aggregate fiscal effects. + """ + out = apply_claiming( + _roster(), + _context(), + np.random.default_rng(0), + schedule=_claiming_schedule(), + ) + entrants = out[out[ENTRY_KIND_COLUMN] == ENTRY_KIND_IMMIGRANT] + over_50 = entrants[entrants["age"] >= 50] + assert len(over_50) == 3 + assert over_50["claim_age"].notna().all() + assert bool(out.loc[out["person_id"] == 104, "claimed"].iloc[0]) is True + + +def test_the_adapter_leaves_entrants_unclaimed_and_incumbents_untouched(): + adapter = edm.EntrantClaimingAdapter( + lambda frame, context, rng: apply_claiming( + frame, context, rng, schedule=_claiming_schedule() + ) + ) + out = adapter(_roster(), _context(), np.random.default_rng(0)) + entrants = out[out[ENTRY_KIND_COLUMN] == ENTRY_KIND_IMMIGRANT] + incumbents = out[out[ENTRY_KIND_COLUMN].isna()] + + assert entrants["claim_age"].isna().all() + assert not entrants["claimed"].any() + assert entrants["claim_year"].isna().all() + # The incumbents get exactly what the historical adapter gives them. + assert incumbents.loc[incumbents["age"] >= 50, "claim_age"].notna().all() + + +def test_the_adapter_is_a_passthrough_without_entrants(): + frame = _roster().drop(columns=[ENTRY_KIND_COLUMN]) + step = lambda f, c, r: apply_claiming( # noqa: E731 + f, c, r, schedule=_claiming_schedule() + ) + adapter = edm.EntrantClaimingAdapter(step) + direct = step(frame, _context(), np.random.default_rng(0)) + through = adapter(frame, _context(), np.random.default_rng(0)) + pd.testing.assert_frame_equal(direct, through) + + +def test_the_adapter_preserves_the_roster_and_person_sort(): + adapter = edm.EntrantClaimingAdapter( + lambda frame, context, rng: apply_claiming( + frame, context, rng, schedule=_claiming_schedule() + ) + ) + out = adapter(_roster(), _context(), np.random.default_rng(0)) + assert len(out) == 8 + assert out["person_id"].is_monotonic_increasing + + +# -------------------------------------------------------------------------- +# Earnings domain +# -------------------------------------------------------------------------- +def test_entrants_marked_inside_the_earnings_domain_are_rejected(): + frame = _roster() + frame[EARNINGS_DOMAIN_COLUMN] = [True] * 4 + [False, False, True, False] + with pytest.raises(ValueError, match="marked inside the fitted earnings"): + edm.assert_entrants_out_of_earnings_domain(frame) + + +def test_entrants_outside_the_earnings_domain_pass(): + frame = _roster() + frame[EARNINGS_DOMAIN_COLUMN] = [True] * 4 + [False] * 4 + assert edm.assert_entrants_out_of_earnings_domain(frame) == 4 + + +def test_a_frame_without_the_domain_column_still_counts_entrants(): + assert edm.assert_entrants_out_of_earnings_domain(_roster()) == 4 + + +# -------------------------------------------------------------------------- +# Benefit suppression +# -------------------------------------------------------------------------- +def test_entrant_benefit_outputs_are_missing_not_zero(): + frame = _roster() + frame["aime"] = 100.0 + frame["pia"] = 50.0 + out = edm.suppress_entrant_benefit_outputs(frame) + entrants = out[out[ENTRY_KIND_COLUMN] == ENTRY_KIND_IMMIGRANT] + assert entrants["aime"].isna().all() + assert entrants["pia"].isna().all() + # A zero would be a measurement; prior US covered earnings are censored. + assert not (entrants["aime"] == 0).any() + assert (out[out[ENTRY_KIND_COLUMN].isna()]["aime"] == 100.0).all() + + +# -------------------------------------------------------------------------- +# The report +# -------------------------------------------------------------------------- +def test_exclusion_report_covers_all_four_domains(): + report = edm.exclusion_report(_roster()) + assert report.n_entrants == 4 + assert set(report.excluded) == set(edm.EXCLUDED_DOMAINS) + assert ( + report.excluded["claiming_eligibility"][ + "n_would_have_drawn_a_claim_age" + ] + == 3 + ) + + +def test_exclusion_report_states_it_is_report_only_and_not_a_behaviour_model(): + record = edm.exclusion_report(_roster()).as_dict() + assert record["gated"] is False + assert record["execution_verified"] is False + assert record["status"] == "inventory_only" + assert "OUTSIDE the estimand" in record["interpretation"] + assert any( + "mortality" in entry + for entry in record["intended_demographic_domains"] + ) + + +def test_every_excluded_domain_names_the_code_that_makes_it_unfitted(): + for domain, reason in edm.EXCLUDED_DOMAINS.items(): + assert ".py:" in reason, domain + + +@pytest.mark.parametrize( + "column,value", + [ + ("claim_age", 62), + ("claim_year", 2020), + ("claimed", True), + ("claimed", "False"), + ("di_converted", True), + ], +) +def test_existing_entrant_claim_state_is_rejected_before_any_step( + column, value +): + frame = _roster() + frame[column] = pd.Series([pd.NA] * len(frame), dtype="object") + frame.loc[4, column] = value + original = frame.copy(deep=True) + rng = np.random.default_rng(9) + untouched_rng = np.random.default_rng(9) + + def forbidden(*args): + raise AssertionError("contradictory state reached the incumbent step") + + with pytest.raises(ValueError, match="excluded entrants have"): + edm.EntrantClaimingAdapter(forbidden)(frame, _context(), rng) + pd.testing.assert_frame_equal(frame, original) + assert rng.bytes(32) == untouched_rng.bytes(32) + + +def test_an_all_entrant_frame_never_calls_the_incumbent_step(): + frame = _roster().iloc[4:].copy() + + def forbidden(*args): + raise AssertionError("all-entrant frame invoked the incumbent step") + + result = edm.EntrantClaimingAdapter(forbidden)( + frame, _context(), np.random.default_rng(0) + ) + assert result["claim_age"].isna().all() + assert result["claim_year"].isna().all() + assert not result["claimed"].any() + assert result["person_id"].tolist() == frame["person_id"].tolist() + + +@pytest.mark.parametrize("kind", [None, "immgrant_cohort", 1]) +def test_synthetic_entrants_cannot_lose_their_provenance(kind): + frame = _roster().astype({ENTRY_KIND_COLUMN: "object"}) + frame["synthetic_entry"] = [False] * 4 + [True] * 4 + frame.loc[4, ENTRY_KIND_COLUMN] = kind + with pytest.raises(ValueError, match="entry_kind"): + edm.entrant_mask(frame) + + +def test_losing_the_entire_provenance_column_is_detected_for_synthetic_rows(): + frame = _roster().drop(columns=ENTRY_KIND_COLUMN) + frame["synthetic_entry"] = [False] * 4 + [True] * 4 + with pytest.raises(ValueError, match="require explicit entry_kind"): + edm.exclusion_report(frame) + + +def test_incumbent_claims_and_random_consumption_match_direct_subset(): + frame = _roster() + + def step(f, c, r): + return apply_claiming(f, c, r, schedule=_claiming_schedule()) + + first_rng = np.random.default_rng(11) + direct_rng = np.random.default_rng(11) + result = edm.EntrantClaimingAdapter(step)(frame, _context(), first_rng) + direct = step(frame.iloc[:4].copy(), _context(), direct_rng) + pd.testing.assert_frame_equal(result.iloc[:4], direct, check_dtype=False) + assert first_rng.bytes(32) == direct_rng.bytes(32) + + +def test_claiming_inventory_counts_missing_plans_without_claiming_execution(): + frame = _roster() + frame["claim_age"] = pd.array([pd.NA] * len(frame), dtype="Int64") + frame.loc[5, "claim_age"] = 62 + report = edm.exclusion_report(frame).as_dict() + assert ( + report["excluded_domains"]["claiming_eligibility"][ + "n_would_have_drawn_a_claim_age" + ] + == 2 + ) + assert not report["execution_verified"] + + +@pytest.mark.parametrize( + "selectors", [("immgrant_cohort",), "immigrant_cohort", ()] +) +def test_invalid_selectors_cannot_silently_bypass_the_claiming_guard( + selectors, +): + def forbidden(*args): + raise AssertionError("bad selectors reached the incumbent step") + + with pytest.raises(ValueError, match="entry_kind|entry kind"): + edm.EntrantClaimingAdapter(forbidden, entry_kinds=selectors)( + _roster(), _context(), np.random.default_rng(0) + ) + + +def test_birth_provenance_boundary_preserves_historical_materialization(): + frame = _roster() + frame["synthetic_entry"] = [False] * 4 + [True] * 4 + births = pd.DataFrame({"parent_person_id": [1], "birth_year": [2026]}) + context = PeriodContext( + 1, + 2026, + 0, + {"synthetic_id_allocator": SyntheticPersonIdAllocator(105)}, + rng_registry=ProjectionRNGRegistry(0, 2), + person_ordinals={ + pid: index for index, pid in enumerate(frame["person_id"]) + }, + ) + raw_context = PeriodContext( + 1, 2026, 0, {"synthetic_id_allocator": SyntheticPersonIdAllocator(105)} + ) + raw = materialize_maternal_births( + frame, births, raw_context, np.random.default_rng(4) + ) + with pytest.raises(ValueError, match="require explicit entry_kind"): + edm.entrant_mask(raw) + labeled = edm.materialize_births_with_provenance( + frame, births, context, np.random.default_rng(4) + ) + pd.testing.assert_frame_equal( + labeled.drop(columns=ENTRY_KIND_COLUMN), + raw.drop(columns=ENTRY_KIND_COLUMN), + ) + child = labeled.loc[labeled["person_id"] == 105].iloc[0] + assert child[ENTRY_KIND_COLUMN] == ENTRY_KIND_BIRTH + assert child["age"] == 0 + assert child["parent_person_id"] == 1 + assert context.synthetic_id_allocator.next_id == 106 + assert edm.excluded_claiming_ids(labeled) == {101, 102, 103, 104} + + def step(f, c, r): + return apply_claiming(f, c, r, schedule=_claiming_schedule()) + + result = edm.EntrantClaimingAdapter(step)( + labeled, context, np.random.default_rng(0) + ) + direct = step(frame.iloc[:4], context, np.random.default_rng(0)) + pd.testing.assert_frame_equal( + result.iloc[:4][direct.columns], direct, check_dtype=False + ) + assert not result.loc[result["person_id"] == 105, "claimed"].iloc[0] + assert edm.exclusion_report(result).n_entrants == 4 diff --git a/tests/test_entrant_schedule.py b/tests/test_entrant_schedule.py new file mode 100644 index 00000000..a1ae80b2 --- /dev/null +++ b/tests/test_entrant_schedule.py @@ -0,0 +1,550 @@ +"""Tests for the entrant schedule builder and its seam contract. + +The final test runs a real :class:`ProjectionEngine` over a built schedule. +That is the point of the piece: the seam already exists and is tested, so a +schedule builder is only correct if the loop accepts what it produces. +""" + +from __future__ import annotations + +import numpy as np +import pandas as pd +import pytest + +from populace_dynamics.engine import entrant_schedule as esm +from populace_dynamics.engine.entrant_domains import ( + EntrantClaimingAdapter, + suppress_entrant_benefit_outputs, +) +from populace_dynamics.engine.loop import ( + SCHEDULED_ENTRIES_KEY, + MaritalStepResult, + PeriodModules, + ProjectionEngine, + SyntheticPersonIdAllocator, +) +from populace_dynamics.engine.steps import advance_age + + +def _donor(n: int = 40, seed: int = 2) -> pd.DataFrame: + """An explicit donor pool. + + The schedule builder's contract is the donor COLUMNS, not the frame it + came from, so the fixture is built directly rather than sampled out of a + synthetic frame -- that keeps these tests independent of how thinly a + random frame happens to populate the recent-arrival band. + The native frame-reader path remains outside this source-only slice. + """ + rng = np.random.default_rng(seed) + return pd.DataFrame( + { + "person_id": np.arange(n), + "weight": rng.uniform(500.0, 5000.0, size=n), + "entry_age": rng.integers(0, 80, size=n).astype(np.int64), + "is_female": rng.random(n) < 0.5, + "source_year": np.full(n, 2024, dtype=np.int64), + "peinusyr": np.full(n, 28, dtype=np.int64), + "prcitshp": rng.choice([4, 5], size=n).astype(np.int64), + "penatvty": rng.integers(100, 556, size=n).astype(np.int64), + "foreign_born": np.ones(n, dtype=bool), + } + ) + + +def _schedule(years=(2026, 2027), start_id: int = 90_000_000, **kwargs): + donor = _donor() + inflow = {year: 1000.0 + 10 * index for index, year in enumerate(years)} + allocator = SyntheticPersonIdAllocator(start_id) + return ( + esm.build_entrant_schedule( + donor, inflow, allocator=allocator, **kwargs + ), + donor, + allocator, + ) + + +# -------------------------------------------------------------------------- +# The seam contract, which loop.py enforces +# -------------------------------------------------------------------------- +def test_frames_carry_the_year_before_activation(): + schedule, _, _ = _schedule(years=(2026, 2027)) + for year, frame in schedule.frames.items(): + assert set(frame["year"].unique()) == {year - 1} + + +def test_person_ids_are_unique_across_every_activation_year(): + schedule, _, _ = _schedule(years=(2026, 2027, 2028)) + pooled = pd.concat(schedule.frames.values(), ignore_index=True) + assert not pooled["person_id"].duplicated().any() + + +def test_ids_come_from_the_allocator_and_advance_it(): + schedule, donor, allocator = _schedule(years=(2026, 2027)) + assert allocator.next_id == 90_000_000 + 2 * len(donor) + pooled = pd.concat(schedule.frames.values(), ignore_index=True) + assert pooled["person_id"].min() >= 90_000_000 + + +def test_the_allocator_still_guards_the_reserved_real_namespace(): + donor = _donor() + allocator = SyntheticPersonIdAllocator( + 10, reserved_real_ids=frozenset({12}) + ) + with pytest.raises(RuntimeError, match="overlap the reserved real-person"): + esm.build_entrant_schedule(donor, {2026: 1000.0}, allocator=allocator) + + +def test_rows_are_person_sorted(): + schedule, _, _ = _schedule() + for frame in schedule.frames.values(): + assert frame["person_id"].is_monotonic_increasing + + +def test_birth_year_is_consistent_with_the_frame_coordinate(): + schedule, _, _ = _schedule(years=(2026,)) + frame = schedule.frames[2026] + assert ( + frame["birth_year"].to_numpy() + == frame["year"].to_numpy() - frame["age"].to_numpy() + ).all() + + +def test_sex_uses_the_roster_string_encoding(): + schedule, _, _ = _schedule() + for frame in schedule.frames.values(): + assert set(frame["sex"].unique()) <= {"female", "male"} + + +# -------------------------------------------------------------------------- +# Sizing +# -------------------------------------------------------------------------- +def test_cohort_weight_equals_the_control_in_persons(): + schedule, _, _ = _schedule(years=(2026,)) + record = schedule.alignment[2026] + assert record["target_weighted_persons"] == 1000.0 * 1000.0 + assert record["scheduled_weighted_persons"] == pytest.approx( + 1_000_000.0, rel=1e-12 + ) + assert abs(record["relative_residual"]) < 1e-12 + + +def test_reweighting_preserves_the_donor_composition_exactly(): + schedule, donor, _ = _schedule(years=(2026,)) + frame = schedule.frames[2026] + donor_share = donor["weight"].to_numpy() / donor["weight"].sum() + merged = frame.sort_values("donor_person_id") + scheduled_share = merged["weight"].to_numpy() / merged["weight"].sum() + expected = ( + donor.sort_values("person_id")["weight"].to_numpy() + / donor["weight"].sum() + ) + assert np.allclose(np.sort(donor_share), np.sort(scheduled_share)) + assert np.allclose(scheduled_share, expected) + + +def test_no_rng_is_consumed(): + """Two builds from independent allocators agree row for row.""" + donor = _donor() + first = esm.build_entrant_schedule( + donor, {2026: 1340.0}, allocator=SyntheticPersonIdAllocator(1_000) + ) + second = esm.build_entrant_schedule( + donor, {2026: 1340.0}, allocator=SyntheticPersonIdAllocator(1_000) + ) + pd.testing.assert_frame_equal(first.frames[2026], second.frames[2026]) + + +def test_negative_or_nonfinite_control_is_rejected(): + donor = _donor() + with pytest.raises(ValueError, match="negative control inflow"): + esm.build_entrant_schedule( + donor, {2026: -1.0}, allocator=SyntheticPersonIdAllocator(10) + ) + + +def test_empty_donor_is_rejected(): + donor = _donor().iloc[0:0] + with pytest.raises(ValueError, match="donor pool is empty"): + esm.build_entrant_schedule( + donor, {2026: 1.0}, allocator=SyntheticPersonIdAllocator(10) + ) + + +def test_no_activation_years_is_rejected(): + with pytest.raises(ValueError, match="no activation years"): + esm.build_entrant_schedule( + _donor(), {}, allocator=SyntheticPersonIdAllocator(10) + ) + + +# -------------------------------------------------------------------------- +# Provenance counters +# -------------------------------------------------------------------------- +def test_counters_read_the_entry_kind_column_not_id_arithmetic(): + schedule, donor, _ = _schedule(years=(2026, 2027)) + counters = esm.entrant_provenance_counters(schedule.frames) + assert counters["immigrant_cohorts"] == 2 * len(donor) + assert counters["n_rows_by_entry_kind"] == { + esm.ENTRY_KIND_IMMIGRANT: 2 * len(donor) + } + assert "not ID arithmetic" in counters["counter_basis"] + + +def test_counters_reject_a_frame_without_the_provenance_column(): + schedule, _, _ = _schedule(years=(2026,)) + frame = schedule.frames[2026].drop(columns=[esm.ENTRY_KIND_COLUMN]) + with pytest.raises(ValueError, match="entrant provenance cannot be"): + esm.entrant_provenance_counters({2026: frame}) + + +def test_provenance_states_the_sizing_basis_and_exclusions(): + schedule, _, _ = _schedule() + provenance = schedule.provenance + assert provenance["sizing_basis"] == "trustees_va2_gross_positive_inflow" + assert provenance["report_only"] is True + assert provenance["gated"] is False + joined = " ".join(provenance["sizing_excludes"]) + assert "emigration" in joined and "reclassification" in joined + + +def test_provenance_refuses_to_call_the_control_an_arrival_count(): + """The control is a stock-accounting proxy, and must say so. + + V.A2's temporary-or-unlawfully-present inflow counts only those who + remain to year-end, so the gross total is not a count of physical + arrivals. Naming the basis without that qualification would invite the + cohort to be read as arrivals, which is the same class of error as + reading an ssa_area_proxy as resident-aligned. + """ + schedule, _, _ = _schedule() + disclosure = schedule.provenance["sizing_basis_disclosure"] + assert "NOT a count of physical" in disclosure + assert "remain to year-end" in disclosure + + +# -------------------------------------------------------------------------- +# The loop actually accepts it +# -------------------------------------------------------------------------- +def test_a_built_schedule_activates_through_the_real_projection_engine(): + donor = _donor(n=200, seed=4) + allocator = SyntheticPersonIdAllocator(90_000_000) + schedule = esm.build_entrant_schedule( + donor, {2026: 1340.0, 2027: 1350.0}, allocator=allocator + ) + + def mortality(frame, context, rng): + del context, rng + return frame.sort_values("person_id").reset_index(drop=True) + + def marital(frame, context, rng): + del frame, context, rng + return MaritalStepResult(pd.DataFrame(), pd.DataFrame()) + + def passthrough(frame, context, rng): + del context, rng + return frame.copy() + + modules = PeriodModules( + mortality=mortality, + aging=advance_age, + marital_core=marital, + fertility=lambda frame, context, result, rng: frame.copy(), + disability=passthrough, + earnings=passthrough, + claiming=passthrough, + household_composition=( + lambda frame, context, result, rng: frame.copy() + ), + ) + initial = pd.DataFrame( + { + "person_id": [1, 2], + "year": [2025, 2025], + "age": [40, 41], + "sex": ["female", "male"], + "weight": [1.0, 1.0], + } + ) + result = ProjectionEngine(modules).project( + initial, + end_year=2027, + draw_index=0, + metadata={SCHEDULED_ENTRIES_KEY: schedule.as_metadata()}, + ) + + # 2 incumbents, then + one cohort in 2026, then + another in 2027. + assert [len(frame) for frame in result.slices] == [ + 2, + 2 + len(donor), + 2 + 2 * len(donor), + ] + final = result.slices[-1] + assert set(final["year"].unique()) == {2027} + assert not final["person_id"].duplicated().any() + entrants = final[final[esm.ENTRY_KIND_COLUMN].notna()] + assert len(entrants) == 2 * len(donor) + + +def test_the_entrant_faces_mortality_at_its_entry_age_before_aging(): + """The seam's fixed convention, pinned rather than assumed. + + A row scheduled with ``age = entry_age`` at ``year = y - 1`` is seen by + the mortality step at ``entry_age``; only then does aging advance it. + """ + donor = _donor(n=120, seed=6) + schedule = esm.build_entrant_schedule( + donor, {2026: 1000.0}, allocator=SyntheticPersonIdAllocator(90_000_000) + ) + seen: list[np.ndarray] = [] + + def mortality(frame, context, rng): + del context, rng + ordered = frame.sort_values("person_id").reset_index(drop=True) + seen.append(ordered["age"].to_numpy(dtype=np.int64).copy()) + return ordered + + def marital(frame, context, rng): + del frame, context, rng + return MaritalStepResult(pd.DataFrame(), pd.DataFrame()) + + def passthrough(frame, context, rng): + del context, rng + return frame.copy() + + modules = PeriodModules( + mortality=mortality, + aging=advance_age, + marital_core=marital, + fertility=lambda frame, context, result, rng: frame.copy(), + disability=passthrough, + earnings=passthrough, + claiming=passthrough, + household_composition=( + lambda frame, context, result, rng: frame.copy() + ), + ) + initial = pd.DataFrame( + { + "person_id": [1], + "year": [2025], + "age": [30], + "sex": ["female"], + "weight": [1.0], + } + ) + result = ProjectionEngine(modules).project( + initial, + end_year=2026, + draw_index=0, + metadata={SCHEDULED_ENTRIES_KEY: schedule.as_metadata()}, + ) + scheduled_ages = np.sort( + schedule.frames[2026]["age"].to_numpy(dtype=np.int64) + ) + # The mortality step saw the entry ages themselves, alongside the one + # incumbent (age 30). Compare the full multiset so an entrant aged 0 is + # not silently confused with the incumbent by position. + expected_seen = np.sort(np.concatenate([scheduled_ages, [30]])) + assert np.array_equal(np.sort(seen[0]), expected_seen) + # ...and the activation-year slice carries entry_age + 1. + final = result.slices[-1] + entrant_ages = np.sort( + final.loc[final[esm.ENTRY_KIND_COLUMN].notna(), "age"].to_numpy( + dtype=np.int64 + ) + ) + assert np.array_equal(entrant_ages, scheduled_ages + 1) + + +def test_zero_inflow_has_an_alignment_record_without_rows_or_ids(): + allocator = SyntheticPersonIdAllocator(90_000_000) + schedule = esm.build_entrant_schedule( + _donor(), {2026: 0.0, 2027: 0.0}, allocator=allocator + ) + assert schedule.frames == schedule.as_metadata() == {} + assert schedule.total_rows() == schedule.total_weight() == 0 + assert allocator.next_id == 90_000_000 + assert set(schedule.alignment) == {2026, 2027} + assert all(record["n_rows"] == 0 for record in schedule.alignment.values()) + assert schedule.provenance["zero_inflow_years"] == [2026, 2027] + assert schedule.provenance["scheduled_activation_years"] == [] + + +def test_zero_years_and_donors_do_not_change_later_cohort_ids(): + donor = _donor() + donor.loc[0, "weight"] = 0.0 + first = esm.build_entrant_schedule( + donor, + {2026: 0.0, 2027: 1000.0}, + allocator=SyntheticPersonIdAllocator(90_000_000), + ) + direct = esm.build_entrant_schedule( + donor.iloc[1:], + {2027: 1000.0}, + allocator=SyntheticPersonIdAllocator(90_000_000), + ) + assert set(first.frames) == {2027} + pd.testing.assert_frame_equal(first.frames[2027], direct.frames[2027]) + assert first.total_rows() == len(donor) - 1 + assert (first.frames[2027]["weight"] > 0).all() + + +@pytest.mark.parametrize("control", [-1.0, np.nan, np.inf, 1e308]) +def test_all_controls_are_checked_before_allocating_any_ids(control): + allocator = SyntheticPersonIdAllocator(1000) + with pytest.raises(ValueError, match="control inflow"): + esm.build_entrant_schedule( + _donor(), {2026: 1.0, 2027: control}, allocator=allocator + ) + assert allocator.next_id == 1000 + + +@pytest.mark.parametrize("year", [2026.5, "2026", True]) +def test_activation_years_are_not_silently_coerced(year): + allocator = SyntheticPersonIdAllocator(1000) + with pytest.raises(ValueError, match="activation years must be integers"): + esm.build_entrant_schedule(_donor(), {year: 1.0}, allocator=allocator) + assert allocator.next_id == 1000 + + +@pytest.mark.parametrize("age", [-1, 10.5, np.nan, np.inf]) +def test_invalid_donor_ages_do_not_become_integer_demographic_states(age): + donor = _donor().astype({"entry_age": "float64"}) + donor.loc[0, "entry_age"] = age + with pytest.raises(ValueError, match="entry_age must be nonnegative"): + esm.build_entrant_schedule( + donor, {2026: 1.0}, allocator=SyntheticPersonIdAllocator(1000) + ) + + +@pytest.mark.parametrize("column", ["is_female", "foreign_born"]) +@pytest.mark.parametrize("value", ["False", pd.NA, 2]) +def test_donor_flags_are_not_truthiness_coerced(column, value): + donor = _donor().astype({column: "object"}) + donor.loc[0, column] = value + with pytest.raises(ValueError, match=f"{column} must contain booleans"): + esm.build_entrant_schedule( + donor, {2026: 1.0}, allocator=SyntheticPersonIdAllocator(1000) + ) + + +def test_a_mislabeled_immigrant_schedule_is_rejected(): + with pytest.raises(ValueError, match="require immigrant_cohort kind"): + esm.build_entrant_schedule( + _donor(), + {2026: 1.0}, + allocator=SyntheticPersonIdAllocator(1000), + entry_kind=esm.ENTRY_KIND_BIRTH, + ) + + +@pytest.mark.parametrize("kind", [None, "immgrant_cohort"]) +def test_provenance_counts_reject_missing_or_unknown_kinds(kind): + schedule, _, _ = _schedule(years=(2026,)) + schedule.frames[2026].loc[0, esm.ENTRY_KIND_COLUMN] = kind + with pytest.raises(ValueError, match="missing or unknown entry_kind"): + esm.entrant_provenance_counters(schedule.frames) + + +@pytest.mark.parametrize("weight,inflow", [(1e-300, 1e300), (1e300, 1e-300)]) +def test_unrepresentable_scaling_fails_before_cohort_allocation( + weight, inflow +): + donor = _donor(n=1) + donor["weight"] = weight + allocator = SyntheticPersonIdAllocator(1000) + with pytest.raises(ValueError, match="finite positive donor weights"): + esm.build_entrant_schedule(donor, {2026: inflow}, allocator=allocator) + assert allocator.next_id == 1000 + + +def test_large_integer_donor_weights_use_the_validated_float_sum(): + donor = _donor(n=2) + donor["weight"] = np.array([2**62, 2**62], dtype=np.int64) + schedule = esm.build_entrant_schedule( + donor, {2026: 1.0}, allocator=SyntheticPersonIdAllocator(1000) + ) + assert schedule.frames[2026]["weight"].tolist() == [500.0, 500.0] + assert schedule.alignment[2026]["residual_persons"] == 0.0 + + +def test_positive_subnormal_scale_does_not_distort_the_control(): + donor = _donor(n=1) + donor["weight"] = 2.0**1023 + allocator = SyntheticPersonIdAllocator(1000) + schedule = esm.build_entrant_schedule( + donor, {2026: 1.0, 2027: 2.0**-60}, allocator=allocator + ) + assert schedule.frames[2026]["weight"].iloc[0] == 1000.0 + assert schedule.frames[2027]["weight"].iloc[0] == 1000.0 * 2.0**-60 + assert schedule.alignment[2027]["residual_persons"] == 0.0 + assert allocator.next_id == 1002 + + +def test_later_cohorts_activate_after_extinction_with_unknown_benefits(): + """Real loop mechanics with synthetic survival and benefit placeholders. + + The placeholder values test output suppression only; no statutory benefit + calculator or native fitted transition is exercised. + """ + donor = _donor(n=3) + allocator = SyntheticPersonIdAllocator(1000) + schedule = esm.build_entrant_schedule( + donor, {2026: 0.0, 2027: 1.0, 2028: 2.0}, allocator=allocator + ) + seen = {} + + def mortality(frame, context, rng): + seen[context.year] = frame.copy() + if context.year == 2026: + return frame.iloc[:0].copy() + # A later cohort must activate even when the previous cohort dies. + if context.year == 2028: + return frame.loc[frame["entry_year"] == 2028].copy() + return frame.copy() + + def forbidden_claiming(*args): + raise AssertionError("unsupported entrants reached claiming behavior") + + def passthrough(frame, context, rng): + return frame.copy() + + def marital(frame, context, rng): + return MaritalStepResult(frame.copy(), pd.DataFrame()) + + modules = PeriodModules( + mortality=mortality, + aging=advance_age, + marital_core=marital, + fertility=lambda f, c, m, r: f.copy(), + disability=passthrough, + earnings=lambda f, c, r: f.assign(aime=10.0, pia=20.0, benefit=30.0), + claiming=EntrantClaimingAdapter( + lambda f, c, r: f if f.empty else forbidden_claiming(f, c, r) + ), + household_composition=lambda f, c, m, r: suppress_entrant_benefit_outputs( + f + ), + ) + initial = pd.DataFrame( + {"person_id": [1], "year": [2025], "age": [30], "sex": ["female"]} + ) + result = ProjectionEngine(modules).project( + initial, + end_year=2028, + draw_index=0, + metadata={SCHEDULED_ENTRIES_KEY: schedule.as_metadata()}, + ) + assert [len(frame) for frame in result.slices] == [1, 0, 3, 3] + assert allocator.next_id == 1006 + for year in (2027, 2028): + observed = seen[year].loc[seen[year]["entry_year"] == year] + assert observed["age"].tolist() == donor["entry_age"].tolist() + frame = result.slices[year - 2025] + assert (frame["age"] == frame["entry_age"] + 1).all() + assert frame[["aime", "pia", "benefit"]].isna().all().all() + assert frame["claim_age"].isna().all() + assert not frame["claimed"].any() + assert frame["weight"].sum() == pytest.approx((year - 2026) * 1000.0) + assert set(result.slices[-1]["person_id"]) == {1003, 1004, 1005} diff --git a/tests/tier_counts.json b/tests/tier_counts.json index 71769153..48490cd6 100644 --- a/tests/tier_counts.json +++ b/tests/tier_counts.json @@ -1,7 +1,7 @@ { "schema_version": 1, "counts": { - "unit": 1563, + "unit": 1641, "artifact": 2668, "integration_psid": 848, "reproduction_legacy": 520,