diff --git a/changelog.d/679-uk-frs-derived-stages.added.md b/changelog.d/679-uk-frs-derived-stages.added.md new file mode 100644 index 00000000..d17bbfbc --- /dev/null +++ b/changelog.d/679-uk-frs-derived-stages.added.md @@ -0,0 +1 @@ +Add declarative UK FRS derived source stages for employment, council tax, disability, education, legacy-benefit proxies, and education-grant splitting. diff --git a/packages/microcosm-build/src/microcosm/build/source_manifest.py b/packages/microcosm-build/src/microcosm/build/source_manifest.py index 899226d5..a4845005 100644 --- a/packages/microcosm-build/src/microcosm/build/source_manifest.py +++ b/packages/microcosm-build/src/microcosm/build/source_manifest.py @@ -96,6 +96,7 @@ "impute_child_support_to_puf_support", "impute_disability_benefits_to_puf_support", "impute_energy_subsidy_to_puf_support", + "impute_cell_means", "impute_housing_assistance_to_puf_support", "impute_other_health_insurance_premiums_to_puf_support", "impute_prior_year_income_to_puf_support", @@ -192,6 +193,7 @@ class SourceStageSpec: operations: tuple[SourceOperationSpec, ...] outputs: tuple[str, ...] nonnegative_outputs: tuple[str, ...] = () + rewrites: tuple[str, ...] = () notes: str = "" @classmethod @@ -217,6 +219,9 @@ def from_mapping(cls, raw: Mapping[str, Any]) -> SourceStageSpec: key="nonnegative_outputs", ) ) + rewrites = tuple( + _require_string_sequence(raw.get("rewrites", ()), key="rewrites") + ) unknown_nonnegative = sorted(set(nonnegative_outputs) - set(outputs)) if unknown_nonnegative: raise ValueError( @@ -237,6 +242,7 @@ def from_mapping(cls, raw: Mapping[str, Any]) -> SourceStageSpec: operations=operations, outputs=outputs, nonnegative_outputs=nonnegative_outputs, + rewrites=rewrites, notes=notes, ) diff --git a/packages/microcosm-build/src/microcosm/build/uk/gates.json b/packages/microcosm-build/src/microcosm/build/uk/gates.json index 3ba0a828..89c4ff7d 100644 --- a/packages/microcosm-build/src/microcosm/build/uk/gates.json +++ b/packages/microcosm-build/src/microcosm/build/uk/gates.json @@ -101,6 +101,14 @@ "evidence_absent_blocks": true, "notes": "Every fit-produced weight column carries a completed audit record. Armed by the SPI income stage; an absent audit is not a passing audit — and blocks every posture, not only release candidates: the legacy schema-3 path's strictness, ported during the #654 retirement (microcosm#691 review)." }, + { + "id": "uk_nonnegative_columns", + "gate": "nonnegative_columns", + "phase": "terminal", + "criticality": "release_blocking", + "parameters": {}, + "notes": "Every UK source-stage column declared in source_stages.json nonnegative_outputs must be finite and non-negative on the terminal frame; this makes the shared nonnegative column gate live for UK." + }, { "id": "uk_export_surface", "gate": "export_surface", diff --git a/packages/microcosm-build/src/microcosm/build/uk/source_stages.json b/packages/microcosm-build/src/microcosm/build/uk/source_stages.json index 57d99f7f..64ad4a14 100644 --- a/packages/microcosm-build/src/microcosm/build/uk/source_stages.json +++ b/packages/microcosm-build/src/microcosm/build/uk/source_stages.json @@ -6,7 +6,7 @@ { "stage": "frs_spine", "survey": "Family Resources Survey 2023-24", - "source": "Department for Work and Pensions Family Resources Survey 2023-24, UK Data Service series, local licensed 2023_24 tabs; vintage DOI was not found in the local reference metadata reviewed for this port.", + "source": "Department for Work and Pensions Family Resources Survey 2023-24, UK Data Service SN 9367, DOI 10.5255/UKDA-SN-9367-2; local licensed 2023_24 tabs.", "grain": "household", "artifacts": [ { @@ -263,7 +263,6 @@ "is_married", "dependent_children", "household_id", - "household_weight", "region", "tenure_type", "accommodation_type", @@ -283,7 +282,6 @@ "external_child_payments" ], "nonnegative_outputs": [ - "household_weight", "age", "hours_worked", "employment_income", @@ -310,6 +308,271 @@ ], "notes": "Root E2 spine assembly. It carries direct raw mappings only; education-grant aggregate and council-tax reported fields remain raw carriers for E3. Benefit take-up, BRMA/LHA assignment, stochastic flags, and imputations are intentionally absent from this stage." }, + { + "stage": "frs_employment", + "survey": "Family Resources Survey 2023-24", + "source": "Department for Work and Pensions Family Resources Survey 2023-24, UK Data Service SN 9252 local licensed tabs.", + "grain": "person", + "artifacts": [ + { + "role": "frs_table", + "table": "adult", + "kind": "licensed_microdata", + "format": "tab", + "vintage": "2023_24", + "locator": "adult.tab", + "sha256": "e09f9647d03585c81a528636028b2ed495f8f1fbcf64c5e7b4fe521b67367e06", + "size_bytes": 35323384, + "runtime_sha256_required": true + } + ], + "operations": [ + { + "kind": "read_tables", + "format": "tab", + "delimiter": "\t", + "lowercase_columns": true, + "numeric_errors": "coerce", + "runtime_sha256_required": true + }, + { + "kind": "map_coded_amounts", + "scope": "empstati employment status, mjobsect sector, and raw SIC division" + } + ], + "outputs": [ + "employment_status", + "employment_sector", + "sic_industry_division" + ], + "nonnegative_outputs": [ + "sic_industry_division" + ], + "notes": "Ports FRS employment derivations. empstati code 11 preserves the incumbent truncated-map artifact as LONG_TERM_DISABLED, so OTHER_INACTIVE is not emitted; mjobsect and sic are direct-indexed and fail loudly if absent." + }, + { + "stage": "frs_council_tax", + "survey": "Family Resources Survey 2023-24", + "source": "Department for Work and Pensions Family Resources Survey 2023-24, UK Data Service SN 9252 local licensed tabs.", + "grain": "household", + "artifacts": [ + { + "role": "frs_table", + "table": "househol", + "kind": "licensed_microdata", + "format": "tab", + "vintage": "2023_24", + "locator": "househol.tab", + "sha256": "5fd26b8b675f33b3b30c9ac789a18da17de734790f77e00ded287d1c3a187b30", + "size_bytes": 12387117, + "runtime_sha256_required": true + } + ], + "operations": [ + { + "kind": "read_tables", + "format": "tab", + "delimiter": "\t", + "lowercase_columns": true, + "numeric_errors": "coerce", + "runtime_sha256_required": true + }, + { + "kind": "impute_cell_means", + "cells": [ + "gvtregno", + "ctband", + "adulth == 1" + ], + "donor_filter": "raw ctannual > 0", + "missing": "raw ctannual < 0 or NaN", + "value": "Scottish-water-netted CTANNUAL" + } + ], + "outputs": [ + "council_tax" + ], + "nonnegative_outputs": [ + "council_tax" + ], + "notes": "Re-reads raw househol.tab because spine council_tax_reported clips missing values. Scottish Water charges are netted before cell means; no-donor cells impute zero. The dead ct_mean.replace(-1, ...) branch is intentionally dropped." + }, + { + "stage": "frs_disability", + "survey": "Family Resources Survey 2023-24", + "source": "Department for Work and Pensions Family Resources Survey 2023-24, UK Data Service SN 9252 local licensed tabs plus policyengine-uk DWP parameters.", + "grain": "person", + "artifacts": [], + "operations": [ + { + "kind": "derive", + "parameters": "disability category thresholds from baseline.gov.dwp at Jan-1" + }, + { + "kind": "derive", + "parameters": "disability flags from gov.dwp at Jan-1" + } + ], + "outputs": [ + "aa_category", + "dla_sc_category", + "dla_m_category", + "pip_m_category", + "pip_dl_category", + "is_disabled_for_benefits", + "is_enhanced_disabled_for_benefits", + "is_severely_disabled_for_benefits" + ], + "notes": "Consumes E2 reported disability amount carriers. The five internal amount carriers are retained through E7 and stripped at E10; export allowlists stay fail-closed." + }, + { + "stage": "frs_education", + "survey": "Family Resources Survey 2023-24", + "source": "Department for Work and Pensions Family Resources Survey 2023-24, UK Data Service SN 9252 local licensed tabs.", + "grain": "person", + "artifacts": [ + { + "role": "frs_table", + "table": "adult", + "kind": "licensed_microdata", + "format": "tab", + "vintage": "2023_24", + "locator": "adult.tab", + "sha256": "e09f9647d03585c81a528636028b2ed495f8f1fbcf64c5e7b4fe521b67367e06", + "size_bytes": 35323384, + "runtime_sha256_required": true + }, + { + "role": "frs_table", + "table": "child", + "kind": "licensed_microdata", + "format": "tab", + "vintage": "2023_24", + "locator": "child.tab", + "sha256": "b5dc84fe8b002ee925e61fae23fed27b11537af9fb174f1d07d9cc1748b9702e", + "size_bytes": 2913156, + "runtime_sha256_required": true + } + ], + "operations": [ + { + "kind": "read_tables", + "format": "tab", + "delimiter": "\t", + "lowercase_columns": true, + "numeric_errors": "coerce", + "runtime_sha256_required": true + }, + { + "kind": "derive", + "scope": "current/highest education, QYP inputs, EMA cell-mean degenerate fills, and benefits-in-own-right flag" + }, + { + "kind": "impute_cell_means", + "cells": [ + "single cell: EMA participants (code == 1; adema/ademaamt pair, eduma/edumaamt when adema is absent; chema/chemaamt for children)" + ], + "donor_filter": "participants with non-negative reported amounts", + "missing": "participants with sentinel negative reported amounts", + "value": "donor-mean fill, floored at zero, annualized with 365.25 / 7" + } + ], + "outputs": [ + "current_education", + "highest_education", + "is_in_non_advanced_education", + "is_in_approved_training", + "age_started_or_accepted_current_education_or_training", + "is_before_universal_credit_qualifying_young_person_terminal_date", + "adult_ema", + "child_ema", + "receives_benefits_in_own_right" + ], + "nonnegative_outputs": [ + "adult_ema", + "child_ema", + "age_started_or_accepted_current_education_or_training" + ], + "notes": "Ports the incumbent education cascade including its unreachable POST_SECONDARY branch order. EDUCQUAL_MAP carries the corrected highest-qualification codeframe (1 = Doctorate, descending) per the FRS 2023-24 data dictionary (UK Data Service SN 9367, DOI 10.5255/UKDA-SN-9367-2, adult table), corroborated against the raw aggregates and adopted as a signed difference (PR #703); an upstream defect report records the incumbent inversion. Code 87 is undocumented in the dictionary and falls to the default. EMA uses the shared weeks-in-year constant." + }, + { + "stage": "frs_legacy_proxies", + "survey": "Family Resources Survey 2023-24", + "source": "Department for Work and Pensions Family Resources Survey 2023-24, UK Data Service SN 9252 local licensed tabs plus policyengine-uk DWP parameters.", + "grain": "person", + "artifacts": [ + { + "role": "frs_table", + "table": "adult", + "kind": "licensed_microdata", + "format": "tab", + "vintage": "2023_24", + "locator": "adult.tab", + "sha256": "e09f9647d03585c81a528636028b2ed495f8f1fbcf64c5e7b4fe521b67367e06", + "size_bytes": 35323384, + "runtime_sha256_required": true + } + ], + "operations": [ + { + "kind": "read_tables", + "format": "tab", + "delimiter": "\t", + "lowercase_columns": true, + "numeric_errors": "coerce", + "runtime_sha256_required": true + }, + { + "kind": "materialize_rules_engine_predictors", + "predictors": [ + "state_pension_age" + ], + "consumed_only": true + }, + { + "kind": "derive", + "scope": "legacy JSA and ESA claimant-state proxies" + } + ], + "outputs": [ + "legacy_jobseeker_proxy", + "esa_health_condition_proxy", + "esa_support_group_proxy" + ], + "notes": "The proxies are labels, not entitlement determinations. JSA hours compare against 16 * (365.25 / 7) on the E2 spine hours scale; state_pension_age is consumed but not persisted." + }, + { + "stage": "frs_education_grant_split", + "survey": "Family Resources Survey 2023-24", + "source": "Department for Work and Pensions Family Resources Survey 2023-24, UK Data Service SN 9252 local licensed tabs plus policyengine-uk DfE grant parameters.", + "grain": "person", + "artifacts": [], + "operations": [ + { + "kind": "materialize_rules_engine_predictors", + "predictors": [ + "childcare_grant", + "parents_learning_allowance", + "adult_dependants_grant" + ], + "consumed_only": true + }, + { + "kind": "derive", + "scope": "proportional split of aggregate education_grants and DSA residual capacity" + } + ], + "outputs": [ + "disabled_students_allowance_eligible_expenses" + ], + "rewrites": [ + "education_grants" + ], + "nonnegative_outputs": [ + "disabled_students_allowance_eligible_expenses" + ], + "notes": "Runs before BRMA and always runs; both are signed inert differences for 2023-24. Pre-2025 DSA capacity is an aligned zero vector rather than an engine shape-sizing read." + }, { "stage": "frs_hmrc_retained_leaves", "survey": "Family Resources Survey 2023-24", @@ -1011,7 +1274,7 @@ "support_channel_column": "person_support_channel", "required_support_channel": "spi", "mass_share_denominator": "all_person_effective_mass", - "minimum_nondefault_mass_share": 1e-06, + "minimum_nondefault_mass_share": 0.000001, "fail_below_floor": true } ], 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 17980279..609b8d43 100644 --- a/packages/microcosm-build/src/microcosm/build/uk_runtime/__init__.py +++ b/packages/microcosm-build/src/microcosm/build/uk_runtime/__init__.py @@ -62,6 +62,48 @@ UK_CGT_TARGET_SPECS, UK_FISCAL_TARGET_REGISTRY, ) +from microcosm.build.uk_runtime.frs_council_tax import ( + FRS_COUNCIL_TAX_OUTPUT_COLUMNS, + UKFRSCouncilTaxStageTransform, + add_frs_council_tax, + derive_council_tax, +) +from microcosm.build.uk_runtime.frs_disability import ( + FRS_DISABILITY_OUTPUT_COLUMNS, + UK_INTERNAL_DISABILITY_REPORTED_COLUMNS, + UKDWPBaselineDisabilityRates, + UKDWPDisabilityFlagRates, + UKFRSDisabilityStageTransform, + add_frs_disability, + derive_frs_disability, + uk_dwp_baseline_disability_rates, + uk_dwp_disability_flag_rates, +) +from microcosm.build.uk_runtime.frs_education import ( + EDUCQUAL_MAP, + FRS_EDUCATION_OUTPUT_COLUMNS, + UKFRSEducationStageTransform, + add_frs_education, + derive_current_education, + derive_frs_education, +) +from microcosm.build.uk_runtime.frs_education_grants import ( + FRS_EDUCATION_GRANT_OUTPUT_COLUMNS, + FRS_EDUCATION_GRANT_REWRITES, + UK_EDUCATION_GRANT_CAPACITY_PREDICTORS, + UKDSAPolicy, + UKFRSEducationGrantSplitStageTransform, + add_frs_education_grant_split, + allocate_reported_education_grants, + disabled_students_allowance_capacity, + uk_dsa_policy, +) +from microcosm.build.uk_runtime.frs_employment import ( + FRS_EMPLOYMENT_OUTPUT_COLUMNS, + UKFRSEmploymentStageTransform, + add_frs_employment, + derive_frs_employment, +) from microcosm.build.uk_runtime.frs_hmrc_leaves import ( FRS_HMRC_INCPBEN_COLUMN, FRS_HMRC_OSSBEN_IDENTIFIABLE_SUBSET_COLUMN, @@ -74,6 +116,14 @@ UKFRSHMRCRetainedLeavesStageTransform, retain_uk_frs_hmrc_leaves, ) +from microcosm.build.uk_runtime.frs_legacy_proxies import ( + FRS_LEGACY_PROXY_OUTPUT_COLUMNS, + UK_LEGACY_PROXY_PREDICTORS, + UKFRSLegacyProxiesStageTransform, + UKLegacyJSAPolicy, + derive_frs_legacy_proxies, + uk_legacy_jsa_policy, +) from microcosm.build.uk_runtime.geography_ladder import ( GEOGRAPHY_LADDER_ARTIFACT_SHA256_ATTR, GEOGRAPHY_LADDER_VINTAGES_ATTR, @@ -617,12 +667,25 @@ "compute_household_metrics", "constituency_household_targets", "create_uk_spi_support_tables", + "derive_council_tax", + "derive_current_education", + "derive_frs_disability", + "derive_frs_education", + "derive_frs_employment", + "derive_frs_legacy_proxies", "employment_band_name", "fill_support_channel_from_source", "format_uk_release_id", "impute_uk_spi_income_support", "replace_uk_spi_support_tables", "retain_uk_frs_hmrc_leaves", + "add_frs_council_tax", + "add_frs_disability", + "add_frs_education", + "add_frs_education_grant_split", + "add_frs_employment", + "allocate_reported_education_grants", + "disabled_students_allowance_capacity", "generate_base_firms", "generate_input_values", "generate_uk_firm_population", @@ -683,6 +746,10 @@ "uk_firm_source_data_from_ledger_facts", "uk_calibration_diagnostics_payload", "uk_frame_content_identity", + "uk_dsa_policy", + "uk_dwp_baseline_disability_rates", + "uk_dwp_disability_flag_rates", + "uk_legacy_jsa_policy", "uk_geography_ladder_assignment_summary", "uk_geography_ladder_gate", "uk_release_input_coverage_gate", @@ -712,9 +779,30 @@ "write_uk_rowwise_dataset", "write_hmrc_replay_report", "UK_DEFAULT_ZERO_WEIGHT_STRATA", + "EDUCQUAL_MAP", + "FRS_COUNCIL_TAX_OUTPUT_COLUMNS", + "FRS_DISABILITY_OUTPUT_COLUMNS", + "FRS_EDUCATION_GRANT_OUTPUT_COLUMNS", + "FRS_EDUCATION_GRANT_REWRITES", + "FRS_EDUCATION_OUTPUT_COLUMNS", + "FRS_EMPLOYMENT_OUTPUT_COLUMNS", + "FRS_LEGACY_PROXY_OUTPUT_COLUMNS", "UK_MAX_TARGET_ABS_RELATIVE_ERROR", "UKInputMassParityPolicy", + "UKDSAPolicy", + "UKDWPBaselineDisabilityRates", + "UKDWPDisabilityFlagRates", + "UKFRSCouncilTaxStageTransform", + "UKFRSDisabilityStageTransform", + "UKFRSEducationGrantSplitStageTransform", + "UKFRSEducationStageTransform", + "UKFRSEmploymentStageTransform", + "UKFRSLegacyProxiesStageTransform", "UKInputMassReference", + "UKLegacyJSAPolicy", + "UK_INTERNAL_DISABILITY_REPORTED_COLUMNS", + "UK_EDUCATION_GRANT_CAPACITY_PREDICTORS", + "UK_LEGACY_PROXY_PREDICTORS", "UKQRFTailConcentrationPolicy", "UKZeroWeightStratumDeclaration", "load_uk_input_mass_reference", 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 19514fdb..8eae3616 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 @@ -38,13 +38,18 @@ EvidenceContext, GateBinding, ) -from microcosm.build.gates import GateResult, weights_audit_gate +from microcosm.build.gates import ( + GateResult, + nonnegative_columns_gate, + weights_audit_gate, +) from microcosm.build.uk_runtime.national_frame import _uk_gate_surface from microcosm.build.uk_runtime.release_input_coverage import ( assert_uk_release_input_coverage_build_stages, assert_uk_release_input_coverage_manifest_current, uk_release_input_coverage_gate, ) +from microcosm.build.uk_runtime.source_runtime import UK_NONNEGATIVE_OUTPUTS_BY_STAGE from microcosm.build.uk_runtime.terminal_gates import ( UKZeroWeightStratumDeclaration, _household_weights, @@ -189,6 +194,39 @@ def _evaluate_source_coverage( ) +def _evaluate_nonnegative_columns( + context: EvidenceContext, parameters: Mapping[str, Any] +) -> GateResult: + if parameters: + raise ValueError( + "uk_nonnegative_columns takes no parameters; the checked columns " + "come from source_stages.json nonnegative_outputs." + ) + # The required set is every nonnegative output declared by a stage the + # build actually scheduled — absence of a scheduled stage's declared + # column is a failure (the shared gate's missing-column path), while + # unscheduled stages' columns are not demanded. Never pre-filter to + # present columns: that would let a frame missing every declared output + # pass with columns_required=0. + stage_names = tuple(context.artifacts["build_stage_names"]) + required = tuple( + dict.fromkeys( + column + for stage in stage_names + for column in UK_NONNEGATIVE_OUTPUTS_BY_STAGE.get(str(stage), ()) + ) + ) + column_values: dict[str, Any] = {} + for entity in context.frame.entities: + table = context.frame.table(entity) + for column in table.columns: + column_values.setdefault(str(column), table[column]) + return nonnegative_columns_gate( + column_values, + required, + ) + + def _stage_names_evidence( context: EvidenceContext, parameters: Mapping[str, Any] ) -> object: @@ -474,6 +512,11 @@ def _evaluate_tail_concentration( needs_frame=False, evidence=_stage_names_evidence, ), + "nonnegative_columns": UKGateBinding( + name="nonnegative_columns", + evaluator=_evaluate_nonnegative_columns, + artifact_keys=frozenset({"build_stage_names"}), + ), "degenerate_release_surface": UKGateBinding( name="degenerate_release_surface", evaluator=_evaluate_degenerate_release_surface, diff --git a/packages/microcosm-build/src/microcosm/build/uk_runtime/frs_council_tax.py b/packages/microcosm-build/src/microcosm/build/uk_runtime/frs_council_tax.py new file mode 100644 index 00000000..d9b2c550 --- /dev/null +++ b/packages/microcosm-build/src/microcosm/build/uk_runtime/frs_council_tax.py @@ -0,0 +1,110 @@ +"""FRS council-tax cell-mean imputation.""" + +from __future__ import annotations + +from collections.abc import Mapping +from pathlib import Path +from typing import Any + +import numpy as np +import pandas as pd + +from microcosm.build.source_manifest import SourceStageSpec +from microcosm.build.uk_runtime.frs_spine import ( + WEEKS_IN_YEAR, + normalize_ids, + read_pinned_tab, +) +from microcosm.build.uk_runtime.national_frame import ( + uk_household_weight_kind, + uk_national_frame, + uk_time_period, + validate_uk_national_frame, +) +from microcosm.frame import Frame + +FRS_COUNCIL_TAX_OUTPUT_COLUMNS = ("council_tax",) +SCOTLAND_GVTREGNO = 12 + + +class UKFRSCouncilTaxStageTransform: + """Whole-stage callable for FRS council-tax imputation.""" + + def __init__(self, raw_dir: str | Path, *, stage: SourceStageSpec) -> None: + self.raw_dir = Path(raw_dir) + self.stage = stage + + def __call__(self, frame: Frame) -> Frame: + return add_frs_council_tax(frame, self.raw_dir, stage=self.stage) + + @staticmethod + def output_columns() -> tuple[str, ...]: + return FRS_COUNCIL_TAX_OUTPUT_COLUMNS + + +def add_frs_council_tax( + frame: Frame, raw_dir: str | Path, *, stage: SourceStageSpec +) -> Frame: + artifacts = _artifact_by_table(stage) + household_raw = normalize_ids( + read_pinned_tab( + Path(raw_dir) / str(artifacts["househol"]["locator"]), + artifacts["househol"], + ) + ) + household = frame.table("household").copy() + derived = derive_council_tax(household, household_raw) + household["council_tax"] = derived.reindex(household["household_id"]).to_numpy() + result = uk_national_frame( + person=frame.table("person"), + benunit=frame.table("benunit"), + household=household, + time_period=uk_time_period(frame), + weight_kind=uk_household_weight_kind(frame), + household_weights=frame.weights_for("household").values, + mass_log=frame.mass_log, + ) + validate_uk_national_frame(result) + return result + + +def derive_council_tax( + household: pd.DataFrame, household_raw: pd.DataFrame +) -> pd.Series: + """Impute raw missing council tax from raw region/band/adult cells.""" + + raw = household_raw.set_index("household_id") + aligned = raw.reindex(household["household_id"]) + ctannual = pd.to_numeric(aligned["ctannual"], errors="coerce") + gvtregno = pd.to_numeric(aligned["gvtregno"], errors="coerce") + ctband = pd.to_numeric(aligned["ctband"], errors="coerce") + single_adult = pd.to_numeric(aligned["adulth"], errors="coerce") == 1 + scottish_water = np.where( + gvtregno == SCOTLAND_GVTREGNO, + ( + np.maximum(pd.to_numeric(aligned["csewamt"], errors="coerce").fillna(0), 0) + + np.maximum( + pd.to_numeric(aligned["cwatamtd"], errors="coerce").fillna(0), 0 + ) + ) + * WEEKS_IN_YEAR, + 0.0, + ) + tax_only = pd.Series(np.maximum(ctannual - scottish_water, 0), index=aligned.index) + donors = ctannual > 0 + cell_mean = tax_only[donors].groupby( + [gvtregno[donors], ctband[donors], single_adult[donors]], dropna=False + ).mean() + keys = pd.MultiIndex.from_arrays([gvtregno, ctband, single_adult]) + imputed = pd.Series(keys.map(cell_mean).to_numpy(dtype=float), index=aligned.index) + imputed = imputed.fillna(0.0).clip(lower=0) + missing = (ctannual < 0) | ctannual.isna() + result = pd.Series(np.where(missing, imputed, tax_only), index=aligned.index) + return result.fillna(0.0).clip(lower=0) + + +def _artifact_by_table(stage: SourceStageSpec) -> dict[str, Mapping[str, Any]]: + by_table = {str(artifact.get("table")): artifact for artifact in stage.artifacts} + if "househol" not in by_table: + raise ValueError("frs_council_tax manifest is missing househol.tab.") + return by_table diff --git a/packages/microcosm-build/src/microcosm/build/uk_runtime/frs_disability.py b/packages/microcosm-build/src/microcosm/build/uk_runtime/frs_disability.py new file mode 100644 index 00000000..d3a8d0c6 --- /dev/null +++ b/packages/microcosm-build/src/microcosm/build/uk_runtime/frs_disability.py @@ -0,0 +1,287 @@ +"""FRS disability benefit category and flag derivations.""" + +from __future__ import annotations + +from dataclasses import dataclass + +import numpy as np +import pandas as pd + +from microcosm.build.source_manifest import SourceStageSpec +from microcosm.build.uk_runtime.frs_spine import WEEKS_IN_YEAR +from microcosm.build.uk_runtime.national_frame import ( + uk_household_weight_kind, + uk_national_frame, + uk_time_period, + validate_uk_national_frame, +) +from microcosm.frame import Frame + +UK_INTERNAL_DISABILITY_REPORTED_COLUMNS = ( + "attendance_allowance_reported", + "dla_sc_reported", + "dla_m_reported", + "pip_m_reported", + "pip_dl_reported", +) +UK_DISABILITY_FLAG_REPORTED_COLUMNS = ( + *UK_INTERNAL_DISABILITY_REPORTED_COLUMNS, + "sda_reported", + "incapacity_benefit_reported", + "iidb_reported", + "afcs_reported", + "esa_contrib_reported", + "esa_income_reported", +) +FRS_DISABILITY_OUTPUT_COLUMNS = ( + "aa_category", + "dla_sc_category", + "dla_m_category", + "pip_m_category", + "pip_dl_category", + "is_disabled_for_benefits", + "is_enhanced_disabled_for_benefits", + "is_severely_disabled_for_benefits", +) + + +@dataclass(frozen=True) +class UKDWPBaselineDisabilityRates: + """The incumbent categories' rates: the pre-fiscal-conversion baseline subtree.""" + + aa_lower: float + aa_higher: float + dla_sc_lower: float + dla_sc_middle: float + dla_sc_higher: float + dla_m_lower: float + dla_m_higher: float + pip_m_standard: float + pip_m_enhanced: float + pip_dl_standard: float + pip_dl_enhanced: float + instant: str + source: str + + +@dataclass(frozen=True) +class UKDWPDisabilityFlagRates: + """The incumbent flags' rates: the fiscal-converted plain tree.""" + + aa_higher: float + dla_sc_higher: float + pip_dl_enhanced: float + instant: str + source: str + + +class UKFRSDisabilityStageTransform: + """Whole-stage callable for FRS disability derivations.""" + + def __init__( + self, + *, + stage: SourceStageSpec, + baseline_rates: UKDWPBaselineDisabilityRates | None = None, + flag_rates: UKDWPDisabilityFlagRates | None = None, + ) -> None: + self.stage = stage + self.baseline_rates = baseline_rates + self.flag_rates = flag_rates + + def __call__(self, frame: Frame) -> Frame: + period = uk_time_period(frame) + return add_frs_disability( + frame, + baseline_rates=self.baseline_rates + or uk_dwp_baseline_disability_rates(period), + flag_rates=self.flag_rates or uk_dwp_disability_flag_rates(period), + ) + + @staticmethod + def output_columns() -> tuple[str, ...]: + return FRS_DISABILITY_OUTPUT_COLUMNS + + +def uk_dwp_baseline_disability_rates( + build_period: int | str, +) -> UKDWPBaselineDisabilityRates: + """Read the incumbent categories' rates: the ``.baseline`` subtree. + + The baseline-vs-plain split in the incumbent is value-bearing, not just + provenance: the ``baseline`` clone is created before policyengine-uk's + fiscal-year parameter conversion, so at build period 2023 it carries the + April-2022-era weekly rates (e.g. 92.40) that the incumbent's category + thresholds actually use, while the plain tree (the flags reader below) + carries the fiscal-2023-24 values (e.g. 101.75). + """ + + try: + import policyengine_uk + except ImportError as exc: + raise ImportError( + "UK DWP disability parameters require `uv sync --all-packages --extra uk`." + ) from exc + + instant = f"{int(build_period)}-01-01" + dwp = ( + policyengine_uk.CountryTaxBenefitSystem() + .parameters(int(build_period)) + .baseline.gov.dwp + ) + return UKDWPBaselineDisabilityRates( + aa_lower=_parameter_value(dwp.attendance_allowance.lower, instant), + aa_higher=_parameter_value(dwp.attendance_allowance.higher, instant), + dla_sc_lower=_parameter_value(dwp.dla.self_care.lower, instant), + dla_sc_middle=_parameter_value(dwp.dla.self_care.middle, instant), + dla_sc_higher=_parameter_value(dwp.dla.self_care.higher, instant), + dla_m_lower=_parameter_value(dwp.dla.mobility.lower, instant), + dla_m_higher=_parameter_value(dwp.dla.mobility.higher, instant), + pip_m_standard=_parameter_value(dwp.pip.mobility.standard, instant), + pip_m_enhanced=_parameter_value(dwp.pip.mobility.enhanced, instant), + pip_dl_standard=_parameter_value(dwp.pip.daily_living.standard, instant), + pip_dl_enhanced=_parameter_value(dwp.pip.daily_living.enhanced, instant), + instant=instant, + source="policyengine-uk parameters " + f"{getattr(policyengine_uk, '__version__', 'unknown')}", + ) + + +def uk_dwp_disability_flag_rates( + build_period: int | str, +) -> UKDWPDisabilityFlagRates: + """Read the incumbent flags' rates: the plain (fiscal-converted) tree. + + The incumbent's flag thresholds come from ``parameters(year).gov.dwp`` + on the model system, whose parameter tree policyengine-uk converts to + fiscal-year snapshots at load — at build period 2023 that is the + fiscal-2023-24 weekly rates (e.g. 101.75), NOT the raw dated files' + January-1 values (92.40). Reading the raw files here produced a + single-row flag divergence against the incumbent's own output, caught + by the licensed head-to-head receipt. + """ + + try: + import policyengine_uk + except ImportError as exc: + raise ImportError( + "UK DWP disability parameters require `uv sync --all-packages --extra uk`." + ) from exc + + instant = f"{int(build_period)}-01-01" + dwp = ( + policyengine_uk.CountryTaxBenefitSystem() + .parameters(int(build_period)) + .gov.dwp + ) + return UKDWPDisabilityFlagRates( + aa_higher=_parameter_value(dwp.attendance_allowance.higher, instant), + dla_sc_higher=_parameter_value(dwp.dla.self_care.higher, instant), + pip_dl_enhanced=_parameter_value(dwp.pip.daily_living.enhanced, instant), + instant=instant, + source="policyengine-uk parameters (fiscal-converted plain tree) " + f"{getattr(policyengine_uk, '__version__', 'unknown')}", + ) + + +def add_frs_disability( + frame: Frame, + *, + baseline_rates: UKDWPBaselineDisabilityRates, + flag_rates: UKDWPDisabilityFlagRates, +) -> Frame: + person = frame.table("person").copy() + derived = derive_frs_disability( + person, baseline_rates=baseline_rates, flag_rates=flag_rates + ) + for column in FRS_DISABILITY_OUTPUT_COLUMNS: + person[column] = derived[column].to_numpy() + result = uk_national_frame( + person=person, + benunit=frame.table("benunit"), + household=frame.table("household"), + time_period=uk_time_period(frame), + weight_kind=uk_household_weight_kind(frame), + household_weights=frame.weights_for("household").values, + mass_log=frame.mass_log, + ) + validate_uk_national_frame(result) + return result + + +def derive_frs_disability( + person: pd.DataFrame, + *, + baseline_rates: UKDWPBaselineDisabilityRates, + flag_rates: UKDWPDisabilityFlagRates, +) -> pd.DataFrame: + values = pd.DataFrame(index=person.index) + values["aa_category"] = _category( + _amount(person, "attendance_allowance_reported"), + (("LOWER", baseline_rates.aa_lower), ("HIGHER", baseline_rates.aa_higher)), + ) + values["dla_sc_category"] = _category( + _amount(person, "dla_sc_reported"), + ( + ("LOWER", baseline_rates.dla_sc_lower), + ("MIDDLE", baseline_rates.dla_sc_middle), + ("HIGHER", baseline_rates.dla_sc_higher), + ), + ) + values["dla_m_category"] = _category( + _amount(person, "dla_m_reported"), + (("LOWER", baseline_rates.dla_m_lower), ("HIGHER", baseline_rates.dla_m_higher)), + ) + values["pip_m_category"] = _category( + _amount(person, "pip_m_reported"), + ( + ("STANDARD", baseline_rates.pip_m_standard), + ("ENHANCED", baseline_rates.pip_m_enhanced), + ), + ) + values["pip_dl_category"] = _category( + _amount(person, "pip_dl_reported"), + ( + ("STANDARD", baseline_rates.pip_dl_standard), + ("ENHANCED", baseline_rates.pip_dl_enhanced), + ), + ) + total = sum(_amount(person, column) for column in UK_DISABILITY_FLAG_REPORTED_COLUMNS) + dla_sc = _amount(person, "dla_sc_reported") + aa = _amount(person, "attendance_allowance_reported") + pip_dl = _amount(person, "pip_dl_reported") + afcs = _amount(person, "afcs_reported") + gap = WEEKS_IN_YEAR + aa_higher = flag_rates.aa_higher * WEEKS_IN_YEAR - gap + dla_sc_higher = flag_rates.dla_sc_higher * WEEKS_IN_YEAR - gap + pip_dl_enhanced = flag_rates.pip_dl_enhanced * WEEKS_IN_YEAR - gap + values["is_disabled_for_benefits"] = total > 0 + values["is_enhanced_disabled_for_benefits"] = ( + (aa >= aa_higher) | (dla_sc > dla_sc_higher) | (pip_dl >= pip_dl_enhanced) + ) + values["is_severely_disabled_for_benefits"] = ( + (aa > 0) | (dla_sc >= dla_sc_higher) | (pip_dl >= pip_dl_enhanced) | (afcs > 0) + ) + return values + + +def _amount(person: pd.DataFrame, column: str) -> pd.Series: + if column not in person: + return pd.Series(0.0, index=person.index) + return pd.to_numeric(person[column], errors="coerce").fillna(0.0) + + +def _category( + reported_amount: pd.Series, thresholds: tuple[tuple[str, float], ...] +) -> np.ndarray: + weekly = reported_amount.to_numpy(dtype=float) / WEEKS_IN_YEAR + category = np.full(len(weekly), "NONE", dtype=object) + for name, weekly_rate in thresholds: + category[weekly >= max(0.0, float(weekly_rate) - 1.0)] = name + return category + + +def _parameter_value(value, instant: str) -> float: + if callable(value): + value = value(instant) + return float(value) diff --git a/packages/microcosm-build/src/microcosm/build/uk_runtime/frs_education.py b/packages/microcosm-build/src/microcosm/build/uk_runtime/frs_education.py new file mode 100644 index 00000000..0b60f8d3 --- /dev/null +++ b/packages/microcosm-build/src/microcosm/build/uk_runtime/frs_education.py @@ -0,0 +1,266 @@ +"""FRS education derivations for the UK source spine.""" + +from __future__ import annotations + +from collections.abc import Mapping +from pathlib import Path +from typing import Any + +import numpy as np +import pandas as pd + +from microcosm.build.source_manifest import SourceStageSpec +from microcosm.build.uk_runtime.frs_spine import ( + WEEKS_IN_YEAR, + normalize_ids, + read_pinned_tab, +) +from microcosm.build.uk_runtime.national_frame import ( + uk_household_weight_kind, + uk_national_frame, + uk_time_period, + validate_uk_national_frame, +) +from microcosm.frame import Frame + +NON_ADVANCED_EDUCATION_LEVELS = ( + "PRE_PRIMARY", + "PRIMARY", + "LOWER_SECONDARY", + "UPPER_SECONDARY", + "POST_SECONDARY", +) +FRS_APPROVED_TRAINING_CODES = tuple(range(1, 10)) +UNKNOWN_QUALIFYING_EDUCATION_OR_TRAINING_ENTRY_AGE = 1000 +BENEFITS_IN_OWN_RIGHT_REPORTED_COLUMNS = ( + "universal_credit_reported", + "jsa_contrib_reported", + "jsa_income_reported", + "esa_contrib_reported", + "esa_income_reported", +) +# FRS EDUCQUAL value labels per the official data dictionary: FRS 2023-24, +# UK Data Service SN 9367, DOI 10.5255/UKDA-SN-9367-2, adult table, EDUCQUAL +# (every mapped label below matches the dictionary verbatim). A signed +# difference vs the incumbent, whose map was inverted (low codes read as +# school-level, 17-21 as degrees); the raw aggregates corroborate — code 1 +# is 1.8% of adults with the highest mean earnings (Doctorate), 18/19 are +# near-empty niche baccalaureates, and the GCSE band carries the mass. +# Code 87 is undocumented in the dictionary (no value label) yet carried by +# ~13% of adults; it is deliberately unmapped and falls to the +# UPPER_SECONDARY fillna default, as the incumbent's default did. +EDUCQUAL_MAP = { + # Degree level and above (TERTIARY) + 1: "TERTIARY", # Doctorate or MPhil + 2: "TERTIARY", # Masters, PGCE or other postgrad + 3: "TERTIARY", # Degree inc foundation degree + 4: "TERTIARY", # Teaching qualification (excl PGCE) + 5: "TERTIARY", # Foreign qualification at degree level + 6: "TERTIARY", # Other work-related qual at degree level + 7: "TERTIARY", # Other professional qual at degree level + # Higher education below degree (POST_SECONDARY) + 8: "POST_SECONDARY", # Other HE qualification below degree + 9: "POST_SECONDARY", # Nursing or other medical + 10: "POST_SECONDARY", # Diploma in higher education + 11: "POST_SECONDARY", # HNC/HND + 12: "POST_SECONDARY", # BTEC higher level + 13: "POST_SECONDARY", # SCOTVEC higher level + 14: "POST_SECONDARY", # NVQ/SVQ Level 4 + 15: "POST_SECONDARY", # NVQ/SVQ Level 5 + 16: "POST_SECONDARY", # RSA higher diploma / OCR Level 4 + # A-level equivalent (UPPER_SECONDARY) + 17: "UPPER_SECONDARY", # A-Level or equivalent + 18: "UPPER_SECONDARY", # Welsh Baccalaureate Advanced + 19: "UPPER_SECONDARY", # Scottish Baccalaureate + 20: "UPPER_SECONDARY", # International Baccalaureate + 21: "UPPER_SECONDARY", # AS-level or equivalent + 22: "UPPER_SECONDARY", # Certificate of 6th Year Studies + 23: "UPPER_SECONDARY", # Access to Higher Education + 24: "UPPER_SECONDARY", # Scottish Higher/Intermediate + 25: "UPPER_SECONDARY", # Skills for work Higher + 26: "POST_SECONDARY", # ONC/OND + 27: "POST_SECONDARY", # BTEC National level + 28: "POST_SECONDARY", # SCOTVEC National level + 29: "UPPER_SECONDARY", # New Diploma Advanced + 30: "UPPER_SECONDARY", # New Diploma Progression + 31: "UPPER_SECONDARY", # NVQ/SVQ Level 3 + 32: "UPPER_SECONDARY", # GNVQ Advanced + 33: "UPPER_SECONDARY", # RSA advanced diploma / OCR Level 3 + 34: "UPPER_SECONDARY", # City and Guilds advanced craft + 35: "UPPER_SECONDARY", # Welsh Baccalaureate Intermediate + # GCSE/O-level equivalent and below-GCSE (LOWER_SECONDARY) + **{code: "LOWER_SECONDARY" for code in range(36, 83)}, + # Entry-level / basic skills + 83: "NOT_COMPLETED_PRIMARY", # Basic Skills (literacy/numeracy) + 84: "NOT_COMPLETED_PRIMARY", # Entry Level Qualifications + 85: "NOT_COMPLETED_PRIMARY", # Award/Certificate at entry level + 86: "LOWER_SECONDARY", # Other professional/vocational/foreign +} +FRS_EDUCATION_OUTPUT_COLUMNS = ( + "current_education", + "highest_education", + "is_in_non_advanced_education", + "is_in_approved_training", + "age_started_or_accepted_current_education_or_training", + "is_before_universal_credit_qualifying_young_person_terminal_date", + "adult_ema", + "child_ema", + "receives_benefits_in_own_right", +) + + +class UKFRSEducationStageTransform: + """Whole-stage callable for FRS education derivations.""" + + def __init__(self, raw_dir: str | Path, *, stage: SourceStageSpec) -> None: + self.raw_dir = Path(raw_dir) + self.stage = stage + + def __call__(self, frame: Frame) -> Frame: + return add_frs_education(frame, self.raw_dir, stage=self.stage) + + @staticmethod + def output_columns() -> tuple[str, ...]: + return FRS_EDUCATION_OUTPUT_COLUMNS + + +def add_frs_education( + frame: Frame, raw_dir: str | Path, *, stage: SourceStageSpec +) -> Frame: + artifacts = _artifact_by_table(stage) + adult = normalize_ids( + read_pinned_tab(Path(raw_dir) / str(artifacts["adult"]["locator"]), artifacts["adult"]) + ) + child = normalize_ids( + read_pinned_tab(Path(raw_dir) / str(artifacts["child"]["locator"]), artifacts["child"]) + ) + raw_person = pd.concat([adult, child], ignore_index=True, sort=False) + person = frame.table("person").copy() + derived = derive_frs_education(person, raw_person) + for column in FRS_EDUCATION_OUTPUT_COLUMNS: + person[column] = derived[column].to_numpy() + result = uk_national_frame( + person=person, + benunit=frame.table("benunit"), + household=frame.table("household"), + time_period=uk_time_period(frame), + weight_kind=uk_household_weight_kind(frame), + household_weights=frame.weights_for("household").values, + mass_log=frame.mass_log, + ) + validate_uk_national_frame(result) + return result + + +def derive_frs_education(person: pd.DataFrame, raw_person: pd.DataFrame) -> pd.DataFrame: + raw = raw_person.set_index("person_id").reindex(person["person_id"]) + values = pd.DataFrame(index=person.index) + age = pd.to_numeric(person["age"], errors="coerce").fillna(0).to_numpy() + fted_source = "fted" if "fted" in raw.columns else "educft" + fted = _num(raw, fted_source).to_numpy() + typeed2 = _num(raw, "typeed2").to_numpy() + current = derive_current_education(fted=fted, typeed2=typeed2, age=age) + values["current_education"] = current + values["highest_education"] = ( + _num(raw, "educqual").astype(int).map(EDUCQUAL_MAP).fillna("UPPER_SECONDARY") + ).to_numpy() + values["is_in_non_advanced_education"] = np.isin( + current, NON_ADVANCED_EDUCATION_LEVELS + ) + train = _num(raw, "train") if "train" in raw.columns else pd.Series(0, index=raw.index) + values["is_in_approved_training"] = train.isin(FRS_APPROVED_TRAINING_CODES).to_numpy() + in_qualifying = ( + values["is_in_non_advanced_education"].to_numpy(dtype=bool) + | values["is_in_approved_training"].to_numpy(dtype=bool) + ) + values["age_started_or_accepted_current_education_or_training"] = np.where( + in_qualifying, + np.minimum(age, 18), + UNKNOWN_QUALIFYING_EDUCATION_OR_TRAINING_ENTRY_AGE, + ) + values["is_before_universal_credit_qualifying_young_person_terminal_date"] = ( + (age == 19) & in_qualifying + ) + if "adema" in raw.columns: + values["adult_ema"] = _ema(raw, code_column="adema", amount_column="ademaamt") + else: + # Vintages without the adema pair carry eduma/edumaamt — the + # incumbent aliases them before its fill; 2023-24 is such a vintage. + values["adult_ema"] = _ema(raw, code_column="eduma", amount_column="edumaamt") + values["child_ema"] = _ema(raw, code_column="chema", amount_column="chemaamt") + benefit_columns = [ + column for column in BENEFITS_IN_OWN_RIGHT_REPORTED_COLUMNS if column in person + ] + values["receives_benefits_in_own_right"] = ( + person[benefit_columns].fillna(0).sum(axis=1) > 0 if benefit_columns else False + ) + return values + + +def derive_current_education(*, fted, typeed2, age) -> np.ndarray: + fted = np.asarray(fted) + typeed2 = np.asarray(typeed2) + age = np.asarray(age) + # The post-secondary branch is deliberately unreachable because the + # incumbent tests upper-secondary first for typeed2 7/8. + return np.select( + [ + np.isin(fted, (2, -1, 0)), + typeed2 == 1, + np.isin(typeed2, (2, 4)) + | (np.isin(typeed2, (3, 8)) & (age < 11)) + | ((typeed2 == 0) & (fted == 1) & (age > 5) & (age < 11)), + np.isin(typeed2, (5, 6)) + | (np.isin(typeed2, (3, 8)) & (age >= 11) & (age <= 16)) + | ((typeed2 == 0) & (fted == 1) & (age <= 16)), + (typeed2 == 7) + | (np.isin(typeed2, (3, 8)) & (age > 16)) + | ((typeed2 == 0) & (fted == 1) & (age > 16)), + np.isin(typeed2, (7, 8)) & (age >= 19), + (typeed2 == 9) | ((typeed2 == 0) & (fted == 1) & (age >= 19)), + ], + [ + "NOT_IN_EDUCATION", + "PRE_PRIMARY", + "PRIMARY", + "LOWER_SECONDARY", + "UPPER_SECONDARY", + "POST_SECONDARY", + "TERTIARY", + ], + default="NOT_IN_EDUCATION", + ) + + +def _ema(raw: pd.DataFrame, *, code_column: str, amount_column: str) -> np.ndarray: + """Participation-gated mean fill for EMA amounts (fill_with_mean port). + + Participants (code == 1) reporting a sentinel negative amount receive + the mean of participants' non-negative reported amounts; everyone else + keeps their reported amount. Floored at zero and annualized at + WEEKS_IN_YEAR (the signed multiplier difference vs the incumbent's + bare 52). A vintage with participants but no valid donor amount fills + with zero — frames refuse NaN, where the incumbent would propagate it. + """ + + code = _num(raw, code_column) + amount = _num(raw, amount_column) + needs_fill = (code == 1) & (amount < 0) + donors = (code == 1) & (amount >= 0) + fill_mean = float(amount[donors].mean()) if bool(donors.any()) else 0.0 + filled = np.where(needs_fill, fill_mean, amount) + return np.maximum(filled, 0.0) * WEEKS_IN_YEAR + + +def _num(frame: pd.DataFrame, column: str) -> pd.Series: + if column not in frame.columns: + return pd.Series(0.0, index=frame.index) + return pd.to_numeric(frame[column], errors="coerce").fillna(0) + + +def _artifact_by_table(stage: SourceStageSpec) -> dict[str, Mapping[str, Any]]: + by_table = {str(artifact.get("table")): artifact for artifact in stage.artifacts} + missing = sorted({"adult", "child"} - set(by_table)) + if missing: + raise ValueError(f"frs_education manifest is missing tab artifact(s): {missing}.") + return by_table diff --git a/packages/microcosm-build/src/microcosm/build/uk_runtime/frs_education_grants.py b/packages/microcosm-build/src/microcosm/build/uk_runtime/frs_education_grants.py new file mode 100644 index 00000000..63b1bcc7 --- /dev/null +++ b/packages/microcosm-build/src/microcosm/build/uk_runtime/frs_education_grants.py @@ -0,0 +1,187 @@ +"""Split aggregate FRS education grants into modelled capacities.""" + +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path + +import numpy as np +import pandas as pd + +from microcosm.build.source_manifest import SourceStageSpec +from microcosm.build.uk_runtime.national_frame import ( + uk_household_weight_kind, + uk_national_frame, + uk_time_period, + validate_uk_national_frame, +) +from microcosm.frame import Frame +from microcosm.frame.rules import assert_rules_engine_country + +FRS_EDUCATION_GRANT_OUTPUT_COLUMNS = ( + "disabled_students_allowance_eligible_expenses", +) +FRS_EDUCATION_GRANT_REWRITES = ("education_grants",) +UK_EDUCATION_GRANT_CAPACITY_PREDICTORS = ( + "childcare_grant", + "parents_learning_allowance", + "adult_dependants_grant", +) +DISABLED_STUDENTS_ALLOWANCE_FIRST_MODELED_YEAR = 2025 +DISABLED_STUDENTS_ALLOWANCE_ELIGIBILITY_VARIABLES = ( + "maintenance_loan_in_england_system", + "disabled_students_allowance_course_eligible", + "disabled_students_allowance_has_qualifying_condition", +) + + +@dataclass(frozen=True) +class UKDSAPolicy: + """DSA maximum read at 1 January of the build year.""" + + maximum: float + instant: str + source: str + + +class UKFRSEducationGrantSplitStageTransform: + """Whole-stage callable for FRS education grant splitting.""" + + def __init__( + self, + *, + stage: SourceStageSpec, + engine: object, + policy: UKDSAPolicy | None = None, + ) -> None: + self.stage = stage + self.engine = engine + self.policy = policy + + def __call__(self, frame: Frame) -> Frame: + period = uk_time_period(frame) + assert_rules_engine_country(self.engine, "uk") + predictors = list(UK_EDUCATION_GRANT_CAPACITY_PREDICTORS) + if int(period) >= DISABLED_STUDENTS_ALLOWANCE_FIRST_MODELED_YEAR: + predictors.extend(DISABLED_STUDENTS_ALLOWANCE_ELIGIBILITY_VARIABLES) + materialized = self.engine.materialize(frame, predictors, period) + policy = self.policy + if policy is None and int(period) < DISABLED_STUDENTS_ALLOWANCE_FIRST_MODELED_YEAR: + policy = UKDSAPolicy( + maximum=0.0, + instant=f"{int(period)}-01-01", + source="pre-2025 DSA not modelled", + ) + return add_frs_education_grant_split( + frame, + capacities=materialized, + policy=policy or uk_dsa_policy(period), + ) + + @staticmethod + def output_columns() -> tuple[str, ...]: + return FRS_EDUCATION_GRANT_OUTPUT_COLUMNS + + +def uk_dsa_policy(build_period: int | str) -> UKDSAPolicy: + """Read DSA maximum at ``{year}-01-01`` for future vintages.""" + + try: + import policyengine_uk + from policyengine_core.parameters import ParameterNode + except ImportError as exc: + raise ImportError( + "UK DSA parameters require `uv sync --all-packages --extra uk`." + ) from exc + + parameters = ParameterNode( + directory_path=str(Path(policyengine_uk.__file__).parent / "parameters") + ) + instant = f"{int(build_period)}-01-01" + return UKDSAPolicy( + maximum=float(parameters.gov.dfe.disabled_students_allowance.maximum(instant)), + instant=instant, + source="policyengine-uk parameters " + f"{getattr(policyengine_uk, '__version__', 'unknown')}", + ) + + +def add_frs_education_grant_split( + frame: Frame, + *, + capacities: dict[str, np.ndarray], + policy: UKDSAPolicy, +) -> Frame: + person = frame.table("person").copy() + period = int(uk_time_period(frame)) + dsa_capacity = disabled_students_allowance_capacity( + person, + capacities=capacities, + policy=policy, + year=period, + ) + grant_capacities = { + name: capacities[name] for name in UK_EDUCATION_GRANT_CAPACITY_PREDICTORS + } + split = allocate_reported_education_grants( + person["education_grants"], {**grant_capacities, "dsa": dsa_capacity} + ) + person["education_grants"] = split["education_grants"] + person["disabled_students_allowance_eligible_expenses"] = split["dsa"] + result = uk_national_frame( + person=person, + benunit=frame.table("benunit"), + household=frame.table("household"), + time_period=str(period), + weight_kind=uk_household_weight_kind(frame), + household_weights=frame.weights_for("household").values, + mass_log=frame.mass_log, + ) + validate_uk_national_frame(result) + return result + + +def allocate_reported_education_grants( + reported_grants, grant_capacities: dict[str, np.ndarray] +) -> dict[str, np.ndarray]: + reported = np.maximum(np.nan_to_num(np.asarray(reported_grants, dtype=float)), 0.0) + capacities = { + name: np.maximum(np.nan_to_num(np.asarray(value, dtype=float)), 0.0) + for name, value in grant_capacities.items() + } + total_capacity = np.zeros_like(reported) + for name, capacity in capacities.items(): + if capacity.shape != reported.shape: + raise ValueError( + f"{name} capacity has shape {capacity.shape}, expected {reported.shape}." + ) + total_capacity += capacity + fraction = np.divide( + reported, + total_capacity, + out=np.zeros_like(reported), + where=total_capacity > 0, + ) + fraction = np.minimum(fraction, 1.0) + result: dict[str, np.ndarray] = {} + allocated = np.zeros_like(reported) + for name, capacity in capacities.items(): + result[name] = capacity * fraction + allocated += result[name] + result["education_grants"] = np.maximum(reported - allocated, 0.0) + return result + + +def disabled_students_allowance_capacity( + person: pd.DataFrame, + *, + capacities: dict[str, np.ndarray], + policy: UKDSAPolicy, + year: int, +) -> np.ndarray: + if year < DISABLED_STUDENTS_ALLOWANCE_FIRST_MODELED_YEAR: + return np.zeros(len(person), dtype=float) + eligible = np.ones(len(person), dtype=bool) + for variable in DISABLED_STUDENTS_ALLOWANCE_ELIGIBILITY_VARIABLES: + eligible &= np.asarray(capacities[variable], dtype=bool) + return np.where(eligible, policy.maximum, 0.0) diff --git a/packages/microcosm-build/src/microcosm/build/uk_runtime/frs_employment.py b/packages/microcosm-build/src/microcosm/build/uk_runtime/frs_employment.py new file mode 100644 index 00000000..9f42a2d6 --- /dev/null +++ b/packages/microcosm-build/src/microcosm/build/uk_runtime/frs_employment.py @@ -0,0 +1,125 @@ +"""FRS derived employment columns for the UK source spine.""" + +from __future__ import annotations + +from collections.abc import Mapping +from pathlib import Path +from typing import Any + +import pandas as pd + +from microcosm.build.source_manifest import SourceStageSpec +from microcosm.build.uk_runtime.frs_spine import ( + normalize_ids, + read_pinned_tab, +) +from microcosm.build.uk_runtime.national_frame import ( + uk_household_weight_kind, + uk_national_frame, + uk_time_period, + validate_uk_national_frame, +) +from microcosm.frame import Frame + +EMPLOYMENT_STATUS_MAP = { + 0: "CHILD", + 1: "FT_EMPLOYED", + 2: "PT_EMPLOYED", + 3: "FT_SELF_EMPLOYED", + 4: "PT_SELF_EMPLOYED", + 5: "UNEMPLOYED", + 6: "RETIRED", + 7: "STUDENT", + 8: "CARER", + 9: "LONG_TERM_DISABLED", + 10: "SHORT_TERM_DISABLED", + 11: "LONG_TERM_DISABLED", +} +EMPLOYMENT_SECTOR_MAP = { + 0: "NOT_EMPLOYED", + 1: "PRIVATE", + 2: "PUBLIC", +} +FRS_EMPLOYMENT_OUTPUT_COLUMNS = ( + "employment_status", + "employment_sector", + "sic_industry_division", +) + + +class UKFRSEmploymentStageTransform: + """Whole-stage callable for FRS employment derivations.""" + + def __init__(self, raw_dir: str | Path, *, stage: SourceStageSpec) -> None: + self.raw_dir = Path(raw_dir) + self.stage = stage + + def __call__(self, frame: Frame) -> Frame: + return add_frs_employment(frame, self.raw_dir, stage=self.stage) + + @staticmethod + def output_columns() -> tuple[str, ...]: + return FRS_EMPLOYMENT_OUTPUT_COLUMNS + + +def add_frs_employment( + frame: Frame, raw_dir: str | Path, *, stage: SourceStageSpec +) -> Frame: + """Add employment status, sector, and SIC from pinned ``adult.tab``.""" + + artifacts = _artifact_by_table(stage) + adult = normalize_ids( + read_pinned_tab(Path(raw_dir) / str(artifacts["adult"]["locator"]), artifacts["adult"]) + ) + derived = derive_frs_employment(frame.table("person"), adult) + person = frame.table("person").copy() + for column in FRS_EMPLOYMENT_OUTPUT_COLUMNS: + person[column] = derived[column].to_numpy() + result = uk_national_frame( + person=person, + benunit=frame.table("benunit"), + household=frame.table("household"), + time_period=uk_time_period(frame), + weight_kind=uk_household_weight_kind(frame), + household_weights=frame.weights_for("household").values, + mass_log=frame.mass_log, + ) + validate_uk_national_frame(result) + return result + + +def derive_frs_employment(person: pd.DataFrame, adult: pd.DataFrame) -> pd.DataFrame: + """Return person-indexed employment derivations. + + ``empstati``, ``mjobsect``, and ``sic`` are intentionally direct-indexed + so missing source columns fail loudly, matching the incumbent FRS port. + """ + + raw = adult.set_index("person_id") + aligned = raw.reindex(person["person_id"]) + values = pd.DataFrame(index=person.index) + empstati = pd.to_numeric(aligned["empstati"], errors="coerce").fillna(0) + # Unmapped codes above the declared domain follow the incumbent's + # post-map fillna to LONG_TERM_DISABLED; 0/NaN rows land on CHILD via + # the explicit map entry. + values["employment_status"] = ( + empstati.astype(int) + .map(EMPLOYMENT_STATUS_MAP) + .fillna("LONG_TERM_DISABLED") + .to_numpy() + ) + sector = pd.to_numeric(aligned["mjobsect"], errors="coerce").fillna(0) + values["employment_sector"] = ( + sector.astype(int).map(EMPLOYMENT_SECTOR_MAP).fillna("NOT_EMPLOYED").to_numpy() + ) + values["sic_industry_division"] = ( + pd.to_numeric(aligned["sic"], errors="coerce").fillna(0).clip(lower=0).astype(int).to_numpy() + ) + return values + + +def _artifact_by_table(stage: SourceStageSpec) -> dict[str, Mapping[str, Any]]: + by_table = {str(artifact.get("table")): artifact for artifact in stage.artifacts} + if "adult" not in by_table: + raise ValueError("frs_employment manifest is missing adult.tab.") + return by_table diff --git a/packages/microcosm-build/src/microcosm/build/uk_runtime/frs_legacy_proxies.py b/packages/microcosm-build/src/microcosm/build/uk_runtime/frs_legacy_proxies.py new file mode 100644 index 00000000..a2911dc8 --- /dev/null +++ b/packages/microcosm-build/src/microcosm/build/uk_runtime/frs_legacy_proxies.py @@ -0,0 +1,174 @@ +"""FRS legacy-benefit claimant-state proxies.""" + +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +import numpy as np +import pandas as pd + +from microcosm.build.source_manifest import SourceStageSpec +from microcosm.build.uk_runtime.frs_spine import ( + WEEKS_IN_YEAR, + normalize_ids, + read_pinned_tab, +) +from microcosm.build.uk_runtime.national_frame import ( + uk_household_weight_kind, + uk_national_frame, + uk_time_period, + validate_uk_national_frame, +) +from microcosm.frame import Frame +from microcosm.frame.rules import assert_rules_engine_country + +FRS_LEGACY_PROXY_OUTPUT_COLUMNS = ( + "legacy_jobseeker_proxy", + "esa_health_condition_proxy", + "esa_support_group_proxy", +) +UK_LEGACY_PROXY_PREDICTORS = ("state_pension_age",) +ESA_HEALTH_EMPLOYMENT_STATUSES = ("LONG_TERM_DISABLED", "SHORT_TERM_DISABLED") + + +@dataclass(frozen=True) +class UKLegacyJSAPolicy: + """JSA hours rule read at 1 January of the build year.""" + + max_weekly_hours_single: float + instant: str + source: str + + +class UKFRSLegacyProxiesStageTransform: + """Whole-stage callable for FRS legacy-benefit proxies.""" + + def __init__( + self, + raw_dir: str | Path, + *, + stage: SourceStageSpec, + engine: object, + policy: UKLegacyJSAPolicy | None = None, + ) -> None: + self.raw_dir = Path(raw_dir) + self.stage = stage + self.engine = engine + self.policy = policy + + def __call__(self, frame: Frame) -> Frame: + period = uk_time_period(frame) + assert_rules_engine_country(self.engine, "uk") + materialized = self.engine.materialize(frame, UK_LEGACY_PROXY_PREDICTORS, period) + return add_frs_legacy_proxies( + frame, + self.raw_dir, + stage=self.stage, + state_pension_age=materialized["state_pension_age"], + policy=self.policy or uk_legacy_jsa_policy(period), + ) + + @staticmethod + def output_columns() -> tuple[str, ...]: + return FRS_LEGACY_PROXY_OUTPUT_COLUMNS + + +def uk_legacy_jsa_policy(build_period: int | str) -> UKLegacyJSAPolicy: + """Read JSA weekly hours at ``{year}-01-01``.""" + + try: + import policyengine_uk + from policyengine_core.parameters import ParameterNode + except ImportError as exc: + raise ImportError( + "UK JSA parameters require `uv sync --all-packages --extra uk`." + ) from exc + + parameters = ParameterNode( + directory_path=str(Path(policyengine_uk.__file__).parent / "parameters") + ) + instant = f"{int(build_period)}-01-01" + return UKLegacyJSAPolicy( + max_weekly_hours_single=float(parameters.gov.dwp.JSA.hours.single(instant)), + instant=instant, + source="policyengine-uk parameters " + f"{getattr(policyengine_uk, '__version__', 'unknown')}", + ) + + +def add_frs_legacy_proxies( + frame: Frame, + raw_dir: str | Path, + *, + stage: SourceStageSpec, + state_pension_age: np.ndarray, + policy: UKLegacyJSAPolicy, +) -> Frame: + artifacts = _artifact_by_table(stage) + adult = normalize_ids( + read_pinned_tab(Path(raw_dir) / str(artifacts["adult"]["locator"]), artifacts["adult"]) + ) + person = frame.table("person").copy() + raw = adult.set_index("person_id").reindex(person["person_id"]) + reported = pd.to_numeric(raw["empstati"], errors="coerce").fillna(0).to_numpy() > 0 + derived = derive_frs_legacy_proxies( + person, + employment_status_reported=reported, + state_pension_age=state_pension_age, + max_annual_hours=policy.max_weekly_hours_single * WEEKS_IN_YEAR, + ) + for column in FRS_LEGACY_PROXY_OUTPUT_COLUMNS: + person[column] = derived[column].to_numpy() + result = uk_national_frame( + person=person, + benunit=frame.table("benunit"), + household=frame.table("household"), + time_period=uk_time_period(frame), + weight_kind=uk_household_weight_kind(frame), + household_weights=frame.weights_for("household").values, + mass_log=frame.mass_log, + ) + validate_uk_national_frame(result) + return result + + +def derive_frs_legacy_proxies( + person: pd.DataFrame, + *, + employment_status_reported, + state_pension_age, + max_annual_hours: float, +) -> pd.DataFrame: + age = pd.to_numeric(person["age"], errors="coerce").fillna(0).to_numpy() + status = person["employment_status"].to_numpy() + hours = pd.to_numeric(person["hours_worked"], errors="coerce").fillna(0).to_numpy() + education = person["current_education"].to_numpy() + reported = np.asarray(employment_status_reported, dtype=bool) + spa = np.asarray(state_pension_age, dtype=float) + values = pd.DataFrame(index=person.index) + values["legacy_jobseeker_proxy"] = ( + reported + & (age >= 18) + & (age < spa) + & (status == "UNEMPLOYED") + & (hours < max_annual_hours) + & (education == "NOT_IN_EDUCATION") + ) + health = reported & (age >= 16) & (age < spa) & np.isin( + status, ESA_HEALTH_EMPLOYMENT_STATUSES + ) + values["esa_health_condition_proxy"] = health + values["esa_support_group_proxy"] = ( + reported & health & (status == "LONG_TERM_DISABLED") & (hours <= 0) + ) + return values + + +def _artifact_by_table(stage: SourceStageSpec) -> dict[str, Mapping[str, Any]]: + by_table = {str(artifact.get("table")): artifact for artifact in stage.artifacts} + if "adult" not in by_table: + raise ValueError("frs_legacy_proxies manifest is missing adult.tab.") + return by_table diff --git a/packages/microcosm-build/src/microcosm/build/uk_runtime/frs_spine.py b/packages/microcosm-build/src/microcosm/build/uk_runtime/frs_spine.py index a70b6228..1ac77a43 100644 --- a/packages/microcosm-build/src/microcosm/build/uk_runtime/frs_spine.py +++ b/packages/microcosm-build/src/microcosm/build/uk_runtime/frs_spine.py @@ -17,6 +17,26 @@ ) from microcosm.frame import Frame, WeightKind +__all__ = [ + "FRS_SPINE_TABLES", + "OUTPUT_COLUMNS", + "REGION_MAP", + "TIME_PERIOD", + "WEEKS_IN_YEAR", + "UKFRSSpineStageTransform", + "artifact_by_table", + "build_uk_frs_spine_frame", + "map_codes", + "normalize_ids", + "number", + "positive", + "raw_number", + "read_pinned_tab", + "reject_nan", + "sum_to_entity", + "uk_frs_spine_seed_frame", +] + FRS_SPINE_TABLES = ( "accounts", "adult", @@ -189,7 +209,6 @@ "is_married", "dependent_children", "household_id", - "household_weight", "region", "tenure_type", "accommodation_type", @@ -790,3 +809,14 @@ def _reject_nan(frame: pd.DataFrame, entity: str) -> None: if frame.isna().any().any(): bad = sorted(frame.columns[frame.isna().any()].tolist()) raise ValueError(f"FRS spine {entity} produced NaN column(s): {bad}.") + + +artifact_by_table = _artifact_by_table +read_pinned_tab = _read_pinned_tab +normalize_ids = _normalize_ids +sum_to_entity = _sum_to_entity +map_codes = _map_codes +number = _number +raw_number = _raw_number +positive = _positive +reject_nan = _reject_nan diff --git a/packages/microcosm-build/src/microcosm/build/uk_runtime/national_build.py b/packages/microcosm-build/src/microcosm/build/uk_runtime/national_build.py index 56d5b83b..9c34ad8d 100644 --- a/packages/microcosm-build/src/microcosm/build/uk_runtime/national_build.py +++ b/packages/microcosm-build/src/microcosm/build/uk_runtime/national_build.py @@ -533,6 +533,10 @@ def build_uk_national_dataset( artifacts: dict[str, object] = { "coverage_engine": engine, "exclusions_evaluated_on": evaluation_date, + # The nonnegative gate derives its required columns from the stages + # this build actually scheduled (same roster the preflight coverage + # gate attests). + "build_stage_names": tuple(stage.name for stage in materialized_stages), } fit_weight_records = _stage_fit_weight_records(materialized_stages) if fit_weight_records is not None: diff --git a/packages/microcosm-build/src/microcosm/build/uk_runtime/source_runtime.py b/packages/microcosm-build/src/microcosm/build/uk_runtime/source_runtime.py index 14b44ea5..6ca259be 100644 --- a/packages/microcosm-build/src/microcosm/build/uk_runtime/source_runtime.py +++ b/packages/microcosm-build/src/microcosm/build/uk_runtime/source_runtime.py @@ -17,23 +17,67 @@ from microcosm.frame.rules import materialize_rules_engine_predictors __all__ = [ + "UK_NONNEGATIVE_OUTPUTS_BY_STAGE", + "UK_NONNEGATIVE_SOURCE_OUTPUTS", "materialize_uk_rules_engine_predictors_from_manifest", "uk_source_operation_handlers", "uk_stage_implementations", ] +def _uk_nonnegative_outputs_by_stage() -> dict[str, tuple[str, ...]]: + from microcosm.build.country_spec import load_country_spec + + spec = load_country_spec("uk") + if spec.sources is None: + return {} + return { + stage.stage: tuple(stage.nonnegative_outputs) + for stage in spec.sources.stages + } + + +UK_NONNEGATIVE_OUTPUTS_BY_STAGE = _uk_nonnegative_outputs_by_stage() +UK_NONNEGATIVE_SOURCE_OUTPUTS = tuple( + dict.fromkeys( + column + for columns in UK_NONNEGATIVE_OUTPUTS_BY_STAGE.values() + for column in columns + ) +) + + def uk_stage_implementations( *, retained_leaves_transform: Callable[[Frame], Frame], hmrc_income_transform: Callable[[Frame], Frame], + frs_spine_transform: Callable[[Frame], Frame] | None = None, + frs_employment_transform: Callable[[Frame], Frame] | None = None, + frs_council_tax_transform: Callable[[Frame], Frame] | None = None, + frs_disability_transform: Callable[[Frame], Frame] | None = None, + frs_education_transform: Callable[[Frame], Frame] | None = None, + frs_legacy_proxies_transform: Callable[[Frame], Frame] | None = None, + frs_education_grant_split_transform: Callable[[Frame], Frame] | None = None, ) -> dict[str, Callable[[Frame], Frame]]: """Return the whole-stage implementation map for the UK source plan.""" - return { + implementations = { "frs_hmrc_retained_leaves": retained_leaves_transform, "hmrc_spi_income": hmrc_income_transform, } + optional = { + "frs_spine": frs_spine_transform, + "frs_employment": frs_employment_transform, + "frs_council_tax": frs_council_tax_transform, + "frs_disability": frs_disability_transform, + "frs_education": frs_education_transform, + "frs_legacy_proxies": frs_legacy_proxies_transform, + "frs_education_grant_split": frs_education_grant_split_transform, + } + implementations.update( + {name: transform for name, transform in optional.items() if transform is not None} + ) + return implementations def uk_source_operation_handlers() -> Mapping[str, SourceOperationHandler]: diff --git a/packages/microcosm-build/tests/test_country_spec.py b/packages/microcosm-build/tests/test_country_spec.py index 9785ebf1..c0e2da2e 100644 --- a/packages/microcosm-build/tests/test_country_spec.py +++ b/packages/microcosm-build/tests/test_country_spec.py @@ -320,6 +320,7 @@ def test_declares_the_full_june_battery(self, manifest) -> None: "uk_weight_ess", "uk_weight_ratio", "uk_weights_audit", + "uk_nonnegative_columns", "uk_export_surface", "uk_target_surface", "uk_target_fit", diff --git a/packages/microcosm-build/tests/test_uk_battery_bindings.py b/packages/microcosm-build/tests/test_uk_battery_bindings.py index 1d0d8c97..4ee9dc04 100644 --- a/packages/microcosm-build/tests/test_uk_battery_bindings.py +++ b/packages/microcosm-build/tests/test_uk_battery_bindings.py @@ -163,6 +163,9 @@ def _run_battery(tables, *, parity=None, fit_records=None, armed=True, clock=CLO artifacts: dict[str, object] = { "coverage_engine": object(), "exclusions_evaluated_on": clock, + # The staging pipeline's two scheduled stages declare no nonnegative + # outputs, so the nonnegative gate passes with zero required columns. + "build_stage_names": ("frs_hmrc_retained_leaves", "hmrc_spi_income"), } if fit_records is not None: artifacts["fit_weight_records"] = fit_records @@ -209,6 +212,79 @@ def test_surface_materializes_the_frame_not_fallbacks(self) -> None: assert surface.household_weight_kind is uk_household_weight_kind(frame) assert surface.mass_log == frame.mass_log + def _nonnegative_frame(self, *, sic: list[float] | None): + person, benunit, household = _tables() + if sic is not None: + person["sic_industry_division"] = sic + return uk_national_frame( + person=person, + benunit=benunit, + household=household, + time_period="2023", + ) + + def test_nonnegative_binding_requires_scheduled_stage_columns(self) -> None: + # frs_employment declares sic_industry_division nonnegative; a build + # that scheduled the stage but lost the column must fail — the + # missing-column path is the reason the required set is never + # pre-filtered to present columns. + binding = UK_GATE_REGISTRY["nonnegative_columns"] + context = EvidenceContext( + frame=self._nonnegative_frame(sic=None), + artifacts={"build_stage_names": ("frs_employment",)}, + ) + + result = binding.evaluate(context, {}) + + assert result.passed is False + assert "sic_industry_division" in result.failures[0] + + def test_nonnegative_binding_fails_on_negative_values(self) -> None: + binding = UK_GATE_REGISTRY["nonnegative_columns"] + context = EvidenceContext( + frame=self._nonnegative_frame(sic=[1.0, -2.0, 3.0, 4.0]), + artifacts={"build_stage_names": ("frs_employment",)}, + ) + + result = binding.evaluate(context, {}) + + assert result.name == "nonnegative_columns" + assert result.passed is False + assert ( + "sic_industry_division: 1 finite value(s) below zero" + in result.failures[0] + ) + + def test_nonnegative_binding_passes_clean_scheduled_columns(self) -> None: + binding = UK_GATE_REGISTRY["nonnegative_columns"] + context = EvidenceContext( + frame=self._nonnegative_frame(sic=[1.0, 0.0, 3.0, 4.0]), + artifacts={"build_stage_names": ("frs_employment",)}, + ) + + result = binding.evaluate(context, {}) + + assert result.passed is True + + def test_nonnegative_binding_does_not_demand_unscheduled_stages(self) -> None: + # The national staging build schedules only the two HMRC stages, + # which declare no nonnegative outputs — the gate passes honestly + # with zero required columns rather than by silent pre-filtering. + binding = UK_GATE_REGISTRY["nonnegative_columns"] + context = EvidenceContext( + frame=self._nonnegative_frame(sic=None), + artifacts={ + "build_stage_names": ( + "frs_hmrc_retained_leaves", + "hmrc_spi_income", + ) + }, + ) + + result = binding.evaluate(context, {}) + + assert result.passed is True + class TestUKCompatibility: """The BE plumbing test, run over the UK spec: an empty evidence context @@ -263,7 +339,9 @@ def test_fully_armed_battery_evaluates_gate_for_gate(self) -> None: passed = [ entry_id for entry_id, o in by_id.items() if o.status is GateStatus.PASSED ] - assert len(passed) == 10 + # 11 with uk_nonnegative_columns: the scheduled stages declare no + # nonnegative outputs, so the gate passes with zero required columns. + assert len(passed) == 11 qrf = by_id["uk_qrf_tail_concentration"] assert qrf.status is GateStatus.FAILED assert "declared QRF output is absent" in qrf.result.failures[0] diff --git a/packages/microcosm-build/tests/test_uk_frs_council_tax.py b/packages/microcosm-build/tests/test_uk_frs_council_tax.py new file mode 100644 index 00000000..ff47a5b0 --- /dev/null +++ b/packages/microcosm-build/tests/test_uk_frs_council_tax.py @@ -0,0 +1,31 @@ +from __future__ import annotations + +import numpy as np +import pandas as pd + +from microcosm.build.uk_runtime.frs_council_tax import derive_council_tax +from microcosm.build.uk_runtime.frs_spine import WEEKS_IN_YEAR + + +def test_council_tax_imputes_raw_missing_from_raw_cells() -> None: + household = pd.DataFrame({"household_id": [1, 2, 3, 4, 5, 6]}) + raw = pd.DataFrame( + { + "household_id": [1, 2, 3, 4, 5, 6], + "gvtregno": [12, 12, 1, 1, 2, 2], + "ctband": [1, 1, np.nan, np.nan, 2, 2], + "adulth": [1, 1, 1, 1, 2, 2], + "ctannual": [1000.0, -1.0, 600.0, np.nan, -1.0, 0.0], + "csewamt": [2.0, 2.0, 0.0, 0.0, 0.0, 0.0], + "cwatamtd": [3.0, 3.0, 0.0, 0.0, 0.0, 0.0], + } + ) + + result = derive_council_tax(household, raw) + + scottish_tax = 1000.0 - (2.0 + 3.0) * WEEKS_IN_YEAR + assert np.isclose(result.loc[1], scottish_tax) + assert np.isclose(result.loc[2], scottish_tax) + assert result.loc[4] == 600.0 + assert result.loc[5] == 0.0 + assert result.loc[6] == 0.0 diff --git a/packages/microcosm-build/tests/test_uk_frs_disability.py b/packages/microcosm-build/tests/test_uk_frs_disability.py new file mode 100644 index 00000000..c0b7d030 --- /dev/null +++ b/packages/microcosm-build/tests/test_uk_frs_disability.py @@ -0,0 +1,112 @@ +from __future__ import annotations + +import numpy as np +import pandas as pd +import pytest + +from microcosm.build.uk_runtime.frs_disability import ( + UKDWPBaselineDisabilityRates, + UKDWPDisabilityFlagRates, + derive_frs_disability, + uk_dwp_baseline_disability_rates, +) +from microcosm.build.uk_runtime.frs_spine import WEEKS_IN_YEAR + + +def _baseline() -> UKDWPBaselineDisabilityRates: + return UKDWPBaselineDisabilityRates( + aa_lower=10, + aa_higher=20, + dla_sc_lower=10, + dla_sc_middle=20, + dla_sc_higher=30, + dla_m_lower=10, + dla_m_higher=20, + pip_m_standard=10, + pip_m_enhanced=20, + pip_dl_standard=10, + pip_dl_enhanced=20, + instant="2023-01-01", + source="fixture", + ) + + +def _flags() -> UKDWPDisabilityFlagRates: + return UKDWPDisabilityFlagRates( + aa_higher=20, + dla_sc_higher=30, + pip_dl_enhanced=20, + instant="2023-01-01", + source="fixture", + ) + + +def test_disability_category_threshold_and_overwrite() -> None: + person = pd.DataFrame( + { + "attendance_allowance_reported": [ + (20 - 1) * WEEKS_IN_YEAR, + (20 - 1.01) * WEEKS_IN_YEAR, + ], + "dla_sc_reported": [0, 0], + "dla_m_reported": [0, 0], + "pip_m_reported": [0, 0], + "pip_dl_reported": [0, 0], + } + ) + + result = derive_frs_disability( + person, baseline_rates=_baseline(), flag_rates=_flags() + ) + + assert result["aa_category"].tolist() == ["HIGHER", "LOWER"] + + +def test_disability_flag_operator_asymmetry_and_afcs() -> None: + boundary = (30 - 1) * WEEKS_IN_YEAR + person = pd.DataFrame( + { + "attendance_allowance_reported": [0, 0], + "dla_sc_reported": [boundary, 0], + "dla_m_reported": [0, 0], + "pip_m_reported": [0, 0], + "pip_dl_reported": [0, 0], + "sda_reported": [0, 0], + "incapacity_benefit_reported": [0, 0], + "iidb_reported": [0, 0], + "afcs_reported": [0, 1], + "esa_contrib_reported": [0, 0], + "esa_income_reported": [0, 0], + } + ) + + result = derive_frs_disability( + person, baseline_rates=_baseline(), flag_rates=_flags() + ) + + assert result["is_enhanced_disabled_for_benefits"].tolist() == [False, False] + assert result["is_severely_disabled_for_benefits"].tolist() == [True, True] + + +def test_dwp_reader_split_is_value_bearing() -> None: + # The readers construct the real engine's parameter tree; the wheel gate + # and the us-extra CI lane run without policyengine-uk, so skip there. + # The baseline-vs-plain split carries different VALUES, not just + # provenance: the baseline clone escapes policyengine-uk's fiscal-year + # conversion (April-2022-era weekly rates at build period 2023), while + # the plain tree the incumbent's flags read is fiscal-2023-24. A licensed + # head-to-head against the incumbent's own output caught the one-row + # flag divergence when both readers used the raw-file January values. + pytest.importorskip("policyengine_uk") + from microcosm.build.uk_runtime.frs_disability import ( + uk_dwp_disability_flag_rates, + ) + + rates = uk_dwp_baseline_disability_rates(2023) + flag_rates = uk_dwp_disability_flag_rates(2023) + + assert rates.instant == "2023-01-01" + assert np.isfinite(rates.aa_lower) + assert rates.aa_higher == pytest.approx(92.40) + assert flag_rates.aa_higher == pytest.approx(101.75) + assert flag_rates.dla_sc_higher > rates.dla_sc_higher diff --git a/packages/microcosm-build/tests/test_uk_frs_education.py b/packages/microcosm-build/tests/test_uk_frs_education.py new file mode 100644 index 00000000..abb27b1a --- /dev/null +++ b/packages/microcosm-build/tests/test_uk_frs_education.py @@ -0,0 +1,130 @@ +from __future__ import annotations + +import numpy as np +import pandas as pd + +from microcosm.build.uk_runtime.frs_education import ( + EDUCQUAL_MAP, + derive_current_education, + derive_frs_education, +) +from microcosm.build.uk_runtime.frs_spine import WEEKS_IN_YEAR + + +def test_current_education_cascade_keeps_dead_post_secondary_shadowed() -> None: + result = derive_current_education( + fted=np.array([2, 1, 1, 1, 1, 1]), + typeed2=np.array([9, 1, 2, 5, 7, 9]), + age=np.array([20, 4, 10, 16, 19, 20]), + ) + + assert result.tolist() == [ + "NOT_IN_EDUCATION", + "PRE_PRIMARY", + "PRIMARY", + "LOWER_SECONDARY", + "UPPER_SECONDARY", + "TERTIARY", + ] + + +def test_educqual_map_pins_the_corrected_frs_value_labels() -> None: + # Signed difference vs the incumbent, whose map was inverted (it read + # low codes as school-level and 17-21 as degrees). Labels verified + # against the FRS 2023-24 data dictionary (UKDS SN 9367, DOI + # 10.5255/UKDA-SN-9367-2, adult table) and corroborated by the raw + # aggregates: code 1 (Doctorate) is ~1.8% of adults at the highest mean + # earnings; 18/19 are near-empty baccalaureates; the GCSE band (36-82) + # carries the mass. Code 87 is undocumented in the dictionary and is + # deliberately unmapped, falling to the fillna default. + assert EDUCQUAL_MAP[1] == "TERTIARY" + assert EDUCQUAL_MAP[3] == "TERTIARY" + assert EDUCQUAL_MAP[11] == "POST_SECONDARY" + assert EDUCQUAL_MAP[17] == "UPPER_SECONDARY" + assert EDUCQUAL_MAP[26] == "POST_SECONDARY" + assert EDUCQUAL_MAP[36] == "LOWER_SECONDARY" + assert EDUCQUAL_MAP[82] == "LOWER_SECONDARY" + assert EDUCQUAL_MAP[85] == "NOT_COMPLETED_PRIMARY" + assert EDUCQUAL_MAP[86] == "LOWER_SECONDARY" + assert 87 not in EDUCQUAL_MAP + assert set(EDUCQUAL_MAP) == set(range(1, 87)) + + +def test_training_qyp_ema_and_benefits_in_own_right() -> None: + person = pd.DataFrame( + { + "person_id": [1, 2, 3], + "age": [19, 17, 20], + "universal_credit_reported": [0, 1, 0], + "jsa_contrib_reported": [0, 0, 0], + "jsa_income_reported": [0, 0, 0], + "esa_contrib_reported": [0, 0, 0], + "esa_income_reported": [0, 0, 0], + } + ) + raw = pd.DataFrame( + { + "person_id": [1, 2, 3], + "fted": [1, 2, 2], + "typeed2": [7, 0, 0], + "educqual": [17, 86, 999], + "train": [9, 10, np.nan], + # 2023-24-shaped vintage: no adema pair, eduma/edumaamt instead. + "eduma": [1, 1, 0], + "edumaamt": [2.0, -1.0, 5.0], + "chema": [0, 1, 1], + "chemaamt": [0.0, 3.0, -1.0], + } + ) + + result = derive_frs_education(person, raw) + + assert result["is_in_approved_training"].tolist() == [True, False, False] + assert result[ + "age_started_or_accepted_current_education_or_training" + ].tolist() == [18, 1000, 1000] + assert result[ + "is_before_universal_credit_qualifying_young_person_terminal_date" + ].tolist() == [True, False, False] + # fill_with_mean semantics: a participant's sentinel −1 receives the + # participant mean (2.0 adult / 3.0 child); a non-participant keeps a + # positive reported amount (no flag-gating of reported values). + assert result["adult_ema"].tolist() == [ + 2.0 * WEEKS_IN_YEAR, + 2.0 * WEEKS_IN_YEAR, + 5.0 * WEEKS_IN_YEAR, + ] + assert result["child_ema"].tolist() == [ + 0.0, + 3.0 * WEEKS_IN_YEAR, + 3.0 * WEEKS_IN_YEAR, + ] + assert result["receives_benefits_in_own_right"].tolist() == [False, True, False] + + +def test_adult_ema_prefers_the_adema_pair_when_present() -> None: + person = pd.DataFrame({"person_id": [1, 2], "age": [17, 18]}) + raw = pd.DataFrame( + { + "person_id": [1, 2], + "fted": [2, 2], + "typeed2": [0, 0], + "educqual": [3, 3], + # adema pair present: it must win over a contradictory eduma pair + # (the incumbent aliases eduma into adema only when adema is + # absent). + "adema": [1, 1], + "ademaamt": [4.0, -1.0], + "eduma": [1, 1], + "edumaamt": [9.0, 9.0], + "chema": [0, 0], + "chemaamt": [0.0, 0.0], + } + ) + + result = derive_frs_education(person, raw) + + assert result["adult_ema"].tolist() == [ + 4.0 * WEEKS_IN_YEAR, + 4.0 * WEEKS_IN_YEAR, + ] diff --git a/packages/microcosm-build/tests/test_uk_frs_education_grants.py b/packages/microcosm-build/tests/test_uk_frs_education_grants.py new file mode 100644 index 00000000..ab400e9b --- /dev/null +++ b/packages/microcosm-build/tests/test_uk_frs_education_grants.py @@ -0,0 +1,41 @@ +from __future__ import annotations + +import numpy as np +import pandas as pd + +from microcosm.build.uk_runtime.frs_education_grants import ( + UKDSAPolicy, + allocate_reported_education_grants, + disabled_students_allowance_capacity, +) + + +def test_grant_allocator_caps_fraction_and_keeps_residual() -> None: + result = allocate_reported_education_grants( + [60.0, 300.0], + { + "childcare_grant": np.array([100.0, 100.0]), + "parents_learning_allowance": np.array([100.0, 0.0]), + "adult_dependants_grant": np.array([0.0, 0.0]), + "dsa": np.array([0.0, 50.0]), + }, + ) + + assert result["childcare_grant"].tolist() == [30.0, 100.0] + assert result["parents_learning_allowance"].tolist() == [30.0, 0.0] + assert result["dsa"].tolist() == [0.0, 50.0] + assert result["education_grants"].tolist() == [0.0, 150.0] + + +def test_pre_2025_dsa_capacity_is_aligned_zero_vector() -> None: + person = pd.DataFrame(index=[10, 20, 30]) + + result = disabled_students_allowance_capacity( + person, + capacities={}, + policy=UKDSAPolicy(maximum=100.0, instant="2023-01-01", source="fixture"), + year=2023, + ) + + assert result.shape == (3,) + assert result.tolist() == [0.0, 0.0, 0.0] diff --git a/packages/microcosm-build/tests/test_uk_frs_employment.py b/packages/microcosm-build/tests/test_uk_frs_employment.py new file mode 100644 index 00000000..23f44ca8 --- /dev/null +++ b/packages/microcosm-build/tests/test_uk_frs_employment.py @@ -0,0 +1,50 @@ +from __future__ import annotations + +import numpy as np +import pandas as pd +import pytest + +from microcosm.build.uk_runtime.frs_employment import derive_frs_employment + + +def test_employment_maps_status_sector_and_sic() -> None: + person = pd.DataFrame({"person_id": [1, 2, 3, 4, 5]}) + adult = pd.DataFrame( + { + "person_id": [1, 2, 3, 4, 5], + "empstati": [0, 10, 11, np.nan, 12], + "mjobsect": [0, 1, 2, np.nan, 1], + "sic": [-5, 84.9, np.nan, 7, 20], + } + ) + + result = derive_frs_employment(person, adult) + + assert result["employment_status"].tolist() == [ + "CHILD", + "SHORT_TERM_DISABLED", + "LONG_TERM_DISABLED", + "CHILD", + # Beyond-domain codes take the incumbent's post-map fillna, not the + # CHILD default reserved for 0/NaN rows. + "LONG_TERM_DISABLED", + ] + assert result["employment_sector"].tolist() == [ + "NOT_EMPLOYED", + "PRIVATE", + "PUBLIC", + "NOT_EMPLOYED", + "PRIVATE", + ] + assert result["sic_industry_division"].tolist() == [0, 84, 0, 7, 20] + + +@pytest.mark.parametrize("missing", ["empstati", "mjobsect", "sic"]) +def test_employment_missing_fail_loud_columns_raise(missing: str) -> None: + person = pd.DataFrame({"person_id": [1]}) + adult = pd.DataFrame( + {"person_id": [1], "empstati": [1], "mjobsect": [1], "sic": [10]} + ).drop(columns=[missing]) + + with pytest.raises(KeyError): + derive_frs_employment(person, adult) diff --git a/packages/microcosm-build/tests/test_uk_frs_legacy_proxies.py b/packages/microcosm-build/tests/test_uk_frs_legacy_proxies.py new file mode 100644 index 00000000..eba5fdd0 --- /dev/null +++ b/packages/microcosm-build/tests/test_uk_frs_legacy_proxies.py @@ -0,0 +1,43 @@ +from __future__ import annotations + +import pandas as pd + +from microcosm.build.uk_runtime.frs_legacy_proxies import derive_frs_legacy_proxies +from microcosm.build.uk_runtime.frs_spine import WEEKS_IN_YEAR + + +def test_legacy_proxy_truth_table_and_jsa_hours_boundary() -> None: + person = pd.DataFrame( + { + "age": [18, 18, 30, 30], + "employment_status": [ + "UNEMPLOYED", + "UNEMPLOYED", + "SHORT_TERM_DISABLED", + "LONG_TERM_DISABLED", + ], + "hours_worked": [ + 15.99 * WEEKS_IN_YEAR, + 16 * WEEKS_IN_YEAR, + 0, + 0, + ], + "current_education": [ + "NOT_IN_EDUCATION", + "NOT_IN_EDUCATION", + "NOT_IN_EDUCATION", + "NOT_IN_EDUCATION", + ], + } + ) + + result = derive_frs_legacy_proxies( + person, + employment_status_reported=[True, True, True, True], + state_pension_age=[66, 66, 66, 66], + max_annual_hours=16 * WEEKS_IN_YEAR, + ) + + assert result["legacy_jobseeker_proxy"].tolist() == [True, False, False, False] + assert result["esa_health_condition_proxy"].tolist() == [False, False, True, True] + assert result["esa_support_group_proxy"].tolist() == [False, False, False, True] diff --git a/packages/microcosm-build/tests/test_uk_frs_spine.py b/packages/microcosm-build/tests/test_uk_frs_spine.py index 5562873d..d9a4cab7 100644 --- a/packages/microcosm-build/tests/test_uk_frs_spine.py +++ b/packages/microcosm-build/tests/test_uk_frs_spine.py @@ -3,6 +3,7 @@ import hashlib import importlib.util import json +import sys from pathlib import Path from types import SimpleNamespace @@ -13,6 +14,11 @@ from microcosm.build.country_spec import country_stage_plan, load_country_spec from microcosm.build.logbook import load_spool_rows from microcosm.build.source_manifest import SourceManifest, SourceStageSpec +from microcosm.build.uk_runtime import ( + frs_disability, + frs_education_grants, + frs_legacy_proxies, +) from microcosm.build.uk_runtime.frs_spine import ( FRS_SPINE_TABLES, REGION_MAP, @@ -101,6 +107,15 @@ def _fixture_tables() -> dict[str, list[dict[str, object]]]: "HRPID": 1, "UPERSON": 1, "MARITAL": 1, + "EMPSTATI": 5, + "MJOBSECT": 1, + "SIC": 84, + "FTED": 2, + "TYPEED2": 0, + "EDUCQUAL": 17, + "TRAIN": 10, + "EMAAMT": 0.0, + "CHEMAAMT": 0.0, "INEARNS": 10.0, "SEINCAM2": 3.0, "MNTUS1": 2, @@ -141,6 +156,12 @@ def _fixture_tables() -> dict[str, list[dict[str, object]]]: "HRPID": 0, "UPERSON": 0, "MARITAL": 2, + "FTED": 1, + "TYPEED2": 2, + "EDUCQUAL": 86, + "TRAIN": 9, + "EMAAMT": 0.0, + "CHEMAAMT": 1.0, } return { "adult": [adult_2, adult_1], @@ -326,18 +347,148 @@ def _manifest_stage() -> SourceStageSpec: def _synthetic_spec(stage: SourceStageSpec) -> SimpleNamespace: + def source_stage( + name: str, + *, + tables: tuple[str, ...] = (), + outputs: tuple[str, ...], + operations: list[dict[str, object]] | None = None, + nonnegative_outputs: tuple[str, ...] = (), + rewrites: tuple[str, ...] = (), + ) -> SourceStageSpec: + artifacts = [ + artifact for artifact in stage.artifacts if artifact["table"] in tables + ] + payload = { + "stage": name, + "survey": "Synthetic FRS", + "source": "local fabricated rows", + "grain": "person", + "artifacts": artifacts, + "operations": operations or [{"kind": "derive"}], + "outputs": list(outputs), + "nonnegative_outputs": list(nonnegative_outputs), + } + if rewrites: + payload["rewrites"] = list(rewrites) + return SourceStageSpec.from_mapping(payload) + return SimpleNamespace( country="uk", sources=SourceManifest( country="uk", version=1, policy="Synthetic FRS spine spec.", - stages=(stage,), + stages=( + stage, + source_stage( + "frs_employment", + tables=("adult",), + operations=[{"kind": "read_tables"}, {"kind": "map_coded_amounts"}], + outputs=( + "employment_status", + "employment_sector", + "sic_industry_division", + ), + nonnegative_outputs=("sic_industry_division",), + ), + source_stage( + "frs_council_tax", + tables=("househol",), + operations=[{"kind": "read_tables"}, {"kind": "impute_cell_means"}], + outputs=("council_tax",), + nonnegative_outputs=("council_tax",), + ), + source_stage( + "frs_disability", + outputs=( + "aa_category", + "dla_sc_category", + "dla_m_category", + "pip_m_category", + "pip_dl_category", + "is_disabled_for_benefits", + "is_enhanced_disabled_for_benefits", + "is_severely_disabled_for_benefits", + ), + ), + source_stage( + "frs_education", + tables=("adult", "child"), + operations=[{"kind": "read_tables"}, {"kind": "derive"}], + outputs=( + "current_education", + "highest_education", + "is_in_non_advanced_education", + "is_in_approved_training", + "age_started_or_accepted_current_education_or_training", + "is_before_universal_credit_qualifying_young_person_terminal_date", + "adult_ema", + "child_ema", + "receives_benefits_in_own_right", + ), + nonnegative_outputs=( + "adult_ema", + "child_ema", + "age_started_or_accepted_current_education_or_training", + ), + ), + source_stage( + "frs_legacy_proxies", + tables=("adult",), + operations=[ + {"kind": "read_tables"}, + { + "kind": "materialize_rules_engine_predictors", + "predictors": ["state_pension_age"], + }, + {"kind": "derive"}, + ], + outputs=( + "legacy_jobseeker_proxy", + "esa_health_condition_proxy", + "esa_support_group_proxy", + ), + ), + source_stage( + "frs_education_grant_split", + operations=[ + { + "kind": "materialize_rules_engine_predictors", + "predictors": [ + "childcare_grant", + "parents_learning_allowance", + "adult_dependants_grant", + ], + }, + {"kind": "derive"}, + ], + outputs=("disabled_students_allowance_eligible_expenses",), + rewrites=("education_grants",), + nonnegative_outputs=( + "disabled_students_allowance_eligible_expenses", + ), + ), + ), ), geography_spine=None, ) +class _FakeUKEngine: + country = "uk" + + def materialize(self, frame, variables, period): + person_count = len(frame.table("person")) + values = {} + for variable in variables: + if variable == "state_pension_age": + values[variable] = np.full(person_count, 66.0) + else: + values[variable] = np.zeros(person_count) + return values + + def test_manifest_stage_and_runtime_agree_on_artifacts_and_operations() -> None: stage = _manifest_stage() @@ -535,6 +686,66 @@ def test_shuffled_household_fixture_produces_identical_output(tmp_path: Path) -> ) +def _stub_policy_readers(monkeypatch: pytest.MonkeyPatch) -> None: + """Make the driver run engine-hermetic. + + The us-extra CI lane and the wheel-gate venv have no policyengine-uk, + so the driver tests stub the three deferred parameter readers and then + block the engine import outright — a regression to eager engine use + fails loudly in every environment, not just engine-less ones. + """ + + monkeypatch.setattr( + frs_disability, + "uk_dwp_baseline_disability_rates", + lambda period: frs_disability.UKDWPBaselineDisabilityRates( + aa_lower=68.1, + aa_higher=101.75, + dla_sc_lower=26.9, + dla_sc_middle=68.1, + dla_sc_higher=101.75, + dla_m_lower=26.9, + dla_m_higher=71.0, + pip_m_standard=26.9, + pip_m_enhanced=71.0, + pip_dl_standard=68.1, + pip_dl_enhanced=101.75, + instant=f"{period}-01-01", + source="test stub", + ), + ) + monkeypatch.setattr( + frs_disability, + "uk_dwp_disability_flag_rates", + lambda period: frs_disability.UKDWPDisabilityFlagRates( + aa_higher=101.75, + dla_sc_higher=101.75, + pip_dl_enhanced=101.75, + instant=f"{period}-01-01", + source="test stub", + ), + ) + monkeypatch.setattr( + frs_legacy_proxies, + "uk_legacy_jsa_policy", + lambda period: frs_legacy_proxies.UKLegacyJSAPolicy( + max_weekly_hours_single=16.0, + instant=f"{period}-01-01", + source="test stub", + ), + ) + monkeypatch.setattr( + frs_education_grants, + "uk_dsa_policy", + lambda period: frs_education_grants.UKDSAPolicy( + maximum=0.0, + instant=f"{period}-01-01", + source="test stub", + ), + ) + monkeypatch.setitem(sys.modules, "policyengine_uk", None) + + def test_driver_writes_spine_h5_sidecars_and_logbook( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: @@ -549,6 +760,8 @@ def test_driver_writes_spine_h5_sidecars_and_logbook( monkeypatch.setattr( tool, "load_country_spec", lambda country: _synthetic_spec(stage) ) + monkeypatch.setattr(tool, "_rules_engine", lambda: _FakeUKEngine()) + _stub_policy_readers(monkeypatch) monkeypatch.delenv("POPULACE_LEDGER_URL", raising=False) monkeypatch.delenv("POPULACE_LEDGER_KEY", raising=False) monkeypatch.delenv("POPULACE_LEDGER_API_KEY", raising=False) @@ -574,6 +787,8 @@ def test_driver_writes_spine_h5_sidecars_and_logbook( assert len(frame.table("household")) == 2 sidecar = json.loads(output.with_suffix(".build.json").read_text()) assert sidecar["pipeline"] == "uk-frs-spine" + assert sidecar["schema_version"] == 2 + assert sidecar["stages"] == list(tool._STAGE_NAMES) assert sidecar["entity_row_counts"] == { "person": 3, "benunit": 2, @@ -582,7 +797,10 @@ def test_driver_writes_spine_h5_sidecars_and_logbook( assert sidecar["household_weight_total"] == 30.0 assert set(sidecar["artifact_pins"]) == set(FRS_SPINE_TABLES) share_payload = json.loads(shares.read_text()) - assert share_payload["employment_income"] == pytest.approx(2 / 3) + assert share_payload["stages"]["frs_spine"]["employment_income"] == pytest.approx( + 2 / 3 + ) + assert "education_grants" in share_payload["final"] rows = load_spool_rows(tmp_path / "logbook-spool") assert len(rows) == 1 assert rows[0].pipeline == "uk-frs-spine" @@ -602,6 +820,8 @@ def test_driver_writes_payload_identical_h5s( monkeypatch.setattr( tool, "load_country_spec", lambda country: _synthetic_spec(stage) ) + monkeypatch.setattr(tool, "_rules_engine", lambda: _FakeUKEngine()) + _stub_policy_readers(monkeypatch) monkeypatch.delenv("POPULACE_LEDGER_URL", raising=False) monkeypatch.delenv("POPULACE_LEDGER_KEY", raising=False) monkeypatch.delenv("POPULACE_LEDGER_API_KEY", raising=False) diff --git a/packages/microcosm-build/tests/test_uk_national_build.py b/packages/microcosm-build/tests/test_uk_national_build.py index fad28bea..ea398799 100644 --- a/packages/microcosm-build/tests/test_uk_national_build.py +++ b/packages/microcosm-build/tests/test_uk_national_build.py @@ -890,6 +890,7 @@ def test_national_build_real_terminal_batch_passes_before_staging( "uk_weight_ess": "passed", "uk_weight_ratio": "passed", "uk_weights_audit": "passed", + "uk_nonnegative_columns": "passed", # The legacy report omitted unevidenced gates; the battery names # every gap — non-blocking off the release-candidate posture. "uk_export_surface": "evidence_absent", diff --git a/packages/microcosm-build/tests/test_uk_source_stages.py b/packages/microcosm-build/tests/test_uk_source_stages.py index 41fe6598..0f327577 100644 --- a/packages/microcosm-build/tests/test_uk_source_stages.py +++ b/packages/microcosm-build/tests/test_uk_source_stages.py @@ -18,6 +18,20 @@ UK_PACKAGE = ROOT / "packages/microcosm-build/src/microcosm/build/uk" FROZEN_SOURCE_STAGES = UK_PACKAGE / "hmrc_income_source_stages.json" CANONICAL_SOURCE_STAGES = UK_PACKAGE / "source_stages.json" +E3_STAGE_NAMES = [ + "frs_employment", + "frs_council_tax", + "frs_disability", + "frs_education", + "frs_legacy_proxies", + "frs_education_grant_split", +] +UK_SOURCE_STAGE_NAMES = [ + "frs_spine", + *E3_STAGE_NAMES, + "frs_hmrc_retained_leaves", + "hmrc_spi_income", +] FROZEN_SOURCE_STAGES_SHA256 = ( "c0341af7166ae3a85a3c1164e7d9e880c4b4aec122f1a8fa90c73b46c596e1ea" ) @@ -57,21 +71,13 @@ def test_source_stages_json_loads_as_shared_manifest(self) -> None: assert manifest.country == "uk" assert manifest.version == 1 - assert [stage.stage for stage in manifest.stages] == [ - "frs_spine", - "frs_hmrc_retained_leaves", - "hmrc_spi_income", - ] + assert [stage.stage for stage in manifest.stages] == UK_SOURCE_STAGE_NAMES def test_country_spec_declares_three_uk_source_stages(self) -> None: spec = load_country_spec("uk") assert spec.sources is not None - assert [stage.stage for stage in spec.sources.stages] == [ - "frs_spine", - "frs_hmrc_retained_leaves", - "hmrc_spi_income", - ] + assert [stage.stage for stage in spec.sources.stages] == UK_SOURCE_STAGE_NAMES def test_copy_is_lockstep_with_frozen_original_except_citation_rewrites( self, @@ -79,7 +85,7 @@ def test_copy_is_lockstep_with_frozen_original_except_citation_rewrites( frozen = _load_json(FROZEN_SOURCE_STAGES) canonical = _load_json(CANONICAL_SOURCE_STAGES) frozen_stage = frozen["stages"][0] - _, stage1, stage2 = canonical["stages"] + stage1, stage2 = canonical["stages"][-2:] expected_operations = copy.deepcopy(frozen_stage["operations"]) predictor_note = expected_operations[6]["reviewed_absent_predictors"][ @@ -144,6 +150,12 @@ def test_country_stage_plan_assembles_two_uk_national_stages(self) -> None: ( { "frs_spine": _identity, + "frs_employment": _identity, + "frs_council_tax": _identity, + "frs_disability": _identity, + "frs_education": _identity, + "frs_legacy_proxies": _identity, + "frs_education_grant_split": _identity, "frs_hmrc_retained_leaves": _identity, "hmrc_spi_income": _identity, "hmrc_spi_income_fallback": _identity, @@ -185,6 +197,146 @@ def test_stage1_outputs_are_exactly_the_retained_leaf_columns(self) -> None: stage1 = stages["frs_hmrc_retained_leaves"] assert stage1.outputs == tuple(FRS_HMRC_RETAINED_LEAF_COLUMNS) + def test_e3_outputs_are_backed_by_runtime_written_columns(self) -> None: + from microcosm.build.uk_runtime.frs_council_tax import ( + FRS_COUNCIL_TAX_OUTPUT_COLUMNS, + ) + from microcosm.build.uk_runtime.frs_disability import ( + FRS_DISABILITY_OUTPUT_COLUMNS, + ) + from microcosm.build.uk_runtime.frs_education import ( + FRS_EDUCATION_OUTPUT_COLUMNS, + ) + from microcosm.build.uk_runtime.frs_education_grants import ( + FRS_EDUCATION_GRANT_OUTPUT_COLUMNS, + FRS_EDUCATION_GRANT_REWRITES, + ) + from microcosm.build.uk_runtime.frs_employment import ( + FRS_EMPLOYMENT_OUTPUT_COLUMNS, + ) + from microcosm.build.uk_runtime.frs_legacy_proxies import ( + FRS_LEGACY_PROXY_OUTPUT_COLUMNS, + ) + + spec = load_country_spec("uk") + stages = {stage.stage: stage for stage in spec.sources.stages} + + assert stages["frs_employment"].outputs == FRS_EMPLOYMENT_OUTPUT_COLUMNS + assert stages["frs_council_tax"].outputs == FRS_COUNCIL_TAX_OUTPUT_COLUMNS + assert stages["frs_disability"].outputs == FRS_DISABILITY_OUTPUT_COLUMNS + assert stages["frs_education"].outputs == FRS_EDUCATION_OUTPUT_COLUMNS + assert ( + stages["frs_legacy_proxies"].outputs == FRS_LEGACY_PROXY_OUTPUT_COLUMNS + ) + assert ( + stages["frs_education_grant_split"].outputs + == FRS_EDUCATION_GRANT_OUTPUT_COLUMNS + ) + assert ( + stages["frs_education_grant_split"].rewrites + == FRS_EDUCATION_GRANT_REWRITES + ) + + +class TestE3ManifestLockstep: + def test_e3_raw_tab_pins_match_spine_artifacts(self) -> None: + spec = load_country_spec("uk") + stages = {stage.stage: stage for stage in spec.sources.stages} + spine_pins = { + artifact["table"]: ( + artifact["locator"], + artifact["sha256"], + artifact["size_bytes"], + ) + for artifact in stages["frs_spine"].artifacts + } + + for stage_name in E3_STAGE_NAMES: + for artifact in stages[stage_name].artifacts: + assert ( + artifact["locator"], + artifact["sha256"], + artifact["size_bytes"], + ) == spine_pins[artifact["table"]] + + def test_e3_operation_kinds_are_declared_in_order(self) -> None: + spec = load_country_spec("uk") + stages = {stage.stage: stage for stage in spec.sources.stages} + + assert [op.kind for op in stages["frs_employment"].operations] == [ + "read_tables", + "map_coded_amounts", + ] + assert [op.kind for op in stages["frs_council_tax"].operations] == [ + "read_tables", + "impute_cell_means", + ] + assert [op.kind for op in stages["frs_disability"].operations] == [ + "derive", + "derive", + ] + assert [op.kind for op in stages["frs_education"].operations] == [ + "read_tables", + "derive", + "impute_cell_means", + ] + assert [op.kind for op in stages["frs_legacy_proxies"].operations] == [ + "read_tables", + "materialize_rules_engine_predictors", + "derive", + ] + assert [ + op.kind for op in stages["frs_education_grant_split"].operations + ] == ["materialize_rules_engine_predictors", "derive"] + + def test_engine_predictor_and_rewrite_constants_match_manifest(self) -> None: + from microcosm.build.uk_runtime.frs_education_grants import ( + FRS_EDUCATION_GRANT_REWRITES, + UK_EDUCATION_GRANT_CAPACITY_PREDICTORS, + ) + from microcosm.build.uk_runtime.frs_legacy_proxies import ( + UK_LEGACY_PROXY_PREDICTORS, + ) + + spec = load_country_spec("uk") + stages = {stage.stage: stage for stage in spec.sources.stages} + + legacy_predictors = stages["frs_legacy_proxies"].operations[1].parameters[ + "predictors" + ] + grant_predictors = stages["frs_education_grant_split"].operations[ + 0 + ].parameters["predictors"] + assert tuple(legacy_predictors) == UK_LEGACY_PROXY_PREDICTORS + assert tuple(grant_predictors) == UK_EDUCATION_GRANT_CAPACITY_PREDICTORS + assert ( + stages["frs_education_grant_split"].rewrites + == FRS_EDUCATION_GRANT_REWRITES + ) + + def test_internal_disability_carriers_stay_out_of_export_registers(self) -> None: + from microcosm.build.uk_runtime.frs_disability import ( + UK_INTERNAL_DISABILITY_REPORTED_COLUMNS, + ) + from microcosm.build.uk_runtime.release_input_coverage import ( + uk_release_input_coverage_required_columns, + ) + from microcosm.build.uk_runtime.terminal_gates import ( + UK_ALLOWED_EXTRA_EXPORT_COLUMNS, + ) + + gates = _load_json(UK_PACKAGE / "gates.json") + export_gate = next( + gate for gate in gates["gates"] if gate["id"] == "uk_export_surface" + ) + allowed_extra = set(export_gate["parameters"]["allowed_extra_columns"]) + allowed_extra.update(UK_ALLOWED_EXTRA_EXPORT_COLUMNS) + required = uk_release_input_coverage_required_columns() + + for column in UK_INTERNAL_DISABILITY_REPORTED_COLUMNS: + assert f"person.{column}" not in allowed_extra + assert column not in required + def test_stage2_outputs_are_backed_by_runtime_written_columns(self) -> None: from microcosm.build.uk_runtime.spi_support import ( SPI_HMRC_DERIVED_AUXILIARY_COLUMNS, diff --git a/packages/microcosm-build/tests/test_us_plan.py b/packages/microcosm-build/tests/test_us_plan.py index 0369d027..9d4a66df 100644 --- a/packages/microcosm-build/tests/test_us_plan.py +++ b/packages/microcosm-build/tests/test_us_plan.py @@ -1018,6 +1018,7 @@ def test_no_incumbent_data_package_references_in_live_tree(self) -> None: ".git" in path.parts or ".venv" in path.parts or ".claude" in path.parts + or ".codex-work" in path.parts or "out" in path.parts # Run scaffolding and staged run outputs (launchers, base-rebuild # summaries) are not shipped source and may record the incumbent diff --git a/packages/microcosm-data/src/microcosm/data/contract.py b/packages/microcosm-data/src/microcosm/data/contract.py index 5537b0b1..225fbfc1 100644 --- a/packages/microcosm-data/src/microcosm/data/contract.py +++ b/packages/microcosm-data/src/microcosm/data/contract.py @@ -329,13 +329,13 @@ # fingerprint derives from the manifest digest. Editing the spec moves all # three here in the same reviewed change. _UK_GATE_BATTERY_POLICY_SHA256 = ( - "b147b50369e1f8b851f843e89b4a490b0a8c3b6a92e32b6bdb7bf40ca454cd0c" + "a852b3de381376ea401b55f4ed98c59f01c2335b0f8483c6ca5b9337c1cbce32" ) _UK_GATE_BATTERY_GATES_MANIFEST_SHA256 = ( - "6a98915343a7add9f469e9adebde5a9c85fdcbb3ecefee870df7b77c2b658e81" + "22e3b51e5886d8cf8bfb9a8a67b84fc99b112fbfd3dd7119fcff8bf79411924c" ) _UK_GATE_BATTERY_SPEC_FINGERPRINT = ( - "da0039af7d84d0dd7c5bac2016aa94dee938c787ff2400743e1b875cb81adfc6" + "63e7977a2fdcaaed2fe167ef5703e66f8ce90b2a0f0fc98c1645212266bc7c55" ) #: Spec entry id -> the legacy gate name whose observable detail checks #: apply unchanged (the battery re-keys the report by entry id; the gate @@ -347,6 +347,7 @@ "uk_weight_ess": "weight_ess", "uk_weight_ratio": "weight_ratio", "uk_weights_audit": "weights_audit", + "uk_nonnegative_columns": "nonnegative_columns", "uk_export_surface": "export_surface", "uk_target_surface": "target_surface", "uk_target_fit": "target_fit", @@ -369,6 +370,7 @@ "uk_weight_ess": ("weight_ess", "terminal"), "uk_weight_ratio": ("weight_ratio", "terminal"), "uk_weights_audit": ("weights_audit", "terminal"), + "uk_nonnegative_columns": ("nonnegative_columns", "terminal"), "uk_export_surface": ("export_surface", "terminal"), "uk_target_surface": ("target_surface", "terminal"), "uk_target_fit": ("target_fit", "terminal"), diff --git a/packages/microcosm-data/tests/test_contract.py b/packages/microcosm-data/tests/test_contract.py index 5e6626cf..a3b404fd 100644 --- a/packages/microcosm-data/tests/test_contract.py +++ b/packages/microcosm-data/tests/test_contract.py @@ -101,13 +101,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 = ( - "b147b50369e1f8b851f843e89b4a490b0a8c3b6a92e32b6bdb7bf40ca454cd0c" + "a852b3de381376ea401b55f4ed98c59f01c2335b0f8483c6ca5b9337c1cbce32" ) UK_GATE_BATTERY_GATES_MANIFEST_SHA256 = ( - "6a98915343a7add9f469e9adebde5a9c85fdcbb3ecefee870df7b77c2b658e81" + "22e3b51e5886d8cf8bfb9a8a67b84fc99b112fbfd3dd7119fcff8bf79411924c" ) UK_GATE_BATTERY_SPEC_FINGERPRINT = ( - "da0039af7d84d0dd7c5bac2016aa94dee938c787ff2400743e1b875cb81adfc6" + "63e7977a2fdcaaed2fe167ef5703e66f8ce90b2a0f0fc98c1645212266bc7c55" ) UK_GATE_BATTERY_DEGENERATE_EVIDENCE_SHA256 = ( "d0d024043132fa07c378c393dbe2b24fe99bf19e876bcc39997d2c80cc9bd4f6" @@ -137,6 +137,11 @@ def _trusted_terminal_gate_signing_key(monkeypatch) -> None: "uk_weight_ess": ("weight_ess", "terminal", "weight_ess"), "uk_weight_ratio": ("weight_ratio", "terminal", "weight_ratio"), "uk_weights_audit": ("weights_audit", "terminal", "weights_audit"), + "uk_nonnegative_columns": ( + "nonnegative_columns", + "terminal", + "nonnegative_columns", + ), "uk_export_surface": ("export_surface", "terminal", "export_surface"), "uk_target_surface": ("target_surface", "terminal", "target_surface"), "uk_target_fit": ("target_fit", "terminal", "target_fit"), @@ -709,6 +714,16 @@ def _terminal_gate_details(name: str) -> dict: "allowed_unweighted": {}, "unused_allowed_unweighted": [], } + if name == "nonnegative_columns": + return { + "columns_checked": 1, + "negative_counts": {}, + "minima": {"employment_income": 1.0}, + "reviewed_exclusions": {}, + "unused_reviewed_exclusions": [], + "atol": 0.0, + "chunk_size": 1_000_000, + } if name == "export_surface": return { "candidate_columns": 1, diff --git a/packages/microcosm-frame/src/microcosm/frame/adapters/policyengine_uk.py b/packages/microcosm-frame/src/microcosm/frame/adapters/policyengine_uk.py index b7763317..aacf2377 100644 --- a/packages/microcosm-frame/src/microcosm/frame/adapters/policyengine_uk.py +++ b/packages/microcosm-frame/src/microcosm/frame/adapters/policyengine_uk.py @@ -148,7 +148,7 @@ def _build_dataset(self, bundle: Frame, period: int | str) -> Any: person=tables["person"].copy(), benunit=tables["benunit"].copy(), household=tables["household"].copy(), - time_period=str(period), + fiscal_year=int(period), ) diff --git a/packages/microcosm-frame/tests/test_policyengine_uk_adapter.py b/packages/microcosm-frame/tests/test_policyengine_uk_adapter.py index ce376da4..719dc558 100644 --- a/packages/microcosm-frame/tests/test_policyengine_uk_adapter.py +++ b/packages/microcosm-frame/tests/test_policyengine_uk_adapter.py @@ -26,3 +26,40 @@ def test_policyengine_uk_adapter_export_side_is_not_implemented() -> None: with pytest.raises(NotImplementedError, match="write_uk_national_frame"): adapter.write_dataset(object(), "unused.h5", period=2023) # type: ignore[arg-type] + + +def test_policyengine_uk_adapter_builds_a_real_engine_dataset() -> None: + # Regression: the real UKSingleYearDataset constructor takes fiscal_year, + # not time_period — a kwarg mismatch the protocol tests above cannot see + # because they never import the engine. Dataset construction is cheap + # (no simulation), so this runs wherever the uk extra is installed. + pytest.importorskip("policyengine_uk") + import numpy as np + import pandas as pd + + from microcosm.frame import EntitySchema, Frame, WeightKind, Weights + + person = pd.DataFrame( + { + "person_id": np.asarray([10, 11, 12], dtype="int64"), + "person_benunit_id": np.asarray([1, 1, 2], dtype="int64"), + "person_household_id": np.asarray([5, 5, 6], dtype="int64"), + "age": [34.0, 3.0, 61.0], + } + ) + benunit = pd.DataFrame({"benunit_id": np.asarray([1, 2], dtype="int64")}) + household = pd.DataFrame({"household_id": np.asarray([5, 6], dtype="int64")}) + frame = Frame( + tables={"person": person, "benunit": benunit, "household": household}, + schema=EntitySchema(group_entities=("benunit", "household")), + weights={ + "household": Weights( + values=np.array([120.0, 250.0]), kind=WeightKind.DESIGN + ) + }, + ) + + dataset = PolicyEngineUKEngine()._build_dataset(frame, 2023) + + stamp = str(getattr(dataset, "time_period", getattr(dataset, "fiscal_year", ""))) + assert stamp.startswith("2023") diff --git a/tools/build_uk_frs_spine.py b/tools/build_uk_frs_spine.py index 7a471949..8c8c31bb 100644 --- a/tools/build_uk_frs_spine.py +++ b/tools/build_uk_frs_spine.py @@ -26,16 +26,37 @@ sha256_argument, write_error_receipt, ) +from microcosm.build.uk_runtime.frs_council_tax import UKFRSCouncilTaxStageTransform +from microcosm.build.uk_runtime.frs_disability import UKFRSDisabilityStageTransform +from microcosm.build.uk_runtime.frs_education import UKFRSEducationStageTransform +from microcosm.build.uk_runtime.frs_education_grants import ( + FRS_EDUCATION_GRANT_REWRITES, + UKFRSEducationGrantSplitStageTransform, +) +from microcosm.build.uk_runtime.frs_employment import UKFRSEmploymentStageTransform +from microcosm.build.uk_runtime.frs_legacy_proxies import ( + UKFRSLegacyProxiesStageTransform, +) from microcosm.build.uk_runtime.frs_spine import ( UKFRSSpineStageTransform, uk_frs_spine_seed_frame, ) from microcosm.build.uk_runtime.national_build import write_uk_national_frame from microcosm.build.uk_runtime.national_frame import uk_household_weight_kind +from microcosm.frame.adapters.policyengine_uk import PolicyEngineUKEngine _PIPELINE = "uk-frs-spine" _RUNG = "f100" _REPOSITORY = Path(__file__).resolve().parents[1] +_STAGE_NAMES = ( + "frs_spine", + "frs_employment", + "frs_council_tax", + "frs_disability", + "frs_education", + "frs_legacy_proxies", + "frs_education_grant_split", +) def _parse_args(argv: list[str] | None = None) -> argparse.Namespace: @@ -98,16 +119,33 @@ def _validate_args(args: argparse.Namespace) -> None: resolved[target] = label -def _artifact_pins(stage) -> dict[str, dict[str, object]]: +def _artifact_pins(stages) -> dict[str, dict[str, object]]: pins = {} - for artifact in stage.artifacts: - table = str(artifact["table"]) - pins[table] = { + for stage in stages: + for artifact in stage.artifacts: + table = str(artifact["table"]) + pin = { + "locator": str(artifact["locator"]), + "sha256": str(artifact["sha256"]), + "size_bytes": int(artifact["size_bytes"]), + } + if table in pins and pins[table] != pin: + raise ValueError( + f"FRS tab {table!r} has inconsistent artifact pins across stages." + ) + pins[table] = pin + return dict(sorted(pins.items())) + + +def _stage_artifact_pins(stage) -> dict[str, dict[str, object]]: + return { + str(artifact["table"]): { "locator": str(artifact["locator"]), "sha256": str(artifact["sha256"]), "size_bytes": int(artifact["size_bytes"]), } - return dict(sorted(pins.items())) + for artifact in stage.artifacts + } def _role_pins(pins: dict[str, dict[str, object]]) -> dict[str, dict[str, object]]: @@ -124,17 +162,42 @@ def _entity_row_counts(frame) -> dict[str, int]: return {entity: int(len(frame.table(entity))) for entity in frame.entities} -def _build_sidecar(*, frame, stage, records, artifact_pins) -> dict[str, object]: +def _rules_engine() -> PolicyEngineUKEngine: + try: + import policyengine_uk # noqa: F401 + except ImportError as exc: + raise ImportError( + "build_uk_frs_spine requires the microcosm-build 'uk' extra " + "(policyengine-uk). Run: uv sync --all-packages --extra uk" + ) from exc + return PolicyEngineUKEngine() + + +def _rules_engine_provenance() -> dict[str, str]: + try: + import policyengine_uk + except ImportError: + return {"package": "policyengine-uk", "version": "unavailable"} + return { + "package": "policyengine-uk", + "version": str(getattr(policyengine_uk, "__version__", "unknown")), + } + + +def _build_sidecar(*, frame, stages, records, artifact_pins) -> dict[str, object]: household_weight = frame.weights_for("household") return { - "schema_version": 1, + "schema_version": 2, "pipeline": _PIPELINE, - "stage": "frs_spine", + "stages": [stage.stage for stage in stages], "time_period": str(frame.metadata["time_period"]), "household_weight_kind": uk_household_weight_kind(frame).value, "household_weight_total": float(household_weight.values.sum()), "entity_row_counts": _entity_row_counts(frame), "artifact_pins": artifact_pins, + "stage_artifact_pins": { + stage.stage: _stage_artifact_pins(stage) for stage in stages + }, "stage_records": [ { "stage": record.stage, @@ -144,10 +207,30 @@ def _build_sidecar(*, frame, stage, records, artifact_pins) -> dict[str, object] } for record in records ], - "operations": [operation.kind for operation in stage.operations], + "operations": { + stage.stage: [operation.kind for operation in stage.operations] + for stage in stages + }, + "rules_engine": _rules_engine_provenance(), } +def _nonzero_shares(frame, columns: list[str]) -> dict[str, float]: + shares: dict[str, float] = {} + for column in columns: + for entity in frame.entities: + table = frame.table(entity) + if column not in table.columns: + continue + values = table[column] + if values.dtype == object: + shares[column] = float(values.astype(str).ne("").mean()) + else: + shares[column] = float((values != 0).mean()) + break + return shares + + def _new_build_id(timestamp: datetime) -> str: return f"uk-frs-spine-{timestamp.strftime('%Y%m%dT%H%M%SZ')}" @@ -203,12 +286,13 @@ def main(argv: list[str] | None = None) -> int: spec = load_country_spec("uk") if spec.sources is None: raise ValueError("UK country spec has no source stages.") - stage = spec.sources.stage_map()["frs_spine"] - artifact_pins = _artifact_pins(stage) + stages_by_name = spec.sources.stage_map() + stages = [stages_by_name[name] for name in _STAGE_NAMES] + artifact_pins = _artifact_pins(stages) state.input_pins_digest = role_pins_digest(_role_pins(artifact_pins)) run_config = { "pipeline": _PIPELINE, - "stage": "frs_spine", + "stages": list(_STAGE_NAMES), "artifact_pins_digest": state.input_pins_digest, "spine_h5": str(args.spine_h5), } @@ -216,15 +300,42 @@ def main(argv: list[str] | None = None) -> int: canonical_json_bytes(run_config) ).hexdigest() append_phase(state, "inputs_pinned") + engine = _rules_engine() plan = country_stage_plan( spec, { "frs_spine": UKFRSSpineStageTransform( args.frs_raw_dir, - stage=stage, - ) + stage=stages_by_name["frs_spine"], + ), + "frs_employment": UKFRSEmploymentStageTransform( + args.frs_raw_dir, + stage=stages_by_name["frs_employment"], + ), + "frs_council_tax": UKFRSCouncilTaxStageTransform( + args.frs_raw_dir, + stage=stages_by_name["frs_council_tax"], + ), + "frs_disability": UKFRSDisabilityStageTransform( + stage=stages_by_name["frs_disability"], + ), + "frs_education": UKFRSEducationStageTransform( + args.frs_raw_dir, + stage=stages_by_name["frs_education"], + ), + "frs_legacy_proxies": UKFRSLegacyProxiesStageTransform( + args.frs_raw_dir, + stage=stages_by_name["frs_legacy_proxies"], + engine=engine, + ), + "frs_education_grant_split": ( + UKFRSEducationGrantSplitStageTransform( + stage=stages_by_name["frs_education_grant_split"], + engine=engine, + ) + ), }, - stage_names=("frs_spine",), + stage_names=_STAGE_NAMES, ) frame, records = plan.run(uk_frs_spine_seed_frame()) append_phase(state, "spine_built") @@ -237,14 +348,32 @@ def main(argv: list[str] | None = None) -> int: sidecar_path = output.with_suffix(".build.json") sidecar = _build_sidecar( frame=frame, - stage=stage, + stages=stages, records=records, artifact_pins=artifact_pins, ) atomic_write_json(sidecar_path, sidecar) append_phase(state, "build_sidecar_written") if args.emit_nonzero_shares is not None: - atomic_write_json(args.emit_nonzero_shares, dict(records[0].nonzero_share)) + final_columns = list( + dict.fromkeys( + [ + column + for record in records + for column in record.produced + ] + + list(FRS_EDUCATION_GRANT_REWRITES) + ) + ) + atomic_write_json( + args.emit_nonzero_shares, + { + "stages": { + record.stage: dict(record.nonzero_share) for record in records + }, + "final": _nonzero_shares(frame, final_columns), + }, + ) append_phase(state, "nonzero_shares_written") state.artifact_location = local_artifact_reference( output,