Draw UK QRF stages from uniforms keyed on each record's id, and tie pension contributions to pay

changelog.d/uk-keyed-qrf-draws.changed.md

@@ -0,0 +1 @@
+UK QRF stages draw from uniforms keyed on each record's own id, so a household's imputed wealth, consumption, services, UC report, SPI fill and salary sacrifice move only when its own inputs do: an engine change that moved a few households' net income no longer redraws the whole spine through a shifted positional stream. The switch itself redraws these stages once for every household, benefit unit and person, so a comparison of this release with the previous one shows that one-off shift. Each stage refuses ids that are not unique integers, since a float id would key a different draw.

changelog.d/uk-pension-contribution-coherence.fixed.md

@@ -0,0 +1 @@
+UK employee pension contributions are capped at the pay they come out of, on FRS rows and on SPI-channel rows. On an SPI-channel row the forest still decides, from the row's own drawn pay, whether it has an employer contribution; a member whose FRS twin has an ASHE rate takes that rate times its own pay, so the amount follows the redrawn pay, and any other member keeps the forest's amount, bounded by ASHE's top rate.

docs/evidence/spec-engine/us-f0-coverage.json

@@ -1656,13 +1656,13 @@
           "compiler_ir.node_slices"
         ],
         "expected": {
-          "map_sha256": "b3d3f769647212a81d796a28f2b5ff759f513572bd3c70e253816c88ff496091",
-          "protocol_sha256": "d756be07ab66c179f8612eb5c614527586004506d88be3e1ab62b934de0b61ce"
+          "map_sha256": "0db599df419dd16779097361dc72afdefb8df2abf8b134078da9c52d31d8e017",
+          "protocol_sha256": "2b7bdc1d6b0da1bdaf54b4cac29f10f2d3c8bed2e177729c2c89e6b33a099cc5"
         },
         "failures": [],
         "observed": {
-          "map_sha256": "b3d3f769647212a81d796a28f2b5ff759f513572bd3c70e253816c88ff496091",
-          "protocol_sha256": "d756be07ab66c179f8612eb5c614527586004506d88be3e1ab62b934de0b61ce"
+          "map_sha256": "0db599df419dd16779097361dc72afdefb8df2abf8b134078da9c52d31d8e017",
+          "protocol_sha256": "2b7bdc1d6b0da1bdaf54b4cac29f10f2d3c8bed2e177729c2c89e6b33a099cc5"
         },
         "status": "covered"
       },
@@ -1677,7 +1677,7 @@
           "compiler_ir.seed_stream_map"
         ],
         "expected": {
-          "implementation_sha256": "d756be07ab66c179f8612eb5c614527586004506d88be3e1ab62b934de0b61ce",
+          "implementation_sha256": "2b7bdc1d6b0da1bdaf54b4cac29f10f2d3c8bed2e177729c2c89e6b33a099cc5",
           "protocol": "legacy-v1",
           "streams": [
             "build_model",
@@ -1698,7 +1698,7 @@
         },
         "failures": [],
         "observed": {
-          "implementation_sha256": "d756be07ab66c179f8612eb5c614527586004506d88be3e1ab62b934de0b61ce",
+          "implementation_sha256": "2b7bdc1d6b0da1bdaf54b4cac29f10f2d3c8bed2e177729c2c89e6b33a099cc5",
           "protocol": "legacy-v1",
           "streams": [
             "build_model",
@@ -2599,7 +2599,7 @@
       "country": "us",
       "schema_id": "country_spec",
       "schema_version": 1,
-      "spec_sha256": "740d1ff5b2e9fa6da3154c44f8676f80109d404e6aa7a16c48ed8db83055b15d"
+      "spec_sha256": "19460fc168a6f9ed44e580d0e36777cce78bb33cb52b75630ee86203fa795684"
     }
   },
   "report_schema_version": 3,
@@ -2609,7 +2609,7 @@
     "country": "us",
     "schema_id": "country_spec",
     "schema_version": 1,
-    "spec_sha256": "740d1ff5b2e9fa6da3154c44f8676f80109d404e6aa7a16c48ed8db83055b15d"
+    "spec_sha256": "19460fc168a6f9ed44e580d0e36777cce78bb33cb52b75630ee86203fa795684"
   },
   "status": "pass"
 }

packages/microcosm-build/src/microcosm/build/spec_engine/inventory_coverage.py

@@ -367,8 +367,8 @@
     # moves both again: calibrate gains budget_search_bracket, whose default
     # keeps the US bracket, and the budget search, which the US solve does
     # not run, changes its probes, so again no US solve changes.
