diff --git a/changelog.d/802-uk-cross-grain-reconciliation.added.md b/changelog.d/802-uk-cross-grain-reconciliation.added.md new file mode 100644 index 0000000000..2a4ba8b5ee --- /dev/null +++ b/changelog.d/802-uk-cross-grain-reconciliation.added.md @@ -0,0 +1 @@ +Add a country-agnostic cross-grain target reconciliation pass and the UK standing rule that gives country precedence over constituency and local-authority targets, with exact measurement-signature detection, declared household and UC bridges, fail-closed geography-leg rescaling, and mass-log-style receipts in every UK rowwise dry-run plan and manifest. The current candidate binds no national targets, so its target values remain numerically unchanged; the census disclosure-control fence and adjudication now state how a future same-concept national control governs published constituency values. diff --git a/docs/uk-local-cross-grain-reconciliation-802.md b/docs/uk-local-cross-grain-reconciliation-802.md new file mode 100644 index 0000000000..39e1921665 --- /dev/null +++ b/docs/uk-local-cross-grain-reconciliation-802.md @@ -0,0 +1,72 @@ +# UK local cross-grain reconciliation (#802) + +The UK local build applies one standing rule before constructing a calibration +matrix: when the same measurement concept is bound at more than one grain, the +highest bound grain wins in the order country, constituency, then local +authority. The shared operator in `microcosm.build.cross_grain` changes only +the lower-grain target values. It never changes loss weights, scales, caps, or +the solve signature. + +## Detection and bridges + +Exact matches use the measurement fields `concept`, `entity`, `map_to`, and +`filters`. Two explicit UK bridges cover relationships that those fields +cannot express on their own: + +- The 10-cell `ons.household_composition.*` partition sums to the national + household-count control and bridges to the ladder-derived + `census_households/households` metric. +- `dwp.uc.households` bridges to `dwp.uc.households_by_area`. The four + `dwp.uc.payment_distribution_*` rows also match the by-area target exactly + and form a separate exhaustive national partition. + +These declarations live beside the UK Ledger target orchestration in +`uk_runtime/ledger_targets.py`. Tests pin the precedence and complete bridge +membership so a rule change requires an explicit doctrine-constants review. + +## Geography legs and refusals + +The bound control's geography defines the factor legs. A UK control produces +one factor over England, Wales, Scotland, and Northern Ireland. A GB control +covers England, Wales, and Scotland; a surface that also contains Northern +Ireland then fails as unparented. Country controls produce one factor for each +declared country. Area codes map to countries using the same ONS-prefix logic +as the local target runtime. + +The pass also refuses a partially bound declared partition, a target covered +by two bridges, incompatible controls at the same winning grain, an empty or +unparented leg, a zero lower total with a nonzero control, a sign flip, and any +non-finite input or result. Every successful factor records the parent, +constituent target ids, area count, old and new totals, relative shift, and +declared factor. Dry-run plans and build manifests carry the pass receipt even +when no inconsistency is in force. + +## Census evidence and current effect + +Published constituency household counts are disclosure-controlled. At review, +the measured constituency-sum differences from the corresponding national +census totals were E&W +105, Scotland -554, and Northern Ireland +3 — a +relative magnitude around 2e-5. The tension the reconciliation actually +addresses is a separate, larger one: the roughly 1–2% vintage gap between +census-day 2021/2022 counts and any bound 2023 national household count. The +published local values still bind as published when no same-concept national +control is bound. If one is bound in the same solve, country wins and the +standing rule rescales the local values before calibration. + +Today the rowwise candidate binds only `census_households/constituency` and +declares no bound national targets. The pass therefore records an absence +receipt and leaves all target values numerically identical. Increment #762 +must extend the mixed-grain surface through +`apply_uk_cross_grain_reconciliation`; the existing rowwise area-type fence +remains the structural backstop. + +## Rescope from the issue text + +Issue #802 originally called for a committed enumeration artifact and a +per-family ruling register. This implementation deliberately replaces both +with a live pass and a standing, pinned rule. The receipt enumerates the +inconsistencies actually present in each assembled surface, avoiding a second +artifact that could become stale. The precedence and bridges are the ruling, +while the refreshed `census_disclosure_control_noise` adjudication records the +specific disclosure-control acceptance. The issue text should be updated to +reflect this rescope during review. diff --git a/packages/microcosm-build/src/microcosm/build/cross_grain.py b/packages/microcosm-build/src/microcosm/build/cross_grain.py new file mode 100644 index 0000000000..cfc240c578 --- /dev/null +++ b/packages/microcosm-build/src/microcosm/build/cross_grain.py @@ -0,0 +1,719 @@ +"""Country-agnostic reconciliation of targets bound at multiple grains. + +The operator in this module changes target values, never loss weights. A +country supplies the measurement signature, explicit bridges for known +partition relationships, grain precedence, and geography-to-leg mappings. +""" + +from __future__ import annotations + +import hashlib +import json +from collections.abc import Callable, Iterable, Mapping +from dataclasses import dataclass +from typing import Any + +import numpy as np +import pandas as pd + +# Closure of a rescaled leg onto its control is the property this pass exists +# to establish, so it is asserted rather than assumed. The tolerance matches +# the frame kernel's mass-conservation bound (``_MASS_CONSERVE_RTOL``). +_CLOSURE_RTOL = 1e-9 +# A leg whose members cancel to near-nothing would yield an arbitrarily large +# factor, so the vanishing-total refusal is relative to the control rather than +# an exact-zero test. +_MIN_LEG_SUM_RTOL = 1e-9 +# Absent, null, and empty signature values are one canonical "unspecified", so +# two spellings of the same measurement cannot land in different groups. +_UNSPECIFIED = ("",) + + +@dataclass(frozen=True) +class CrossGrainBridge: + """Declare one higher-control partition as the identity of a lower side.""" + + bridge_id: str + concept: str + higher_target_ids: tuple[str, ...] + lower_side: str + + +@dataclass(frozen=True) +class CrossGrainRule: + """All country-owned declarations needed by the shared operator.""" + + grain_precedence: tuple[str, ...] + signature_fields: tuple[str, ...] + bridges: tuple[CrossGrainBridge, ...] + leg_of_area: Callable[[str], str] + parent_geography_legs: Mapping[str, tuple[str, ...]] + + +@dataclass(frozen=True) +class CrossGrainInconsistency: + """Detected rows for one winning-grain/lower-grain reconciliation.""" + + inconsistency_id: str + bridge_id: str | None + signature: tuple[tuple[str, Any], ...] + winning_grain: str + lower_grain: str + winning_positions: tuple[int, ...] + lower_positions: tuple[int, ...] + higher_target_ids: tuple[str, ...] + + +def detect_cross_grain_inconsistencies( + local_frame: pd.DataFrame, + bound_higher_targets: Iterable[str], + contract_signatures: Mapping[str, Mapping[str, Any]], + rule: CrossGrainRule, +) -> tuple[CrossGrainInconsistency, ...]: + """Detect exact-signature and explicitly bridged cross-grain groups. + + ``local_frame`` is a long target surface. It must expose ``value`` and + ``target_id`` plus either the canonical ``grain``/``geography_id`` columns + or the rowwise aliases ``area_type``/``area_code``. Contract sides may be + written as either ```` or ``contract:``; external + sides use their declared ``external:...`` bridge name. + """ + + columns = _surface_columns(local_frame) + _validate_rule(rule) + bound = tuple(str(target_id) for target_id in bound_higher_targets) + _require_unique_nonblank(bound, label="bound_higher_targets") + unknown = sorted(set(bound) - set(contract_signatures)) + if unknown: + raise ValueError( + "cross-grain bound higher target(s) are absent from the contract: " + f"{unknown}." + ) + + bridge_by_side = _bridge_by_side(rule) + bound_set = set(bound) + for bridge in rule.bridges: + selected = bound_set & set(bridge.higher_target_ids) + if selected and selected != set(bridge.higher_target_ids): + missing = sorted(set(bridge.higher_target_ids) - selected) + raise ValueError( + f"cross-grain bridge {bridge.bridge_id!r} is partially bound; " + f"missing higher target(s) {missing}." + ) + + grouped: dict[ + tuple[str, str, tuple[tuple[str, Any], ...]], + list[tuple[int, str, str, str, float]], + ] = {} + seen_bound: set[str] = set() + values = local_frame[columns["value"]].to_numpy(dtype=np.float64) + if not np.isfinite(values).all(): + raise ValueError("cross-grain target values must all be finite.") + + for position, row in enumerate(local_frame.to_dict(orient="records")): + grain = str(row[columns["grain"]]) + geography_id = str(row[columns["geography_id"]]) + side = str(row[columns["target_id"]]) + if grain not in rule.grain_precedence: + raise ValueError( + f"cross-grain surface row {position} uses unknown grain " + f"{grain!r}; expected one of {rule.grain_precedence}." + ) + if not geography_id: + raise ValueError( + f"cross-grain surface row {position} has a blank geography id." + ) + bridge = bridge_by_side.get(side) + contract_id = _contract_target_id(side) + if bridge is None and contract_id is not None: + bridge = bridge_by_side.get(contract_id) + if bridge is None: + bridge = bridge_by_side.get(f"contract:{contract_id}") + + is_bound_higher = contract_id in bound_set + if grain == rule.grain_precedence[0] and not is_bound_higher: + continue + if is_bound_higher: + seen_bound.add(contract_id) + + keys: list[tuple[str, str, tuple[tuple[str, Any], ...]]] = [] + if bridge is not None: + keys.append( + ( + "bridge", + bridge.bridge_id, + (("concept", _freeze(bridge.concept)),), + ) + ) + # A declared bridge is additive to exact-signature detection. This + # matters when one lower side has both a bridged control and a second, + # exact-signature partition (the UK UC case): both controls must be + # seen so incompatible same-grain values fail closed. + if contract_id is not None and contract_id in contract_signatures: + keys.append( + ( + "signature", + "", + _measurement_signature( + contract_signatures[contract_id], rule.signature_fields + ), + ) + ) + for key in dict.fromkeys(keys): + grouped.setdefault(key, []).append( + (position, grain, geography_id, side, float(values[position])) + ) + + missing_rows = sorted(bound_set - seen_bound) + if missing_rows: + raise ValueError( + "cross-grain bound higher target(s) have no value rows in the " + f"surface: {missing_rows}." + ) + + inconsistencies: list[CrossGrainInconsistency] = [] + precedence = {grain: index for index, grain in enumerate(rule.grain_precedence)} + for (kind, bridge_id, signature), rows in grouped.items(): + grains = sorted({row[1] for row in rows}, key=precedence.__getitem__) + if len(grains) < 2: + continue + winning_grain = grains[0] + winning = tuple(row[0] for row in rows if row[1] == winning_grain) + higher_ids = tuple( + sorted( + { + target_id + for position in winning + if ( + target_id := _contract_target_id( + str(local_frame.iloc[position][columns["target_id"]]) + ) + ) + is not None + } + ) + ) + identity = _inconsistency_identity(kind, bridge_id, signature) + for lower_grain in grains[1:]: + lower = tuple(row[0] for row in rows if row[1] == lower_grain) + inconsistencies.append( + CrossGrainInconsistency( + inconsistency_id=( + f"{identity}:{winning_grain}_over_{lower_grain}" + ), + bridge_id=bridge_id or None, + signature=signature, + winning_grain=winning_grain, + lower_grain=lower_grain, + winning_positions=winning, + lower_positions=lower, + higher_target_ids=higher_ids, + ) + ) + return tuple( + sorted(inconsistencies, key=lambda group: group.inconsistency_id) + ) + + +def reconcile_cross_grain_surface( + local_frame: pd.DataFrame, + groups: Iterable[CrossGrainInconsistency], + rule: CrossGrainRule, +) -> tuple[pd.DataFrame, list[dict[str, Any]]]: + """Return a copied surface with every detected lower grain rescaled.""" + + columns = _surface_columns(local_frame) + _validate_rule(rule) + materialized_groups = tuple(groups) + _assert_compatible_overlapping_groups( + local_frame, materialized_groups, columns, rule + ) + reconciled = local_frame.copy(deep=True) + receipts: list[dict[str, Any]] = [] + for group in materialized_groups: + controls = _winning_controls(reconciled, group, columns, rule) + lower_by_leg: dict[str, list[int]] = {} + for position in group.lower_positions: + area = str(reconciled.iloc[position][columns["geography_id"]]) + leg = str(rule.leg_of_area(area)) + if not leg: + raise ValueError( + f"cross-grain inconsistency {group.inconsistency_id!r} " + f"maps area {area!r} to a blank leg." + ) + lower_by_leg.setdefault(leg, []).append(position) + + assigned_controls: dict[str, dict[str, Any]] = {} + for control in controls: + for leg in control["covered_legs"]: + existing = assigned_controls.get(leg) + if existing is not None: + if ( + existing["covered_legs"] != control["covered_legs"] + or existing["value"] != control["value"] + ): + raise ValueError( + "cross-grain inconsistency " + f"{group.inconsistency_id!r} has two different " + f"control values at grain {group.winning_grain!r} " + f"for leg {leg!r}." + ) + continue + assigned_controls[leg] = control + + unparented = sorted(set(lower_by_leg) - set(assigned_controls)) + if unparented: + raise ValueError( + f"cross-grain inconsistency {group.inconsistency_id!r} has " + f"unparented lower-grain leg(s) {unparented}." + ) + + leg_receipts: list[dict[str, Any]] = [] + used_controls: set[tuple[str, tuple[str, ...]]] = set() + for control in controls: + control_key = ( + str(control["parent_geography_id"]), + tuple(control["covered_legs"]), + ) + if control_key in used_controls: + continue + used_controls.add(control_key) + positions = [ + position + for leg in control["covered_legs"] + for position in lower_by_leg.get(leg, ()) + ] + if not positions: + raise ValueError( + f"cross-grain inconsistency {group.inconsistency_id!r} " + f"has an empty leg for parent geography " + f"{control['parent_geography_id']!r}." + ) + raw_values = reconciled.iloc[positions][columns["value"]].to_numpy( + dtype=np.float64 + ) + raw_sum = float(raw_values.sum()) + parent_value = float(control["value"]) + if not np.isfinite(raw_sum) or not np.isfinite(parent_value): + raise ValueError( + f"cross-grain inconsistency {group.inconsistency_id!r} " + "contains a non-finite leg total." + ) + # Offsetting members are legitimate for a net-valued concept, so + # only the conditioning matters: a leg that cancels to nothing + # against a nonzero control would rescale by an arbitrary factor. + if abs(raw_sum) < _MIN_LEG_SUM_RTOL * abs(parent_value): + raise ValueError( + f"cross-grain inconsistency {group.inconsistency_id!r} " + f"cannot scale a vanishing lower-leg total {raw_sum!r} to " + f"control {parent_value!r}." + ) + # Live only when the control is itself zero (the floor above is + # vacuous there): a zero-summing leg against a zero control is a + # no-op, where `parent_value / raw_sum` would be 0/0. + factor = 1.0 if raw_sum == 0.0 else parent_value / raw_sum + if not np.isfinite(factor): + raise ValueError( + f"cross-grain inconsistency {group.inconsistency_id!r} " + "produced a non-finite reconciliation factor." + ) + if factor < 0.0: + raise ValueError( + f"cross-grain inconsistency {group.inconsistency_id!r} " + "cannot reconcile opposite-signed lower and control targets." + ) + reconciled.iloc[ + positions, reconciled.columns.get_loc(columns["value"]) + ] = raw_values * factor + new_total = float( + reconciled.iloc[positions][columns["value"]] + .to_numpy(dtype=np.float64) + .sum() + ) + if not np.isfinite(new_total): + raise ValueError( + f"cross-grain inconsistency {group.inconsistency_id!r} " + "produced a non-finite reconciled total." + ) + if not np.isclose( + new_total, parent_value, rtol=_CLOSURE_RTOL, atol=0.0 + ): + raise ValueError( + f"cross-grain inconsistency {group.inconsistency_id!r} " + f"left leg {'+'.join(control['covered_legs'])!r} off its " + f"control: reconciled total {new_total!r} against control " + f"{parent_value!r}." + ) + areas = { + str(reconciled.iloc[position][columns["geography_id"]]) + for position in positions + } + leg_receipts.append( + { + "leg": "+".join(control["covered_legs"]), + "parent_geography_id": str(control["parent_geography_id"]), + "higher_target_ids": list(control["higher_target_ids"]), + "n_areas": len(areas), + "old_total": raw_sum, + "new_total": new_total, + "relative_shift": ( + 0.0 if raw_sum == 0.0 else (new_total - raw_sum) / raw_sum + ), + "declared_factor": factor, + "reason": ( + "standing cross-grain rule: " + f"{group.winning_grain} controls {group.lower_grain}" + ), + } + ) + receipts.append( + { + "inconsistency_id": group.inconsistency_id, + "bridge_id": group.bridge_id, + "winning_grain": group.winning_grain, + "legs": leg_receipts, + } + ) + return reconciled, receipts + + +def apply_cross_grain_reconciliation( + local_frame: pd.DataFrame, + bound_higher_targets: Iterable[str], + contract_signatures: Mapping[str, Mapping[str, Any]], + rule: CrossGrainRule, +) -> tuple[pd.DataFrame, dict[str, Any]]: + """Detect, reconcile, and return the always-present pass receipt.""" + + bound = tuple(str(target_id) for target_id in bound_higher_targets) + groups = detect_cross_grain_inconsistencies( + local_frame, + bound, + contract_signatures, + rule, + ) + reconciled, group_receipts = reconcile_cross_grain_surface( + local_frame, groups, rule + ) + receipt = { + "bound_higher_targets": list(bound), + "inconsistencies_in_force": [ + group.inconsistency_id for group in groups + ], + "groups": group_receipts, + "absence": ( + None + if groups + else "No cross-grain inconsistencies are in force on this surface." + ), + } + return reconciled, receipt + + +def _surface_columns(frame: pd.DataFrame) -> dict[str, str]: + aliases = { + "grain": ("grain", "area_type"), + "geography_id": ("geography_id", "area_code"), + "target_id": ("target_id",), + "value": ("value",), + } + resolved: dict[str, str] = {} + for canonical, candidates in aliases.items(): + present = [candidate for candidate in candidates if candidate in frame.columns] + if not present: + raise ValueError( + f"cross-grain surface must expose {canonical!r}; accepted " + f"column name(s): {list(candidates)}." + ) + resolved[canonical] = present[0] + return resolved + + +def _validate_rule(rule: CrossGrainRule) -> None: + _require_unique_nonblank(rule.grain_precedence, label="grain_precedence") + _require_unique_nonblank(rule.signature_fields, label="signature_fields") + if not rule.grain_precedence: + raise ValueError("cross-grain grain_precedence must not be empty.") + bridge_ids = tuple(bridge.bridge_id for bridge in rule.bridges) + _require_unique_nonblank(bridge_ids, label="bridge ids") + _bridge_by_side(rule) + for bridge in rule.bridges: + if not bridge.concept or not bridge.lower_side: + raise ValueError( + f"cross-grain bridge {bridge.bridge_id!r} has a blank declaration." + ) + _require_unique_nonblank( + bridge.higher_target_ids, + label=f"bridge {bridge.bridge_id!r} higher_target_ids", + ) + if not bridge.higher_target_ids: + raise ValueError( + f"cross-grain bridge {bridge.bridge_id!r} has no higher targets." + ) + for geography_id, legs in rule.parent_geography_legs.items(): + if not str(geography_id) or not legs or any(not str(leg) for leg in legs): + raise ValueError( + "cross-grain parent_geography_legs must map nonblank geography " + "ids to nonempty, nonblank leg tuples." + ) + if len(set(legs)) != len(legs): + raise ValueError( + f"cross-grain parent geography {geography_id!r} repeats a leg." + ) + + +def _bridge_by_side(rule: CrossGrainRule) -> dict[str, CrossGrainBridge]: + result: dict[str, CrossGrainBridge] = {} + for bridge in rule.bridges: + sides = (*bridge.higher_target_ids, bridge.lower_side) + for side in sides: + aliases = {side} + if side.startswith("contract:"): + aliases.add(side.removeprefix("contract:")) + elif not side.startswith("external:"): + aliases.add(f"contract:{side}") + for alias in aliases: + existing = result.get(alias) + if existing is not None and existing.bridge_id != bridge.bridge_id: + raise ValueError( + f"cross-grain target side {side!r} is matched by two " + f"bridges: {existing.bridge_id!r} and {bridge.bridge_id!r}." + ) + result[alias] = bridge + return result + + +def _measurement_signature( + contract: Mapping[str, Any], fields: tuple[str, ...] +) -> tuple[tuple[str, Any], ...]: + """Canonicalize one contract entry's measurement into a grouping key. + + A signature field that is absent, null, or an empty container collapses to + one canonical "unspecified" value: two spellings of the same measurement + must never land in different groups, because a silent split lets the joint + solve reconcile the pair implicitly. Sequence-valued fields are compared + as unordered collections, since a filter list is a conjunction. + """ + + measurement = contract.get("measurement") + if measurement is None: + raise ValueError( + "cross-grain contract entry must carry a 'measurement' mapping; " + "a top-level fallback would let two contract spellings of the same " + f"measurement split silently. Got keys {sorted(contract)}." + ) + if not isinstance(measurement, Mapping): + raise ValueError("cross-grain contract measurement must be a mapping.") + return tuple( + (field, _canonical_signature_value(measurement.get(field))) + for field in fields + ) + + +def _canonical_signature_value(value: Any) -> Any: + frozen = _freeze(value) + if frozen is None or frozen == () or frozen == "": + return _UNSPECIFIED + if isinstance(value, (list, tuple, set, frozenset)): + # Order-insensitive at this level only: a signature field holding a + # sequence is a conjunction (the filter list), so its members are the + # same measurement however the contract orders them. Ordering *inside* + # a member stays significant — a `between` payload's bounds are data, + # not a conjunction — so `_freeze` below is order-preserving. + return tuple(sorted(frozen, key=_sort_key)) + return frozen + + +def _sort_key(member: Any) -> str: + return json.dumps(_json_safe(member), sort_keys=True, separators=(",", ":")) + + +def _freeze(value: Any) -> Any: + if isinstance(value, Mapping): + return tuple(sorted((str(key), _freeze(member)) for key, member in value.items())) + if isinstance(value, (set, frozenset)): + # A set has no order to preserve; sort for a deterministic signature. + return tuple(sorted((_freeze(member) for member in value), key=_sort_key)) + if isinstance(value, (list, tuple)): + return tuple(_freeze(member) for member in value) + if isinstance(value, np.generic): + value = value.item() + if isinstance(value, bool): + return value + if isinstance(value, (int, float)): + # 0 and 0.0 are the same signature value; keep one numeric spelling. + return float(value) + return value + + +def _contract_target_id(side: str) -> str | None: + if side.startswith("external:"): + return None + return side.removeprefix("contract:") + + +def _inconsistency_identity( + kind: str, + bridge_id: str, + signature: tuple[tuple[str, Any], ...], +) -> str: + if kind == "bridge": + return bridge_id + payload = json.dumps(_json_safe(signature), sort_keys=True, separators=(",", ":")) + digest = hashlib.sha256(payload.encode()).hexdigest()[:12] + concept = next((str(value) for field, value in signature if field == "concept"), "target") + return f"exact:{concept}:{digest}" + + +def _json_safe(value: Any) -> Any: + if isinstance(value, tuple): + return [_json_safe(member) for member in value] + if isinstance(value, Mapping): + return {str(key): _json_safe(member) for key, member in value.items()} + return value + + +def _winning_controls( + frame: pd.DataFrame, + group: CrossGrainInconsistency, + columns: Mapping[str, str], + rule: CrossGrainRule, +) -> list[dict[str, Any]]: + rows = [ + { + "position": position, + "geography_id": str(frame.iloc[position][columns["geography_id"]]), + "target_id": str(frame.iloc[position][columns["target_id"]]), + "value": float(frame.iloc[position][columns["value"]]), + } + for position in group.winning_positions + ] + if group.winning_grain == rule.grain_precedence[0]: + by_geography: dict[str, list[dict[str, Any]]] = {} + for row in rows: + by_geography.setdefault(row["geography_id"], []).append(row) + controls = [] + for geography_id, members in sorted(by_geography.items()): + deduplicated: dict[str, float] = {} + for member in members: + target_id = _contract_target_id(member["target_id"]) + if target_id is None: + raise ValueError( + f"cross-grain winning row {member['target_id']!r} is not " + "a contract target." + ) + existing = deduplicated.get(target_id) + if existing is not None and existing != member["value"]: + raise ValueError( + f"cross-grain target {target_id!r} has two different " + f"control values at {geography_id!r}." + ) + deduplicated[target_id] = member["value"] + if group.bridge_id is not None: + bridge = next( + bridge + for bridge in rule.bridges + if bridge.bridge_id == group.bridge_id + ) + missing = sorted( + set(bridge.higher_target_ids) - set(deduplicated) + ) + if missing: + raise ValueError( + f"cross-grain bridge {bridge.bridge_id!r} is partially " + f"valued at {geography_id!r}; missing {missing}." + ) + covered_legs = rule.parent_geography_legs.get(geography_id) + if covered_legs is None: + raise ValueError( + f"cross-grain parent geography {geography_id!r} declares no legs." + ) + controls.append( + { + "parent_geography_id": geography_id, + "covered_legs": tuple(covered_legs), + "higher_target_ids": tuple(sorted(deduplicated)), + "value": float(sum(deduplicated.values())), + } + ) + return controls + + by_leg: dict[str, list[dict[str, Any]]] = {} + for row in rows: + leg = str(rule.leg_of_area(row["geography_id"])) + if not leg: + raise ValueError( + f"cross-grain winning area {row['geography_id']!r} maps to a " + "blank leg." + ) + by_leg.setdefault(leg, []).append(row) + return [ + { + "parent_geography_id": f"area-derived:{leg}", + "covered_legs": (leg,), + "higher_target_ids": tuple( + sorted( + { + target_id + for member in members + if ( + target_id := _contract_target_id(member["target_id"]) + ) + is not None + } + ) + ), + "value": float(sum(member["value"] for member in members)), + } + for leg, members in sorted(by_leg.items()) + ] + + +def _assert_compatible_overlapping_groups( + frame: pd.DataFrame, + groups: tuple[CrossGrainInconsistency, ...], + columns: Mapping[str, str], + rule: CrossGrainRule, +) -> None: + """Refuse competing same-grain controls over any shared lower row.""" + + for index, left in enumerate(groups): + left_positions = set(left.lower_positions) + left_controls = _winning_controls(frame, left, columns, rule) + for right in groups[index + 1 :]: + if left.winning_grain != right.winning_grain: + continue + if not (left_positions & set(right.lower_positions)): + continue + right_controls = _winning_controls(frame, right, columns, rule) + left_by_coverage = { + tuple(control["covered_legs"]): float(control["value"]) + for control in left_controls + } + right_by_coverage = { + tuple(control["covered_legs"]): float(control["value"]) + for control in right_controls + } + for coverage in set(left_by_coverage) & set(right_by_coverage): + if left_by_coverage[coverage] != right_by_coverage[coverage]: + raise ValueError( + "cross-grain groups " + f"{left.inconsistency_id!r} and " + f"{right.inconsistency_id!r} have two different " + f"control values at grain {left.winning_grain!r} " + f"for legs {coverage}." + ) + + +def _require_unique_nonblank(values: Iterable[str], *, label: str) -> None: + materialized = tuple(str(value) for value in values) + blanks = [value for value in materialized if not value] + if blanks: + raise ValueError(f"cross-grain {label} must not contain blank values.") + duplicates = sorted( + {value for value in materialized if materialized.count(value) > 1} + ) + if duplicates: + raise ValueError( + f"cross-grain {label} must be unique; duplicate(s) {duplicates}." + ) diff --git a/packages/microcosm-build/src/microcosm/build/uk/local_binding_adjudications.json b/packages/microcosm-build/src/microcosm/build/uk/local_binding_adjudications.json index 1478e6c0ce..703e4f9154 100644 --- a/packages/microcosm-build/src/microcosm/build/uk/local_binding_adjudications.json +++ b/packages/microcosm-build/src/microcosm/build/uk/local_binding_adjudications.json @@ -3,11 +3,11 @@ "description": "Reviewed adjudications of the UK local binding fences (uk_local_target_census.json binding_fences). Binding a target family whose fence has no in-force entry here fails in the rowwise doctrine solve before any solve runs. Keys are fence ids; the reason states the ruling (accept the stated basis, or replace it with the named alternative) and cites the evidence.", "exclusions": { "census_disclosure_control_noise": { - "reason": "Accept the stated basis: constituency household counts bind to the published disclosure-controlled census values as-is — ONS/NRS cell-key perturbation and NISRA flexible-table-builder noise is documented, the counts are target values, never exact controls; the measured cross-grain deltas at review (constituency sums vs national totals: E&W +105, Scotland -554, NI +3) are accepted; constituency is the winning grain for the rowwise arm, whose targets are the sha-pinned OA ladder's occupied-household counts summed at constituency grain (ladder_targets.constituency_household_targets, provenance recorded per build) with no cross-grain reconciliation. Evidence: uk_local_target_census.json#/binding_fences/census_disclosure_control_noise.", + "reason": "Accept the stated basis: constituency household counts bind to the published disclosure-controlled census values as-is — ONS/NRS cell-key perturbation and NISRA flexible-table-builder noise is documented, and the measured cross-grain deltas at review (constituency sums vs national totals: E&W +105, Scotland -554, NI +3) are accepted. When a national same-concept control is bound in the same solve, the standing cross-grain rule rescales the constituency values to it before calibration; country is the winning grain. The rule and bridges are declared in microcosm.build.uk_runtime.ledger_targets and implemented by microcosm.build.cross_grain; rationale and evidence are recorded in docs/uk-local-cross-grain-reconciliation-802.md. Fence evidence: uk_local_target_census.json#/binding_fences/census_disclosure_control_noise.", "approved_by": "juaristi22", - "adjudication": "microcosm#760", - "approved_on": "2026-08-27", - "expires_on": "2026-11-27" + "adjudication": "microcosm#802", + "approved_on": "2026-08-31", + "expires_on": "2026-11-30" } } } diff --git a/packages/microcosm-build/src/microcosm/build/uk/uk_local_target_census.json b/packages/microcosm-build/src/microcosm/build/uk/uk_local_target_census.json index dc1b1df353..5c84040830 100644 --- a/packages/microcosm-build/src/microcosm/build/uk/uk_local_target_census.json +++ b/packages/microcosm-build/src/microcosm/build/uk/uk_local_target_census.json @@ -90,7 +90,7 @@ "enforcement": "review_required_before_binding", "fence_id": "census_disclosure_control_noise", "fenced_fact_count": null, - "rule": "All three census household-count legs are disclosure-controlled (ONS/NRS cell-key perturbation; NISRA flexible-table-builder controls), so area counts carry small deliberate noise and do not add exactly across grains (measured constituency-sum vs national-total deltas at review: E&W +105, Scotland -554, NI +3). Binding treats the published counts as the target values with that noise documented \u2014 never as exact controls \u2014 and any cross-grain reconciliation must name which grain wins." + "rule": "All three census household-count legs are disclosure-controlled (ONS/NRS cell-key perturbation; NISRA flexible-table-builder controls), so area counts carry small deliberate noise and do not add exactly across grains (measured constituency-sum vs national-total deltas at review: E&W +105, Scotland -554, NI +3). Binding treats the published counts as the target values with that noise documented \u2014 never as exact controls \u2014 and any cross-grain reconciliation applies the standing cross-grain rule declared in uk_runtime.ledger_targets: country wins, so a bound national same-concept control rescales the constituency values before the solve." }, { "authority": "FRS background and methodology (private-household universe); ONS population estimates methodology (communal-establishment population); microcosm#495 scoping (cross-family review finding).", diff --git a/packages/microcosm-build/src/microcosm/build/uk_runtime/__init__.py b/packages/microcosm-build/src/microcosm/build/uk_runtime/__init__.py index a43624a88a..fc6b4e13cb 100644 --- a/packages/microcosm-build/src/microcosm/build/uk_runtime/__init__.py +++ b/packages/microcosm-build/src/microcosm/build/uk_runtime/__init__.py @@ -251,8 +251,12 @@ local_authority_household_targets, ) from microcosm.build.uk_runtime.ledger_targets import ( + UK_CROSS_GRAIN_BRIDGES, + UK_CROSS_GRAIN_GRAIN_PRECEDENCE, + UK_CROSS_GRAIN_RULE, UKFrameTargetAdapter, UKLedgerTargetCompilation, + apply_uk_cross_grain_reconciliation, compile_uk_local_target_registry, compile_uk_target_registry, load_uk_local_area_crosswalk, @@ -525,6 +529,10 @@ "UK_FISCAL_TARGET_REGISTRY", "UKFrameTargetAdapter", "UKLedgerTargetCompilation", + "UK_CROSS_GRAIN_BRIDGES", + "UK_CROSS_GRAIN_GRAIN_PRECEDENCE", + "UK_CROSS_GRAIN_RULE", + "apply_uk_cross_grain_reconciliation", "compile_uk_target_registry", "materialize_uk_ledger_targets", "AGE_BANDS", diff --git a/packages/microcosm-build/src/microcosm/build/uk_runtime/ledger_targets.py b/packages/microcosm-build/src/microcosm/build/uk_runtime/ledger_targets.py index 21bc915882..b9d00adf70 100644 --- a/packages/microcosm-build/src/microcosm/build/uk_runtime/ledger_targets.py +++ b/packages/microcosm-build/src/microcosm/build/uk_runtime/ledger_targets.py @@ -12,6 +12,11 @@ import pandas as pd from microcosm.build.country_spec import load_country_spec +from microcosm.build.cross_grain import ( + CrossGrainBridge, + CrossGrainRule, + apply_cross_grain_reconciliation, +) from microcosm.build.ledger_targets import ( LedgerTargetReference, _fact_matches_selector, @@ -22,7 +27,10 @@ materialize_target_bindings, ) from microcosm.build.uk_runtime.cgt_calibration import uk_cgt_annual_exempt_amount -from microcosm.build.uk_runtime.local_targets import load_uk_local_geography_contract +from microcosm.build.uk_runtime.local_targets import ( + area_groups_from_codes, + load_uk_local_geography_contract, +) from microcosm.calibrate import TargetRegistry from microcosm.frame import Frame @@ -40,6 +48,63 @@ class UKLedgerTargetCompilation: UK_NATIONAL_TARGET_GEOGRAPHY_LEVELS = frozenset({"country", "region"}) +def _uk_cross_grain_leg_of_area(area_code: str) -> str: + # area_groups_from_codes maps code -> country group, so the single value is + # this code's leg. It refuses an unknown prefix itself; the explicit miss + # below keeps the refusal fail-closed rather than a bare StopIteration if + # that mapping ever returns nothing for a code. + leg = next(iter(area_groups_from_codes((area_code,)).values()), "") + if not leg: + raise ValueError( + f"UK cross-grain area code {area_code!r} maps to no country leg." + ) + return leg + + +UK_CROSS_GRAIN_GRAIN_PRECEDENCE = ("country", "constituency", "la") +UK_CROSS_GRAIN_BRIDGES = ( + CrossGrainBridge( + bridge_id="national_household_composition_partition_vs_census_households", + concept="uk.household.count", + higher_target_ids=( + "ons.household_composition.lone_households_under_65", + "ons.household_composition.lone_households_over_65", + "ons.household_composition.unrelated_adult_households", + "ons.household_composition.couple_no_children_households", + "ons.household_composition.couple_under_3_children_households", + "ons.household_composition.couple_3_plus_children_households", + "ons.household_composition.couple_non_dependent_children_only_households", + "ons.household_composition.lone_parent_dependent_children_households", + "ons.household_composition.lone_parent_non_dependent_children_households", + "ons.household_composition.multi_family_households", + ), + lower_side="external:census_households/households", + ), + CrossGrainBridge( + bridge_id="national_uc_caseload_vs_uc_households_by_area", + concept="uk.benefit_unit.count", + higher_target_ids=("dwp.uc.households",), + lower_side="contract:dwp.uc.households_by_area", + ), +) +# A future move of these declarations into country-package spec JSON follows +# the country-owned specification direction established in microcosm#159. +UK_CROSS_GRAIN_RULE = CrossGrainRule( + grain_precedence=UK_CROSS_GRAIN_GRAIN_PRECEDENCE, + signature_fields=("concept", "entity", "map_to", "filters"), + bridges=UK_CROSS_GRAIN_BRIDGES, + leg_of_area=_uk_cross_grain_leg_of_area, + parent_geography_legs={ + "K02000001": ("England", "Wales", "Scotland", "Northern Ireland"), + "K03000001": ("England", "Wales", "Scotland"), + "E92000001": ("England",), + "W92000004": ("Wales",), + "S92000003": ("Scotland",), + "N92000002": ("Northern Ireland",), + }, +) + + def align_uk_local_registry_parity_fixture( fixture: Mapping[str, Any], ) -> dict[str, Any]: @@ -554,7 +619,10 @@ def to_frame(self) -> Frame: ) -def _uk_contract_targets() -> dict[str, Mapping[str, Any]]: +def _uk_contract_targets( + *, + national_only: bool = True, +) -> dict[str, Mapping[str, Any]]: payload = ( importlib_resources.files("microcosm.build.uk") .joinpath(UK_POPULATION_TARGETS_RESOURCE) @@ -565,11 +633,58 @@ def _uk_contract_targets() -> dict[str, Mapping[str, Any]]: return { target["target_id"]: target for target in contract["targets"] - if set(target.get("geography_levels") or ()) + if not national_only + or set(target.get("geography_levels") or ()) <= UK_NATIONAL_TARGET_GEOGRAPHY_LEVELS } +def apply_uk_cross_grain_reconciliation( + local_frame: pd.DataFrame, + bound_higher_targets: Iterable[str], +) -> tuple[pd.DataFrame, dict[str, Any]]: + """Apply the standing UK rule to a bound mixed-grain target surface. + + Increment #762 may extend the grains in the UK rowwise solve only through + this front door, so detection, reconciliation, and the manifest receipt + cannot be bypassed. + """ + + return apply_cross_grain_reconciliation( + local_frame, + bound_higher_targets, + _uk_contract_targets(national_only=False), + UK_CROSS_GRAIN_RULE, + ) + + +def _validate_uk_cross_grain_declarations() -> None: + contract = _uk_contract_targets(national_only=False) + matched_sides: dict[str, str] = {} + unknown: list[str] = [] + for bridge in UK_CROSS_GRAIN_BRIDGES: + for target_id in bridge.higher_target_ids: + if target_id not in contract: + unknown.append(target_id) + lower_target_id = bridge.lower_side.removeprefix("contract:") + if bridge.lower_side.startswith("contract:") and lower_target_id not in contract: + unknown.append(lower_target_id) + for side in (*bridge.higher_target_ids, bridge.lower_side): + canonical = side.removeprefix("contract:") + existing = matched_sides.get(canonical) + if existing is not None: + raise ValueError( + f"UK cross-grain target {canonical!r} is covered by both " + f"{existing!r} and {bridge.bridge_id!r}." + ) + matched_sides[canonical] = bridge.bridge_id + if unknown: + raise ValueError( + "UK cross-grain bridge target id(s) are absent from the committed " + f"contract: {sorted(set(unknown))}." + ) + + def _warn_on_undeclared_geography(targets) -> None: # Ruling on PR #795 review finding 3: an absent geography_levels reads as # national by doctrine (the empty set is a subset of the national levels), @@ -643,3 +758,6 @@ def _compare_series( if operator not in operations: raise ValueError(f"Unsupported UK target condition operator {operator!r}.") return operations[operator](expected) + + +_validate_uk_cross_grain_declarations() diff --git a/packages/microcosm-build/src/microcosm/build/uk_runtime/local_target_census.py b/packages/microcosm-build/src/microcosm/build/uk_runtime/local_target_census.py index 126825a8bc..4adb60676d 100644 --- a/packages/microcosm-build/src/microcosm/build/uk_runtime/local_target_census.py +++ b/packages/microcosm-build/src/microcosm/build/uk_runtime/local_target_census.py @@ -544,7 +544,10 @@ "national-total deltas at review: E&W +105, Scotland -554, NI " "+3). Binding treats the published counts as the target values " "with that noise documented — never as exact controls — and any " - "cross-grain reconciliation must name which grain wins." + "cross-grain reconciliation applies the standing cross-grain rule " + "declared in uk_runtime.ledger_targets: country wins, so a bound " + "national same-concept control rescales the constituency values " + "before the solve." ), "authority": ( "ONS/NRS/NISRA statistical disclosure control documentation; " diff --git a/packages/microcosm-build/tests/test_cross_grain.py b/packages/microcosm-build/tests/test_cross_grain.py new file mode 100644 index 0000000000..511fa89528 --- /dev/null +++ b/packages/microcosm-build/tests/test_cross_grain.py @@ -0,0 +1,568 @@ +from __future__ import annotations + +import numpy as np +import pandas as pd +import pytest + +from microcosm.build.cross_grain import ( + CrossGrainBridge, + CrossGrainRule, + apply_cross_grain_reconciliation, + detect_cross_grain_inconsistencies, +) + + +def _leg(area: str) -> str: + return area[0] + + +def _rule(*, bridges: tuple[CrossGrainBridge, ...] = ()) -> CrossGrainRule: + return CrossGrainRule( + grain_precedence=("country", "constituency", "la"), + signature_fields=("concept", "entity", "map_to", "filters"), + bridges=bridges, + leg_of_area=_leg, + parent_geography_legs={ + "UK": ("E", "W", "S", "N"), + "GB": ("E", "W", "S"), + "E": ("E",), + "S": ("S",), + }, + ) + + +def _signature(concept: str = "households") -> dict[str, object]: + return { + "measurement": { + "concept": concept, + "entity": "household", + "map_to": None, + "filters": [], + } + } + + +def test_exact_signature_rescales_to_country_and_receipts_without_mutation(): + surface = pd.DataFrame( + [ + ("country", "UK", "national", 120.0), + ("constituency", "E1", "local", 40.0), + ("constituency", "S1", "local", 20.0), + ], + columns=["grain", "geography_id", "target_id", "value"], + ) + original = surface.copy(deep=True) + signatures = {"national": _signature(), "local": _signature()} + + reconciled, receipt = apply_cross_grain_reconciliation( + surface, ("national",), signatures, _rule() + ) + + pd.testing.assert_frame_equal(surface, original) + assert reconciled["value"].tolist() == [120.0, 80.0, 40.0] + assert len(receipt["inconsistencies_in_force"]) == 1 + assert receipt["absence"] is None + group = receipt["groups"][0] + assert group["bridge_id"] is None + assert group["winning_grain"] == "country" + assert group["legs"] == [ + { + "leg": "E+W+S+N", + "parent_geography_id": "UK", + "higher_target_ids": ["national"], + "n_areas": 2, + "old_total": 60.0, + "new_total": 120.0, + "relative_shift": 1.0, + "declared_factor": 2.0, + "reason": "standing cross-grain rule: country controls constituency", + } + ] + + +def test_bridge_sums_exhaustive_higher_partition_to_one_control(): + bridge = CrossGrainBridge( + "partition_vs_external", + concept="households", + higher_target_ids=("part_a", "part_b"), + lower_side="external:census/households", + ) + surface = pd.DataFrame( + [ + ("country", "UK", "part_a", 30.0), + ("country", "UK", "part_b", 70.0), + ("constituency", "E1", "external:census/households", 30.0), + ("constituency", "W1", "external:census/households", 20.0), + ], + columns=["grain", "geography_id", "target_id", "value"], + ) + + reconciled, receipt = apply_cross_grain_reconciliation( + surface, + ("part_a", "part_b"), + {"part_a": _signature(), "part_b": _signature()}, + _rule(bridges=(bridge,)), + ) + + assert reconciled["value"].tolist() == [30.0, 70.0, 60.0, 40.0] + assert receipt["groups"][0]["bridge_id"] == "partition_vs_external" + assert receipt["groups"][0]["legs"][0]["higher_target_ids"] == [ + "part_a", + "part_b", + ] + + +def test_middle_grain_wins_when_country_is_absent(): + surface = pd.DataFrame( + [ + ("constituency", "E1", "same", 75.0), + ("constituency", "S1", "same", 25.0), + ("la", "E9", "same", 30.0), + ("la", "S9", "same", 20.0), + ], + columns=["grain", "geography_id", "target_id", "value"], + ) + + reconciled, receipt = apply_cross_grain_reconciliation( + surface, (), {"same": _signature()}, _rule() + ) + + assert reconciled["value"].tolist() == [75.0, 25.0, 75.0, 25.0] + assert receipt["groups"][0]["winning_grain"] == "constituency" + assert [leg["declared_factor"] for leg in receipt["groups"][0]["legs"]] == [ + 2.5, + 1.25, + ] + + +def test_absence_receipt_exists_when_no_higher_target_is_bound(): + surface = pd.DataFrame( + [("constituency", "E1", "local", 10.0)], + columns=["grain", "geography_id", "target_id", "value"], + ) + + reconciled, receipt = apply_cross_grain_reconciliation( + surface, (), {"local": _signature()}, _rule() + ) + + pd.testing.assert_frame_equal(reconciled, surface) + assert receipt == { + "bound_higher_targets": [], + "inconsistencies_in_force": [], + "groups": [], + "absence": "No cross-grain inconsistencies are in force on this surface.", + } + + +def test_partially_bound_declared_partition_is_refused(): + bridge = CrossGrainBridge( + "partition", + "households", + ("part_a", "part_b"), + "external:census/households", + ) + surface = pd.DataFrame( + [("country", "UK", "part_a", 10.0)], + columns=["grain", "geography_id", "target_id", "value"], + ) + + with pytest.raises(ValueError, match="partially bound"): + detect_cross_grain_inconsistencies( + surface, + ("part_a",), + {"part_a": _signature(), "part_b": _signature()}, + _rule(bridges=(bridge,)), + ) + + +def test_target_matched_by_two_bridges_is_refused(): + bridges = ( + CrossGrainBridge("one", "households", ("national",), "contract:local"), + CrossGrainBridge("two", "households", ("other",), "contract:local"), + ) + surface = pd.DataFrame( + [("constituency", "E1", "local", 10.0)], + columns=["grain", "geography_id", "target_id", "value"], + ) + + with pytest.raises(ValueError, match="matched by two bridges"): + detect_cross_grain_inconsistencies( + surface, + (), + { + "national": _signature(), + "other": _signature(), + "local": _signature(), + }, + _rule(bridges=bridges), + ) + + +@pytest.mark.parametrize( + ("country_value", "local_values", "message"), + [ + (10.0, (0.0, 0.0), "vanishing lower-leg total"), + (-10.0, (4.0, 6.0), "opposite-signed"), + ], +) +def test_invalid_factor_math_is_refused(country_value, local_values, message): + surface = pd.DataFrame( + [ + ("country", "UK", "national", country_value), + ("constituency", "E1", "local", local_values[0]), + ("constituency", "S1", "local", local_values[1]), + ], + columns=["grain", "geography_id", "target_id", "value"], + ) + + with pytest.raises(ValueError, match=message): + apply_cross_grain_reconciliation( + surface, + ("national",), + {"national": _signature(), "local": _signature()}, + _rule(), + ) + + +def test_unparented_and_empty_legs_are_refused(): + unparented = pd.DataFrame( + [ + ("country", "GB", "national", 10.0), + ("constituency", "N1", "local", 10.0), + ], + columns=["grain", "geography_id", "target_id", "value"], + ) + signatures = {"national": _signature(), "local": _signature()} + with pytest.raises(ValueError, match="unparented"): + apply_cross_grain_reconciliation( + unparented, ("national",), signatures, _rule() + ) + + empty = pd.DataFrame( + [ + ("country", "E", "national", 10.0), + ("country", "S", "national", 5.0), + ("constituency", "E1", "local", 8.0), + ], + columns=["grain", "geography_id", "target_id", "value"], + ) + with pytest.raises(ValueError, match="empty leg"): + apply_cross_grain_reconciliation(empty, ("national",), signatures, _rule()) + + +def test_two_different_controls_at_same_grain_are_refused(): + surface = pd.DataFrame( + [ + ("country", "UK", "national", 10.0), + ("country", "UK", "national", 11.0), + ("constituency", "E1", "local", 10.0), + ], + columns=["grain", "geography_id", "target_id", "value"], + ) + with pytest.raises(ValueError, match="two different control values"): + apply_cross_grain_reconciliation( + surface, + ("national",), + {"national": _signature(), "local": _signature()}, + _rule(), + ) + + +def test_incompatible_exact_and_bridged_controls_are_refused(): + bridge = CrossGrainBridge( + "alternate_control", + "households", + ("bridged_national",), + "contract:local", + ) + surface = pd.DataFrame( + [ + ("country", "UK", "exact_partition", 100.0), + ("country", "UK", "bridged_national", 90.0), + ("constituency", "E1", "local", 50.0), + ], + columns=["grain", "geography_id", "target_id", "value"], + ) + signatures = { + "exact_partition": _signature(), + "local": _signature(), + "bridged_national": _signature("alternate"), + } + + with pytest.raises(ValueError, match="two different control values"): + apply_cross_grain_reconciliation( + surface, + ("exact_partition", "bridged_national"), + signatures, + _rule(bridges=(bridge,)), + ) + + +def test_non_finite_targets_are_refused(): + surface = pd.DataFrame( + [("constituency", "E1", "local", np.nan)], + columns=["grain", "geography_id", "target_id", "value"], + ) + with pytest.raises(ValueError, match="finite"): + apply_cross_grain_reconciliation( + surface, (), {"local": _signature()}, _rule() + ) + + +def test_off_control_reconciliation_is_refused(monkeypatch): + """Fault injection: if a rescaled leg lands off its control, the pass fails. + + Closure is the property the pass exists to establish, so it is asserted + after the write rather than inferred from the factor arithmetic. + """ + + from microcosm.build import cross_grain as module + + surface = pd.DataFrame( + [ + ("country", "UK", "national", 120.0), + ("constituency", "E1", "local", 40.0), + ("constituency", "S1", "local", 20.0), + ], + columns=["grain", "geography_id", "target_id", "value"], + ) + signatures = {"national": _signature(), "local": _signature()} + + monkeypatch.setattr( + module.np, "isclose", lambda *args, **kwargs: False + ) + with pytest.raises(ValueError, match="off its control"): + apply_cross_grain_reconciliation( + surface, ("national",), signatures, _rule() + ) + + +def test_closure_holds_across_many_legs_with_awkward_floats(): + """The closure assertion must not false-positive on ordinary float drift.""" + + rows = [("country", "UK", "national", 28_356_000.0)] + values = [1234.567_89 + index * 3.141_59 for index in range(200)] + for index, value in enumerate(values): + leg = "EWSN"[index % 4] + rows.append(("constituency", f"{leg}{index}", "local", value)) + surface = pd.DataFrame( + rows, columns=["grain", "geography_id", "target_id", "value"] + ) + + reconciled, receipt = apply_cross_grain_reconciliation( + surface, + ("national",), + {"national": _signature(), "local": _signature()}, + _rule(), + ) + + local = reconciled.loc[reconciled["grain"] == "constituency", "value"] + assert local.sum() == pytest.approx(28_356_000.0, rel=1e-12) + assert receipt["groups"][0]["legs"][0]["new_total"] == pytest.approx( + 28_356_000.0, rel=1e-12 + ) + + +def test_absent_and_empty_signature_spellings_group_together(): + """`filters: []` and an omitted `filters` are the same measurement.""" + + surface = pd.DataFrame( + [ + ("country", "UK", "national", 120.0), + ("constituency", "E1", "local", 40.0), + ("constituency", "S1", "local", 20.0), + ], + columns=["grain", "geography_id", "target_id", "value"], + ) + spelled_empty = { + "measurement": { + "concept": "households", + "entity": "household", + "map_to": None, + "filters": [], + } + } + spelled_absent = {"measurement": {"concept": "households", "entity": "household"}} + + reconciled, receipt = apply_cross_grain_reconciliation( + surface, + ("national",), + {"national": spelled_empty, "local": spelled_absent}, + _rule(), + ) + + assert len(receipt["groups"]) == 1 + assert reconciled.loc[1:, "value"].tolist() == [80.0, 40.0] + + +def test_filter_order_does_not_split_a_signature(): + conditions = [ + {"concept": "uk.benefits.universal_credit.amount", "op": ">", "value": 0}, + {"concept": "uk.household.tenure", "op": "==", "value": "rented"}, + ] + surface = pd.DataFrame( + [ + ("country", "UK", "national", 100.0), + ("constituency", "E1", "local", 25.0), + ], + columns=["grain", "geography_id", "target_id", "value"], + ) + forward = { + "measurement": { + "concept": "households", + "entity": "household", + "filters": conditions, + } + } + reversed_order = { + "measurement": { + "concept": "households", + "entity": "household", + "filters": list(reversed(conditions)), + } + } + + _, receipt = apply_cross_grain_reconciliation( + surface, + ("national",), + {"national": forward, "local": reversed_order}, + _rule(), + ) + + assert len(receipt["groups"]) == 1 + + +def test_contract_entry_without_measurement_block_is_refused(): + surface = pd.DataFrame( + [("constituency", "E1", "local", 10.0)], + columns=["grain", "geography_id", "target_id", "value"], + ) + with pytest.raises(ValueError, match="must carry a 'measurement' mapping"): + apply_cross_grain_reconciliation( + surface, + (), + {"local": {"concept": "households", "entity": "household"}}, + _rule(), + ) + + +def test_vanishing_lower_leg_total_is_refused(): + surface = pd.DataFrame( + [ + ("country", "E", "national", 100.0), + ("constituency", "E1", "local", 1e-18), + ], + columns=["grain", "geography_id", "target_id", "value"], + ) + with pytest.raises(ValueError, match="vanishing lower-leg total"): + apply_cross_grain_reconciliation( + surface, + ("national",), + {"national": _signature(), "local": _signature()}, + _rule(), + ) + + +def test_offsetting_members_reconcile_when_the_leg_total_is_well_conditioned(): + """A net-valued concept may legitimately hold offsetting members. + + Conditioning is the only hazard, and the relative floor is that test, so a + leg summing cleanly to its control must reconcile rather than be refused + for containing both signs. + """ + + surface = pd.DataFrame( + [ + ("country", "E", "national", 100.0), + ("constituency", "E1", "local", 200.0), + ("constituency", "E2", "local", -100.0), + ], + columns=["grain", "geography_id", "target_id", "value"], + ) + + reconciled, receipt = apply_cross_grain_reconciliation( + surface, + ("national",), + {"national": _signature(), "local": _signature()}, + _rule(), + ) + + assert receipt["groups"][0]["legs"][0]["declared_factor"] == pytest.approx(1.0) + assert reconciled.loc[1:, "value"].tolist() == [200.0, -100.0] + + +def test_cancelling_leg_is_still_refused_by_the_conditioning_floor(): + surface = pd.DataFrame( + [ + ("country", "E", "national", 100.0), + ("constituency", "E1", "local", 1e-3), + ("constituency", "E2", "local", -1e-3), + ], + columns=["grain", "geography_id", "target_id", "value"], + ) + with pytest.raises(ValueError, match="vanishing lower-leg total"): + apply_cross_grain_reconciliation( + surface, + ("national",), + {"national": _signature(), "local": _signature()}, + _rule(), + ) + + +def test_zero_control_over_a_zero_summing_leg_is_a_no_op(): + """The zero branch is live: it is the 0/0 case, not a dead special case.""" + + surface = pd.DataFrame( + [ + ("country", "E", "national", 0.0), + ("constituency", "E1", "local", 5.0), + ("constituency", "E2", "local", -5.0), + ], + columns=["grain", "geography_id", "target_id", "value"], + ) + + reconciled, receipt = apply_cross_grain_reconciliation( + surface, + ("national",), + {"national": _signature(), "local": _signature()}, + _rule(), + ) + + assert receipt["groups"][0]["legs"][0]["declared_factor"] == pytest.approx(1.0) + assert reconciled.loc[1:, "value"].tolist() == [5.0, -5.0] + + +def test_ordered_payload_inside_a_filter_does_not_false_collide(): + """`between [0, 100]` and `between [100, 0]` are different measurements. + + Order-insensitivity is justified at the conjunction level the filter list + occupies; applying it inside a condition would merge two distinct + measurements into one group and rescale onto the wrong control. + """ + + def _between(lower: float, upper: float) -> dict[str, object]: + return { + "measurement": { + "concept": "households", + "entity": "household", + "filters": [{"op": "between", "value": [lower, upper]}], + } + } + + surface = pd.DataFrame( + [ + ("country", "UK", "national", 100.0), + ("constituency", "E1", "local", 25.0), + ], + columns=["grain", "geography_id", "target_id", "value"], + ) + + _, receipt = apply_cross_grain_reconciliation( + surface, + ("national",), + {"national": _between(0.0, 100.0), "local": _between(100.0, 0.0)}, + _rule(), + ) + + assert receipt["groups"] == [] + assert receipt["absence"] diff --git a/packages/microcosm-build/tests/test_uk_ledger_targets.py b/packages/microcosm-build/tests/test_uk_ledger_targets.py index 6253870f1a..22841a8ca6 100644 --- a/packages/microcosm-build/tests/test_uk_ledger_targets.py +++ b/packages/microcosm-build/tests/test_uk_ledger_targets.py @@ -3,10 +3,15 @@ from types import SimpleNamespace import numpy as np +import pandas as pd import pytest from microcosm.build.ledger_targets import LedgerTargetReference from microcosm.build.uk_runtime.ledger_targets import ( + UK_CROSS_GRAIN_BRIDGES, + UK_CROSS_GRAIN_GRAIN_PRECEDENCE, + UK_CROSS_GRAIN_RULE, + apply_uk_cross_grain_reconciliation, compile_uk_local_target_registry, compile_uk_target_registry, materialize_uk_ledger_targets, @@ -489,3 +494,199 @@ def test_household_condition_still_reduces_group_entities(): ) assert list(single) == [True, False] + + +def test_uk_cross_grain_rule_constants_are_review_pinned(): + assert UK_CROSS_GRAIN_GRAIN_PRECEDENCE == ( + "country", + "constituency", + "la", + ) + assert UK_CROSS_GRAIN_RULE.grain_precedence == ( + "country", + "constituency", + "la", + ) + assert [bridge.bridge_id for bridge in UK_CROSS_GRAIN_BRIDGES] == [ + "national_household_composition_partition_vs_census_households", + "national_uc_caseload_vs_uc_households_by_area", + ] + assert UK_CROSS_GRAIN_BRIDGES[0].higher_target_ids == ( + "ons.household_composition.lone_households_under_65", + "ons.household_composition.lone_households_over_65", + "ons.household_composition.unrelated_adult_households", + "ons.household_composition.couple_no_children_households", + "ons.household_composition.couple_under_3_children_households", + "ons.household_composition.couple_3_plus_children_households", + "ons.household_composition.couple_non_dependent_children_only_households", + "ons.household_composition.lone_parent_dependent_children_households", + "ons.household_composition.lone_parent_non_dependent_children_households", + "ons.household_composition.multi_family_households", + ) + assert UK_CROSS_GRAIN_BRIDGES[1].higher_target_ids == ( + "dwp.uc.households", + ) + + +def test_committed_contract_detects_exact_uc_payment_partition(): + payment_ids = ( + "dwp.uc.payment_distribution_single", + "dwp.uc.payment_distribution_lone_parent", + "dwp.uc.payment_distribution_couple_no_children", + "dwp.uc.payment_distribution_couple_with_children", + ) + surface = pd.DataFrame( + [ + *[ + { + "grain": "country", + "geography_id": "K02000001", + "target_id": target_id, + "value": 25.0, + } + for target_id in payment_ids + ], + { + "grain": "constituency", + "geography_id": "E14000001", + "target_id": "dwp.uc.households_by_area", + "value": 30.0, + }, + { + "grain": "constituency", + "geography_id": "S14000001", + "target_id": "dwp.uc.households_by_area", + "value": 20.0, + }, + ] + ) + + reconciled, receipt = apply_uk_cross_grain_reconciliation( + surface, payment_ids + ) + + assert len(receipt["groups"]) == 1 + assert receipt["groups"][0]["bridge_id"] is None + assert reconciled.loc[4:, "value"].tolist() == [60.0, 40.0] + + +def test_real_uk_bridges_resolve_contract_and_external_lower_sides(): + household_bridge, uc_bridge = UK_CROSS_GRAIN_BRIDGES + household_surface = pd.DataFrame( + [ + *[ + { + "grain": "country", + "geography_id": "K02000001", + "target_id": target_id, + "value": 10.0, + } + for target_id in household_bridge.higher_target_ids + ], + { + "grain": "constituency", + "geography_id": "E14000001", + "target_id": "external:census_households/households", + "value": 40.0, + }, + { + "grain": "constituency", + "geography_id": "S14000001", + "target_id": "external:census_households/households", + "value": 10.0, + }, + ] + ) + reconciled, receipt = apply_uk_cross_grain_reconciliation( + household_surface, household_bridge.higher_target_ids + ) + assert receipt["groups"][0]["bridge_id"] == household_bridge.bridge_id + assert reconciled.loc[10:, "value"].tolist() == [80.0, 20.0] + + uc_surface = pd.DataFrame( + [ + { + "grain": "country", + "geography_id": "K02000001", + "target_id": "dwp.uc.households", + "value": 90.0, + }, + { + "grain": "constituency", + "geography_id": "E14000001", + "target_id": "dwp.uc.households_by_area", + "value": 30.0, + }, + { + "grain": "constituency", + "geography_id": "S14000001", + "target_id": "dwp.uc.households_by_area", + "value": 15.0, + }, + ] + ) + reconciled, receipt = apply_uk_cross_grain_reconciliation( + uc_surface, uc_bridge.higher_target_ids + ) + assert receipt["groups"][0]["bridge_id"] == uc_bridge.bridge_id + assert reconciled.loc[1:, "value"].tolist() == [60.0, 30.0] + + +def test_uk_front_door_reconciles_per_country_legs_and_builds_uniform_surface(): + surface = pd.DataFrame( + [ + { + "grain": "country", + "geography_id": "E92000001", + "target_id": "dwp.uc.households", + "value": 60.0, + }, + { + "grain": "country", + "geography_id": "S92000003", + "target_id": "dwp.uc.households", + "value": 40.0, + }, + { + "grain": "constituency", + "geography_id": "E14000001", + "target_id": "dwp.uc.households_by_area", + "value": 30.0, + }, + { + "grain": "constituency", + "geography_id": "S14000001", + "target_id": "dwp.uc.households_by_area", + "value": 10.0, + }, + ] + ) + reconciled, _ = apply_uk_cross_grain_reconciliation( + surface, ("dwp.uc.households",) + ) + local = reconciled.loc[reconciled["grain"] == "constituency"] + targets = local.rename( + columns={"geography_id": "code", "value": "uc_households"} + )[["code", "uc_households"]] + metrics = pd.DataFrame( + {"uc_households": [1.0, 1.0]}, + index=pd.Index([1, 2], name="household_id"), + ) + assigned = pd.Series( + ["E14000001", "S14000001"], + index=metrics.index, + ) + + from microcosm.build.uk_runtime.local_rowwise import ( + _require_uniform_target_surface, + build_uk_rowwise_local_matrix, + ) + + problem = build_uk_rowwise_local_matrix( + metrics, + assigned, + targets, + area_type="constituency", + ) + _require_uniform_target_surface(problem) + assert problem.targets.tolist() == [60.0, 40.0] diff --git a/packages/microcosm-build/tests/test_uk_local_rowwise.py b/packages/microcosm-build/tests/test_uk_local_rowwise.py index 69e58fbb8d..c40b324e6e 100644 --- a/packages/microcosm-build/tests/test_uk_local_rowwise.py +++ b/packages/microcosm-build/tests/test_uk_local_rowwise.py @@ -212,8 +212,9 @@ def test_rowwise_doctrine_solve_uses_base_weights_directly() -> None: stood_on = receipt["stood_on"]["census_households/constituency"] seed = stood_on["census_disclosure_control_noise"] assert seed["approved_by"] == "juaristi22" - assert seed["approved_on"] == "2026-08-27" - assert seed["expires_on"] == "2026-11-27" + assert seed["adjudication"] == "microcosm#802" + assert seed["approved_on"] == "2026-08-31" + assert seed["expires_on"] == "2026-11-30" assert receipt["stood_on"]["tenure/constituency"] == {} diff --git a/packages/microcosm-build/tests/test_uk_local_target_census.py b/packages/microcosm-build/tests/test_uk_local_target_census.py index d3fe3d2d94..69089f967d 100644 --- a/packages/microcosm-build/tests/test_uk_local_target_census.py +++ b/packages/microcosm-build/tests/test_uk_local_target_census.py @@ -154,6 +154,16 @@ def test_fences_declare_enforcement_and_gate_reviewed_families() -> None: assert census["scope"], "census must declare its metric-surface scope" +def test_census_disclosure_fence_names_country_as_winning_grain() -> None: + census = build_uk_local_target_census() + fences = {row["fence_id"]: row for row in census["binding_fences"]} + rule = fences["census_disclosure_control_noise"]["rule"] + assert "standing cross-grain rule" in rule + assert "uk_runtime.ledger_targets" in rule + assert "country wins" in rule + assert "national same-concept control" in rule + + def test_unclassified_metric_fails_closed(monkeypatch: pytest.MonkeyPatch) -> None: from microcosm.build.uk_runtime import local_target_census as module diff --git a/packages/microcosm-build/tests/test_uk_rowwise_candidate.py b/packages/microcosm-build/tests/test_uk_rowwise_candidate.py index f98c54f1ca..0aea2a5390 100644 --- a/packages/microcosm-build/tests/test_uk_rowwise_candidate.py +++ b/packages/microcosm-build/tests/test_uk_rowwise_candidate.py @@ -285,9 +285,16 @@ def test_candidate_build_writes_calibrated_h5_and_evidence(tmp_path) -> None: "census_disclosure_control_noise" ] assert seed["approved_by"] == "juaristi22" - assert seed["approved_on"] == "2026-08-27" - assert seed["expires_on"] == "2026-11-27" + assert seed["adjudication"] == "microcosm#802" + assert seed["approved_on"] == "2026-08-31" + assert seed["expires_on"] == "2026-11-30" assert adjudications["dormant"] == [] + cross_grain = manifest["cross_grain"] + assert cross_grain["bound_national_targets"] == [] + assert cross_grain["bound_higher_targets"] == [] + assert cross_grain["inconsistencies_in_force"] == [] + assert cross_grain["groups"] == [] + assert cross_grain["absence"] assert manifest["ladder_target_provenance"] == ladder_target_provenance(ladder) assert manifest["gate"]["passed"] is True assert manifest["gate"]["phase"] == "post_calibration" @@ -430,6 +437,12 @@ def forbidden(*_args, **_kwargs): in adjudications["stood_on"]["census_households/constituency"] ) assert adjudications["dormant"] == [] + cross_grain = plan["cross_grain"] + assert cross_grain["bound_national_targets"] == [] + assert cross_grain["bound_higher_targets"] == [] + assert cross_grain["inconsistencies_in_force"] == [] + assert cross_grain["groups"] == [] + assert cross_grain["absence"] assert plan["ladder_target_provenance"] == ladder_target_provenance(ladder) assert plan["shapes"]["person"][0] == 8 assert plan["shapes"]["benunit"][0] == 8 diff --git a/tools/build_uk_rowwise_candidate.py b/tools/build_uk_rowwise_candidate.py index cc39600e09..7dea45ff2c 100644 --- a/tools/build_uk_rowwise_candidate.py +++ b/tools/build_uk_rowwise_candidate.py @@ -59,6 +59,7 @@ UkOaLadder, UKRowwiseDoctrineSolve, UKRowwiseLocalMatrix, + apply_uk_cross_grain_reconciliation, build_uk_rowwise_local_matrix, clone_uk_dataset_with_ladder_geography, constituency_household_targets, @@ -76,6 +77,7 @@ from microcosm.frame import MassChangeRecord BOUND_TARGET_FAMILIES = ("census_households/constituency",) +BOUND_NATIONAL_TARGETS: tuple[str, ...] = () CANDIDATE_FILENAME_TEMPLATE = "populace_uk_{source_year}_rowwise_candidate.h5" MANIFEST_FILENAME = "rowwise_candidate_manifest.json" SOLVE_DIAGNOSTICS_FILENAME = "solve_diagnostics.csv" @@ -367,7 +369,7 @@ def _run_candidate( append_phase(state, "cloned") print("binding census household targets...", file=sys.stderr, flush=True) - household, problem = _build_bound_problem( + household, problem, cross_grain = _build_bound_problem( assignment, target_ladder=ladder, ) @@ -394,6 +396,7 @@ def _run_candidate( ladder_artifact=ladder_artifact, target_provenance=target_provenance, binding_adjudications=binding_adjudications, + cross_grain=cross_grain, ) print(_json_text(plan), end="") return 0 @@ -495,6 +498,7 @@ def _run_candidate( input_artifact=input_artifact, ladder_artifact=ladder_artifact, target_provenance=target_provenance, + cross_grain=cross_grain, ) append_phase(state, "published") manifest_path = output_paths["manifest"] @@ -584,7 +588,7 @@ def _build_bound_problem( assignment: _LadderAssignment, *, target_ladder: UkOaLadder, -) -> tuple[pd.DataFrame, UKRowwiseLocalMatrix]: +) -> tuple[pd.DataFrame, UKRowwiseLocalMatrix, dict[str, Any]]: """Bind the one target family, refusing separately loaded ladders.""" if assignment.ladder is not target_ladder: @@ -607,6 +611,22 @@ def _build_bound_problem( name="constituency_code", ) targets = constituency_household_targets(target_ladder) + local_surface = pd.DataFrame( + { + "grain": "constituency", + "geography_id": targets["code"].astype(str), + "target_id": "external:census_households/households", + "value": targets["households"].to_numpy(dtype=np.float64), + } + ) + reconciled_surface, cross_grain = apply_uk_cross_grain_reconciliation( + local_surface, + BOUND_NATIONAL_TARGETS, + ) + targets = targets.copy() + targets["households"] = reconciled_surface["value"].to_numpy( + dtype=np.float64 + ) problem = build_uk_rowwise_local_matrix( metrics, assigned, @@ -614,7 +634,10 @@ def _build_bound_problem( area_type="constituency", code_column="code", ) - return household, problem + return household, problem, { + "bound_national_targets": list(BOUND_NATIONAL_TARGETS), + **cross_grain, + } def _dry_run_plan( @@ -627,6 +650,7 @@ def _dry_run_plan( ladder_artifact: Mapping[str, Any], target_provenance: Mapping[str, Any], binding_adjudications: Mapping[str, Any], + cross_grain: Mapping[str, Any], ) -> dict[str, Any]: return { "schema_version": 1, @@ -635,6 +659,7 @@ def _dry_run_plan( "candidate_scope": "adjudicated_partial", "bound_target_families": list(BOUND_TARGET_FAMILIES), "binding_adjudications": dict(binding_adjudications), + "cross_grain": dict(cross_grain), "ladder_target_provenance": dict(target_provenance), "inputs": { "dataset": dict(input_artifact), @@ -666,6 +691,7 @@ def _write_output_bundle( input_artifact: Mapping[str, Any], ladder_artifact: Mapping[str, Any], target_provenance: Mapping[str, Any], + cross_grain: Mapping[str, Any], ) -> dict[str, Any]: """Stage the complete bundle, then publish atomically per file.""" @@ -719,6 +745,7 @@ def _write_output_bundle( input_artifact=input_artifact, ladder_artifact=ladder_artifact, target_provenance=target_provenance, + cross_grain=cross_grain, outputs=outputs, ) staged["manifest"].write_text(_json_text(manifest)) @@ -741,6 +768,7 @@ def _manifest( input_artifact: Mapping[str, Any], ladder_artifact: Mapping[str, Any], target_provenance: Mapping[str, Any], + cross_grain: Mapping[str, Any], outputs: Mapping[str, Any], ) -> dict[str, Any]: abs_errors = solve.diagnostics["abs_relative_error"].to_numpy(dtype=np.float64) @@ -755,6 +783,7 @@ def _manifest( "git_commit": _git_commit(), "bound_target_families": list(BOUND_TARGET_FAMILIES), "binding_adjudications": dict(solve.binding_adjudications), + "cross_grain": dict(cross_grain), "ladder_target_provenance": dict(target_provenance), "parameters": _parameters(args, source_year=source_year), "inputs": { diff --git a/tools/ci_test_groups.py b/tools/ci_test_groups.py index 6c2d712805..01fffb2c28 100644 --- a/tools/ci_test_groups.py +++ b/tools/ci_test_groups.py @@ -27,6 +27,13 @@ "packages/microcosm-build/tests/test_us_release_head_to_head_scorer.py", ) +# Shared-engine behavioral contracts that deliberately do not carry a country +# prefix or the spec-engine ``test_spec_*`` prefix. Listing them makes their +# shared-lane placement reviewed rather than a silent classifier default. +EXPLICIT_SHARED_SPEC = ( + "packages/microcosm-build/tests/test_cross_grain.py", +) + PROCESSES = { "trade": ("main",), "spine-uk": ("main",), @@ -241,7 +248,9 @@ def defaulted_engine_files(grouped: dict[str, list[str]]) -> list[str]: defaulted = [] for path in grouped["shared-spec"]: name = basename(path) - if not (is_build(path) and name.startswith("test_spec_")): + if path not in EXPLICIT_SHARED_SPEC and not ( + is_build(path) and name.startswith("test_spec_") + ): defaulted.append(path) return defaulted