diff --git a/changelog.d/147-uk-local-gate-battery.added.md b/changelog.d/147-uk-local-gate-battery.added.md new file mode 100644 index 000000000..3e159bfab --- /dev/null +++ b/changelog.d/147-uk-local-gate-battery.added.md @@ -0,0 +1 @@ +Add a signed six-gate battery and calibration diagnostics bundle for UK local candidates. diff --git a/docs/gate-battery-contract.md b/docs/gate-battery-contract.md index 8f340e0b0..a39380cb8 100644 --- a/docs/gate-battery-contract.md +++ b/docs/gate-battery-contract.md @@ -224,10 +224,10 @@ stretches on declaration alone: | | BE | UK | | --- | --- | --- | -| phases | `terminal` | `preflight`, `terminal` | -| entries | 9 | 13 | +| phases | `terminal` | `preflight`, `assembled`, `transferred`, `terminal` | +| entries | 9 | 43 (37 national-build entries plus 6 local-candidate entries) | | incumbent posture | none — incumbent-comparison gates deliberately not selected; external oracles replace self-parity | full — export/target parity against the pinned enhanced-FRS incumbent | -| criticality mix | 8 blocking + 1 diagnostic | all blocking | +| criticality mix | 8 blocking + 1 diagnostic | 39 blocking + 4 local diagnostics | The differences are entirely in the two country inputs — the spec file and the registry — which is the point. The UK national build is the @@ -236,6 +236,14 @@ build and runs both phases under `BLOCKS_ARTIFACT` (preflight before the frame loads, terminal immediately before the staging writer), while BE remains spec-only. +UK evaluates the declaration through four producer scopes: spine, +calibration seam, national release cut, and local candidate. The first three +compose the national release certification. The six local-candidate entries +are explicitly classified as excluded from that certification until the +local publication work in #146; they are evaluated in their own signed +`local_candidate` report rather than silently disappearing from union +accounting. + ## The reference rule A parity gate is only as honest as its reference: comparing a calibrated diff --git a/packages/microcosm-build/src/microcosm/build/__init__.py b/packages/microcosm-build/src/microcosm/build/__init__.py index 624cdd3ed..352f53b74 100644 --- a/packages/microcosm-build/src/microcosm/build/__init__.py +++ b/packages/microcosm-build/src/microcosm/build/__init__.py @@ -81,6 +81,7 @@ def _assert_frame_compatible(version: str, required: tuple[int, int]) -> None: TargetCoverageRequirement, TargetFitRequirement, aggregate_admin_gate, + area_support_gate, column_implication_gate, default_valued_columns_gate, enum_domain_gate, @@ -204,6 +205,7 @@ def _assert_frame_compatible(version: str, required: tuple[int, int]) -> None: "PreparedMonetaryMeasure", "add_ledger_artifact_args", "aggregate_admin_gate", + "area_support_gate", "column_implication_gate", "apply_ledger_target_profile", "bind_monetary_target", diff --git a/packages/microcosm-build/src/microcosm/build/country_spec.py b/packages/microcosm-build/src/microcosm/build/country_spec.py index 44448fa9f..61431d830 100644 --- a/packages/microcosm-build/src/microcosm/build/country_spec.py +++ b/packages/microcosm-build/src/microcosm/build/country_spec.py @@ -111,6 +111,7 @@ ALLOWED_GATE_FUNCTIONS = frozenset( { "aggregate_admin", + "area_support", "calibration_reference_coverage", "column_implication", "degenerate_release_surface", diff --git a/packages/microcosm-build/src/microcosm/build/gates.py b/packages/microcosm-build/src/microcosm/build/gates.py index eeafe83e2..a74a95b00 100644 --- a/packages/microcosm-build/src/microcosm/build/gates.py +++ b/packages/microcosm-build/src/microcosm/build/gates.py @@ -57,6 +57,7 @@ "GateResult", "GateReport", "FitWeightRecord", + "area_support_gate", "default_valued_columns_gate", "enum_domain_gate", "export_surface_gate", @@ -2420,6 +2421,160 @@ def support_gate( ) +def area_support_gate( + area_support: pd.DataFrame, + *, + area_roster: Mapping[str, Iterable[str]], + geography_levels: Iterable[str], + minimum_rows: int, + minimum_effective_sample_size: float, + minimum_distinct_sources: int, +) -> GateResult: + """Require adequate positive-weight support in every declared local area. + + ``area_support`` is the long-form candidate receipt. The independently + declared ``area_roster`` is load-bearing: a missing weak area cannot make + the minima look better by disappearing from the evidence table. + """ + + levels = tuple(str(level) for level in geography_levels) + if not levels or len(set(levels)) != len(levels): + raise ValueError("geography_levels must be non-empty and unique.") + if ( + isinstance(minimum_rows, bool) + or not isinstance(minimum_rows, int) + or minimum_rows < 1 + ): + raise ValueError("minimum_rows must be a positive integer.") + if ( + isinstance(minimum_distinct_sources, bool) + or not isinstance(minimum_distinct_sources, int) + or minimum_distinct_sources < 1 + ): + raise ValueError("minimum_distinct_sources must be a positive integer.") + if ( + not math.isfinite(float(minimum_effective_sample_size)) + or minimum_effective_sample_size <= 0 + ): + raise ValueError("minimum_effective_sample_size must be positive and finite.") + if not isinstance(area_support, pd.DataFrame): + raise TypeError("area_support must be a pandas DataFrame.") + required_columns = { + "geography_level", + "area_code", + "nonzero_households", + "nonzero_source_households", + "effective_sample_size", + } + missing_columns = sorted(required_columns - set(area_support.columns)) + if missing_columns: + raise ValueError( + f"area_support is missing required column(s): {missing_columns}." + ) + unexpected_levels = sorted( + set(area_support["geography_level"].astype(str)) - set(levels) + ) + if unexpected_levels: + raise ValueError( + "area_support contains geography level(s) outside the declared " + f"scope: {unexpected_levels}." + ) + + rows = area_support.copy() + rows["geography_level"] = rows["geography_level"].astype(str) + rows["area_code"] = rows["area_code"].astype(str) + duplicate_mask = rows.duplicated(["geography_level", "area_code"], keep=False) + if duplicate_mask.any(): + duplicates = sorted( + { + f"{row.geography_level}/{row.area_code}" + for row in rows.loc[ + duplicate_mask, ["geography_level", "area_code"] + ].itertuples(index=False) + } + ) + raise ValueError(f"area_support contains duplicate area rows: {duplicates}.") + + expected_pairs: set[tuple[str, str]] = set() + for level in levels: + if level not in area_roster: + raise ValueError(f"area_roster is missing geography level {level!r}.") + codes = tuple(str(code) for code in area_roster[level]) + if not codes or len(set(codes)) != len(codes): + raise ValueError(f"area_roster[{level!r}] must be non-empty and unique.") + expected_pairs.update((level, code) for code in codes) + actual_pairs = set(zip(rows["geography_level"], rows["area_code"], strict=True)) + missing_areas = sorted(expected_pairs - actual_pairs) + extra_areas = sorted(actual_pairs - expected_pairs) + if missing_areas or extra_areas: + raise ValueError( + "area_support must exactly cover the declared roster; " + f"missing={missing_areas[:10]}, extra={extra_areas[:10]}." + ) + + numeric_columns = ( + "nonzero_households", + "nonzero_source_households", + "effective_sample_size", + ) + values = rows.loc[:, list(numeric_columns)].to_numpy(dtype=np.float64) + if not np.isfinite(values).all() or (values < 0).any(): + raise ValueError("area_support contains invalid support values.") + + failures: list[str] = [] + for row in rows.sort_values(["geography_level", "area_code"]).itertuples( + index=False + ): + shortfalls = [] + if row.nonzero_households < minimum_rows: + shortfalls.append(f"rows {int(row.nonzero_households)} < {minimum_rows}") + if row.effective_sample_size < minimum_effective_sample_size: + shortfalls.append( + "ESS " + f"{float(row.effective_sample_size):.6g} < " + f"{float(minimum_effective_sample_size):.6g}" + ) + if row.nonzero_source_households < minimum_distinct_sources: + shortfalls.append( + "distinct sources " + f"{int(row.nonzero_source_households)} < " + f"{minimum_distinct_sources}" + ) + if shortfalls: + failures.append( + f"{row.geography_level}/{row.area_code}: " + ", ".join(shortfalls) + ) + + by_level: dict[str, dict[str, object]] = {} + for level in levels: + level_rows = rows.loc[rows["geography_level"] == level] + by_level[level] = { + "areas_checked": int(len(level_rows)), + "minimum_rows": int(level_rows["nonzero_households"].min()), + "minimum_effective_sample_size": float( + level_rows["effective_sample_size"].min() + ), + "minimum_distinct_sources": int( + level_rows["nonzero_source_households"].min() + ), + } + return GateResult( + name="area_support", + passed=not failures, + failures=tuple(failures), + details={ + "floors": { + "minimum_rows": minimum_rows, + "minimum_effective_sample_size": float(minimum_effective_sample_size), + "minimum_distinct_sources": minimum_distinct_sources, + }, + "by_geography_level": by_level, + "areas_checked": len(expected_pairs), + "areas_failed": len(failures), + }, + ) + + def aggregate_admin_gate( aggregates: Mapping[str, float], anchors: Iterable[TargetSpec], diff --git a/packages/microcosm-build/src/microcosm/build/uk/gates.json b/packages/microcosm-build/src/microcosm/build/uk/gates.json index 42c9e42b2..d290835b5 100644 --- a/packages/microcosm-build/src/microcosm/build/uk/gates.json +++ b/packages/microcosm-build/src/microcosm/build/uk/gates.json @@ -712,6 +712,75 @@ "min_nonzero_records": 104 }, "notes": "Weighted tail-concentration audit of the QRF-imputed output columns derived from the HMRC source manifest. Armed from the #686 L3 baselines per #757 B4, each threshold recording the run that measured it: the 47-column qrf_tail grids of uk_weighted_integrity_baselines_686.json (sha256 8b3f6c4e9c522445bf3e05d3451ee8557f4e14d25524ceac3af8bfe05401a146, licensed acceptance dir 686-spine-swap), measured on spine-a.h5 (sha256 a65f2132736d1dcecb91709a513a0d54a5887635ea03760629cc888d188ee306, 113,649 person rows, design weights) for the #609/#578 threshold adjudication. top_k 100 is the measurement grid anchor [10, 100, 500, 1000]; max_top_share is the exact measured maximum over the checked surface (hmrc_spi_other_social_security_income, 104 carriers), no headroom; min_nonzero_records is the thinnest measured column above the grid anchor (the same column) - the policy domain requires min_nonzero_records > top_k, so the three saturated sub-anchor columns (hmrc_spi_taxable_termination_pay 12, charitable_investment_gifts 22, sda_reported 24 carriers, each top-100 share 1.0) sit below it and go thin visibly in the signed details on every run, never a silent pass. The baselines are design-weight measurements and the terminal gate runs on the calibrated release frame; a calibrated-weight breach names its column and is a finding, not noise. The 12 household_qrf_tail grids (was_wealth surface) are measured in the same baselines file and not yet armed - a household-surface gate entry is a declared follow-up. Reviewed exclusions live in the named register resource of this package." + }, + { + "id": "uk_local_geography_ladder_post_calibration", + "gate": "spine_agreement", + "phase": "terminal", + "criticality": "release_blocking", + "parameters": {}, + "notes": "The existing UK geography-ladder evaluator is rerun on the locally calibrated candidate weights and adapted through UKGateBinding; no second ladder verdict implementation is introduced." + }, + { + "id": "uk_local_area_support", + "gate": "area_support", + "phase": "terminal", + "criticality": "release_blocking", + "parameters": { + "crosswalk_resource": "local_area_crosswalk.json", + "geography_levels": [ + "constituency", + "local_authority" + ], + "minimum_rows": 50, + "minimum_effective_sample_size": 50.0, + "minimum_distinct_sources": 50 + }, + "notes": "Both local grains are blocking from birth. Each declared area needs at least 50 positive-weight rows, Kish ESS 50, and 50 distinct positive-weight source households. The constituency K=4 receipt measured minima of 101 rows, ESS 86.4, and 97 sources; a local-authority miss reopens K or requires a separately reviewed exclusion." + }, + { + "id": "uk_local_target_fit", + "gate": "target_fit", + "phase": "terminal", + "criticality": "diagnostic", + "parameters": { + "surface": "local_candidate", + "max_abs_relative_error": 0.25 + }, + "notes": "Report every bound local target outside 25% absolute relative error. Diagnostic until the first #762 measured-fit receipt arms it." + }, + { + "id": "uk_local_per_family_fit", + "gate": "per_family_fit", + "phase": "terminal", + "criticality": "diagnostic", + "parameters": { + "within": 0.1, + "min_family_share": 0.5, + "hard_within": 0.25, + "min_hard_family_share": 0.5 + }, + "notes": "Report 10% fit shares per bound family and the 25% broad-fit share, so a weak family cannot hide in the aggregate. Diagnostic until the first #762 measured-fit receipt arms it." + }, + { + "id": "uk_local_weight_ratio", + "gate": "weight_ratio", + "phase": "terminal", + "criticality": "diagnostic", + "parameters": { + "maximum_max_to_median_ratio": 100.0 + }, + "notes": "The reviewed local doctrine ratio is 100.0. It ships diagnostic until #762 records the first measured candidate and arms the threshold." + }, + { + "id": "uk_local_weight_ess", + "gate": "weight_ess", + "phase": "terminal", + "criticality": "diagnostic", + "parameters": { + "minimum_ess_fraction": 0.01 + }, + "notes": "The local ESS-fraction diagnostic uses the national analogue's 0.01 floor and remains diagnostic until #762's measured arming review." } ] } 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 fc6b4e13c..1255ee5b7 100644 --- a/packages/microcosm-build/src/microcosm/build/uk_runtime/__init__.py +++ b/packages/microcosm-build/src/microcosm/build/uk_runtime/__init__.py @@ -29,6 +29,8 @@ UK_DIAGNOSTICS_SCHEMA_VERSION, UK_TARGET_GEOGRAPHY_LEVELS, uk_calibration_diagnostics_payload, + uk_weakest_areas_by_fit, + uk_weakest_families, uk_weight_summary, uk_zero_weight_strata, write_uk_calibration_diagnostics, @@ -273,11 +275,14 @@ ) from microcosm.build.uk_runtime.local_rowwise import ( UK_LOCAL_BINDING_ADJUDICATION_REGISTER_RESOURCE, + UK_LOCAL_HOLDOUT_FOLDS, + UK_LOCAL_HOLDOUT_SEED, UKRowwiseDoctrineSolve, UKRowwiseLocalMatrix, build_uk_rowwise_local_matrix, past_cap_census, require_adjudicated_uk_local_binding, + rotated_uk_local_holdout, rowwise_area_support_summary, rowwise_calibration_mass_reason, solve_uk_rowwise_weights_under_doctrine, @@ -509,6 +514,8 @@ "UK_FRAME_METADATA_KEY", "UK_LOCAL_MAX_WEIGHT_RATIO", "UK_LOCAL_BINDING_ADJUDICATION_REGISTER_RESOURCE", + "UK_LOCAL_HOLDOUT_FOLDS", + "UK_LOCAL_HOLDOUT_SEED", "UK_LOCAL_SOLVE_DOCTRINE", "UK_LOCAL_TARGET_LOSS_CAP", "UK_CGT_GAINS_AMOUNT_COLUMN", @@ -521,6 +528,7 @@ "materialize_uk_cgt_calibration_frame", "require_adjudicated_uk_local_binding", "rowwise_calibration_mass_reason", + "rotated_uk_local_holdout", "uk_cgt_annual_exempt_amount", "UK_CGT_REQUIRED_COLUMNS", "UK_CGT_TARGET_COVERAGE_REQUIREMENTS", @@ -838,6 +846,8 @@ "uk_release_input_coverage_reviewed_exclusions", "uk_stage_metadata", "uk_weight_summary", + "uk_weakest_areas_by_fit", + "uk_weakest_families", "uk_zero_weight_strata", "update_england_wales_lad_codes", "validate_uk_firm_population", diff --git a/packages/microcosm-build/src/microcosm/build/uk_runtime/battery_bindings.py b/packages/microcosm-build/src/microcosm/build/uk_runtime/battery_bindings.py index a11f01ab7..8651723e7 100644 --- a/packages/microcosm-build/src/microcosm/build/uk_runtime/battery_bindings.py +++ b/packages/microcosm-build/src/microcosm/build/uk_runtime/battery_bindings.py @@ -48,15 +48,18 @@ from microcosm.build.gates import ( GateResult, aggregate_admin_gate, + area_support_gate, column_implication_gate, enum_domain_gate, ledger_compile_parity_gate, nonnegative_columns_gate, + per_family_fit_gate, support_gate, target_surface_gate, weights_audit_gate, ) from microcosm.build.uk_runtime.frs_take_up import uk_take_up_signal_gate +from microcosm.build.uk_runtime.geography_ladder import uk_geography_ladder_gate from microcosm.build.uk_runtime.ledger_targets import ( LOCAL_REGISTRY_PARITY_FIXTURE_RESOURCE, align_uk_local_registry_parity_fixture, @@ -772,6 +775,90 @@ def _evaluate_weight_ratio( return uk_weight_ratio_gate(weights, **dict(parameters)) +def _evaluate_geography_ladder( + context: EvidenceContext, parameters: Mapping[str, Any] +) -> GateResult: + household = _uk_gate_surface(context.frame).household + weights = _household_weights(household) + return uk_geography_ladder_gate(household, weights, **dict(parameters)) + + +def _local_area_roster( + resource: str, levels: tuple[str, ...] +) -> dict[str, tuple[str, ...]]: + payload = json.loads(files("microcosm.build.uk").joinpath(resource).read_text()) + declared = payload.get("levels") + if not isinstance(declared, Mapping): + raise ValueError(f"{resource} must expose a levels object.") + roster: dict[str, tuple[str, ...]] = {} + for level in levels: + row = declared.get(level) + if not isinstance(row, Mapping): + raise ValueError(f"{resource} must expose level {level!r}.") + area_ids = row.get("area_ids") + if not isinstance(area_ids, (list, tuple)): + raise ValueError(f"{resource} level {level!r} must expose area_ids.") + roster[level] = tuple(str(area_id) for area_id in area_ids) + return roster + + +def _evaluate_area_support( + context: EvidenceContext, parameters: Mapping[str, Any] +) -> GateResult: + kwargs = dict(parameters) + resource = str(kwargs.pop("crosswalk_resource")) + levels = tuple(str(level) for level in kwargs["geography_levels"]) + return area_support_gate( + context.artifacts["uk_area_support_summary"], + area_roster=_local_area_roster(resource, levels), + **kwargs, + ) + + +def _local_target_error_items(evidence: object) -> tuple[tuple[str, float], ...]: + if hasattr(evidence, "to_dict"): + rows = evidence.to_dict(orient="records") + elif isinstance(evidence, (list, tuple)): + rows = evidence + else: + raise TypeError( + "local_target_diagnostics must be a DataFrame or a sequence of rows." + ) + items: list[tuple[str, float]] = [] + for index, row in enumerate(rows): + if not isinstance(row, Mapping): + raise TypeError(f"local target diagnostic row {index} must be a mapping.") + family = str(row.get("family", "")).strip() + area_code = str(row.get("area_code", "")).strip() + metric = str(row.get("metric", "")).strip() + if not family or not area_code or not metric: + raise ValueError( + f"local target diagnostic row {index} must name family, " + "area_code, and metric." + ) + error = float(row["relative_error"]) + if not np.isfinite(error): + raise ValueError( + f"local target diagnostic {family}/{area_code}/{metric} has " + "a non-finite relative_error." + ) + items.append((f"{family}/{area_code}/{metric}", error)) + if not items: + raise ValueError("local_target_diagnostics must not be empty.") + return tuple(items) + + +def _evaluate_per_family_fit( + context: EvidenceContext, parameters: Mapping[str, Any] +) -> GateResult: + items = _local_target_error_items(context.artifacts["local_target_diagnostics"]) + return per_family_fit_gate( + (name for name, _ in items), + (error for _, error in items), + **dict(parameters), + ) + + def _evaluate_weights_audit( context: EvidenceContext, parameters: Mapping[str, Any] ) -> GateResult: @@ -1000,8 +1087,22 @@ def _local_metric_by_target_id() -> dict[str, str]: def _evaluate_target_fit( context: EvidenceContext, parameters: Mapping[str, Any] ) -> GateResult: + kwargs = dict(parameters) + if kwargs.pop("surface", None) == "local_candidate": + errors = dict( + _local_target_error_items(context.artifacts["local_target_diagnostics"]) + ) + return uk_target_fit_gate(errors, **kwargs) parity = context.artifacts["parity_evidence"] - return uk_target_fit_gate(parity.target_relative_errors, **dict(parameters)) + return uk_target_fit_gate(parity.target_relative_errors, **kwargs) + + +def _target_fit_required_artifacts( + parameters: Mapping[str, Any], +) -> frozenset[str]: + if parameters.get("surface") == "local_candidate": + return frozenset({"local_target_diagnostics"}) + return frozenset({"parity_evidence"}) def _evaluate_input_mass_parity( @@ -1347,6 +1448,41 @@ def _ledger_compile_parity_required_artifacts( evaluator=_evaluate_weight_ratio, parameter_keys=frozenset({"maximum_max_to_median_ratio"}), ), + "spine_agreement": UKGateBinding( + name="spine_agreement", + evaluator=_evaluate_geography_ladder, + legacy_name="uk_geography_ladder", + ), + "area_support": UKGateBinding( + name="area_support", + evaluator=_evaluate_area_support, + parameter_keys=frozenset( + { + "crosswalk_resource", + "geography_levels", + "minimum_rows", + "minimum_effective_sample_size", + "minimum_distinct_sources", + } + ), + artifact_keys=frozenset({"uk_area_support_summary"}), + needs_frame=False, + ), + "per_family_fit": UKGateBinding( + name="per_family_fit", + evaluator=_evaluate_per_family_fit, + parameter_keys=frozenset( + { + "within", + "min_family_share", + "hard_within", + "min_hard_family_share", + "min_family_size", + } + ), + artifact_keys=frozenset({"local_target_diagnostics"}), + needs_frame=False, + ), "export_surface": UKGateBinding( name="export_surface", evaluator=_evaluate_export_surface, @@ -1373,8 +1509,10 @@ def _ledger_compile_parity_required_artifacts( "target_fit": UKGateBinding( name="target_fit", evaluator=_evaluate_target_fit, - parameter_keys=frozenset({"max_abs_relative_error", "reviewed_exclusions"}), - artifact_keys=frozenset({"parity_evidence"}), + parameter_keys=frozenset( + {"max_abs_relative_error", "reviewed_exclusions", "surface"} + ), + artifact_selector=_target_fit_required_artifacts, needs_frame=False, ), "input_mass_parity": UKGateBinding( diff --git a/packages/microcosm-build/src/microcosm/build/uk_runtime/calibration_run.py b/packages/microcosm-build/src/microcosm/build/uk_runtime/calibration_run.py index 654006311..6864a9eff 100644 --- a/packages/microcosm-build/src/microcosm/build/uk_runtime/calibration_run.py +++ b/packages/microcosm-build/src/microcosm/build/uk_runtime/calibration_run.py @@ -96,6 +96,15 @@ class UKCalibrationRunResult: "uk_calibration_reference_coverage", ) +UK_LOCAL_GATE_SCOPE = ( + "uk_local_geography_ladder_post_calibration", + "uk_local_area_support", + "uk_local_target_fit", + "uk_local_per_family_fit", + "uk_local_weight_ratio", + "uk_local_weight_ess", +) + UK_SPINE_GATE_SCOPE = ( "uk_stage_was_wealth_support", "uk_stage_lcfs_consumption_support", @@ -159,6 +168,7 @@ def _scope_exclusions() -> dict[str, str]: spine = set(UK_SPINE_GATE_SCOPE) national = set(UK_NATIONAL_GATE_SCOPE) calibration = set(UK_CALIBRATION_GATE_SCOPE) + local = set(UK_LOCAL_GATE_SCOPE) # Closed-world means both halves: every gate owned by someone (below), and # no gate owned twice without saying so. Coverage alone would let an # accidental overlap through, and the certification union is exactly where @@ -166,7 +176,10 @@ def _scope_exclusions() -> dict[str, str]: for left_name, left, right_name, right in ( ("calibration", calibration, "spine", spine), ("calibration", calibration, "national", national), + ("calibration", calibration, "local", local), ("spine", spine, "national", national), + ("spine", spine, "local", local), + ("national", national, "local", local), ): undeclared = (left & right) - UK_SHARED_GATE_IDS if undeclared: @@ -175,7 +188,7 @@ def _scope_exclusions() -> dict[str, str]: f"{sorted(undeclared)} without declaring them in " "UK_SHARED_GATE_IDS." ) - classified = calibration | spine | national + classified = calibration | spine | national | local rationales: dict[str, str] = {} for gate_id in sorted(full - set(UK_CALIBRATION_GATE_SCOPE)): if gate_id in spine: @@ -187,6 +200,12 @@ def _scope_exclusions() -> dict[str, str]: "owned by the release-cut certification producer; runner lands " "with the certification, June runner retired" ) + elif gate_id in local: + reason = ( + "local-candidate gate; owned by the rowwise candidate's " + "scoped battery and excluded from national certification " + "until microcosm#146." + ) elif "parity" in gate_id or gate_id in _SWAP_ACCEPTANCE_GATE_IDS: reason = "swap-acceptance evidence; produced by the swap lane, not the calibration seam." else: @@ -200,6 +219,27 @@ def _scope_exclusions() -> dict[str, str]: UK_CALIBRATION_GATE_SCOPE_EXCLUSIONS = _scope_exclusions() +def uk_local_gate_scope_exclusions() -> dict[str, str]: + """Classify every declared entry the local-candidate battery does not run.""" + + full = {entry.id for entry in load_country_spec("uk").gates.gates} + local = set(UK_LOCAL_GATE_SCOPE) + exclusions: dict[str, str] = {} + for gate_id in sorted(full - local): + if gate_id in UK_SPINE_GATE_SCOPE: + reason = "spine-construction gate; owned by the spine build." + elif gate_id in UK_CALIBRATION_GATE_SCOPE: + reason = "national calibration-seam gate; outside the local candidate." + elif gate_id in UK_NATIONAL_GATE_SCOPE: + reason = "national release-cut gate; outside the local candidate." + else: # pragma: no cover - _scope_exclusions enforces closed-world ownership + raise RuntimeError(f"UK gate {gate_id!r} belongs to no declared scope.") + exclusions[gate_id] = reason + if local | set(exclusions) != full: + raise RuntimeError("UK local gate scope does not classify every gate id.") + return exclusions + + def run_uk_calibration( *, paths: UKCalibrationRunPaths, diff --git a/packages/microcosm-build/src/microcosm/build/uk_runtime/diagnostics.py b/packages/microcosm-build/src/microcosm/build/uk_runtime/diagnostics.py index 70b5dbcdd..7907b4d71 100644 --- a/packages/microcosm-build/src/microcosm/build/uk_runtime/diagnostics.py +++ b/packages/microcosm-build/src/microcosm/build/uk_runtime/diagnostics.py @@ -39,6 +39,8 @@ "UK_TARGET_GEOGRAPHY_LEVELS", "uk_calibration_diagnostics_payload", "uk_target_geography_levels", + "uk_weakest_areas_by_fit", + "uk_weakest_families", "uk_weight_summary", "uk_zero_weight_strata", "write_uk_calibration_diagnostics", @@ -64,6 +66,218 @@ ) _TARGET_PASS_RELATIVE_ERROR = 0.10 +_AREA_FIT_LIMIT = 15 + +_COUNTRY_BY_AREA_PREFIX = { + "E": "England", + "N": "Northern Ireland", + "S": "Scotland", + "W": "Wales", +} + + +def _finite_target_error(row: Mapping[str, object]) -> tuple[str, float]: + name = str(row.get("name") or "") + raw_error = row.get("relative_error") + if ( + not name + or not isinstance(raw_error, (int, float)) + or isinstance(raw_error, bool) + ): + raise ValueError("UK target rollups require named finite relative errors.") + error = float(raw_error) + if not math.isfinite(error): + raise ValueError("UK target rollups require named finite relative errors.") + return name, error + + +def _finite_loss_contribution(row: Mapping[str, object]) -> float: + raw = row.get("final_loss_contribution") + if not isinstance(raw, (int, float)) or isinstance(raw, bool): + raise ValueError( + "UK target rollups require schema-v6 final_loss_contribution values." + ) + value = float(raw) + if not math.isfinite(value) or value < 0.0: + raise ValueError( + "UK target rollups require finite non-negative loss contributions." + ) + return value + + +def uk_weakest_families( + target_rows: Sequence[Mapping[str, object]], +) -> list[dict[str, object]]: + """Rank every scored family by its schema-v6 loss contribution.""" + + buckets: dict[str, list[tuple[str, float, float]]] = {} + for row in target_rows: + registry = row.get("registry") + if not isinstance(registry, Mapping) or not str(registry.get("family") or ""): + raise ValueError( + "UK target rollups require a registry family on every row." + ) + name, error = _finite_target_error(row) + buckets.setdefault(str(registry["family"]), []).append( + (name, abs(error), _finite_loss_contribution(row)) + ) + total_loss = math.fsum( + contribution for rows in buckets.values() for _, _, contribution in rows + ) + result: list[dict[str, object]] = [] + for family, rows in buckets.items(): + worst_name, worst_error, _ = max(rows, key=lambda item: (item[1], item[0])) + contribution = math.fsum(item[2] for item in rows) + within = sum(item[1] <= _TARGET_PASS_RELATIVE_ERROR for item in rows) + result.append( + { + "family": family, + "n_targets": len(rows), + "n_within_10pct": within, + "pass_rate": within / len(rows), + "worst_target": worst_name, + "worst_abs_relative_error": worst_error, + "loss_contribution": contribution, + "loss_share": contribution / total_loss if total_loss else 0.0, + } + ) + result.sort( + key=lambda row: ( + -float(row["loss_contribution"]), + str(row["family"]), + ) + ) + return result + + +def uk_weakest_areas_by_fit( + target_rows: Sequence[Mapping[str, object]], + area_support: pd.DataFrame, + *, + limit: int = _AREA_FIT_LIMIT, +) -> dict[str, object]: + """Return bottom-by-fit local areas plus country-level fit rollups.""" + + if isinstance(limit, bool) or not isinstance(limit, int) or limit <= 0: + raise ValueError("UK weakest-area limit must be a positive integer.") + required = { + "geography_level", + "area_code", + "nonzero_households", + "nonzero_source_households", + "effective_sample_size", + } + missing = sorted(required - set(area_support.columns)) + if missing: + raise ValueError(f"UK area support is missing rollup column(s): {missing}.") + support_rows = { + (str(row.geography_level), str(row.area_code)): row + for row in area_support.itertuples(index=False) + } + if len(support_rows) != len(area_support): + raise ValueError("UK area support contains duplicate geography/area rows.") + + grouped: dict[tuple[str, str], list[tuple[str, float, float]]] = {} + for row in target_rows: + metadata = row.get("metadata") + if not isinstance(metadata, Mapping): + raise ValueError("UK area rollups require target metadata.") + level = _normalize_geography_level(metadata.get("area_type")) + area_code = str(metadata.get("area_code") or "") + if level not in {"constituency", "local_authority"} or not area_code: + raise ValueError("UK area rollups require local target area metadata.") + name, error = _finite_target_error(row) + grouped.setdefault((level, area_code), []).append( + (name, abs(error), _finite_loss_contribution(row)) + ) + + area_rows: list[dict[str, object]] = [] + for (level, area_code), rows in grouped.items(): + support = support_rows.get((level, area_code)) + if support is None: + raise ValueError( + "UK area rollups require exact support for every scored area; " + f"missing {(level, area_code)!r}." + ) + worst_name, worst_error, _ = max(rows, key=lambda item: (item[1], item[0])) + within = sum(item[1] <= _TARGET_PASS_RELATIVE_ERROR for item in rows) + area_rows.append( + { + "geography_level": level, + "area_code": area_code, + "country": _country_for_area(area_code), + "n_targets": len(rows), + "n_within_10pct": within, + "pass_rate": within / len(rows), + "worst_target": worst_name, + "worst_abs_relative_error": worst_error, + "loss_contribution": math.fsum(item[2] for item in rows), + "nonzero_households": int(support.nonzero_households), + "nonzero_source_households": int(support.nonzero_source_households), + "effective_sample_size": float(support.effective_sample_size), + } + ) + area_rows.sort( + key=lambda row: ( + -float(row["worst_abs_relative_error"]), + str(row["geography_level"]), + str(row["area_code"]), + ) + ) + + countries: list[dict[str, object]] = [] + for country in _COUNTRY_BY_AREA_PREFIX.values(): + for level in ("constituency", "local_authority"): + members = [ + row + for row in area_rows + if row["country"] == country and row["geography_level"] == level + ] + if not members: + continue + n_targets = sum(int(row["n_targets"]) for row in members) + n_within = sum(int(row["n_within_10pct"]) for row in members) + worst = max( + members, + key=lambda row: ( + float(row["worst_abs_relative_error"]), + str(row["worst_target"]), + ), + ) + countries.append( + { + "country": country, + "geography_level": level, + "n_areas": len(members), + "n_targets": n_targets, + "n_within_10pct": n_within, + "pass_rate": n_within / n_targets, + "worst_target": worst["worst_target"], + "worst_abs_relative_error": worst["worst_abs_relative_error"], + "loss_contribution": math.fsum( + float(row["loss_contribution"]) for row in members + ), + } + ) + countries.sort(key=lambda row: (str(row["country"]), str(row["geography_level"]))) + # Keyed by role, not by a count: the limit is a parameter, so a fixed + # "bottom_15" name would disagree with the list whenever a caller passes + # anything else, or whenever fewer areas were scored than the limit. + return { + "limit": limit, + "n_areas_scored": len(area_rows), + "bottom_by_fit": area_rows[:limit], + "countries": countries, + } + + +def _country_for_area(area_code: str) -> str: + try: + return _COUNTRY_BY_AREA_PREFIX[area_code[0]] + except (IndexError, KeyError) as error: + raise ValueError( + f"Cannot derive UK country from area code {area_code!r}." + ) from error def _as_weights(values: Sequence[float] | np.ndarray) -> np.ndarray: @@ -438,6 +652,8 @@ def uk_calibration_diagnostics_payload( target_registry: TargetRegistry, stratum_columns: Sequence[str] = _UK_DEFAULT_ZERO_WEIGHT_STRATUM_COLUMNS, build: dict[str, Any] | None = None, + local_area_support: pd.DataFrame | None = None, + rotated_holdout: Mapping[str, object] | None = None, ) -> dict[str, object]: """Render shared diagnostics plus the versioned UK release evidence. @@ -480,12 +696,21 @@ def uk_calibration_diagnostics_payload( stratum_columns=stratum_columns, ) - payload["uk_diagnostics"] = { + uk_diagnostics: dict[str, object] = { "schema_version": UK_DIAGNOSTICS_SCHEMA_VERSION, "weights": weights, "zero_weight_rows_by_stratum": strata, "target_pass_rates_by_geography_level": pass_rates, } + if local_area_support is not None: + uk_diagnostics["weakest_families"] = uk_weakest_families(target_rows) + uk_diagnostics["weakest_areas_by_fit"] = uk_weakest_areas_by_fit( + target_rows, + local_area_support, + ) + if rotated_holdout is not None: + uk_diagnostics["rotated_holdout"] = dict(rotated_holdout) + payload["uk_diagnostics"] = uk_diagnostics return payload @@ -498,6 +723,8 @@ def write_uk_calibration_diagnostics( target_registry: TargetRegistry, stratum_columns: Sequence[str] = _UK_DEFAULT_ZERO_WEIGHT_STRATUM_COLUMNS, build: dict[str, Any] | None = None, + local_area_support: pd.DataFrame | None = None, + rotated_holdout: Mapping[str, object] | None = None, ) -> Path: """Atomically write strict shared-plus-UK diagnostics.""" @@ -510,6 +737,8 @@ def write_uk_calibration_diagnostics( target_registry=target_registry, stratum_columns=stratum_columns, build=build, + local_area_support=local_area_support, + rotated_holdout=rotated_holdout, ), indent=1, allow_nan=False, diff --git a/packages/microcosm-build/src/microcosm/build/uk_runtime/local_rowwise.py b/packages/microcosm-build/src/microcosm/build/uk_runtime/local_rowwise.py index fa0b5a52e..4db180858 100644 --- a/packages/microcosm-build/src/microcosm/build/uk_runtime/local_rowwise.py +++ b/packages/microcosm-build/src/microcosm/build/uk_runtime/local_rowwise.py @@ -30,9 +30,11 @@ import pandas as pd from scipy import sparse as sp +from microcosm.build.holdout import rotated_folds, summarize_rotations from microcosm.build.uk_runtime import local_target_census from microcosm.build.uk_runtime.local_doctrine import ( UK_LOCAL_SOLVE_DOCTRINE, + UK_LOCAL_TARGET_LOSS_CAP, ) from microcosm.build.uk_runtime.local_targets import AREA_TYPES from microcosm.build.uk_runtime.national_frame import ( @@ -48,6 +50,7 @@ from microcosm.calibrate.solve import ( CONSERVE_MASS, FREE_MASS, + CalibrationResult, calibrate, default_target_loss_scales, relative_error_loss, @@ -65,10 +68,14 @@ "uk_area_support_summary", "uk_ladder_area_support_summary", "rowwise_calibration_mass_reason", + "rotated_uk_local_holdout", "rowwise_area_support_summary", "solve_uk_rowwise_weights_under_doctrine", ] +UK_LOCAL_HOLDOUT_FOLDS = 5 +UK_LOCAL_HOLDOUT_SEED = 20260529 + UK_LOCAL_BINDING_ADJUDICATION_REGISTER_RESOURCE = "local_binding_adjudications.json" @@ -113,6 +120,7 @@ class UKRowwiseDoctrineSolve: """ frame: Frame + calibration_result: CalibrationResult weights: np.ndarray initial_weights: np.ndarray diagnostics: pd.DataFrame @@ -430,9 +438,7 @@ def require_adjudicated_uk_local_binding( """Require in-force review records before binding fenced UK local families.""" census_payload = ( - local_target_census.load_uk_local_target_census() - if census is None - else census + local_target_census.load_uk_local_target_census() if census is None else census ) family_rows = _uk_local_census_family_rows(census_payload) declared, parsed = _normalise_uk_local_bound_families( @@ -580,8 +586,7 @@ def _normalise_uk_local_bound_families( duplicates = sorted({name for name in declared if declared.count(name) > 1}) if duplicates: raise ValueError( - "UK local binding declarations: duplicate bound family(ies) " - f"{duplicates}." + f"UK local binding declarations: duplicate bound family(ies) {duplicates}." ) parsed: dict[str, tuple[str, str]] = {} @@ -922,6 +927,7 @@ def solve_uk_rowwise_weights_under_doctrine( ) return UKRowwiseDoctrineSolve( frame=finished, + calibration_result=result, weights=np.asarray(result.weights, dtype=np.float64), initial_weights=np.asarray(result.initial_weights, dtype=np.float64), diagnostics=diagnostics, @@ -934,6 +940,110 @@ def solve_uk_rowwise_weights_under_doctrine( ) +def rotated_uk_local_holdout( + frame: Frame, + problem: UKRowwiseLocalMatrix, + *, + epochs: int = 512, + learning_rate: float = 0.15, + conserve_mass: bool = False, + target_records: int | None = None, + l0_lambda: float = 0.0, + budget_iters: int = 10, + solve_seed: int = 0, +) -> dict[str, object]: + """Run the fixed five-fold target rotation through actual local solves.""" + + folds = rotated_folds( + len(problem.targets), + n_folds=UK_LOCAL_HOLDOUT_FOLDS, + seed=UK_LOCAL_HOLDOUT_SEED, + ) + all_indices = np.arange(len(problem.targets), dtype=np.int64) + fold_rows: list[dict[str, object]] = [] + for fold_index, holdout_indices in enumerate(folds): + train_indices = np.setdiff1d( + all_indices, + holdout_indices, + assume_unique=True, + ) + train_problem = _subset_rowwise_problem(problem, train_indices) + train_families = sorted( + { + f"{local_target_census.family_for_metric(str(row.metric))}/" + f"{row.area_type}" + for row in train_problem.target_frame.itertuples(index=False) + } + ) + train_solve = solve_uk_rowwise_weights_under_doctrine( + frame, + train_problem, + bound_families=train_families, + epochs=epochs, + learning_rate=learning_rate, + conserve_mass=conserve_mass, + target_records=target_records, + l0_lambda=l0_lambda, + budget_iters=budget_iters, + seed=solve_seed, + ) + held_targets = problem.targets[holdout_indices] + held_estimates = np.asarray( + problem.matrix[holdout_indices] @ train_solve.weights, + dtype=np.float64, + ).reshape(-1) + loss = relative_error_loss( + held_estimates, + held_targets, + target_loss_cap=UK_LOCAL_TARGET_LOSS_CAP, + ) + fold_rows.append( + { + "fold": fold_index, + "n_train_targets": int(len(train_indices)), + "n_holdout_targets": int(len(holdout_indices)), + "holdout_target_indices": holdout_indices.tolist(), + "holdout_loss": loss, + } + ) + summary = summarize_rotations(row["holdout_loss"] for row in fold_rows) + return { + "report_only": True, + "method": "rotated_folds", + # Declared so a consumer can check that a recorded holdout was + # measured under the same cap it is being reported beside, rather + # than assuming it across the module boundary. + "target_loss_cap": UK_LOCAL_TARGET_LOSS_CAP, + "n_folds": summary.n_folds, + "seed": UK_LOCAL_HOLDOUT_SEED, + "solve_seed": solve_seed, + "mean_holdout_loss": summary.mean_holdout_loss, + "worst_holdout_loss": summary.worst_holdout_loss, + "fold_losses": list(summary.fold_losses), + "folds": fold_rows, + } + + +def _subset_rowwise_problem( + problem: UKRowwiseLocalMatrix, + indices: np.ndarray, +) -> UKRowwiseLocalMatrix: + if indices.ndim != 1 or not len(indices): + raise ValueError("a rotated local training surface must be non-empty.") + target_frame = problem.target_frame.iloc[indices].reset_index(drop=True).copy() + target_frame["target_index"] = np.arange(len(target_frame), dtype=np.int64) + return UKRowwiseLocalMatrix( + matrix=problem.matrix[indices].tocsr(), + targets=np.asarray(problem.targets[indices], dtype=np.float64), + target_frame=target_frame, + area_codes=problem.area_codes, + metric_names=problem.metric_names, + household_ids=problem.household_ids, + assigned_areas=problem.assigned_areas, + metric_values=problem.metric_values, + ) + + def rowwise_area_support_summary( problem: UKRowwiseLocalMatrix, weights: Sequence[float], @@ -1074,7 +1184,9 @@ def uk_ladder_area_support_summary( "'household_id' explicitly for row-grain sources)." ) if weight_column not in household.columns: - raise ValueError(f"household table must contain weight column {weight_column!r}.") + raise ValueError( + f"household table must contain weight column {weight_column!r}." + ) summaries: dict[str, pd.DataFrame] = {} for area_type, assigned_column, ladder_codes in ( diff --git a/packages/microcosm-build/src/microcosm/build/uk_runtime/release_certification.py b/packages/microcosm-build/src/microcosm/build/uk_runtime/release_certification.py index cc6f12d89..278ecccea 100644 --- a/packages/microcosm-build/src/microcosm/build/uk_runtime/release_certification.py +++ b/packages/microcosm-build/src/microcosm/build/uk_runtime/release_certification.py @@ -47,6 +47,7 @@ from microcosm.build.uk_runtime.battery_bindings import UK_GATE_REGISTRY from microcosm.build.uk_runtime.calibration_run import ( UK_CALIBRATION_GATE_SCOPE, + UK_LOCAL_GATE_SCOPE, UK_NATIONAL_GATE_SCOPE, UK_SHARED_GATE_IDS, UK_SPINE_GATE_SCOPE, @@ -71,6 +72,7 @@ UK_RELEASE_CERTIFICATION_SCHEMA_VERSION = 1 UK_RELEASE_CERTIFICATION_KIND = "uk_release_certification" UK_RELEASE_CUT_POSTURE = "release_cut" +UK_CERTIFICATION_EXCLUDED_GATE_IDS = frozenset(UK_LOCAL_GATE_SCOPE) #: Each certification part's declared scope, phases, and manifest policy #: suffix. The suffixes are load-bearing: they are what the producers bake @@ -137,7 +139,12 @@ def uk_release_cut_scope_exclusions() -> dict[str, str]: exclusions[entry.id] = ( "calibration-seam gate; owned by the seam's scoped battery." ) - else: # pragma: no cover - the three-way partition is import-enforced + elif entry.id in UK_CERTIFICATION_EXCLUDED_GATE_IDS: + exclusions[entry.id] = ( + "local-candidate gate; excluded from national certification " + "until microcosm#146." + ) + else: # pragma: no cover - closed-world classification is import-enforced raise RuntimeError( f"UK gate {entry.id!r} belongs to no declared battery scope." ) @@ -388,6 +395,7 @@ def compose_uk_release_certification( {name: (payload, raw) for name, (payload, raw) in parts_raw.items()}, declared_ids=declared_ids, declared_phases=declared_phases, + certification_excluded_ids=UK_CERTIFICATION_EXCLUDED_GATE_IDS, ) _verify_identity_join( spine_report_bytes=parts_raw["spine"][1], @@ -425,6 +433,9 @@ def compose_uk_release_certification( "declared_entry_count": len(declared_ids), "declared_phases": list(declared_phases), "shared_gate_ids": sorted(UK_SHARED_GATE_IDS), + "certification_excluded_gate_ids": sorted( + UK_CERTIFICATION_EXCLUDED_GATE_IDS + ), }, "doctrine": { "payload": dict(run_config.get("doctrine", {})), @@ -585,6 +596,7 @@ def _verify_union( *, declared_ids: set[str], declared_phases: tuple[str, ...], + certification_excluded_ids: frozenset[str] = frozenset(), ) -> None: seen: dict[str, list[str]] = {} phases_covered: set[str] = set() @@ -593,7 +605,19 @@ def _verify_union( seen.setdefault(gate_id, []).append(part_name) phases_covered.update(str(phase) for phase in payload["phases"]) union = set(seen) - gap = sorted(declared_ids - union) + undeclared_exclusions = sorted(certification_excluded_ids - declared_ids) + if undeclared_exclusions: + raise UKReleaseCertificationError( + "certification exclusions name undeclared gate ids: " + f"{undeclared_exclusions}." + ) + evaluated_exclusions = sorted(certification_excluded_ids & union) + if evaluated_exclusions: + raise UKReleaseCertificationError( + "certification-excluded gate ids were evaluated by national parts: " + f"{evaluated_exclusions}." + ) + gap = sorted(declared_ids - union - set(certification_excluded_ids)) if gap: raise UKReleaseCertificationError( f"certification gap: declared gate ids evaluated by no part: {gap}." diff --git a/packages/microcosm-build/tests/test_country_spec.py b/packages/microcosm-build/tests/test_country_spec.py index d12578359..aac5fa036 100644 --- a/packages/microcosm-build/tests/test_country_spec.py +++ b/packages/microcosm-build/tests/test_country_spec.py @@ -1305,10 +1305,27 @@ def test_declares_the_full_june_battery(self, manifest) -> None: "uk_target_fit", "uk_input_mass_parity", "uk_qrf_tail_concentration", + "uk_local_geography_ladder_post_calibration", + "uk_local_area_support", + "uk_local_target_fit", + "uk_local_per_family_fit", + "uk_local_weight_ratio", + "uk_local_weight_ess", ] - # Legacy behaviour: every evaluated failure raises, so every - # declared entry blocks release. - assert all(g.criticality == "release_blocking" for g in manifest.gates) + diagnostic = { + "uk_local_target_fit", + "uk_local_per_family_fit", + "uk_local_weight_ratio", + "uk_local_weight_ess", + } + assert { + gate.id for gate in manifest.gates if gate.criticality == "diagnostic" + } == diagnostic + assert all( + gate.criticality == "release_blocking" + for gate in manifest.gates + if gate.id not in diagnostic + ) def test_ledger_compile_parity_gates_pin_their_fixture_periods( self, manifest diff --git a/packages/microcosm-build/tests/test_gate_battery_contract_pins.py b/packages/microcosm-build/tests/test_gate_battery_contract_pins.py index bc1d37e7f..311e8062f 100644 --- a/packages/microcosm-build/tests/test_gate_battery_contract_pins.py +++ b/packages/microcosm-build/tests/test_gate_battery_contract_pins.py @@ -88,16 +88,15 @@ def test_entry_membership_mirrors_the_committed_spec(self) -> None: } def test_entry_metadata_mirrors_the_committed_spec(self) -> None: - # The verifier pins each entry's gate and phase, refuses any - # non-release_blocking criticality, and refuses not_applicable - # outright; all three must restate the spec exactly. + # The verifier pins each entry's gate, phase, and criticality, and + # refuses not_applicable outright; all three restate the spec exactly. spec = load_country_spec("uk") assert data_contract._UK_GATE_BATTERY_ENTRY_GATES == { entry.id: (entry.gate, entry.phase) for entry in spec.gates.gates } - assert all( - entry.criticality == "release_blocking" for entry in spec.gates.gates - ) + assert data_contract._UK_GATE_BATTERY_DIAGNOSTIC_IDS == { + entry.id for entry in spec.gates.gates if entry.criticality == "diagnostic" + } assert all(entry.not_applicable is None for entry in spec.gates.gates) def test_outcome_envelope_mirrors_the_producer_construction_invariants( @@ -319,6 +318,7 @@ class TestCertificationMirrors: def test_part_scopes_mirror_the_ownership_partition(self) -> None: from microcosm.build.uk_runtime.calibration_run import ( UK_CALIBRATION_GATE_SCOPE, + UK_LOCAL_GATE_SCOPE, UK_NATIONAL_GATE_SCOPE, UK_SHARED_GATE_IDS, UK_SPINE_GATE_SCOPE, @@ -330,12 +330,15 @@ def test_part_scopes_mirror_the_ownership_partition(self) -> None: assert data_contract._UK_CERTIFICATION_PART_SCOPES[ "calibration_seam" ] == frozenset(UK_CALIBRATION_GATE_SCOPE) - assert data_contract._UK_CERTIFICATION_PART_SCOPES[ - "release_cut" - ] == frozenset(UK_NATIONAL_GATE_SCOPE) + assert data_contract._UK_CERTIFICATION_PART_SCOPES["release_cut"] == frozenset( + UK_NATIONAL_GATE_SCOPE + ) assert data_contract._UK_CERTIFICATION_SHARED_GATE_IDS == frozenset( UK_SHARED_GATE_IDS ) + assert data_contract._UK_CERTIFICATION_EXCLUDED_GATE_IDS == frozenset( + UK_LOCAL_GATE_SCOPE + ) def test_part_digests_mirror_the_live_scoped_manifests(self) -> None: from microcosm.build.uk_runtime.release_certification import ( @@ -350,13 +353,13 @@ def test_part_digests_mirror_the_live_scoped_manifests(self) -> None: policy_suffix=str(spec["policy_suffix"]), ) mirrored = data_contract._UK_CERTIFICATION_PART_DIGESTS[part_name] - assert mirrored["gates_manifest_sha256"] == ( - live["gates_manifest_sha256"] + assert ( + mirrored["gates_manifest_sha256"] == (live["gates_manifest_sha256"]) ), part_name assert mirrored["policy_sha256"] == live["policy_sha256"], part_name - assert list( - data_contract._UK_CERTIFICATION_PART_PHASES[part_name] - ) == list(spec["phases"]) + assert list(data_contract._UK_CERTIFICATION_PART_PHASES[part_name]) == list( + spec["phases"] + ) def test_certification_identity_mirrors(self) -> None: from microcosm.build.uk_runtime import release_certification diff --git a/packages/microcosm-build/tests/test_gates.py b/packages/microcosm-build/tests/test_gates.py index d7c781958..88f277ad8 100644 --- a/packages/microcosm-build/tests/test_gates.py +++ b/packages/microcosm-build/tests/test_gates.py @@ -17,6 +17,7 @@ TargetCoverageRequirement, TargetFitRequirement, aggregate_admin_gate, + area_support_gate, column_implication_gate, default_valued_columns_gate, enum_domain_gate, @@ -495,6 +496,72 @@ def test_missing_range_declaration_fails(self) -> None: assert "no donor range declared" in result.failures[0] +class TestAreaSupportGate: + @staticmethod + def _support(**overrides) -> pd.DataFrame: + rows = { + "geography_level": ["constituency", "local_authority"], + "area_code": ["C1", "L1"], + "nonzero_households": [60, 70], + "nonzero_source_households": [55, 65], + "effective_sample_size": [52.0, 63.0], + } + rows.update(overrides) + return pd.DataFrame(rows) + + def test_passes_both_declared_grains(self) -> None: + result = area_support_gate( + self._support(), + area_roster={"constituency": ["C1"], "local_authority": ["L1"]}, + geography_levels=("constituency", "local_authority"), + minimum_rows=50, + minimum_effective_sample_size=50.0, + minimum_distinct_sources=50, + ) + + assert result.passed + assert result.details["areas_checked"] == 2 + assert result.details["by_geography_level"]["constituency"] == { + "areas_checked": 1, + "minimum_rows": 60, + "minimum_effective_sample_size": 52.0, + "minimum_distinct_sources": 55, + } + + def test_names_every_floor_breach(self) -> None: + result = area_support_gate( + self._support( + nonzero_households=[49, 70], + nonzero_source_households=[48, 65], + effective_sample_size=[47.5, 63.0], + ), + area_roster={"constituency": ["C1"], "local_authority": ["L1"]}, + geography_levels=("constituency", "local_authority"), + minimum_rows=50, + minimum_effective_sample_size=50.0, + minimum_distinct_sources=50, + ) + + assert not result.passed + assert result.failures == ( + "constituency/C1: rows 49 < 50, ESS 47.5 < 50, distinct sources 48 < 50", + ) + + def test_roster_omission_refuses_instead_of_shrinking_the_surface(self) -> None: + with pytest.raises(ValueError, match="exactly cover"): + area_support_gate( + self._support().iloc[:1], + area_roster={ + "constituency": ["C1"], + "local_authority": ["L1"], + }, + geography_levels=("constituency", "local_authority"), + minimum_rows=50, + minimum_effective_sample_size=50.0, + minimum_distinct_sources=50, + ) + + class TestAggregateAdminGate: def _stcg_anchor(self) -> TargetSpec: return TargetSpec( diff --git a/packages/microcosm-build/tests/test_uk_battery_bindings.py b/packages/microcosm-build/tests/test_uk_battery_bindings.py index 875051e83..ce2997371 100644 --- a/packages/microcosm-build/tests/test_uk_battery_bindings.py +++ b/packages/microcosm-build/tests/test_uk_battery_bindings.py @@ -563,7 +563,12 @@ def test_fully_armed_battery_evaluates_gate_for_gate(self) -> None: # student-loan enum gate; their evaluators have direct tests. The BRMA # enum gate is no longer among them: it moved to the spine battery's # assembled boundary, where its column is first written. - assert len(passed) == 16 + # 15 before this lane, plus uk_uc_capital_coherence from #829, plus + # the two frame-only local weight diagnostics that also pass in this + # unscoped compatibility probe. The local ladder gate fails because + # this national fixture deliberately carries no ladder columns; the + # three evidence-backed local arms are named gaps below. + assert len(passed) == 18 qrf = by_id["uk_qrf_tail_concentration"] assert qrf.status is GateStatus.FAILED assert "declared QRF output is absent" in qrf.result.failures[0] @@ -620,6 +625,9 @@ def test_battery_records_evidence_absent(self, uk_gates) -> None: "uk_target_fit", "uk_input_mass_parity", "uk_aggregate_admin", + "uk_local_area_support", + "uk_local_target_fit", + "uk_local_per_family_fit", } for reason in absent.values(): assert reason.startswith("missing evidence: ") @@ -630,13 +638,18 @@ def test_battery_records_evidence_absent(self, uk_gates) -> None: o.entry.id for o in battery.blocking_outcomes(release_candidate=False) } assert default_blocked == { + "uk_local_geography_ladder_post_calibration", "uk_qrf_tail_concentration", "uk_weights_audit", } blocked = { o.entry.id for o in battery.blocking_outcomes(release_candidate=True) } - assert blocked == {*absent, "uk_qrf_tail_concentration"} + assert blocked == { + *(set(absent) - {"uk_local_target_fit", "uk_local_per_family_fit"}), + "uk_local_geography_ladder_post_calibration", + "uk_qrf_tail_concentration", + } def test_absent_fit_evidence_is_named(self) -> None: person, benunit, household = _tables() diff --git a/packages/microcosm-build/tests/test_uk_diagnostics.py b/packages/microcosm-build/tests/test_uk_diagnostics.py index af25c584f..5eb8e8f01 100644 --- a/packages/microcosm-build/tests/test_uk_diagnostics.py +++ b/packages/microcosm-build/tests/test_uk_diagnostics.py @@ -13,6 +13,8 @@ UK_DIAGNOSTICS_SCHEMA_VERSION, UK_TARGET_GEOGRAPHY_LEVELS, uk_calibration_diagnostics_payload, + uk_weakest_areas_by_fit, + uk_weakest_families, uk_weight_summary, uk_zero_weight_strata, write_uk_calibration_diagnostics, @@ -30,6 +32,117 @@ _CG_COLUMN = "household_is_capital_gains_clone" +def _local_target_row( + name: str, + *, + family: str, + area_type: str, + area_code: str, + relative_error: float, + loss_contribution: float, +) -> dict[str, object]: + return { + "name": name, + "relative_error": relative_error, + "final_loss_contribution": loss_contribution, + "registry": {"family": family}, + "metadata": {"area_type": area_type, "area_code": area_code}, + } + + +def test_local_weakest_rollups_pin_family_area_and_country_shapes() -> None: + rows = [ + _local_target_row( + "wales-a", + family="income", + area_type="constituency", + area_code="W07000041", + relative_error=0.30, + loss_contribution=0.20, + ), + _local_target_row( + "wales-b", + family="households", + area_type="constituency", + area_code="W07000041", + relative_error=0.05, + loss_contribution=0.01, + ), + _local_target_row( + "england-a", + family="income", + area_type="local_authority", + area_code="E09000001", + relative_error=0.15, + loss_contribution=0.09, + ), + ] + support = pd.DataFrame( + { + "geography_level": ["constituency", "local_authority"], + "area_code": ["W07000041", "E09000001"], + "nonzero_households": [80, 90], + "nonzero_source_households": [70, 75], + "effective_sample_size": [65.0, 72.0], + } + ) + + families = uk_weakest_families(rows) + assert [row["family"] for row in families] == ["income", "households"] + assert set(families[0]) == { + "family", + "n_targets", + "n_within_10pct", + "pass_rate", + "worst_target", + "worst_abs_relative_error", + "loss_contribution", + "loss_share", + } + assert families[0]["n_targets"] == 2 + assert families[0]["loss_contribution"] == pytest.approx(0.29) + + areas = uk_weakest_areas_by_fit(rows, support) + assert set(areas) == {"limit", "n_areas_scored", "bottom_by_fit", "countries"} + # The reported list is keyed by role and carries its own limit, so the key + # never asserts a count the list does not have. + assert areas["limit"] == 15 + assert areas["n_areas_scored"] == 2 + assert [row["area_code"] for row in areas["bottom_by_fit"]] == [ + "W07000041", + "E09000001", + ] + assert set(areas["bottom_by_fit"][0]) == { + "geography_level", + "area_code", + "country", + "n_targets", + "n_within_10pct", + "pass_rate", + "worst_target", + "worst_abs_relative_error", + "loss_contribution", + "nonzero_households", + "nonzero_source_households", + "effective_sample_size", + } + assert [row["country"] for row in areas["countries"]] == [ + "England", + "Wales", + ] + wales = areas["countries"][1] + assert wales["geography_level"] == "constituency" + assert wales["n_targets"] == 2 + assert wales["pass_rate"] == pytest.approx(0.5) + + # A caller-supplied limit is reported alongside the list it produced, + # rather than being contradicted by a fixed key name. + trimmed = uk_weakest_areas_by_fit(rows, support, limit=1) + assert trimmed["limit"] == 1 + assert trimmed["n_areas_scored"] == 2 + assert [row["area_code"] for row in trimmed["bottom_by_fit"]] == ["W07000041"] + + def _diagnostics_case(*, with_skipped: bool = False): levels_and_weights = ( ("national", 11.0), diff --git a/packages/microcosm-build/tests/test_uk_local_candidate_scorer.py b/packages/microcosm-build/tests/test_uk_local_candidate_scorer.py new file mode 100644 index 000000000..96761fea8 --- /dev/null +++ b/packages/microcosm-build/tests/test_uk_local_candidate_scorer.py @@ -0,0 +1,407 @@ +"""The UK local scorer compares one frozen surface on both sides.""" + +from __future__ import annotations + +import importlib.util +from pathlib import Path + +import numpy as np +import pandas as pd +import pytest + +from microcosm.build.holdout import summarize_rotations +from microcosm.build.uk_runtime.local_doctrine import UK_LOCAL_TARGET_LOSS_CAP +from microcosm.calibrate import ( + TargetRegistry, + TargetSpec, + default_target_loss_scales, + relative_error_loss, +) + + +def _load_scorer(): + path = Path(__file__).resolve().parents[3] / "tools/score_uk_local_candidate.py" + spec = importlib.util.spec_from_file_location("score_uk_local_candidate", path) + module = importlib.util.module_from_spec(spec) + assert spec.loader is not None + spec.loader.exec_module(module) + return module + + +def _case(): + specs = ( + TargetSpec( + name="households@E1", + entity="household", + value=10.0, + measure="households", + period=2025, + source="fixture", + family="census_households", + metadata={"ledger_geography_id": "E1"}, + ), + TargetSpec( + name="income@W1", + entity="household", + value=100.0, + measure="income", + period=2025, + source="fixture", + family="hmrc_income", + metadata={"ledger_geography_id": "W1"}, + ), + ) + registry = TargetRegistry(specs, country="uk") + candidate = { + "schema_version": 6, + "targets": [ + {"name": specs[0].to_target().row_name, "final_estimate": 10.0}, + {"name": specs[1].to_target().row_name, "final_estimate": 110.0}, + ], + "uk_diagnostics": { + "rotated_holdout": { + "report_only": True, + "method": "rotated_folds", + "n_folds": 5, + "seed": 20260529, + "target_loss_cap": UK_LOCAL_TARGET_LOSS_CAP, + # Summary closes over the folds, as summarize_rotations + # produces it: mean 0.4, worst 0.6. + "mean_holdout_loss": 0.4, + "worst_holdout_loss": 0.6, + "fold_losses": [0.2, 0.3, 0.4, 0.5, 0.6], + } + }, + } + # Deliberately asymmetric across households: a positional pairing of these + # two tables gives a different, plausible-looking answer than the join, + # which is what the permutation regression below pins. + weights = pd.DataFrame( + { + "household_id": ["h1", "h2"], + "E1": [4.0, 2.0], + "W1": [1.0, 3.0], + } + ) + metrics = pd.DataFrame( + { + "household_id": ["h1", "h2"], + "households": [2.0, 1.0], + "income": [10.0, 20.0], + } + ) + return registry, candidate, weights, metrics + + +def test_local_scorer_reports_per_family_wins_and_the_measured_holdout() -> None: + scorer = _load_scorer() + registry, candidate, weights, metrics = _case() + + result = scorer.score_uk_local_candidate( + candidate_diagnostics=candidate, + incumbent_weights=weights, + incumbent_metrics=metrics, + target_registry=registry, + expected_reference_count=2, + ) + + assert result["holdout_basis"] == "rotated_folds:n_folds=5:seed=20260529" + assert result["candidate_holdout_loss"] == pytest.approx(0.4) + assert result["candidate_holdout"]["worst_holdout_loss"] == pytest.approx(0.6) + # The incumbent is never re-solved, so it has no holdout to compare, and + # the head-to-head counters must say which surface they ran on. + assert result["incumbent_holdout_loss"] is None + assert result["incumbent_holdout_basis"] == "none_available_incumbent_not_resolved" + assert result["loss"]["head_to_head_surface"] == "candidate_fitted_surface" + # Both aggregates come from the canonical objective at the doctrine cap, + # so the holdout beside them is on the same scale. + assert result["loss"]["objective"] == "microcosm.calibrate.relative_error_loss" + assert result["loss"]["target_loss_cap"] == UK_LOCAL_TARGET_LOSS_CAP + assert result["candidate_fitted_surface_loss"] == pytest.approx( + relative_error_loss( + np.array([10.0, 110.0]), + np.array([10.0, 100.0]), + target_loss_cap=UK_LOCAL_TARGET_LOSS_CAP, + ) + ) + assert result["candidate_target_wins"] == 1 + assert result["incumbent_target_wins"] == 0 + assert result["target_wins_by_family"] == { + "census_households": { + "candidate_target_wins": 0, + "incumbent_target_wins": 0, + "ties": 1, + }, + "hmrc_income": { + "candidate_target_wins": 1, + "incumbent_target_wins": 0, + "ties": 0, + }, + } + + +def test_local_scorer_refuses_a_non_frozen_surface() -> None: + scorer = _load_scorer() + registry, candidate, weights, metrics = _case() + + with pytest.raises(ValueError, match="frozen active reference count"): + scorer.score_uk_local_candidate( + candidate_diagnostics=candidate, + incumbent_weights=weights, + incumbent_metrics=metrics, + target_registry=registry, + ) + + +def test_local_scorer_joins_the_incumbent_on_household_id() -> None: + """A reordered metrics file must score identically, not plausibly wrong.""" + + scorer = _load_scorer() + registry, candidate, weights, metrics = _case() + + straight = scorer.score_uk_local_candidate( + candidate_diagnostics=candidate, + incumbent_weights=weights, + incumbent_metrics=metrics, + target_registry=registry, + expected_reference_count=2, + ) + shuffled = scorer.score_uk_local_candidate( + candidate_diagnostics=candidate, + incumbent_weights=weights, + incumbent_metrics=metrics.iloc[::-1].reset_index(drop=True), + target_registry=registry, + expected_reference_count=2, + ) + + assert straight["target_drift"] == shuffled["target_drift"] + assert straight["incumbent_fitted_surface_loss"] == pytest.approx( + shuffled["incumbent_fitted_surface_loss"] + ) + # Pin the joined values themselves, so a silent revert to positional + # pairing (which would give 8.0 and 50.0 here) fails rather than drifts. + incumbent = scorer._incumbent_estimates(registry, weights, metrics) + assert incumbent["households@E1@2025"] == pytest.approx(10.0) + assert incumbent["income@W1@2025"] == pytest.approx(70.0) + + +@pytest.mark.parametrize("side", ["weights", "metrics"]) +def test_local_scorer_refuses_incumbent_tables_without_household_id(side: str) -> None: + scorer = _load_scorer() + registry, candidate, weights, metrics = _case() + if side == "weights": + weights = weights.drop(columns=["household_id"]) + else: + metrics = metrics.drop(columns=["household_id"]) + + with pytest.raises(ValueError, match="positional pairing is not a join"): + scorer.score_uk_local_candidate( + candidate_diagnostics=candidate, + incumbent_weights=weights, + incumbent_metrics=metrics, + target_registry=registry, + expected_reference_count=2, + ) + + +def test_local_scorer_refuses_a_partial_incumbent_join() -> None: + scorer = _load_scorer() + registry, candidate, weights, metrics = _case() + metrics.loc[1, "household_id"] = "h3" + + with pytest.raises(ValueError, match="must cover the same households"): + scorer.score_uk_local_candidate( + candidate_diagnostics=candidate, + incumbent_weights=weights, + incumbent_metrics=metrics, + target_registry=registry, + expected_reference_count=2, + ) + + +def test_local_scorer_refuses_duplicate_incumbent_households() -> None: + scorer = _load_scorer() + registry, candidate, weights, metrics = _case() + weights.loc[1, "household_id"] = "h1" + metrics.loc[1, "household_id"] = "h1" + + with pytest.raises(ValueError, match="repeats 'household_id'"): + scorer.score_uk_local_candidate( + candidate_diagnostics=candidate, + incumbent_weights=weights, + incumbent_metrics=metrics, + target_registry=registry, + expected_reference_count=2, + ) + + +def test_local_scorer_refuses_a_candidate_with_no_measured_holdout() -> None: + """The rotation ships in the same payload; scoring without it is refused.""" + + scorer = _load_scorer() + registry, candidate, weights, metrics = _case() + candidate = { + key: value for key, value in candidate.items() if key != "uk_diagnostics" + } + + with pytest.raises(ValueError, match="must carry a uk_diagnostics block"): + scorer.score_uk_local_candidate( + candidate_diagnostics=candidate, + incumbent_weights=weights, + incumbent_metrics=metrics, + target_registry=registry, + expected_reference_count=2, + ) + + +def test_local_scorer_refuses_a_holdout_with_missing_fold_losses() -> None: + scorer = _load_scorer() + registry, candidate, weights, metrics = _case() + candidate["uk_diagnostics"]["rotated_holdout"]["fold_losses"] = [0.2, 0.3] + + with pytest.raises(ValueError, match="one loss per declared fold"): + scorer.score_uk_local_candidate( + candidate_diagnostics=candidate, + incumbent_weights=weights, + incumbent_metrics=metrics, + target_registry=registry, + expected_reference_count=2, + ) + + +def test_local_scorer_refuses_a_holdout_measured_at_another_cap() -> None: + """The scale agreement across the module boundary is enforced, not assumed. + + When microcosm#762 moves the doctrine cap, the fitted-surface aggregates + move with it but a recorded holdout does not, so a stale rotation must be + refused rather than reported beside them. + """ + + scorer = _load_scorer() + registry, candidate, weights, metrics = _case() + candidate["uk_diagnostics"]["rotated_holdout"]["target_loss_cap"] = ( + UK_LOCAL_TARGET_LOSS_CAP + 5.0 + ) + + with pytest.raises(ValueError, match="re-measure the candidate"): + scorer.score_uk_local_candidate( + candidate_diagnostics=candidate, + incumbent_weights=weights, + incumbent_metrics=metrics, + target_registry=registry, + expected_reference_count=2, + ) + + +def test_local_scorer_refuses_a_holdout_that_declares_no_cap() -> None: + scorer = _load_scorer() + registry, candidate, weights, metrics = _case() + del candidate["uk_diagnostics"]["rotated_holdout"]["target_loss_cap"] + + with pytest.raises(ValueError, match="must declare the target_loss_cap"): + scorer.score_uk_local_candidate( + candidate_diagnostics=candidate, + incumbent_weights=weights, + incumbent_metrics=metrics, + target_registry=registry, + expected_reference_count=2, + ) + + +def test_local_scorer_drift_rows_use_the_canonical_row_scale() -> None: + """Drift rows and aggregates share one scale, imported not restated.""" + + scorer = _load_scorer() + registry, candidate, weights, metrics = _case() + + result = scorer.score_uk_local_candidate( + candidate_diagnostics=candidate, + incumbent_weights=weights, + incumbent_metrics=metrics, + target_registry=registry, + expected_reference_count=2, + ) + + targets = np.array([10.0, 100.0]) + expected = (np.array([10.0, 110.0]) - targets) / default_target_loss_scales(targets) + assert [row["candidate_relative_error"] for row in result["target_drift"]] == [ + pytest.approx(value) for value in expected + ] + + +def test_local_scorer_refuses_a_headline_that_is_not_its_folds() -> None: + """The substitution the required-holdout refusal exists to catch.""" + + scorer = _load_scorer() + registry, candidate, weights, metrics = _case() + # A fitted-looking number swapped in, with the real folds left beside it. + candidate["uk_diagnostics"]["rotated_holdout"]["mean_holdout_loss"] = 0.02 + + with pytest.raises(ValueError, match="does not close over its fold losses"): + scorer.score_uk_local_candidate( + candidate_diagnostics=candidate, + incumbent_weights=weights, + incumbent_metrics=metrics, + target_registry=registry, + expected_reference_count=2, + ) + + +def test_local_scorer_refuses_a_worst_loss_that_is_not_its_worst_fold() -> None: + scorer = _load_scorer() + registry, candidate, weights, metrics = _case() + candidate["uk_diagnostics"]["rotated_holdout"]["worst_holdout_loss"] = 0.61 + + with pytest.raises(ValueError, match="is not its worst fold"): + scorer.score_uk_local_candidate( + candidate_diagnostics=candidate, + incumbent_weights=weights, + incumbent_metrics=metrics, + target_registry=registry, + expected_reference_count=2, + ) + + +@pytest.mark.parametrize("bad", [float("nan"), -0.1]) +def test_local_scorer_refuses_invalid_fold_losses(bad: float) -> None: + scorer = _load_scorer() + registry, candidate, weights, metrics = _case() + candidate["uk_diagnostics"]["rotated_holdout"]["fold_losses"] = [ + bad, + 0.3, + 0.4, + 0.5, + 0.6, + ] + + with pytest.raises(ValueError, match="non-finite or negative fold loss"): + scorer.score_uk_local_candidate( + candidate_diagnostics=candidate, + incumbent_weights=weights, + incumbent_metrics=metrics, + target_registry=registry, + expected_reference_count=2, + ) + + +def test_local_scorer_accepts_a_summary_its_folds_actually_produce() -> None: + """`summarize_rotations` output must pass the closure check unchanged.""" + + scorer = _load_scorer() + registry, candidate, weights, metrics = _case() + folds = [0.11, 0.27, 0.4, 0.52, 0.63] + summary = summarize_rotations(folds) + holdout = candidate["uk_diagnostics"]["rotated_holdout"] + holdout["fold_losses"] = list(summary.fold_losses) + holdout["mean_holdout_loss"] = summary.mean_holdout_loss + holdout["worst_holdout_loss"] = summary.worst_holdout_loss + + result = scorer.score_uk_local_candidate( + candidate_diagnostics=candidate, + incumbent_weights=weights, + incumbent_metrics=metrics, + target_registry=registry, + expected_reference_count=2, + ) + + assert result["candidate_holdout_loss"] == pytest.approx(summary.mean_holdout_loss) diff --git a/packages/microcosm-build/tests/test_uk_local_gate_battery.py b/packages/microcosm-build/tests/test_uk_local_gate_battery.py new file mode 100644 index 000000000..cd700ef75 --- /dev/null +++ b/packages/microcosm-build/tests/test_uk_local_gate_battery.py @@ -0,0 +1,55 @@ +"""Contract tests for the candidate-time UK local gate scope.""" + +from microcosm.build import load_country_spec +from microcosm.build.uk_runtime.calibration_run import ( + UK_LOCAL_GATE_SCOPE, + uk_local_gate_scope_exclusions, + uk_scoped_gate_manifest, +) + + +def test_local_scope_is_the_declared_six_gate_terminal_battery() -> None: + assert UK_LOCAL_GATE_SCOPE == ( + "uk_local_geography_ladder_post_calibration", + "uk_local_area_support", + "uk_local_target_fit", + "uk_local_per_family_fit", + "uk_local_weight_ratio", + "uk_local_weight_ess", + ) + manifest = uk_scoped_gate_manifest( + UK_LOCAL_GATE_SCOPE, + phases=("terminal",), + policy_suffix="local_candidate", + ) + assert tuple(entry.id for entry in manifest.gates) == UK_LOCAL_GATE_SCOPE + assert {entry.phase for entry in manifest.gates} == {"terminal"} + assert { + entry.id for entry in manifest.gates if entry.criticality == "diagnostic" + } == { + "uk_local_target_fit", + "uk_local_per_family_fit", + "uk_local_weight_ratio", + "uk_local_weight_ess", + } + + +def test_local_area_support_parameters_pin_both_grains_and_all_floors() -> None: + entries = {entry.id: entry for entry in load_country_spec("uk").gates.gates} + area_support = entries["uk_local_area_support"] + assert area_support.gate == "area_support" + assert dict(area_support.parameters) == { + "crosswalk_resource": "local_area_crosswalk.json", + "geography_levels": ("constituency", "local_authority"), + "minimum_rows": 50, + "minimum_effective_sample_size": 50.0, + "minimum_distinct_sources": 50, + } + + +def test_local_scope_exclusions_classify_every_other_uk_gate() -> None: + declared = {entry.id for entry in load_country_spec("uk").gates.gates} + exclusions = uk_local_gate_scope_exclusions() + assert set(exclusions) | set(UK_LOCAL_GATE_SCOPE) == declared + assert not set(exclusions) & set(UK_LOCAL_GATE_SCOPE) + assert all(exclusions.values()) diff --git a/packages/microcosm-build/tests/test_uk_local_holdout.py b/packages/microcosm-build/tests/test_uk_local_holdout.py new file mode 100644 index 000000000..71631450b --- /dev/null +++ b/packages/microcosm-build/tests/test_uk_local_holdout.py @@ -0,0 +1,66 @@ +"""The UK local rotated holdout runs five real training solves.""" + +import numpy as np +import pandas as pd + +from microcosm.build.holdout import rotated_folds +from microcosm.build.uk_runtime import ( + UK_LOCAL_TARGET_LOSS_CAP, + build_uk_rowwise_local_matrix, + rotated_uk_local_holdout, + uk_national_frame, +) +from microcosm.frame import WeightKind + + +def test_rotated_local_holdout_runs_fixed_five_fold_actual_solves() -> None: + household_ids = [101, 102, 103, 104, 105] + area_codes = ["E001", "W001", "S001", "N001", "E002"] + frame = uk_national_frame( + person=pd.DataFrame( + { + "person_id": [1, 2, 3, 4, 5], + "person_household_id": household_ids, + "person_benunit_id": [11, 12, 13, 14, 15], + } + ), + benunit=pd.DataFrame({"benunit_id": [11, 12, 13, 14, 15]}), + household=pd.DataFrame( + { + "household_id": household_ids, + "household_weight": [1.0] * 5, + } + ), + time_period="2023", + weight_kind=WeightKind.IMPORTANCE, + ) + problem = build_uk_rowwise_local_matrix( + pd.DataFrame({"households": np.ones(5)}, index=household_ids), + pd.Series(area_codes, index=household_ids), + pd.DataFrame({"code": area_codes, "households": [2.0] * 5}), + ) + + payload = rotated_uk_local_holdout( + frame, + problem, + epochs=2, + solve_seed=17, + ) + + expected_folds = rotated_folds(5, n_folds=5, seed=20260529) + assert payload["report_only"] is True + assert payload["method"] == "rotated_folds" + assert payload["n_folds"] == 5 + assert payload["seed"] == 20260529 + assert payload["solve_seed"] == 17 + assert [row["holdout_target_indices"] for row in payload["folds"]] == [ + fold.tolist() for fold in expected_folds + ] + assert all(row["n_train_targets"] == 4 for row in payload["folds"]) + assert all(row["n_holdout_targets"] == 1 for row in payload["folds"]) + assert payload["fold_losses"] == [0.5] * 5 + assert payload["mean_holdout_loss"] == 0.5 + assert payload["worst_holdout_loss"] == 0.5 + # The cap the folds were actually measured under travels with them, so a + # consumer can enforce the scale agreement instead of assuming it. + assert payload["target_loss_cap"] == UK_LOCAL_TARGET_LOSS_CAP diff --git a/packages/microcosm-build/tests/test_uk_release_certification.py b/packages/microcosm-build/tests/test_uk_release_certification.py index 69316ae5d..086987c3c 100644 --- a/packages/microcosm-build/tests/test_uk_release_certification.py +++ b/packages/microcosm-build/tests/test_uk_release_certification.py @@ -16,6 +16,7 @@ from microcosm.build.uk_runtime import release_certification from microcosm.build.uk_runtime.calibration_run import ( UK_CALIBRATION_GATE_SCOPE, + UK_LOCAL_GATE_SCOPE, UK_NATIONAL_GATE_SCOPE, UK_SHARED_GATE_IDS, UK_SPINE_GATE_SCOPE, @@ -144,7 +145,12 @@ def test_compose_green_certification(green_certification_inputs): union = set() for part in certification["parts"].values(): union.update(part["entry_ids"]) - assert union == declared + assert union | set(certification["spec"]["certification_excluded_gate_ids"]) == ( + declared + ) + assert certification["spec"]["certification_excluded_gate_ids"] == sorted( + UK_LOCAL_GATE_SCOPE + ) assert certification["spec"]["shared_gate_ids"] == sorted(UK_SHARED_GATE_IDS) assert certification["doctrine"]["overrides"] == { "epochs": {"default": 256, "effective": 1500} @@ -313,7 +319,9 @@ def test_release_cut_battery_runs_and_signs(tmp_path: Path, monkeypatch): assert set(payload["gates"]) == set(UK_NATIONAL_GATE_SCOPE) assert payload["blocked_at_phase"] is None assert set(payload["scope_exclusions"]) == ( - set(UK_SPINE_GATE_SCOPE) | set(UK_CALIBRATION_GATE_SCOPE) + set(UK_SPINE_GATE_SCOPE) + | set(UK_CALIBRATION_GATE_SCOPE) + | set(UK_LOCAL_GATE_SCOPE) ) - set(UK_NATIONAL_GATE_SCOPE) on_disk = json.loads(report_path.read_text(encoding="utf-8")) assert on_disk["attestation"]["signature"] == payload["attestation"]["signature"] diff --git a/packages/microcosm-build/tests/test_uk_rowwise_candidate.py b/packages/microcosm-build/tests/test_uk_rowwise_candidate.py index 0aea2a539..2d5a8933b 100644 --- a/packages/microcosm-build/tests/test_uk_rowwise_candidate.py +++ b/packages/microcosm-build/tests/test_uk_rowwise_candidate.py @@ -2,9 +2,11 @@ from __future__ import annotations +import base64 import hashlib import importlib.util import json +from dataclasses import replace from pathlib import Path import numpy as np @@ -29,6 +31,10 @@ def _spool_only_by_default(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.delenv("POPULACE_LEDGER_KEY", raising=False) monkeypatch.delenv("POPULACE_LEDGER_API_KEY", raising=False) monkeypatch.delenv("POPULACE_LOGBOOK_PREV_ROW_DIGEST", raising=False) + monkeypatch.setenv( + "MICROCOSM_UK_TERMINAL_GATE_SIGNING_KEY", + base64.b64encode(b"\x07" * 32).decode("ascii"), + ) def _spool_rows(output_dir: Path): @@ -164,27 +170,40 @@ def _write_ladder( return load_uk_oa_ladder(path) -def _write_staging_h5(path: Path) -> None: +def _write_staging_h5(path: Path, *, households_per_region: int = 1) -> None: + region_names = ( + "LONDON", + "WALES", + "SCOTLAND", + "NORTHERN_IRELAND", + ) + region_masses = (3.0, 10.0, 10.0, 10.0) + household_ids = list(range(1, 4 * households_per_region + 1)) household = pd.DataFrame( { - "household_id": [1, 2, 3, 4], - "household_weight": [3.0, 10.0, 10.0, 10.0], + "household_id": household_ids, + "household_weight": [ + mass / households_per_region + for mass in region_masses + for _ in range(households_per_region) + ], "region": [ - "LONDON", - "WALES", - "SCOTLAND", - "NORTHERN_IRELAND", + region for region in region_names for _ in range(households_per_region) ], + "household_is_spi_synthetic": [False] * len(household_ids), + "household_is_capital_gains_clone": [False] * len(household_ids), } ) + person_ids = [10_000 + household_id for household_id in household_ids] + benunit_ids = [20_000 + household_id for household_id in household_ids] person = pd.DataFrame( { - "person_id": [11, 21, 31, 41], - "person_household_id": [1, 2, 3, 4], - "person_benunit_id": [101, 201, 301, 401], + "person_id": person_ids, + "person_household_id": household_ids, + "person_benunit_id": benunit_ids, } ) - benunit = pd.DataFrame({"benunit_id": [101, 201, 301, 401]}) + benunit = pd.DataFrame({"benunit_id": benunit_ids}) dataset = uk_national_frame( person=person, benunit=benunit, @@ -204,15 +223,43 @@ def _write_staging_h5(path: Path) -> None: write_uk_national_frame(dataset, path) -def test_candidate_build_writes_calibrated_h5_and_evidence(tmp_path) -> None: +def test_candidate_build_writes_calibrated_h5_and_evidence( + monkeypatch, tmp_path +) -> None: pytest.importorskip("tables") pytest.importorskip("h5py") builder = _load_builder_module() input_h5 = tmp_path / "staging.h5" ladder_path = tmp_path / "ladder.npz" output_dir = tmp_path / "candidate" - _write_staging_h5(input_h5) + _write_staging_h5(input_h5, households_per_region=50) ladder = _write_ladder(ladder_path) + import microcosm.build.uk_runtime.battery_bindings as battery_bindings + + monkeypatch.setattr( + battery_bindings, + "_local_area_roster", + lambda _resource, levels: { + "constituency": tuple(sorted(set(ladder.constituency_code))), + "local_authority": tuple(sorted(set(ladder.local_authority_code))), + }, + ) + holdout = { + "report_only": True, + "method": "rotated_folds", + "n_folds": 5, + "seed": 20260529, + "solve_seed": 7, + "mean_holdout_loss": 0.1, + "worst_holdout_loss": 0.2, + "fold_losses": [0.1, 0.1, 0.2, 0.05, 0.05], + "folds": [], + } + monkeypatch.setattr( + builder, + "rotated_uk_local_holdout", + lambda *_args, **_kwargs: holdout, + ) assert ( builder.main( @@ -242,6 +289,8 @@ def test_candidate_build_writes_calibrated_h5_and_evidence(tmp_path) -> None: builder.SOLVE_DIAGNOSTICS_FILENAME, builder.AREA_SUPPORT_FILENAME, builder.PAST_CAP_FILENAME, + builder.CALIBRATION_DIAGNOSTICS_FILENAME, + builder.LOCAL_GATE_REPORT_FILENAME_TEMPLATE.format(source_year=2023), } assert candidate_h5.exists() assert expected_sidecars <= {path.name for path in output_dir.iterdir()} @@ -253,12 +302,8 @@ def test_candidate_build_writes_calibrated_h5_and_evidence(tmp_path) -> None: candidate_household = store["household"] assert candidate_kind is WeightKind.CALIBRATED assert candidate_household["source_year"].unique().tolist() == [2023] - assert set(candidate_household["source_household_key"]) == { - "2023:1", - "2023:2", - "2023:3", - "2023:4", - } + assert len(set(candidate_household["source_household_key"])) == 200 + assert {"2023:1", "2023:200"} <= set(candidate_household["source_household_key"]) assert len(candidate_mass_log) == 3 calibration_records = [ record @@ -277,9 +322,7 @@ def test_candidate_build_writes_calibrated_h5_and_evidence(tmp_path) -> None: assert manifest["bound_target_families"] == ["census_households/constituency"] adjudications = manifest["binding_adjudications"] assert adjudications["register_resource"] == "local_binding_adjudications.json" - assert adjudications["bound_families"] == [ - "census_households/constituency" - ] + assert adjudications["bound_families"] == ["census_households/constituency"] assert adjudications["evaluated_on"] seed = adjudications["stood_on"]["census_households/constituency"][ "census_disclosure_control_noise" @@ -336,23 +379,46 @@ def test_candidate_build_writes_calibrated_h5_and_evidence(tmp_path) -> None: "target_weight_rule": "uniform", } assert manifest["solve"]["n_targets"] == 4 - assert manifest["solve"]["n_households"] == 8 + assert manifest["solve"]["n_households"] == 400 assert np.isfinite(manifest["solve"]["initial_loss"]) assert np.isfinite(manifest["solve"]["final_loss"]) assert np.isfinite(manifest["solve"]["max_abs_relative_error"]) assert np.isfinite(manifest["solve"]["median_abs_relative_error"]) assert manifest["solve"]["past_cap"]["n_targets"] == 4 - assert manifest["support"]["min_assigned_households"] == 2 - assert manifest["support"]["min_nonzero_households"] == 2 - assert manifest["support"]["min_effective_sample_size"] == pytest.approx(2.0) + assert manifest["support"]["min_assigned_households"] == 100 + assert manifest["support"]["min_nonzero_households"] == 100 + assert manifest["support"]["min_effective_sample_size"] == pytest.approx(100.0) diagnostics = pd.read_csv(output_dir / builder.SOLVE_DIAGNOSTICS_FILENAME) support = pd.read_csv(output_dir / builder.AREA_SUPPORT_FILENAME) past_cap = json.loads((output_dir / builder.PAST_CAP_FILENAME).read_text()) + calibration_diagnostics = json.loads( + (output_dir / builder.CALIBRATION_DIAGNOSTICS_FILENAME).read_text() + ) assert len(diagnostics) == 4 assert diagnostics["metric"].unique().tolist() == ["households"] - assert len(support) == 4 + assert len(support) == 8 assert past_cap["n_targets"] == 4 + assert calibration_diagnostics["schema_version"] == 6 + uk_diagnostics = calibration_diagnostics["uk_diagnostics"] + assert len(uk_diagnostics["weakest_families"]) == 1 + assert len(uk_diagnostics["weakest_areas_by_fit"]["bottom_by_fit"]) == 4 + assert uk_diagnostics["weakest_areas_by_fit"]["n_areas_scored"] == 4 + assert { + row["country"] for row in uk_diagnostics["weakest_areas_by_fit"]["countries"] + } == { + "England", + "Northern Ireland", + "Scotland", + "Wales", + } + assert ( + manifest["diagnostics"]["weakest_families"] + == uk_diagnostics["weakest_families"] + ) + assert uk_diagnostics["rotated_holdout"] == holdout + assert manifest["diagnostics"]["rotated_holdout"] == holdout + assert "calibration_diagnostics" in manifest["outputs"] rows = _spool_rows(output_dir) assert len(rows) == 1 row = rows[0] @@ -361,23 +427,12 @@ def test_candidate_build_writes_calibrated_h5_and_evidence(tmp_path) -> None: assert row.seed == 7 assert row.disposition == "iterating" assert row.artifact_location == _local_ref(candidate_h5) - assert row.gate_verdicts == { - "uk_geography_ladder_post_calibration": { - "verdict": "passed", - "receipt": f"{_local_ref(output_dir / builder.MANIFEST_FILENAME)}#/gate", - }, - "uk_target_fit": { - "verdict": "passed", - "receipt": ( - f"{_local_ref(output_dir / builder.MANIFEST_FILENAME)}" - "#/solve/max_abs_relative_error" - ), - }, - "uk_area_support": { - "verdict": "passed", - "receipt": f"{_local_ref(output_dir / builder.MANIFEST_FILENAME)}#/support", - }, - } + assert set(row.gate_verdicts) == set(builder.UK_LOCAL_GATE_SCOPE) + assert {item["verdict"] for item in row.gate_verdicts.values()} == {"passed"} + assert all( + ".local_gates.json#/gates/" in item["receipt"] + for item in row.gate_verdicts.values() + ) def test_candidate_dry_run_plans_without_solve_or_write( @@ -428,9 +483,7 @@ def forbidden(*_args, **_kwargs): assert plan["bound_target_families"] == ["census_households/constituency"] adjudications = plan["binding_adjudications"] assert adjudications["register_resource"] == "local_binding_adjudications.json" - assert adjudications["bound_families"] == [ - "census_households/constituency" - ] + assert adjudications["bound_families"] == ["census_households/constituency"] assert adjudications["evaluated_on"] assert ( "census_disclosure_control_noise" @@ -468,15 +521,25 @@ def test_candidate_refusal_records_receipt_and_reraises( def failing_gate(*_args, **_kwargs): return builder.GateResult( - name="uk_geography_ladder", + name="spine_agreement", passed=False, failures=("post-calibration coverage failed",), details={"minimum": 0}, ) - monkeypatch.setattr(builder, "uk_geography_ladder_gate", failing_gate) + original = builder.UK_GATE_REGISTRY["spine_agreement"] + monkeypatch.setattr( + builder, + "UK_GATE_REGISTRY", + { + **builder.UK_GATE_REGISTRY, + "spine_agreement": replace(original, evaluator=failing_gate), + }, + ) - with pytest.raises(ValueError, match="post-calibration coverage failed"): + with pytest.raises( + builder.GateBatteryBlockedError, match="post-calibration coverage failed" + ): builder.main( [ "--input-h5", @@ -498,16 +561,16 @@ def failing_gate(*_args, **_kwargs): assert len(rows) == 1 row = rows[0] assert row.disposition == "failed" - refusal_path = ( - output_dir / "logbook-receipts" / row.build_id / "candidate-refusal.json" + gate_report_path = output_dir / builder.LOCAL_GATE_REPORT_FILENAME_TEMPLATE.format( + source_year=2023 ) - assert refusal_path.exists() - refusal = json.loads(refusal_path.read_text()) - assert refusal["gate"]["phase"] == "post_calibration" - assert refusal["gate"]["passed"] is False - assert row.gate_verdicts["uk_geography_ladder_post_calibration"] == { + assert gate_report_path.exists() + assert row.gate_verdicts["uk_local_geography_ladder_post_calibration"] == { "verdict": "failed", - "receipt": f"{_local_ref(refusal_path)}#/gate", + "receipt": ( + f"{_local_ref(gate_report_path)}" + "#/gates/uk_local_geography_ladder_post_calibration" + ), } assert row.gate_verdicts["pipeline_error"]["verdict"] == "error" assert row.gate_verdicts["pipeline_error"]["receipt"].endswith("#/error_type") diff --git a/packages/microcosm-data/src/microcosm/data/contract.py b/packages/microcosm-data/src/microcosm/data/contract.py index f3306e803..91376f846 100644 --- a/packages/microcosm-data/src/microcosm/data/contract.py +++ b/packages/microcosm-data/src/microcosm/data/contract.py @@ -375,13 +375,13 @@ # fingerprint derives from the manifest digest. Editing the spec moves all # three here in the same reviewed change. _UK_GATE_BATTERY_POLICY_SHA256 = ( - "12aab28f1e8e49347887c53fe1fabd228a5eda045964d65224390e0ce8b118d5" + "bcbcfd552424313b0843bc68ad64afea04c3b4146b01458cc9fdaf54b85aea82" ) _UK_GATE_BATTERY_GATES_MANIFEST_SHA256 = ( - "efdb12a1f97421197871aefbb7de4be90e5d9a4f0461e6c6e72e5dcc8cf65089" + "76c861d48d48d73fd8f18f3d5cddac2855294c2839105ca2b6ac9a8a9d4b5f2f" ) _UK_GATE_BATTERY_SPEC_FINGERPRINT = ( - "96186a467471393be608dc638f8288db9ebfdcf2f54a1afbaf8f070db6716746" + "dfbad2aa2930ef7e615742bd43bccd6cadaeb77ceb46f9161bb81b715dda20aa" ) #: Spec entry id -> the legacy gate name whose observable detail checks #: apply unchanged (the battery re-keys the report by entry id; the gate @@ -407,9 +407,7 @@ "uk_qrf_tail_concentration": "qrf_tail_concentration", } #: Spec entry id -> (gate, phase), mirrored per entry so a report cannot -#: relabel an entry's identity. Every entry in this vintage is -#: release_blocking with no declared excuse, so criticality and -#: not_applicable are enforced globally rather than per entry. +#: relabel an entry's identity. _UK_GATE_BATTERY_ENTRY_GATES = { "uk_release_input_coverage_manifest_current": ( "release_input_coverage", @@ -476,8 +474,25 @@ "uk_target_fit": ("target_fit", "terminal"), "uk_input_mass_parity": ("input_mass_parity", "terminal"), "uk_qrf_tail_concentration": ("tail_concentration", "terminal"), + "uk_local_geography_ladder_post_calibration": ( + "spine_agreement", + "terminal", + ), + "uk_local_area_support": ("area_support", "terminal"), + "uk_local_target_fit": ("target_fit", "terminal"), + "uk_local_per_family_fit": ("per_family_fit", "terminal"), + "uk_local_weight_ratio": ("weight_ratio", "terminal"), + "uk_local_weight_ess": ("weight_ess", "terminal"), } _UK_GATE_BATTERY_ENTRY_IDS = frozenset(_UK_GATE_BATTERY_ENTRY_GATES) +_UK_GATE_BATTERY_DIAGNOSTIC_IDS = frozenset( + { + "uk_local_target_fit", + "uk_local_per_family_fit", + "uk_local_weight_ratio", + "uk_local_weight_ess", + } +) #: The entries whose bindings contribute an evidence digest; their keys are #: the only ones a schema-4 ``evidence_sha256`` may carry, and each appears #: exactly when its entry evaluated. @@ -557,6 +572,16 @@ _UK_RELEASE_CERTIFICATION_SCHEMA_VERSION = 1 _UK_RELEASE_CERTIFICATION_KIND = "uk_release_certification" _UK_CERTIFICATION_SHARED_GATE_IDS = frozenset({"uk_aggregate_admin"}) +_UK_CERTIFICATION_EXCLUDED_GATE_IDS = frozenset( + { + "uk_local_geography_ladder_post_calibration", + "uk_local_area_support", + "uk_local_target_fit", + "uk_local_per_family_fit", + "uk_local_weight_ratio", + "uk_local_weight_ess", + } +) _UK_CERTIFICATION_PART_PHASES: Mapping[str, tuple[str, ...]] = { "spine": ("assembled", "transferred"), "calibration_seam": ("terminal",), @@ -2562,15 +2587,29 @@ def _check_uk_gate_battery_report( f"{owner}.phase must be {pinned_phase!r} per the committed " f"spec, got {outcome.get('phase')!r}." ) + # Criticality is pinned per entry against the committed spec, so a + # relabel in either direction is a failure and cannot dodge the + # shippability recompute below. `unreached`, `not_applicable` and + # any status outside the taxonomy are already refused above for every + # entry, diagnostic ones included; what the diagnostic label buys is + # only that `failed`/`evidence_absent` do not block, which is the + # declared posture for the four local fit gates until microcosm#762 + # arms them. + expected_criticality = ( + "diagnostic" + if entry_id in _UK_GATE_BATTERY_DIAGNOSTIC_IDS + else "release_blocking" + ) criticality = outcome.get("criticality") - if criticality != "release_blocking": - # Every entry in this vintage blocks; a relabel to diagnostic - # would dodge the shippability recompute below. + if criticality != expected_criticality: failures.append( - f"{owner}.criticality must be 'release_blocking' per the " + f"{owner}.criticality must be {expected_criticality!r} per the " f"committed spec, got {criticality!r}." ) - elif status not in _UK_GATE_BATTERY_SHIPPABLE_STATUSES: + elif ( + criticality == "release_blocking" + and status not in _UK_GATE_BATTERY_SHIPPABLE_STATUSES + ): # Shippability is recomputed here, per entry, instead of # trusting the report's own shippable flag. failures.append( @@ -2947,9 +2986,12 @@ def _check_uk_release_certification( for scope in _UK_CERTIFICATION_PART_SCOPES.values(): for gate_id in scope: union[gate_id] = union.get(gate_id, 0) + 1 - if set(union) != _UK_GATE_BATTERY_ENTRY_IDS: + if set(union) | set(_UK_CERTIFICATION_EXCLUDED_GATE_IDS) != ( + _UK_GATE_BATTERY_ENTRY_IDS + ): failures.append( - f"{file} mirrored part scopes do not union to the declared gate-entry set." + f"{file} mirrored part scopes plus certification exclusions do not " + "cover the declared gate-entry set." ) overlap = sorted( gate_id @@ -2992,6 +3034,13 @@ def _check_uk_release_certification( f"{file} spec.shared_gate_ids must be " f"{sorted(_UK_CERTIFICATION_SHARED_GATE_IDS)}." ) + if list(spec.get("certification_excluded_gate_ids", ())) != sorted( + _UK_CERTIFICATION_EXCLUDED_GATE_IDS + ): + failures.append( + f"{file} spec.certification_excluded_gate_ids must be " + f"{sorted(_UK_CERTIFICATION_EXCLUDED_GATE_IDS)}." + ) if ( calibration_diagnostics_sha256 is not None diff --git a/packages/microcosm-data/tests/test_contract.py b/packages/microcosm-data/tests/test_contract.py index 4b1a88596..9c536c332 100644 --- a/packages/microcosm-data/tests/test_contract.py +++ b/packages/microcosm-data/tests/test_contract.py @@ -137,13 +137,13 @@ def _trusted_terminal_gate_signing_key(monkeypatch) -> None: UK_GATE_BATTERY_PRODUCER = "microcosm.build.gate_battery" UK_GATE_BATTERY_SIGNING_KEY_ENV = "MICROCOSM_UK_TERMINAL_GATE_SIGNING_KEY" UK_GATE_BATTERY_POLICY_SHA256 = ( - "12aab28f1e8e49347887c53fe1fabd228a5eda045964d65224390e0ce8b118d5" + "bcbcfd552424313b0843bc68ad64afea04c3b4146b01458cc9fdaf54b85aea82" ) UK_GATE_BATTERY_GATES_MANIFEST_SHA256 = ( - "efdb12a1f97421197871aefbb7de4be90e5d9a4f0461e6c6e72e5dcc8cf65089" + "76c861d48d48d73fd8f18f3d5cddac2855294c2839105ca2b6ac9a8a9d4b5f2f" ) UK_GATE_BATTERY_SPEC_FINGERPRINT = ( - "96186a467471393be608dc638f8288db9ebfdcf2f54a1afbaf8f070db6716746" + "dfbad2aa2930ef7e615742bd43bccd6cadaeb77ceb46f9161bb81b715dda20aa" ) UK_GATE_BATTERY_DEGENERATE_EVIDENCE_SHA256 = ( "d0d024043132fa07c378c393dbe2b24fe99bf19e876bcc39997d2c80cc9bd4f6" @@ -271,6 +271,16 @@ def _trusted_terminal_gate_signing_key(monkeypatch) -> None: "terminal", "qrf_tail_concentration", ), + "uk_local_geography_ladder_post_calibration": ( + "spine_agreement", + "terminal", + None, + ), + "uk_local_area_support": ("area_support", "terminal", None), + "uk_local_target_fit": ("target_fit", "terminal", None), + "uk_local_per_family_fit": ("per_family_fit", "terminal", None), + "uk_local_weight_ratio": ("weight_ratio", "terminal", None), + "uk_local_weight_ess": ("weight_ess", "terminal", None), } @@ -1202,6 +1212,11 @@ def _gate_battery_payload( } elif entry_id == "uk_calibration_reference_coverage": details = {"activated": 388, "resolved": 388, "matrix": 388} + elif entry_id.startswith("uk_local_"): + # Local candidate gates are explicitly excluded from national + # certification; this full-report fixture needs only their + # authenticated envelope, not candidate-only evidence details. + details = {} elif gate == "stage_health": details = { "stage": stage_health_stages[entry_id], @@ -1212,7 +1227,17 @@ def _gate_battery_payload( gates[entry_id] = { "gate": gate, "phase": phase, - "criticality": "release_blocking", + "criticality": ( + "diagnostic" + if entry_id + in { + "uk_local_target_fit", + "uk_local_per_family_fit", + "uk_local_weight_ratio", + "uk_local_weight_ess", + } + else "release_blocking" + ), "status": "passed", "failures": [], "details": details, @@ -2225,9 +2250,7 @@ def test_uk_national_release_requires_signed_evidence_files(tmp_path: Path) -> N directory = _write_uk_national_release_dir(tmp_path) (directory / "terminal_gates.json").unlink() - with pytest.raises( - ReleaseContractError, match="missing 'terminal_gates.json'" - ): + with pytest.raises(ReleaseContractError, match="missing 'terminal_gates.json'"): validate_release_dir(directory) @@ -2257,9 +2280,7 @@ def test_uk_national_release_refuses_unbindable_score_digest( ) release_path.write_text(json.dumps(release)) - with pytest.raises( - ReleaseContractError, match=r"score_receipt\.sha256 is not a" - ): + with pytest.raises(ReleaseContractError, match=r"score_receipt\.sha256 is not a"): validate_release_dir(directory) @@ -4294,6 +4315,36 @@ def test_exact_k_uk_gate_battery_recomputes_shippability(tmp_path: Path) -> None assert "release-blocking with status 'failed'" in failures +def test_exact_k_uk_gate_battery_status_checks_cover_diagnostic_entries( + tmp_path: Path, +) -> None: + # The diagnostic label exempts an entry from the shippability recompute + # and nothing else: a diagnostic gate that compared nothing, or that + # carries a status outside the taxonomy, is still refused. + for status, expected in ( + ("not_applicable", "claims not_applicable"), + ("unreached", "is unreached"), + ("error", "outside the taxonomy"), + ): + directory, payload = _write_battery_release(tmp_path / status) + payload["gates"]["uk_local_target_fit"]["status"] = status + _rewrite_battery_report(directory, payload) + + assert expected in _battery_failures(directory) + + +def test_exact_k_uk_gate_battery_rejects_relabelled_diagnostic_criticality( + tmp_path: Path, +) -> None: + # Criticality is pinned per entry, so a blocking gate cannot be relabelled + # diagnostic to dodge the recompute, nor the reverse. + directory, payload = _write_battery_release(tmp_path) + payload["gates"]["uk_local_area_support"]["criticality"] = "diagnostic" + _rewrite_battery_report(directory, payload) + + assert "criticality must be 'release_blocking'" in _battery_failures(directory) + + def test_exact_k_uk_gate_battery_rejects_passed_entry_with_failures( tmp_path: Path, ) -> None: @@ -5233,6 +5284,9 @@ def _green_uk_certification( "declared_entry_count": len(contract._UK_GATE_BATTERY_ENTRY_IDS), "declared_phases": list(contract._UK_GATE_BATTERY_PHASES), "shared_gate_ids": sorted(contract._UK_CERTIFICATION_SHARED_GATE_IDS), + "certification_excluded_gate_ids": sorted( + contract._UK_CERTIFICATION_EXCLUDED_GATE_IDS + ), }, "doctrine": {"payload": {"epochs": 1500}, "overrides": {}}, "diagnostics_sha256": diagnostics_sha256, diff --git a/tools/build_uk_rowwise_candidate.py b/tools/build_uk_rowwise_candidate.py index 7dea45ff2..8061967ab 100644 --- a/tools/build_uk_rowwise_candidate.py +++ b/tools/build_uk_rowwise_candidate.py @@ -34,6 +34,12 @@ import numpy as np import pandas as pd +from microcosm.build.gate_battery import ( + BlockingMode, + EvidenceContext, + GateBatteryBlockedError, + GateBatteryRun, +) from microcosm.build.gates import GateResult from microcosm.build.logbook import canonical_json_bytes from microcosm.build.logbook_adoption import ( @@ -52,6 +58,7 @@ write_error_receipt, ) from microcosm.build.uk_runtime import ( + UK_GATE_REGISTRY, UK_LOCAL_MAX_WEIGHT_RATIO, UK_LOCAL_SOLVE_DOCTRINE, UK_LOCAL_TARGET_LOSS_CAP, @@ -66,21 +73,34 @@ ladder_target_provenance, load_uk_national_frame, load_uk_oa_ladder, + local_target_census, require_adjudicated_uk_local_binding, - rowwise_area_support_summary, + rotated_uk_local_holdout, solve_uk_rowwise_weights_under_doctrine, - uk_geography_ladder_gate, uk_household_weight_kind, + uk_ladder_area_support_summary, uk_time_period, + write_uk_calibration_diagnostics, write_uk_rowwise_dataset, ) +from microcosm.build.uk_runtime.calibration_run import ( + UK_LOCAL_GATE_SCOPE, + finalize_uk_scoped_gate_report, + uk_local_gate_scope_exclusions, + uk_scoped_gate_manifest, +) +from microcosm.calibrate import TargetRegistry, TargetSpec 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" +LOCAL_GATE_REPORT_FILENAME_TEMPLATE = ( + "populace_uk_{source_year}_rowwise_candidate.local_gates.json" +) MANIFEST_FILENAME = "rowwise_candidate_manifest.json" SOLVE_DIAGNOSTICS_FILENAME = "solve_diagnostics.csv" +CALIBRATION_DIAGNOSTICS_FILENAME = "calibration_diagnostics.json" AREA_SUPPORT_FILENAME = "area_support_summary.csv" PAST_CAP_FILENAME = "past_cap_census.json" @@ -89,6 +109,7 @@ _L0_LAMBDA = 0.0 _BUDGET_ITERS = 10 _UK_CANDIDATE_PIPELINE = "uk-local-candidate" +_LOCAL_GATE_POLICY_SUFFIX = "local_candidate" _REPOSITORY = Path(__file__).resolve().parents[1] _PAST_CAP_COUNT_KEYS = ( "n_targets", @@ -441,43 +462,53 @@ def _run_candidate( "calibrated frame's latest mass record is not the calibration " f"record: {calibration_record.reason!r}." ) - candidate_gate = uk_geography_ladder_gate( + support = _candidate_area_support( solve.frame.table("household"), - np.asarray(solve.weights, dtype=np.float64), + ladder, + weights=solve.weights, ) - if not candidate_gate.passed: - refusal_path = ( - out_dir / "logbook-receipts" / state.build_id / "candidate-refusal.json" - ) - atomic_write_json( - refusal_path, - {"gate": _gate_payload(candidate_gate, phase="post_calibration")}, + _validate_support_summary(support) + local_diagnostics = _local_gate_diagnostics(solve.diagnostics) + target_registry, target_geography_levels = _local_diagnostics_registry( + solve, + problem, + ) + try: + gate_report, candidate_gate = _run_local_gate_battery( + frame=solve.frame, + support=support, + diagnostics=local_diagnostics, + report_path=output_paths["local_gates"], + release_id=state.build_id, ) - state.gate_verdicts["uk_geography_ladder_post_calibration"] = { - "verdict": "failed", - "receipt": ( - f"{local_artifact_reference(refusal_path, repository_hint=_REPOSITORY)}" - "#/gate" - ), - } - raise ValueError( - "UK geography ladder gate failed on calibrated candidate weights: " - + "; ".join(candidate_gate.failures) + except GateBatteryBlockedError: + _apply_gate_verdicts( + state, + json.loads(output_paths["local_gates"].read_text(encoding="utf-8")), + output_paths["local_gates"], ) + raise + _apply_gate_verdicts(state, gate_report, output_paths["local_gates"]) append_phase(state, "candidate_gated") + rotated_holdout = rotated_uk_local_holdout( + clone.frame, + problem, + epochs=args.epochs, + learning_rate=args.learning_rate, + conserve_mass=_CONSERVE_MASS, + target_records=_TARGET_RECORDS, + l0_lambda=_L0_LAMBDA, + budget_iters=_BUDGET_ITERS, + solve_seed=args.seed, + ) + candidate = dataclasses.replace( clone, frame=solve.frame, gate=candidate_gate, output_path=None, ) - support = rowwise_area_support_summary( - problem, - solve.weights, - source_household_ids=household["source_household_id"].tolist(), - ) - _validate_support_summary(support) _assert_artifacts_unchanged( input_h5=input_h5, input_artifact=input_artifact, @@ -491,6 +522,10 @@ def _run_candidate( clone=clone, problem=problem, solve=solve, + local_diagnostics=local_diagnostics, + target_registry=target_registry, + target_geography_levels=target_geography_levels, + rotated_holdout=rotated_holdout, support=support, calibration_record=calibration_record, source_year=source_year, @@ -501,24 +536,6 @@ def _run_candidate( cross_grain=cross_grain, ) append_phase(state, "published") - manifest_path = output_paths["manifest"] - state.gate_verdicts = { - "uk_geography_ladder_post_calibration": { - "verdict": "passed", - "receipt": f"{local_artifact_reference(manifest_path, repository_hint=_REPOSITORY)}#/gate", - }, - "uk_target_fit": { - "verdict": "passed", - "receipt": ( - f"{local_artifact_reference(manifest_path, repository_hint=_REPOSITORY)}" - "#/solve/max_abs_relative_error" - ), - }, - "uk_area_support": { - "verdict": "passed", - "receipt": f"{local_artifact_reference(manifest_path, repository_hint=_REPOSITORY)}#/support", - }, - } state.artifact_location = local_artifact_reference( output_paths["dataset"], repository_hint=_REPOSITORY, @@ -640,6 +657,172 @@ def _build_bound_problem( } +def _candidate_area_support( + household: pd.DataFrame, + ladder: UkOaLadder, + *, + weights: np.ndarray, +) -> pd.DataFrame: + weighted_household = household.copy() + weighted_household["household_weight"] = np.asarray(weights, dtype=np.float64) + summaries = uk_ladder_area_support_summary(weighted_household, ladder) + return pd.concat( + ( + summaries["constituency"].assign(geography_level="constituency"), + summaries["la"].assign(geography_level="local_authority"), + ), + ignore_index=True, + )[ + [ + "geography_level", + "area_code", + "assigned_households", + "nonzero_households", + "nonzero_source_households", + "weight_sum", + "max_weight", + "effective_sample_size", + ] + ] + + +def _local_gate_diagnostics(diagnostics: pd.DataFrame) -> pd.DataFrame: + family_by_area_type: dict[str, str] = {} + for bound_family in BOUND_TARGET_FAMILIES: + family, separator, area_type = bound_family.partition("/") + if not family or not separator or not area_type: + raise ValueError( + f"bound target family {bound_family!r} must be family/area_type." + ) + if area_type in family_by_area_type: + raise ValueError( + f"multiple bound target families claim area type {area_type!r}." + ) + family_by_area_type[area_type] = family + result = diagnostics.copy() + result["family"] = result["area_type"].map(family_by_area_type) + if result["family"].isna().any(): + unknown = sorted(result.loc[result["family"].isna(), "area_type"].unique()) + raise ValueError( + f"local diagnostics contain unclassified area type(s): {unknown}." + ) + return result + + +def _local_diagnostics_registry( + solve: UKRowwiseDoctrineSolve, + problem: UKRowwiseLocalMatrix, +) -> tuple[TargetRegistry, dict[str, str]]: + targets = tuple(solve.calibration_result.problem.targets) + if len(targets) != len(problem.target_frame): + raise RuntimeError("local diagnostics registry is not aligned to the solve.") + specs: list[TargetSpec] = [] + geography: dict[str, str] = {} + for target, row in zip( + targets, + problem.target_frame.itertuples(index=False), + strict=True, + ): + metric = str(row.metric) + spec = TargetSpec( + name=str(target.name), + entity=str(target.entity), + value=float(target.value), + measure=f"rowwise_metric:{metric}", + filter=f"rowwise_area:{row.area_code}", + period=target.period, + source=str(target.source), + family=local_target_census.family_for_metric(metric), + metadata={key: str(value) for key, value in target.metadata.items()}, + ) + specs.append(spec) + geography[spec.to_target().row_name] = str(row.area_type) + return TargetRegistry(specs, country="uk"), geography + + +def _run_local_gate_battery( + *, + frame: Any, + support: pd.DataFrame, + diagnostics: pd.DataFrame, + report_path: Path, + release_id: str, +) -> tuple[dict[str, object], GateResult]: + manifest = uk_scoped_gate_manifest( + UK_LOCAL_GATE_SCOPE, + phases=("terminal",), + policy_suffix=_LOCAL_GATE_POLICY_SUFFIX, + ) + battery = GateBatteryRun( + manifest, + release_id=release_id, + report_path=report_path, + release_candidate=False, + registry=UK_GATE_REGISTRY, + ) + phase = battery.run_phase( + "terminal", + EvidenceContext( + frame=frame, + artifacts={ + "uk_area_support_summary": support, + "local_target_diagnostics": diagnostics, + }, + ), + ) + try: + battery.enforce("terminal", mode=BlockingMode.BLOCKS_ARTIFACT) + except GateBatteryBlockedError: + payload = battery.report_payload() + finalize_uk_scoped_gate_report( + payload, + posture="local_candidate", + scope_exclusions=uk_local_gate_scope_exclusions(), + aggregate_admin_measurement=None, + ) + atomic_write_json(report_path, payload) + raise + payload = battery.report_payload() + finalize_uk_scoped_gate_report( + payload, + posture="local_candidate", + scope_exclusions=uk_local_gate_scope_exclusions(), + aggregate_admin_measurement=None, + ) + atomic_write_json(report_path, payload) + ladder = next( + outcome + for outcome in phase.outcomes + if outcome.entry.id == "uk_local_geography_ladder_post_calibration" + ) + if ladder.result is None or not ladder.result.passed: + raise RuntimeError( + "a non-passing local geography-ladder result escaped battery enforcement." + ) + return payload, ladder.result + + +def _apply_gate_verdicts( + state: AttemptState, + report: Mapping[str, object], + report_path: Path, +) -> None: + gates = report.get("gates") + if not isinstance(gates, Mapping) or set(gates) != set(UK_LOCAL_GATE_SCOPE): + raise RuntimeError("local gate report does not cover the declared scope.") + receipt = local_artifact_reference(report_path, repository_hint=_REPOSITORY) + state.gate_verdicts = { + gate_id: { + "verdict": str(payload["status"]), + "receipt": f"{receipt}#/gates/{gate_id}", + } + for gate_id, payload in gates.items() + if isinstance(payload, Mapping) + } + if set(state.gate_verdicts) != set(UK_LOCAL_GATE_SCOPE): + raise RuntimeError("local gate verdicts are malformed.") + + def _dry_run_plan( args: argparse.Namespace, *, @@ -684,6 +867,10 @@ def _write_output_bundle( clone: UKLadderRowwiseDatasetResult, problem: UKRowwiseLocalMatrix, solve: UKRowwiseDoctrineSolve, + local_diagnostics: pd.DataFrame, + target_registry: TargetRegistry, + target_geography_levels: Mapping[str, str], + rotated_holdout: Mapping[str, object], support: pd.DataFrame, calibration_record: MassChangeRecord, source_year: int, @@ -711,9 +898,25 @@ def _write_output_bundle( flush=True, ) write_uk_rowwise_dataset(candidate, staged["dataset"]) - solve.diagnostics.to_csv(staged["diagnostics"], index=False) + local_diagnostics.to_csv(staged["diagnostics"], index=False) support.to_csv(staged["support"], index=False) staged["past_cap"].write_text(_json_text(dict(solve.past_cap_census or {}))) + write_uk_calibration_diagnostics( + solve.calibration_result, + staged["calibration_diagnostics"], + solve.frame, + target_geography_levels=target_geography_levels, + target_registry=target_registry, + local_area_support=support, + rotated_holdout=rotated_holdout, + build={ + "build_kind": "uk_rowwise_calibrated_candidate", + "candidate_scope": "adjudicated_partial", + }, + ) + calibration_diagnostics = json.loads( + staged["calibration_diagnostics"].read_text(encoding="utf-8") + ) outputs = { "dataset": _artifact_info( @@ -732,6 +935,11 @@ def _write_output_bundle( staged["past_cap"], reported_path=output_paths["past_cap"], ), + "calibration_diagnostics": _artifact_info( + staged["calibration_diagnostics"], + reported_path=output_paths["calibration_diagnostics"], + ), + "local_gate_report": _artifact_info(output_paths["local_gates"]), } manifest = _manifest( args, @@ -746,6 +954,7 @@ def _write_output_bundle( ladder_artifact=ladder_artifact, target_provenance=target_provenance, cross_grain=cross_grain, + calibration_diagnostics=calibration_diagnostics, outputs=outputs, ) staged["manifest"].write_text(_json_text(manifest)) @@ -769,6 +978,7 @@ def _manifest( ladder_artifact: Mapping[str, Any], target_provenance: Mapping[str, Any], cross_grain: Mapping[str, Any], + calibration_diagnostics: Mapping[str, Any], outputs: Mapping[str, Any], ) -> dict[str, Any]: abs_errors = solve.diagnostics["abs_relative_error"].to_numpy(dtype=np.float64) @@ -829,6 +1039,19 @@ def _manifest( "n_nonzero": int(solve.n_nonzero), "past_cap": {key: int(past_cap[key]) for key in _PAST_CAP_COUNT_KEYS}, }, + "diagnostics": { + "schema_version": calibration_diagnostics["schema_version"], + "target_registry": calibration_diagnostics["target_registry"], + "weakest_families": calibration_diagnostics["uk_diagnostics"][ + "weakest_families" + ], + "weakest_areas_by_fit": calibration_diagnostics["uk_diagnostics"][ + "weakest_areas_by_fit" + ], + "rotated_holdout": calibration_diagnostics["uk_diagnostics"][ + "rotated_holdout" + ], + }, "support": { "min_assigned_households": int(support["assigned_households"].min()), "min_nonzero_households": int(support["nonzero_households"].min()), @@ -909,8 +1132,11 @@ def _validate_solve_result( def _validate_support_summary(support: pd.DataFrame) -> None: required = { + "geography_level", + "area_code", "assigned_households", "nonzero_households", + "nonzero_source_households", "effective_sample_size", } missing = sorted(required - set(support.columns)) @@ -918,7 +1144,8 @@ def _validate_support_summary(support: pd.DataFrame) -> None: raise RuntimeError( f"area support summary is empty or missing required columns: {missing}." ) - values = support[list(required)].to_numpy(dtype=np.float64) + numeric = sorted(required - {"geography_level", "area_code"}) + values = support[numeric].to_numpy(dtype=np.float64) if not np.isfinite(values).all() or (values < 0).any(): raise RuntimeError("area support summary contains invalid values.") @@ -939,13 +1166,16 @@ def _validate_cli_args(args: argparse.Namespace) -> None: def _output_paths(out_dir: Path, *, source_year: int) -> dict[str, Path]: + dataset = out_dir / CANDIDATE_FILENAME_TEMPLATE.format(source_year=source_year) return { - "dataset": out_dir - / CANDIDATE_FILENAME_TEMPLATE.format(source_year=source_year), + "dataset": dataset, "manifest": out_dir / MANIFEST_FILENAME, "diagnostics": out_dir / SOLVE_DIAGNOSTICS_FILENAME, "support": out_dir / AREA_SUPPORT_FILENAME, "past_cap": out_dir / PAST_CAP_FILENAME, + "calibration_diagnostics": out_dir / CALIBRATION_DIAGNOSTICS_FILENAME, + "local_gates": out_dir + / LOCAL_GATE_REPORT_FILENAME_TEMPLATE.format(source_year=source_year), } @@ -984,6 +1214,7 @@ def _publish_staged_files( "diagnostics", "support", "past_cap", + "calibration_diagnostics", "manifest", ) published: list[Path] = [] diff --git a/tools/score_uk_local_candidate.py b/tools/score_uk_local_candidate.py new file mode 100644 index 000000000..2729a9563 --- /dev/null +++ b/tools/score_uk_local_candidate.py @@ -0,0 +1,488 @@ +"""Score a UK local candidate against incumbent wide-format area weights. + +The candidate side is read from its schema-v6 calibration diagnostics. The +incumbent side is deliberately explicit: a household-grain metric table and a +wide weight table with one column per local area. Both are evaluated on the +same frozen UK TargetRegistry; no fitted row is allowed to disappear. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import math +from collections.abc import Mapping +from pathlib import Path +from typing import Any + +import numpy as np +import pandas as pd + +from microcosm.build.uk_runtime.local_doctrine import UK_LOCAL_TARGET_LOSS_CAP +from microcosm.calibrate import ( + CALIBRATION_DIAGNOSTICS_SCHEMA_VERSION, + TargetRegistry, + default_target_loss_scales, + relative_error_loss, +) + +UK_LOCAL_ACTIVE_REFERENCE_COUNT = 17_077 +UK_LOCAL_SCORE_TARGET_PERIOD = 2025 +#: The incumbent is scored from published weights, never re-solved, so no +#: incumbent holdout exists to place beside the candidate's rotation. +UK_LOCAL_INCUMBENT_HOLDOUT_BASIS = "none_available_incumbent_not_resolved" +_HOUSEHOLD_ID_COLUMN = "household_id" + + +def _sha256_file(path: str | Path) -> str: + digest = hashlib.sha256() + with Path(path).open("rb") as stream: + for chunk in iter(lambda: stream.read(1 << 20), b""): + digest.update(chunk) + return digest.hexdigest() + + +def _verify_artifact(path: str | Path, expected_sha256: str) -> dict[str, object]: + artifact = Path(path) + measured = _sha256_file(artifact) + if measured != expected_sha256: + raise ValueError( + f"artifact sha mismatch for {artifact}: measured {measured}, " + f"pinned {expected_sha256}" + ) + return { + "path": str(artifact), + "sha256": measured, + "size_bytes": artifact.stat().st_size, + } + + +def _candidate_estimates( + diagnostics: Mapping[str, object], + registry: TargetRegistry, +) -> dict[str, float]: + if diagnostics.get("schema_version") != CALIBRATION_DIAGNOSTICS_SCHEMA_VERSION: + raise ValueError("UK local scoring requires schema-v6 candidate diagnostics.") + rows = diagnostics.get("targets") + if not isinstance(rows, list): + raise ValueError("candidate diagnostics must contain target rows.") + estimates: dict[str, float] = {} + for row in rows: + if not isinstance(row, Mapping): + raise ValueError("candidate diagnostics contain a malformed target row.") + name = str(row.get("name") or "") + raw = row.get("final_estimate") + if ( + not name + or not isinstance(raw, (int, float)) + or isinstance(raw, bool) + or not math.isfinite(float(raw)) + or name in estimates + ): + raise ValueError("candidate diagnostics contain invalid target estimates.") + estimates[name] = float(raw) + expected = {spec.to_target().row_name for spec in registry.specs} + if set(estimates) != expected: + missing = sorted(expected - set(estimates)) + extra = sorted(set(estimates) - expected) + raise ValueError( + "candidate diagnostics must exactly cover the frozen local register; " + f"missing={missing[:10]}, extra={extra[:10]}." + ) + return estimates + + +def _candidate_holdout(diagnostics: Mapping[str, object]) -> dict[str, object]: + """Read the candidate's measured rotated holdout out of its diagnostics. + + The candidate driver runs the rotation and publishes it in the same + schema-v6 payload this scorer already reads, so a receipt that reported + ``none_declared`` beside it would be understating what was measured. The + block is required: a candidate whose diagnostics carry no rotation is + refused rather than scored on its fitted surface alone. + """ + + uk_diagnostics = diagnostics.get("uk_diagnostics") + if not isinstance(uk_diagnostics, Mapping): + raise ValueError("candidate diagnostics must carry a uk_diagnostics block.") + holdout = uk_diagnostics.get("rotated_holdout") + if not isinstance(holdout, Mapping): + raise ValueError( + "candidate diagnostics must carry uk_diagnostics.rotated_holdout; " + "scoring a candidate with no measured holdout is refused." + ) + method = str(holdout.get("method") or "") + n_folds = holdout.get("n_folds") + seed = holdout.get("seed") + if ( + not method + or isinstance(n_folds, bool) + or not isinstance(n_folds, int) + or n_folds < 2 + or isinstance(seed, bool) + or not isinstance(seed, int) + ): + raise ValueError("candidate rotated holdout declares no usable basis.") + # The holdout is a number recorded upstream, under whatever cap that run + # used; the fitted-surface aggregates are computed here under the current + # doctrine cap. Requiring the recorded cap to match makes the agreement + # an enforced invariant instead of a convention that quietly lapses the + # first time microcosm#762 moves the constant — at which point the + # candidate needs re-measuring, not re-reporting. + declared_cap = holdout.get("target_loss_cap") + if ( + not isinstance(declared_cap, (int, float)) + or isinstance(declared_cap, bool) + or not math.isfinite(float(declared_cap)) + ): + raise ValueError( + "candidate rotated holdout must declare the target_loss_cap it was " + "measured under." + ) + if float(declared_cap) != float(UK_LOCAL_TARGET_LOSS_CAP): + raise ValueError( + "candidate rotated holdout was measured at target_loss_cap " + f"{float(declared_cap)!r}, but this scorer reports its aggregates " + f"at {float(UK_LOCAL_TARGET_LOSS_CAP)!r}; re-measure the candidate " + "rather than reporting the two on different scales." + ) + losses: dict[str, float] = {} + for key in ("mean_holdout_loss", "worst_holdout_loss"): + raw = holdout.get(key) + if ( + not isinstance(raw, (int, float)) + or isinstance(raw, bool) + or not math.isfinite(float(raw)) + ): + raise ValueError(f"candidate rotated holdout has an invalid {key}.") + losses[key] = float(raw) + fold_losses = holdout.get("fold_losses") + if not isinstance(fold_losses, list) or len(fold_losses) != n_folds: + raise ValueError( + "candidate rotated holdout must report one loss per declared fold." + ) + folds: list[float] = [] + for value in fold_losses: + if ( + not isinstance(value, (int, float)) + or isinstance(value, bool) + or not math.isfinite(float(value)) + or float(value) < 0.0 + ): + raise ValueError( + "candidate rotated holdout has a non-finite or negative fold loss." + ) + folds.append(float(value)) + # The summary is derived from these folds by `summarize_rotations`, so it + # must still close over them. Without this, a headline number lifted + # from somewhere else — a fitted loss, say — passes every other check + # with plausible folds sitting beside it, which is the substitution the + # required-holdout refusal exists to catch. + if not math.isclose( + losses["mean_holdout_loss"], + math.fsum(folds) / n_folds, + rel_tol=1e-9, + abs_tol=1e-12, + ): + raise ValueError( + "candidate rotated holdout mean does not close over its fold losses." + ) + if not math.isclose( + losses["worst_holdout_loss"], + max(folds), + rel_tol=1e-9, + abs_tol=1e-12, + ): + raise ValueError("candidate rotated holdout worst loss is not its worst fold.") + return { + "basis": f"{method}:n_folds={n_folds}:seed={seed}", + "method": method, + "target_loss_cap": float(declared_cap), + "n_folds": n_folds, + "seed": seed, + "fold_losses": folds, + **losses, + } + + +def _align_on_household_id( + incumbent_weights: pd.DataFrame, + incumbent_metrics: pd.DataFrame, +) -> tuple[pd.DataFrame, pd.DataFrame]: + """Join the two incumbent tables on ``household_id``. + + A row-count check passes for any permutation of the right size, so pairing + these tables by position scores the incumbent from mismatched households + and reports it as a win or a loss rather than as an error + (``uk-data#468``). The join is the check: it must be total and unique on + both sides, and both tables are returned on one household order. + """ + + aligned: list[pd.DataFrame] = [] + for label, frame in ( + ("incumbent wide weights", incumbent_weights), + ("incumbent household metrics", incumbent_metrics), + ): + if _HOUSEHOLD_ID_COLUMN not in frame.columns: + raise ValueError( + f"{label} must carry a {_HOUSEHOLD_ID_COLUMN!r} column; " + "positional pairing is not a join." + ) + keys = frame[_HOUSEHOLD_ID_COLUMN] + if keys.isna().any(): + raise ValueError(f"{label} has missing {_HOUSEHOLD_ID_COLUMN} values.") + indexed = frame.set_index(_HOUSEHOLD_ID_COLUMN) + if not indexed.index.is_unique: + duplicates = sorted( + {str(key) for key in indexed.index[indexed.index.duplicated()]} + ) + raise ValueError( + f"{label} repeats {_HOUSEHOLD_ID_COLUMN!r} value(s): {duplicates[:10]}." + ) + aligned.append(indexed) + + weights, metrics = aligned + if set(weights.index) != set(metrics.index): + missing = sorted({str(key) for key in set(weights.index) - set(metrics.index)}) + extra = sorted({str(key) for key in set(metrics.index) - set(weights.index)}) + raise ValueError( + "incumbent weights and household metrics must cover the same " + f"households; missing_from_metrics={missing[:10]}, " + f"missing_from_weights={extra[:10]}." + ) + order = weights.index + return weights, metrics.reindex(order) + + +def _incumbent_estimates( + registry: TargetRegistry, + incumbent_weights: pd.DataFrame, + incumbent_metrics: pd.DataFrame, +) -> dict[str, float]: + incumbent_weights, incumbent_metrics = _align_on_household_id( + incumbent_weights, + incumbent_metrics, + ) + estimates: dict[str, float] = {} + for spec in registry.specs: + area = str(spec.metadata.get("ledger_geography_id") or "") + if not area: + raise ValueError( + f"local target {spec.to_target().row_name!r} has no " + "ledger_geography_id metadata." + ) + if area not in incumbent_weights.columns: + raise ValueError(f"incumbent wide weights are missing area {area!r}.") + if spec.measure not in incumbent_metrics.columns: + raise ValueError( + f"incumbent household metrics are missing measure {spec.measure!r}." + ) + weights = incumbent_weights[area].to_numpy(dtype=np.float64) + metric = incumbent_metrics[spec.measure].to_numpy(dtype=np.float64) + if ( + not np.isfinite(weights).all() + or (weights < 0.0).any() + or not np.isfinite(metric).all() + ): + raise ValueError( + f"incumbent inputs for {spec.to_target().row_name!r} are invalid." + ) + estimates[spec.to_target().row_name] = float(np.dot(weights, metric)) + return estimates + + +def _relative_errors(estimates: np.ndarray, targets: np.ndarray) -> np.ndarray: + """Signed, uncapped per-row misses on the canonical row scale. + + A different quantity from the aggregate objective — signed and uncapped, + for the drift rows and the head-to-head counters — but deliberately not a + different *scale*: the denominator is imported from + :func:`default_target_loss_scales` rather than restated, so these rows + cannot drift away from the aggregates printed beside them. + """ + + return (estimates - targets) / default_target_loss_scales(targets) + + +def score_uk_local_candidate( + *, + candidate_diagnostics: Mapping[str, object], + incumbent_weights: pd.DataFrame, + incumbent_metrics: pd.DataFrame, + target_registry: TargetRegistry, + expected_reference_count: int = UK_LOCAL_ACTIVE_REFERENCE_COUNT, + target_period: int = UK_LOCAL_SCORE_TARGET_PERIOD, +) -> dict[str, Any]: + """Score both sides on the exact frozen local target surface.""" + + if target_registry.country != "uk": + raise ValueError("UK local scoring requires a UK TargetRegistry.") + if len(target_registry) != expected_reference_count: + raise ValueError( + "UK local scoring requires the frozen active reference count " + f"{expected_reference_count}, got {len(target_registry)}." + ) + wrong_period = [ + spec.to_target().row_name + for spec in target_registry.specs + if spec.period != target_period + ] + if wrong_period: + raise ValueError( + f"UK local scoring requires target period {target_period}; " + f"mismatches={wrong_period[:10]}." + ) + holdout = _candidate_holdout(candidate_diagnostics) + candidate = _candidate_estimates(candidate_diagnostics, target_registry) + incumbent = _incumbent_estimates( + target_registry, + incumbent_weights, + incumbent_metrics, + ) + + families: dict[str, dict[str, int]] = {} + drift: list[dict[str, object]] = [] + targets = np.array([spec.value for spec in target_registry.specs], dtype=np.float64) + candidate_estimates = np.array( + [candidate[spec.to_target().row_name] for spec in target_registry.specs], + dtype=np.float64, + ) + incumbent_estimates = np.array( + [incumbent[spec.to_target().row_name] for spec in target_registry.specs], + dtype=np.float64, + ) + candidate_errors = _relative_errors(candidate_estimates, targets) + incumbent_errors = _relative_errors(incumbent_estimates, targets) + candidate_wins = 0 + incumbent_wins = 0 + for index, spec in enumerate(target_registry.specs): + name = spec.to_target().row_name + candidate_error = float(candidate_errors[index]) + incumbent_error = float(incumbent_errors[index]) + bucket = families.setdefault( + spec.family, + {"candidate_target_wins": 0, "incumbent_target_wins": 0, "ties": 0}, + ) + if abs(candidate_error) < abs(incumbent_error): + winner = "candidate" + candidate_wins += 1 + bucket["candidate_target_wins"] += 1 + elif abs(incumbent_error) < abs(candidate_error): + winner = "incumbent" + incumbent_wins += 1 + bucket["incumbent_target_wins"] += 1 + else: + winner = "tie" + bucket["ties"] += 1 + drift.append( + { + "target": name, + "family": spec.family, + "candidate_relative_error": candidate_error, + "incumbent_relative_error": incumbent_error, + "winner": winner, + } + ) + # Both aggregates, and the holdout the candidate driver measured, go + # through the one canonical objective at the one declared doctrine cap, + # so the numbers in this receipt are on a single scale and stay there + # when microcosm#762 adjudicates the cap. + candidate_loss = relative_error_loss( + candidate_estimates, + targets, + target_loss_cap=UK_LOCAL_TARGET_LOSS_CAP, + ) + incumbent_loss = relative_error_loss( + incumbent_estimates, + targets, + target_loss_cap=UK_LOCAL_TARGET_LOSS_CAP, + ) + return { + "candidate_fitted_surface_loss": candidate_loss, + "candidate_holdout_loss": holdout["mean_holdout_loss"], + "incumbent_fitted_surface_loss": incumbent_loss, + "incumbent_holdout_loss": None, + "candidate_target_wins": candidate_wins, + "incumbent_target_wins": incumbent_wins, + "holdout_basis": holdout["basis"], + "incumbent_holdout_basis": UK_LOCAL_INCUMBENT_HOLDOUT_BASIS, + "candidate_holdout": holdout, + "loss": { + # Names the function that actually produced every loss above, + # including the candidate's holdout, at its declared cap. + "objective": "microcosm.calibrate.relative_error_loss", + "target_loss_cap": UK_LOCAL_TARGET_LOSS_CAP, + "target_loss_cap_source": "UK_LOCAL_TARGET_LOSS_CAP", + # The head-to-head counters below compare both sides on the + # surface the candidate was fitted to; only the candidate has a + # held-out measurement, and it is reported beside them, never as + # part of them. + "head_to_head_surface": "candidate_fitted_surface", + }, + "register": { + "country": target_registry.country, + "version": target_registry.version, + "n_specs": len(target_registry), + "target_period": target_period, + "surface": "active_local_reference", + }, + "target_wins_by_family": families, + "target_drift": drift, + } + + +def _read_json(path: Path) -> dict[str, object]: + payload = json.loads(path.read_text(encoding="utf-8")) + if not isinstance(payload, dict): + raise ValueError(f"{path} must contain a JSON object.") + return payload + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--candidate-diagnostics-json", required=True, type=Path) + parser.add_argument("--candidate-diagnostics-sha256", required=True) + parser.add_argument("--incumbent-weights-csv", required=True, type=Path) + parser.add_argument("--incumbent-weights-sha256", required=True) + parser.add_argument("--incumbent-household-metrics-csv", required=True, type=Path) + parser.add_argument("--incumbent-household-metrics-sha256", required=True) + parser.add_argument("--registry-json", required=True, type=Path) + parser.add_argument("--registry-sha256", required=True) + parser.add_argument("--output-json", required=True, type=Path) + args = parser.parse_args(argv) + + artifacts = { + "candidate_diagnostics": _verify_artifact( + args.candidate_diagnostics_json, + args.candidate_diagnostics_sha256, + ), + "incumbent_wide_weights": _verify_artifact( + args.incumbent_weights_csv, + args.incumbent_weights_sha256, + ), + "incumbent_household_metrics": _verify_artifact( + args.incumbent_household_metrics_csv, + args.incumbent_household_metrics_sha256, + ), + "target_registry": _verify_artifact( + args.registry_json, + args.registry_sha256, + ), + } + score = score_uk_local_candidate( + candidate_diagnostics=_read_json(args.candidate_diagnostics_json), + incumbent_weights=pd.read_csv(args.incumbent_weights_csv), + incumbent_metrics=pd.read_csv(args.incumbent_household_metrics_csv), + target_registry=TargetRegistry.from_json(args.registry_json), + ) + score["artifacts"] = artifacts + args.output_json.write_text( + json.dumps(score, indent=2, sort_keys=True, allow_nan=False), + encoding="utf-8", + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())