-    "seed_map": "b3d3f769647212a81d796a28f2b5ff759f513572bd3c70e253816c88ff496091",
-    "seed_protocol": "d756be07ab66c179f8612eb5c614527586004506d88be3e1ab62b934de0b61ce",
+    "seed_map": "0db599df419dd16779097361dc72afdefb8df2abf8b134078da9c52d31d8e017",
+    "seed_protocol": "2b7bdc1d6b0da1bdaf54b4cac29f10f2d3c8bed2e177729c2c89e6b33a099cc5",
     "source_manifest": "f4f8bfeb79268a4043a843b47cec1d1c0e58e9f5dd6bc23417ada95446dd7b8e",
     "take_up": "9522ce40f7dea569312dd7a7beb474e5a5afadd2f1e10cb5876534c3ef623d35",
     "tail": "ac92829c88a1a4fb6460d61190918d5d99c6c377fc8dd8f62f02b332d09bf59c",

packages/microcosm-build/src/microcosm/build/stochastic_assignment.py

@@ -13,6 +13,7 @@
     "assign_binary_from_rate",
     "assign_binary_with_anchored_residual",
     "clipped_normal_from_uniforms",
+    "integer_record_keys",
     "sample_categorical_from_counts",
     "stable_identity_uniforms",
 ]
@@ -44,6 +45,25 @@ def stable_identity_uniforms(
     )
 
 
+def integer_record_keys(
+    ids: Sequence[object] | np.ndarray, *, label: str
+) -> np.ndarray:
+    """Record ids to key ``stable_identity_uniforms`` on: integers, each once.
+
+    The key string formats each id, so the integer ``1001`` and the float
+    ``1001.0`` draw differently, and a merge that turned an id column into
+    floats would silently redraw every record. A repeated id would give two
+    records the same uniforms.
+    """
+
+    keys = np.asarray(ids)
+    if keys.ndim != 1 or not np.issubdtype(keys.dtype, np.integer):
+        raise TypeError(f"{label} must be a one-dimensional array of integer ids.")
+    if np.unique(keys).size != keys.size:
+        raise ValueError(f"{label} must name each record once.")
+    return keys
+
+
 def assign_binary_from_rate(
     draws: Sequence[float] | np.ndarray,
     rate: float,

packages/microcosm-build/src/microcosm/build/uk_runtime/etb_services.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import json
+from collections.abc import Sequence
 from dataclasses import dataclass, field
 from importlib.resources import files
 from pathlib import Path
@@ -12,6 +13,10 @@
 
 from microcosm.build.gates import FitWeightRecord
 from microcosm.build.source_manifest import SourceStageSpec
+from microcosm.build.stochastic_assignment import (
+    integer_record_keys,
+    stable_identity_uniforms,
+)
 from microcosm.build.uk_runtime.bus_support_pricing import (
     PRICE_BUS_SUPPORT_KIND,
     bus_support_pricing_operation,
@@ -186,7 +191,10 @@ def __call__(self, frame: Frame) -> Frame:
         )
         predictors = recipient_predictors(frame, self.engine)
         draws, records = impute_etb_services(
-            donor, predictors, seed=_qrf_seed(self.stage)
+            donor,
+            predictors,
+            seed=_qrf_seed(self.stage),
+            record_ids=frame.table("household")["household_id"].to_numpy(),
         )
         clip_result = support_clip_to_donor(
             draws, donor, exempt=support_clip_exempt(self.stage)
@@ -436,11 +444,28 @@ def person_sum(values: np.ndarray) -> np.ndarray:
 
 
 def impute_etb_services(
-    donor: pd.DataFrame, recipient: pd.DataFrame, *, seed: int, n_estimators: int = 100
+    donor: pd.DataFrame,
+    recipient: pd.DataFrame,
+    *,
+    seed: int,
+    record_ids: Sequence[object] | np.ndarray,
+    n_estimators: int = 100,
 ) -> tuple[pd.DataFrame, tuple[FitWeightRecord, ...]]:
+    """Chain-draw the ETB services targets from household-keyed uniforms.
+
+    Each draw reads uniforms keyed on the household's own id (``record_ids``,
+    in recipient row order) and the target, so a household's draws move only
+    when its own predictors do, whatever happens to the other recipients.
+    """
+
     from microcosm.fit import RegimeGatedQRF
 
     targets = UK_ETB_SERVICES_HOUSEHOLD_OUTPUT_COLUMNS[:3]
+    ids = integer_record_keys(record_ids, label="ETB record_ids")
+    if ids.shape != (len(recipient),) or pd.Series(ids).duplicated().any():
+        raise ValueError(
+            "ETB record_ids must name every recipient once, in recipient row order."
+        )
     model = RegimeGatedQRF(n_estimators=n_estimators, seed=seed)
     state = model.start_chain(
         donor,
@@ -451,12 +476,15 @@ def impute_etb_services(
     raw = pd.DataFrame(index=recipient.index)
     records: list[FitWeightRecord] = []
     for target in targets:
+        salt = f"{UK_ETB_SERVICES_FIT_NAME}:{target}"
         result = model.fit_draw_next(
             donor,
             recipient.loc[:, list(UK_ETB_SERVICES_PREDICTORS)],
             raw,
             state=state,
             weights="weight",
+            quantiles=stable_identity_uniforms(ids, seed=seed, salt=salt),
+            sign_uniforms=stable_identity_uniforms(ids, seed=seed, salt=f"{salt}:sign"),
         )
         raw[target] = result.raw_draw
         records.append(

packages/microcosm-build/src/microcosm/build/uk_runtime/frs_spine.py

@@ -24,6 +24,10 @@
     uk_national_frame,
     validate_uk_national_frame,
 )
+from microcosm.build.uk_runtime.pension_coherence import (
+    cap_net_pay_contributions,
+    coherence_receipt,
+)
 from microcosm.build.uk_runtime.uc_relationships import frs_uc_claimant_mask
 from microcosm.frame import Frame, WeightKind
 
@@ -319,6 +323,7 @@ def __init__(self, raw_dir: str | Path, *, stage: SourceStageSpec) -> None:
         self.stage = stage
         self._sentinel_mapped_rows: int | None = None
         self._access_fund: dict[str, object] | None = None
+        self._net_pay: dict[str, object] | None = None
 
     def __call__(self, frame: Frame) -> Frame:
         evidence: dict[str, object] = {}
@@ -328,6 +333,7 @@ def __call__(self, frame: Frame) -> Frame:
         capital = result.table("benunit")["frs_benunit_capital"]
         self._sentinel_mapped_rows = int((capital == UC_CAPITAL_UNAVAILABLE).sum())
         self._access_fund = dict(evidence["access_fund"])
+        self._net_pay = dict(evidence["net_pay_contributions"])
         return result
 
     @staticmethod
@@ -338,7 +344,11 @@ def checkpoint_metadata(self) -> dict[str, object]:
         """Report how loudly the capital availability rule and the
         access-fund repair fired."""
 
-        if self._sentinel_mapped_rows is None or self._access_fund is None:
+        if (
+            self._sentinel_mapped_rows is None
+            or self._access_fund is None
+            or self._net_pay is None
+        ):
             raise RuntimeError("checkpoint metadata requires a completed stage run.")
         return {
             "evidence": {
@@ -348,6 +358,7 @@ def checkpoint_metadata(self) -> dict[str, object]:
                     "mapped_rows": self._sentinel_mapped_rows,
                 },
                 "access_fund": dict(self._access_fund),
+                "net_pay_contributions": dict(self._net_pay),
             }
         }
 
@@ -585,7 +596,7 @@ def _assemble_frame(
     pe_person["would_claim_carers_allowance"] = (
         pe_person["carers_allowance_reported"] > 0
     )
-    _add_person_expenses(pe_person, person, frs)
+    _add_person_expenses(pe_person, person, frs, evidence=evidence)
 
     pe_benunit["is_married"] = _number(benunit_raw, "famtypb2").isin([5, 7])
     pe_benunit["dependent_children"] = dependent_children
@@ -1004,7 +1015,11 @@ def _add_benefits(
 
 
 def _add_person_expenses(
-    pe_person: pd.DataFrame, person: pd.DataFrame, frs: Mapping[str, pd.DataFrame]
+    pe_person: pd.DataFrame,
+    person: pd.DataFrame,
+    frs: Mapping[str, pd.DataFrame],
+    *,
+    evidence: dict[str, object] | None = None,
 ) -> None:
     household = frs["househol"].set_index("household_id").sort_index()
     ctrebamt = pd.Series(_number(household, "ctrebamt").values, index=household.index)
@@ -1055,7 +1070,7 @@ def _add_person_expenses(
         * WEEKS_IN_YEAR,
     )
     job = frs["job"]
-    pe_person["employee_pension_contributions"] = np.maximum(
+    deducted = np.maximum(
         0,
         _sum_to_entity(
             _number(job, "deduc1").fillna(0),
@@ -1064,6 +1079,36 @@ def _add_person_expenses(
         )
         * WEEKS_IN_YEAR,
     )
+    # A net-pay deduction cannot exceed the pay it comes out of; the FRS
+    # records each job's deduction apart from the adult's usual earnings, so
+    # the deduction is capped at them (uk_runtime.pension_coherence).
+    capped, changed = cap_net_pay_contributions(
+        np.asarray(deducted, dtype=np.float64),
+        pe_person["employment_income"].to_numpy(dtype=np.float64),
+    )
+    pe_person["employee_pension_contributions"] = capped
+    if evidence is not None:
+        # A deduction beside no recorded pay may stand for pay the adult tab
+        # left blank, not for a contribution above pay, so the two cases are
+        # counted apart.
+        recorded_pay = pe_person["employment_income"].to_numpy(dtype=np.float64) > 0.0
+        receipt = {
+            "before": np.asarray(deducted, dtype=np.float64),
+            "after": capped,
+            "weights": person["household_id"]
+            .map(_raw_number(household, "gross4"))
+            .fillna(0.0)
+            .to_numpy(dtype=np.float64),
+            "source_ids": person["household_id"].to_numpy(),
+        }
+        evidence["net_pay_contributions"] = {
+            "above_recorded_pay": coherence_receipt(
+                changed=changed & recorded_pay, **receipt
+            ),
+            "beside_no_recorded_pay": coherence_receipt(
+                changed=changed & ~recorded_pay, **receipt
+            ),
+        }
     pe_person["pension_contributions_via_salary_sacrifice"] = np.maximum(
         0,
         _sum_to_entity(

packages/microcosm-build/src/microcosm/build/uk_runtime/lcfs_consumption.py

@@ -13,7 +13,10 @@
 
 from microcosm.build.gates import EXPLICIT_KIND, FitWeightRecord
 from microcosm.build.source_manifest import SourceStageSpec
-from microcosm.build.stochastic_assignment import stable_identity_uniforms
+from microcosm.build.stochastic_assignment import (
+    integer_record_keys,
+    stable_identity_uniforms,
+)
 from microcosm.build.uk_runtime.bus_fare_pricing import (
     PRICE_BUS_JOURNEYS_KIND,
     UK_DFT_BUS_JOURNEYS_RESOURCE,
@@ -575,6 +578,7 @@ def __call__(self, frame: Frame) -> Frame:
             recipient,
             seed=_operation_seed(self.stage, "fit_weighted_qrf_chain"),
             n_estimators=_qrf_n_estimators(self.stage),
+            record_ids=frame.table("household")["household_id"].to_numpy(),
         )
         clip_result = support_clip_to_donor(
             imputation.draws, donor, exempt=support_clip_exempt(self.stage)
@@ -3108,19 +3112,29 @@ def impute_lcfs_consumption(
     *,
     seed: int,
     n_estimators: int,
+    record_ids: Sequence[object] | np.ndarray,
 ) -> UKLCFSConsumptionImputationResult:
     """Chain-draw the targets in the declared order.
 
     ``bus_fare_spending`` stays a chain target so later targets condition on
     the same raw draw; the emitted fares are then replaced by the priced
     value where an area prices (microcosm#930).
+
+    Each draw reads uniforms keyed on the household's own id (``record_ids``,
+    in recipient row order) and the target, never a positional stream, so a
+    household's consumption moves only when its own predictors do.
     """
 
     from microcosm.fit import RegimeGatedQRF
 
     donor_encoded, recipient_encoded, predictors = _encode_consumption_predictors(
         donor, recipient_predictor_frame
     )
+    ids = integer_record_keys(record_ids, label="LCFS record_ids")
+    if ids.shape != (len(recipient_encoded),) or pd.Series(ids).duplicated().any():
+        raise ValueError(
+            "LCFS record_ids must name every recipient once, in recipient row order."
+        )
     model = RegimeGatedQRF(n_estimators=n_estimators, seed=seed)
     state = model.start_chain(
         donor_encoded,
@@ -3133,12 +3147,15 @@ def impute_lcfs_consumption(
     fit_records: list[FitWeightRecord] = []
     for target in UK_LCFS_CONSUMPTION_TARGET_COLUMNS:
         features = recipient_encoded.loc[:, list(predictors)]
+        salt = f"{UK_LCFS_CONSUMPTION_FIT_NAME}:{target}"
         result = model.fit_draw_next(
             donor_encoded,
             features,
             raw,
             state=state,
             weights="household_weight",
+            quantiles=stable_identity_uniforms(ids, seed=seed, salt=salt),
+            sign_uniforms=stable_identity_uniforms(ids, seed=seed, salt=f"{salt}:sign"),
         )
         raw[target] = result.raw_draw
         draws[target] = result.raw_draw