From 4da1b0991fb9d878f4d84402e02d7b0b6dcd22eb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mar=C3=ADa=20Juaristi?= <127882282+juaristi22@users.noreply.github.com> Date: Tue, 18 Aug 2026 22:49:09 +0200 Subject: [PATCH 1/2] Create the SPI income channel on the raw spine as declarative source stages Three new manifest stages ahead of the untouched certified pair: frs_hmrc_spine_leaves (retained HMRC leaves aligned by raw person_id + the employer-pension derive), spi_support_channel (stack 10,000 zero-weight SPI-synthetic households at the pre-clone grain, gate the stratum count, allocate the 0.5 prior-mass share per region with the post-#710 semantics), and hmrc_spi_income_spine (SPI stage-1 QRF, FRS-only stage-2, base-channel dividend redraw after stage-2 training, band materialization, 208-fact classification, effective-mass gate). Four new generic operation kinds including the first row-appending manifest primitive (stack_zero_weight_donors). The spine driver runs all 14 stages and emits the .hmrc_replay.json sidecar. The certified national path, uk/gates.json, the data-shard contract, and the frozen HMRC manifest are all byte-unchanged. Refs #683. Co-Authored-By: Claude Fable 5 --- changelog.d/683-uk-spi-spine-channel.added.md | 1 + .../src/microcosm/build/source_manifest.py | 4 + .../src/microcosm/build/uk/source_stages.json | 882 +++++++++++++++++- .../microcosm/build/uk_runtime/__init__.py | 32 + .../microcosm/build/uk_runtime/hmrc_replay.py | 6 +- .../build/uk_runtime/source_runtime.py | 6 + .../microcosm/build/uk_runtime/spi_income.py | 70 ++ .../microcosm/build/uk_runtime/spi_spine.py | 708 ++++++++++++++ .../microcosm/build/uk_runtime/spi_support.py | 85 +- .../tests/test_country_spec.py | 30 + .../tests/test_uk_frs_spine.py | 344 ++++++- .../tests/test_uk_hmrc_replay.py | 8 + .../tests/test_uk_source_stages.py | 106 ++- .../tests/test_uk_spi_spine.py | 423 +++++++++ .../tests/test_uk_spi_support.py | 84 ++ tools/build_uk_frs_spine.py | 60 ++ 16 files changed, 2835 insertions(+), 14 deletions(-) create mode 100644 changelog.d/683-uk-spi-spine-channel.added.md create mode 100644 packages/microcosm-build/src/microcosm/build/uk_runtime/spi_spine.py create mode 100644 packages/microcosm-build/tests/test_uk_spi_spine.py diff --git a/changelog.d/683-uk-spi-spine-channel.added.md b/changelog.d/683-uk-spi-spine-channel.added.md new file mode 100644 index 000000000..d883c8e2c --- /dev/null +++ b/changelog.d/683-uk-spi-spine-channel.added.md @@ -0,0 +1 @@ +Create the SPI income channel on the raw UK spine (#683): three new manifest stages (`frs_hmrc_spine_leaves`, `spi_support_channel`, `hmrc_spi_income_spine`) stack 10,000 zero-weight SPI-synthetic households at the pre-clone grain, run the SPI stage-1 and FRS-only stage-2 QRFs, re-impute base-channel dividends, and emit the 208-fact HMRC replay sidecar from the spine driver — with four new generic operation kinds including the first row-appending manifest primitive (`stack_zero_weight_donors`). The certified-H5 path and the frozen HMRC manifest are unchanged. diff --git a/packages/microcosm-build/src/microcosm/build/source_manifest.py b/packages/microcosm-build/src/microcosm/build/source_manifest.py index 8e0a274d3..92823f9f2 100644 --- a/packages/microcosm-build/src/microcosm/build/source_manifest.py +++ b/packages/microcosm-build/src/microcosm/build/source_manifest.py @@ -49,6 +49,7 @@ "assign_clipped_normal", "assign_uniform_draw", "aggregate_person_to_benunit", + "allocate_zero_weight_prior_mass", "annualize_periodic_amounts", "assemble_group_entities", "attribute_self_employed_health_premiums", @@ -94,6 +95,7 @@ "fit_weighted_qrf_stage2", "fold_into", "gate_distributional_effective_mass", + "gate_zero_weight_strata", "head_carry", "join", "impute_retirement_contributions_to_puf_support", @@ -115,11 +117,13 @@ "read_table", "read_tables", "read_acs_rent_donor", + "redraw_columns_from_fitted_qrf", "replace_zero_weight_spi_support", "retain_adjudicated_frs_hmrc_leaves", "sample_categorical_from_count_table", "replace_sentinels", "split_component_by_share", + "stack_zero_weight_donors", "strict_read_private_table", "support_clip", "uprate", 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 e275af106..393def1b1 100644 --- a/packages/microcosm-build/src/microcosm/build/uk/source_stages.json +++ b/packages/microcosm-build/src/microcosm/build/uk/source_stages.json @@ -803,6 +803,886 @@ ], "notes": "Identity-keyed seed 0 streams replace the incumbent BRMA seed 0 sequential generator. Household collapse uses salt brma:household_pick." }, + { + "stage": "frs_hmrc_spine_leaves", + "survey": "Family Resources Survey 2023-24", + "source": "Department for Work and Pensions Family Resources Survey 2023-24 raw adult.tab and benefits.tab, caller-supplied local input", + "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": "benefits", + "kind": "licensed_microdata", + "format": "tab", + "vintage": "2023_24", + "locator": "benefits.tab", + "sha256": "ff30d054cc659bcf23b44c492d98cfd701c0bfdb63e8e9aa9769b490ba9d636b", + "size_bytes": 4460292, + "runtime_sha256_required": true + } + ], + "operations": [ + { + "kind": "retain_adjudicated_frs_hmrc_leaves", + "population": "uk_frs_raw_spine", + "source_vintage": "2023-24", + "mapped_build_period": 2023, + "annualization": "weekly raw FRS amounts * (365.25 / 7)", + "status": "adjudicated_partial_replay", + "retained_full_constituents": { + "hmrc_spi_pay": { + "spi_concept": "PAY", + "scope": "full", + "raw_sources": [ + "ADULT.INEARNS" + ], + "formula": "max(0, ADULT.INEARNS) * (365.25 / 7)" + }, + "hmrc_spi_unemployment_benefit_income": { + "spi_concept": "UBISJA", + "scope": "full", + "raw_sources": [ + "BENEFITS.BENEFIT=14:BENAMT", + "BENEFITS.BENEFIT=19:BENAMT" + ], + "formula": "sum(BENAMT where BENEFIT in {14, 19}) * (365.25 / 7)" + }, + "hmrc_spi_incapacity_benefit_income": { + "spi_concept": "INCPBEN", + "scope": "full", + "raw_sources": [ + "BENEFITS.BENEFIT=17:BENAMT" + ], + "formula": "sum(BENAMT where BENEFIT == 17) * (365.25 / 7)", + "observed_support": "structural zero in the audited 2023-24 FRS; retained so future vintages flow" + } + }, + "retained_named_subsets": { + "ossben_identifiable_subset": { + "spi_concept": "OSSBEN", + "raw_sources": [ + "BENEFITS.BENEFIT=13:BENAMT", + "BENEFITS.BENEFIT=16,VAR2 in {1,3}:BENAMT" + ], + "formula": "sum(BENAMT where BENEFIT == 13 or (BENEFIT == 16 and VAR2 in {1, 3})) * (365.25 / 7)", + "scope": "identifiable_subset" + }, + "srp_regular_code5": { + "spi_concept": "SRP", + "raw_sources": [ + "BENEFITS.BENEFIT=5:BENAMT" + ], + "formula": "sum(BENAMT where BENEFIT == 5) * (365.25 / 7)", + "scope": "regular_code5_subset" + } + }, + "source_absent_full_constituents": [ + "EPB", + "EXPS", + "TAXTERM", + "MOTHINC", + "OTHERINC" + ], + "full_concepts_forbidden_on_frs": [ + "hmrc_spi_employment_benefits", + "hmrc_spi_employment_expenses", + "hmrc_spi_taxable_termination_pay", + "hmrc_spi_miscellaneous_employment_income", + "hmrc_spi_other_income", + "hmrc_spi_other_social_security_income", + "hmrc_spi_state_pension_income" + ], + "forbid_proxy_substitution": [ + "employment_income", + "miscellaneous_income" + ], + "fail_on_missing_retained_constituent": true, + "fail_on_full_concept_alias": true + }, + { + "kind": "derive", + "output": "employer_pension_contributions", + "formula": "3 * employee_pension_contributions", + "source": "incumbent enhanced-FRS frs.py employer-pension contribution estimate" + } + ], + "outputs": [ + "hmrc_spi_pay", + "hmrc_spi_unemployment_benefit_income", + "hmrc_spi_incapacity_benefit_income", + "ossben_identifiable_subset", + "srp_regular_code5", + "employer_pension_contributions" + ], + "nonnegative_outputs": [ + "hmrc_spi_pay", + "hmrc_spi_unemployment_benefit_income", + "hmrc_spi_incapacity_benefit_income", + "ossben_identifiable_subset", + "srp_regular_code5", + "employer_pension_contributions" + ], + "notes": "Retains adjudicated raw-FRS HMRC leaves on the raw FRS spine, where person_id equals the raw sernum*1000+person identity, then ports the incumbent employer-pension-contributions estimate." + }, + { + "stage": "spi_support_channel", + "survey": "Family Resources Survey 2023-24", + "source": "Synthetic SPI support channel sampled uniformly without replacement from the raw FRS spine before cloning.", + "grain": "household", + "artifacts": [], + "operations": [ + { + "kind": "stack_zero_weight_donors", + "count": 10000, + "seed": 42, + "flag_column": "household_is_spi_synthetic", + "channels": { + "base": "frs", + "synthetic": "spi" + }, + "draw": "uniform_without_replacement" + }, + { + "kind": "gate_zero_weight_strata", + "gate": "uk_zero_weight_strata_gate", + "declarations": [ + { + "name": "e7_spi_synthetic_preclone", + "selector": { + "household_is_spi_synthetic": true + }, + "maximum_zero_weight_rows": 10000, + "reason": "E7 stacks exactly 10,000 zero-weight SPI-synthetic households on the raw pre-clone UK FRS spine." + } + ] + }, + { + "kind": "allocate_zero_weight_prior_mass", + "share": 0.5, + "strata": [ + "region" + ], + "weight_kind_out": "importance", + "conservation": "exact_total" + } + ], + "outputs": [ + "household_is_spi_synthetic", + "person_support_channel", + "person_support_clone_index", + "person_source_id", + "benunit_support_channel", + "benunit_support_clone_index", + "benunit_source_id", + "household_support_channel", + "household_support_clone_index", + "household_source_id", + "source_household_id", + "source_year", + "source_household_key" + ], + "nonnegative_outputs": [ + "person_support_clone_index", + "person_source_id", + "benunit_support_clone_index", + "benunit_source_id", + "household_support_clone_index", + "household_source_id", + "source_household_id", + "source_year" + ], + "notes": "Stacks the pre-clone SPI support channel, gates the 10,000 zero-weight synthetic households under a stage-local declaration, and allocates 50 percent of each region stratum prior mass to the SPI channel with exact national conservation." + }, + { + "stage": "hmrc_spi_income_spine", + "survey": "Survey of Personal Incomes Public Use Tape 2022-23 and HMRC Personal Incomes Tables 3.6/3.7 2023-24", + "source": "https://assets.publishing.service.gov.uk/media/69f1f12d2fae53a03709682f/Collated_Tables_3_1_to_3_11_2324.ods", + "grain": "person", + "artifacts": [ + { + "role": "qrf_donor", + "kind": "private_microdata", + "format": "tab_delimited", + "survey": "Survey of Personal Incomes Public Use Tape 2022-23", + "vintage": "2022-23", + "tax_year_start": 2022, + "ukds_study_number": "SN 9422", + "doi": "10.5255/UKDA-SN-9422-1", + "filename": "put2223uk.tab", + "sha256": "5ef829461060c91a2a47be59ad541d9b519fc3976d66ca80d4920f711bb96f66", + "size_bytes": 141323762, + "reviewed_source": "PolicyEngine licensed UKDS mirror (private Hugging Face repository), spi_2022_23.zip", + "access": "private_local_input", + "locator": "caller-supplied local input", + "runtime_sha256_required": true + }, + { + "role": "published_fact_surface", + "kind": "administrative_table", + "format": "ods", + "survey": "HMRC Personal Incomes Tables 3.6 and 3.7", + "publication": "https://www.gov.uk/government/statistics/personal-incomes-statistics-for-the-tax-year-2023-to-2024", + "vintage": "2023-24", + "tax_year_start": 2023, + "locator": "https://assets.publishing.service.gov.uk/media/69f1f12d2fae53a03709682f/Collated_Tables_3_1_to_3_11_2324.ods", + "sha256": "ad063b06b2bdeef8600dbbb09d48153337a4966f8c7eea50df7a2e0304ebd73e", + "size_bytes": 166693, + "mime_type": "application/vnd.oasis.opendocument.spreadsheet", + "sheets": [ + "Table_3_6", + "Table_3_7" + ], + "mapped_build_period": 2023, + "period_mapping": "tax_year_start", + "runtime_sha256_required": true + } + ], + "operations": [ + { + "kind": "verify_pinned_hmrc_source_pair", + "artifact_roles": [ + "qrf_donor", + "published_fact_surface" + ], + "require_before_source_read": true, + "runtime_sha256_required": true, + "fail_on_mismatch": true + }, + { + "kind": "strict_read_private_table", + "artifact_role": "qrf_donor", + "filename": "put2223uk.tab", + "delimiter": "\t", + "weight": "FACT", + "required_columns": [ + "AGERANGE", + "GORCODE", + "SEX", + "FACT", + "PAY", + "EPB", + "EXPS", + "TAXTERM", + "INCPBEN", + "OSSBEN", + "UBISJA", + "MOTHINC", + "OTHERINC", + "PROFITS", + "CAPALL", + "LOSSBF", + "SRP", + "INCBBS", + "DIVIDENDS", + "PENSION", + "INCPROP", + "OTHERINV", + "GIFTAID", + "GIFTINV", + "TEI", + "TII", + "TI" + ], + "runtime_sha256_required": true, + "fail_on_missing_file": true, + "fail_on_missing_columns": true, + "fail_on_invalid_weight": true, + "seed": 42 + }, + { + "kind": "fit_weighted_qrf_stage1", + "training_artifact_role": "qrf_donor", + "predictors": [ + "age", + "gender", + "region" + ], + "categorical_predictors": [ + "gender", + "region" + ], + "source_sampling_weight": "FACT", + "sample_size": 100000, + "sample_with_replacement": true, + "post_sample_fit_weight": "uniform", + "fit_weight_kind": "design", + "double_apply_source_weight": false, + "source_columns": { + "self_employment_income": [ + "PROFITS", + "CAPALL", + "LOSSBF" + ], + "savings_interest_income": [ + "INCBBS" + ], + "dividend_income": [ + "DIVIDENDS" + ], + "private_pension_income": [ + "PENSION" + ], + "property_income": [ + "INCPROP" + ], + "other_investment_income": [ + "OTHERINV" + ], + "gift_aid": [ + "GIFTAID" + ], + "charitable_investment_gifts": [ + "GIFTINV" + ], + "hmrc_spi_pay": [ + "PAY" + ], + "hmrc_spi_employment_benefits": [ + "EPB" + ], + "hmrc_spi_employment_expenses": [ + "EXPS" + ], + "hmrc_spi_incapacity_benefit_income": [ + "INCPBEN" + ], + "hmrc_spi_other_social_security_income": [ + "OSSBEN" + ], + "hmrc_spi_taxable_termination_pay": [ + "TAXTERM" + ], + "hmrc_spi_unemployment_benefit_income": [ + "UBISJA" + ], + "hmrc_spi_miscellaneous_employment_income": [ + "MOTHINC" + ], + "hmrc_spi_other_income": [ + "OTHERINC" + ], + "hmrc_spi_state_pension_income": [ + "SRP" + ] + }, + "derived_policyengine_outputs": { + "employment_income": { + "source_columns": [ + "PAY", + "EPB", + "TAXTERM" + ], + "formula": "hmrc_spi_pay + hmrc_spi_employment_benefits + hmrc_spi_taxable_termination_pay", + "derive_after_draw": true + } + }, + "outputs": [ + "self_employment_income", + "savings_interest_income", + "dividend_income", + "private_pension_income", + "property_income", + "other_investment_income", + "gift_aid", + "charitable_investment_gifts", + "hmrc_spi_pay", + "hmrc_spi_employment_benefits", + "hmrc_spi_employment_expenses", + "hmrc_spi_incapacity_benefit_income", + "hmrc_spi_other_social_security_income", + "hmrc_spi_taxable_termination_pay", + "hmrc_spi_unemployment_benefit_income", + "hmrc_spi_miscellaneous_employment_income", + "hmrc_spi_other_income", + "hmrc_spi_state_pension_income" + ], + "joint_draw": true, + "savings_interest_source_semantics": "INCBBS is taxable bank/building-society interest before reconstruction to the PolicyEngine gross input", + "employment_income_source_semantics": "PolicyEngine input = PAY + EPB + TAXTERM, matching the pinned enhanced-FRS pipeline; it is not the Table 3.6 measure", + "hmrc_employed_income_source_semantics": "Derived after each draw as max(0, PAY + EPB - EXPS) + INCPBEN + OSSBEN + TAXTERM + UBISJA + MOTHINC, using normalized leaves identically on FRS and SPI channels", + "self_employment_income_source_semantics": "max(0, PROFITS - CAPALL - LOSSBF)", + "assessable_income_source_semantics": "QRF draws leaves only; TEI, TII, and TI are deterministic post-draw accounting aggregates and TI equals TEI + TII exactly", + "source_ti_identity_fields": [ + "TI", + "TEI", + "TII" + ], + "source_leaf_reconciliation": { + "documentation_url": "https://doc.ukdataservice.ac.uk/doc/9422/mrdoc/pdf/9422_put_2223_full_documentation.pdf", + "composite_indicator": "AGERANGE == -1", + "formulas": { + "TEI": "max(0, PAY + EPB - EXPS) + INCPBEN + OSSBEN + TAXTERM + UBISJA + MOTHINC + OTHERINC + SRP + PENSION + max(0, PROFITS - CAPALL - LOSSBF)", + "TII": "OTHERINV + DIVIDENDS + INCPROP + INCBBS", + "TI": "TEI + TII" + }, + "maximum_absolute_difference_gbp": { + "ordinary": { + "TEI": 15, + "TII": 10, + "TI": 20 + }, + "composite": { + "TEI": 180, + "TII": 10, + "TI": 180 + } + }, + "rationale": "The official PUT rounds source fields, averages documented composite records, then rounds remaining income fields to GBP 5. These are the observed envelopes in the exact sha-pinned donor; post-draw synthetic identities remain exact." + }, + "ti_identity_absolute_tolerance_gbp": 5, + "stochastic_aggregates_forbidden": [ + "hmrc_spi_employed_income", + "hmrc_spi_total_earned_income", + "hmrc_spi_total_investment_income", + "hmrc_spi_assessable_income" + ], + "require_all_predictors": true, + "require_all_outputs": true, + "initialize_frs_channel_columns": { + "gift_aid": 0.0, + "charitable_investment_gifts": 0.0 + }, + "n_estimators": 100, + "seed": 42 + }, + { + "kind": "fit_weighted_qrf_stage2", + "training_population": "uk_frs_raw_spine_base_support_channel", + "target_population": "spi_synthetic_support_channel", + "predictors": [ + "age", + "gender", + "region", + "employment_income", + "self_employment_income", + "savings_interest_income", + "dividend_income", + "private_pension_income", + "property_income" + ], + "reviewed_absent_predictors": { + "other_investment_income": "This remains a stage-1 SPI draw and an official HMRC fact component, but it is not an FRS-only stage-2 predictor: the incumbent UK data build's frs_only.py defines exactly six income predictors and the certified Microcosm UK base candidate has no other_investment_income column." + }, + "categorical_predictors": [ + "gender", + "region" + ], + "weight": "household_weight", + "weight_mapping": "household_to_person", + "outputs": [ + "employee_pension_contributions", + "employer_pension_contributions", + "personal_pension_contributions", + "pension_contributions_via_salary_sacrifice", + "tax_free_savings_income", + "universal_credit_reported", + "pension_credit_reported", + "child_benefit_reported", + "housing_benefit_reported", + "income_support_reported", + "working_tax_credit_reported", + "child_tax_credit_reported", + "attendance_allowance_reported", + "state_pension_reported", + "dla_sc_reported", + "dla_m_reported", + "pip_m_reported", + "pip_dl_reported", + "sda_reported", + "carers_allowance_reported", + "iidb_reported", + "afcs_reported", + "bsp_reported", + "winter_fuel_allowance_reported", + "council_tax_benefit_reported", + "jsa_contrib_reported", + "jsa_income_reported", + "esa_contrib_reported", + "esa_income_reported" + ], + "reviewed_absent_outputs": { + "incapacity_benefit_reported": "Absent/all-default on the pinned enhanced-FRS export and certified Microcosm UK base; not a populated loader layer.", + "maternity_allowance_reported": "Absent from the pinned enhanced-FRS export and certified Microcosm UK base; no training source can be materialized for this stage." + }, + "postprocess": { + "gross_savings_interest_income": "stage1 INCBBS draw + stage2 tax_free_savings_income", + "refresh_disability_categories": [ + "aa_category", + "dla_sc_category", + "dla_m_category", + "pip_m_category", + "pip_dl_category" + ], + "refresh_disability_flags": [ + "is_disabled_for_benefits", + "is_enhanced_disabled_for_benefits", + "is_severely_disabled_for_benefits" + ] + }, + "joint_draw": true, + "require_all_predictors": true, + "require_all_materializable_outputs": true, + "require_all_outputs": false, + "n_estimators": 100, + "seed": 43 + }, + { + "kind": "redraw_columns_from_fitted_qrf", + "fit": "stage1", + "columns": [ + "dividend_income" + ], + "rows": "base_support_channel" + }, + { + "kind": "materialize_hmrc_income_bands_fail_closed", + "artifact_role": "published_fact_surface", + "mapped_build_period": 2023, + "period_mapping": "tax_year_start", + "column_index_base": 0, + "data_row_start_index": 5, + "stop_label": "All ranges", + "count_unit_multiplier": 1000, + "amount_unit_multiplier": 1000000, + "component_columns": { + "employment_income": { + "sheet": "Table_3_6", + "count_column_index": 4, + "amount_column_index": 5 + }, + "self_employment_income": { + "sheet": "Table_3_6", + "count_column_index": 1, + "amount_column_index": 2 + }, + "state_pension": { + "sheet": "Table_3_6", + "count_column_index": 7, + "amount_column_index": 8 + }, + "private_pension_income": { + "sheet": "Table_3_6", + "count_column_index": 10, + "amount_column_index": 11 + }, + "property_income": { + "sheet": "Table_3_7", + "count_column_index": 1, + "amount_column_index": 2 + }, + "savings_interest_income": { + "sheet": "Table_3_7", + "count_column_index": 4, + "amount_column_index": 5 + }, + "dividend_income": { + "sheet": "Table_3_7", + "count_column_index": 7, + "amount_column_index": 8 + }, + "other_investment_income": { + "sheet": "Table_3_7", + "count_column_index": 10, + "amount_column_index": 11 + } + }, + "required_band_lower_bounds_gbp": [ + 12570, + 15000, + 20000, + 30000, + 40000, + 50000, + 70000, + 100000, + 150000, + 200000, + 300000, + 500000, + 1000000 + ], + "required_measures": [ + "count", + "amount" + ], + "fail_on_missing_sheet": true, + "fail_on_missing_component": true, + "fail_on_missing_band": true, + "fail_on_non_numeric_value": true + }, + { + "kind": "classify_hmrc_income_facts_with_reviewed_fences", + "target_operation": "materialize_hmrc_income_bands_fail_closed", + "components": [ + "employment_income", + "self_employment_income", + "state_pension", + "private_pension_income", + "property_income", + "savings_interest_income", + "dividend_income", + "other_investment_income" + ], + "breakdown_dependency": "hmrc_spi_assessable_income", + "frs_breakdown_status": "unavailable_full_measure", + "input_weight_kind": "importance", + "output_weight_kind": "importance", + "calibration_permitted": false, + "required_fact_count": 208, + "outcome_counts": { + "exact_pass": 0, + "exact_fail": 0, + "directional_pass": 0, + "directional_fail": 0, + "excluded_with_fence": 208 + }, + "classification_rationale": "Every published fact uses non-overlapping total-income bands. The FRS channel cannot materialize full TEI, and omitted income can move a person between bands, so neither an exact fact nor a per-band directional bound is valid.", + "reviewed_fences": [ + { + "fence_id": "frs_epb_source_absent", + "constituents": [ + "EPB" + ], + "raw_sources_searched": [ + "JOB.EXPBEN01-EXPBEN13", + "JOB.CARVAL", + "JOB.CARAMT", + "JOB.FUELAMT", + "JOB.VCHAMT", + "JOB.CHVAMT" + ], + "finding": "Missing. EXPBEN* are receipt flags, and the amount fields cover only selected benefits; they cannot produce complete taxable expenses payments and benefits.", + "mass_implication": "12.9485464% of certified-candidate FRS effective person mass has at least one receipt flag, but this is not monetary support.", + "rationale": "Receipt flags and selected benefit amounts cannot be promoted to the SPI EPB monetary concept without an imputation or proxy.", + "dependent_fence_ids": [] + }, + { + "fence_id": "frs_exps_source_absent", + "constituents": [ + "EXPS" + ], + "raw_sources_searched": [ + "JOB.EXPBEN04/EXPBEN05", + "JOB.MILEAMT/JOB.MOTAMT", + "JOB.UMILEAMT/JOB.UMOTAMT", + "JOB.DEDUC1-DEDUC9", + "JOB.UDEDUC1-UDEDUC9" + ], + "finding": "Missing. These fields describe reimbursements or payroll deductions, not the complete tax-deductible employment-expense amount required by SPI.", + "mass_implication": "5.1302528% of certified-candidate FRS effective person mass has an adjacent reimbursement flag; the true EXPS mass is not estimable.", + "rationale": "The nearby fields do not measure the required deductible amount, and EXPS enters the employed-income identity with a negative sign.", + "dependent_fence_ids": [] + }, + { + "fence_id": "frs_taxterm_source_absent", + "constituents": [ + "TAXTERM" + ], + "raw_sources_searched": [ + "ADULT.REDAMT", + "ADULT and JOB taxable-termination split search" + ], + "finding": "Missing. REDAMT is gross redundancy pay and has neither the taxable amount nor non-redundancy termination pay.", + "mass_implication": "0.3746084% of certified-candidate FRS effective person mass has positive gross redundancy pay; taxable mass is unknown.", + "rationale": "Gross redundancy pay cannot be relabeled as taxable termination pay.", + "dependent_fence_ids": [] + }, + { + "fence_id": "frs_mothinc_source_absent", + "constituents": [ + "MOTHINC" + ], + "raw_sources_searched": [ + "ODDJOB.OJAMT/ODDJOB.OJNOW", + "ADULT.ALLPAY2", + "ADULT.ROYYR2-ROYYR4", + "JOB.OWNOTHER" + ], + "finding": "Missing. The fields are heterogeneous and belong to distinct income concepts; assigning their union to SPI miscellaneous employment income would be a proxy.", + "mass_implication": "Odd-job-only effective person mass is 0.1724207%; the broader unresolved miscellaneous pool is 1.4650566%.", + "rationale": "The FRS instrument cannot separate the SPI miscellaneous-employment concept source-faithfully.", + "dependent_fence_ids": [] + }, + { + "fence_id": "frs_otherinc_source_absent", + "constituents": [ + "OTHERINC" + ], + "raw_sources_searched": [ + "ADULT, ODDJOB, and JOB miscellaneous fields", + "PENSION", + "ACCOUNTS", + "ASSETS", + "BENEFITS" + ], + "finding": "Missing. No person-level raw FRS variable has SPI OTHERINC semantics, and the miscellaneous pool cannot be split between MOTHINC and OTHERINC from source evidence.", + "mass_implication": "No separable mass estimate exists; the unresolved miscellaneous pool is 1.4650566% of certified-candidate FRS effective person mass.", + "rationale": "A union of heterogeneous residual fields would be a new proxy, not a retained source constituent.", + "dependent_fence_ids": [] + }, + { + "fence_id": "frs_ossben_identifiable_subset", + "constituents": [ + "OSSBEN", + "ossben_identifiable_subset" + ], + "raw_sources_searched": [ + "BENEFITS.BENAMT", + "BENEFITS.BENEFIT", + "BENEFITS.VAR2", + "BENEFITS codes 13, 16, 6, and 30" + ], + "finding": "Incomplete. Carer's Allowance and contribution-based ESA form an identifiable subset, but code 6 mixes tax treatments and code 30 is an undifferentiated catch-all, so the complete taxable family cannot be emitted.", + "mass_implication": "1.8045088% of certified-candidate FRS effective person mass carries the identifiable lower-bound subset; it is not full OSSBEN support.", + "rationale": "The retained column must remain explicitly named as a subset and cannot satisfy the full SPI concept.", + "dependent_fence_ids": [] + }, + { + "fence_id": "frs_srp_regular_code5_subset", + "constituents": [ + "SRP", + "srp_regular_code5" + ], + "raw_sources_searched": [ + "BENEFITS.BENAMT where BENEFIT == 5", + "BENEFITS codes 6 and 9" + ], + "finding": "Incomplete. Code 5 supplies regular State Pension, but the FRS source does not identify the full SPI combination of State Pension lump sums and widow's pension; code 6 mixes benefits and code 9 is tax-free War Widow's Pension.", + "mass_implication": "18.1567916% of certified-candidate FRS effective person mass carries regular code-5 State Pension; it is not complete SRP support.", + "rationale": "The retained column must remain explicitly named as a subset and cannot be reported as the full published state-pension measure.", + "dependent_fence_ids": [] + }, + { + "fence_id": "full_frs_tei_band_unavailable", + "constituents": [ + "EPB", + "EXPS", + "TAXTERM", + "MOTHINC", + "OTHERINC", + "OSSBEN", + "SRP" + ], + "raw_sources_searched": [], + "finding": "The complete FRS TEI measure cannot be materialized from retained source constituents, so exact HMRC total-income band assignment is unavailable on the FRS channel.", + "mass_implication": "Every one of the 208 published facts is banded by total income and therefore depends on this unavailable like-for-like measure.", + "rationale": "A component-level subset does not imply a per-band lower bound: omitted income can move a taxpayer into or out of any non-overlapping published band. Biased partial bands are not emitted as estimates.", + "dependent_fence_ids": [ + "frs_epb_source_absent", + "frs_exps_source_absent", + "frs_taxterm_source_absent", + "frs_mothinc_source_absent", + "frs_otherinc_source_absent", + "frs_ossben_identifiable_subset", + "frs_srp_regular_code5_subset" + ] + } + ], + "fact_fence_id": "full_frs_tei_band_unavailable", + "blocked_dependency": "hmrc_spi_assessable_income", + "fail_on_unfenced_exclusion": true, + "fail_on_fact_count_mismatch": true, + "forbid_biased_estimate_or_delta": true + }, + { + "kind": "gate_distributional_effective_mass", + "columns": [ + "gift_aid", + "charitable_investment_gifts" + ], + "weight": "household_weight", + "weight_mapping": "household_to_person", + "support_channel_column": "person_support_channel", + "required_support_channel": "spi", + "mass_share_denominator": "all_person_effective_mass", + "minimum_nondefault_mass_share": 1e-06, + "fail_below_floor": true + } + ], + "outputs": [ + "other_investment_income", + "gift_aid", + "charitable_investment_gifts", + "hmrc_spi_employment_benefits", + "hmrc_spi_employment_expenses", + "hmrc_spi_other_social_security_income", + "hmrc_spi_taxable_termination_pay", + "hmrc_spi_miscellaneous_employment_income", + "hmrc_spi_other_income", + "hmrc_spi_state_pension_income", + "hmrc_spi_employed_income", + "hmrc_spi_total_earned_income", + "hmrc_spi_total_investment_income", + "hmrc_spi_assessable_income" + ], + "nonnegative_outputs": [ + "other_investment_income", + "gift_aid", + "charitable_investment_gifts", + "hmrc_spi_employment_benefits", + "hmrc_spi_employment_expenses", + "hmrc_spi_other_social_security_income", + "hmrc_spi_taxable_termination_pay", + "hmrc_spi_other_income", + "hmrc_spi_state_pension_income", + "hmrc_spi_employed_income", + "hmrc_spi_total_earned_income", + "hmrc_spi_total_investment_income", + "hmrc_spi_assessable_income" + ], + "rewrites": [ + "employment_income", + "self_employment_income", + "savings_interest_income", + "dividend_income", + "private_pension_income", + "property_income", + "employee_pension_contributions", + "employer_pension_contributions", + "personal_pension_contributions", + "pension_contributions_via_salary_sacrifice", + "tax_free_savings_income", + "universal_credit_reported", + "pension_credit_reported", + "child_benefit_reported", + "housing_benefit_reported", + "income_support_reported", + "working_tax_credit_reported", + "child_tax_credit_reported", + "attendance_allowance_reported", + "state_pension_reported", + "dla_sc_reported", + "dla_m_reported", + "pip_m_reported", + "pip_dl_reported", + "sda_reported", + "carers_allowance_reported", + "iidb_reported", + "afcs_reported", + "bsp_reported", + "winter_fuel_allowance_reported", + "council_tax_benefit_reported", + "jsa_contrib_reported", + "jsa_income_reported", + "esa_contrib_reported", + "esa_income_reported", + "hmrc_spi_pay", + "hmrc_spi_unemployment_benefit_income", + "hmrc_spi_incapacity_benefit_income" + ], + "notes": "Runs the SPI-trained income QRFs on the raw-spine support channel, initializes FRS charity columns to zero, trains FRS-only stage 2 before redrawing base-channel dividends, and emits a sidecar-only 208-fact replay report for the spine path." + }, { "stage": "frs_hmrc_retained_leaves", "survey": "Family Resources Survey 2023-24", @@ -1504,7 +2384,7 @@ "support_channel_column": "person_support_channel", "required_support_channel": "spi", "mass_share_denominator": "all_person_effective_mass", - "minimum_nondefault_mass_share": 0.000001, + "minimum_nondefault_mass_share": 1e-06, "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 609b8d434..e6e823792 100644 --- a/packages/microcosm-build/src/microcosm/build/uk_runtime/__init__.py +++ b/packages/microcosm-build/src/microcosm/build/uk_runtime/__init__.py @@ -396,6 +396,22 @@ impute_uk_spi_income_support, verify_spi_donor_identity, ) +from microcosm.build.uk_runtime.spi_spine import ( + EMPLOYER_PENSION_CONTRIBUTIONS_COLUMN, + UK_FRS_HMRC_SPINE_LEAF_OUTPUT_COLUMNS, + UK_FRS_HMRC_SPINE_LEAVES_STAGE_NAME, + UK_HMRC_SPI_INCOME_SPINE_STAGE_NAME, + UK_HMRC_SPI_SPINE_REPLAY_REPORT_KIND, + UK_SPI_INCOME_SPINE_NONNEGATIVE_OUTPUT_COLUMNS, + UK_SPI_INCOME_SPINE_OUTPUT_COLUMNS, + UK_SPI_INCOME_SPINE_REWRITE_COLUMNS, + UK_SPI_SUPPORT_CHANNEL_OUTPUT_COLUMNS, + UKFRSHMRCSpineLeavesResult, + UKFRSHMRCSpineLeavesStageTransform, + UKSPIIncomeSpineResult, + UKSPIIncomeSpineStageTransform, + UKSPISupportChannelStageTransform, +) from microcosm.build.uk_runtime.spi_support import ( BASE_FRS_SUPPORT_CHANNEL, DEFAULT_SPI_PRIOR_MASS_SHARE, @@ -411,6 +427,7 @@ SPI_SYNTHETIC_SUPPORT_CHANNEL, UK_SPI_SUPPORT_STAGE_NAME, UKSPISupportResult, + build_uk_spi_support_channel, create_uk_spi_support_tables, fill_support_channel_from_source, replace_uk_spi_support_tables, @@ -622,9 +639,17 @@ "UKEffectiveMassCoveragePolicy", "UKSPISupportResult", "UKSPIIncomeImputationResult", + "UKSPIIncomeSpineResult", "CERTIFIED_UK_CANDIDATE_SHA256", "CERTIFIED_UK_CANDIDATE_TIER", "UK_SINGLE_YEAR_TABLES", + "UK_FRS_HMRC_SPINE_LEAVES_STAGE_NAME", + "UK_HMRC_SPI_INCOME_SPINE_STAGE_NAME", + "UK_HMRC_SPI_SPINE_REPLAY_REPORT_KIND", + "UK_SPI_INCOME_SPINE_NONNEGATIVE_OUTPUT_COLUMNS", + "UK_SPI_INCOME_SPINE_OUTPUT_COLUMNS", + "UK_SPI_INCOME_SPINE_REWRITE_COLUMNS", + "UK_SPI_SUPPORT_CHANNEL_OUTPUT_COLUMNS", "UK_SPI_SUPPORT_STAGE_NAME", "UK_TARGET_GEOGRAPHY_LEVELS", "VAT_LIABILITY_BANDS", @@ -647,6 +672,7 @@ "assign_vat_flags", "build_firm_target_matrix", "build_conservative_hmrc_replay_report", + "build_uk_spi_support_channel", "build_uk_local_target_census", "build_uk_national_dataset", "calibrate_uk_hmrc_income", @@ -787,6 +813,8 @@ "FRS_EDUCATION_OUTPUT_COLUMNS", "FRS_EMPLOYMENT_OUTPUT_COLUMNS", "FRS_LEGACY_PROXY_OUTPUT_COLUMNS", + "EMPLOYER_PENSION_CONTRIBUTIONS_COLUMN", + "UK_FRS_HMRC_SPINE_LEAF_OUTPUT_COLUMNS", "UK_MAX_TARGET_ABS_RELATIVE_ERROR", "UKInputMassParityPolicy", "UKDSAPolicy", @@ -797,7 +825,11 @@ "UKFRSEducationGrantSplitStageTransform", "UKFRSEducationStageTransform", "UKFRSEmploymentStageTransform", + "UKFRSHMRCSpineLeavesResult", + "UKFRSHMRCSpineLeavesStageTransform", "UKFRSLegacyProxiesStageTransform", + "UKSPIIncomeSpineStageTransform", + "UKSPISupportChannelStageTransform", "UKInputMassReference", "UKLegacyJSAPolicy", "UK_INTERNAL_DISABILITY_REPORTED_COLUMNS", diff --git a/packages/microcosm-build/src/microcosm/build/uk_runtime/hmrc_replay.py b/packages/microcosm-build/src/microcosm/build/uk_runtime/hmrc_replay.py index d3c48f605..c086af959 100644 --- a/packages/microcosm-build/src/microcosm/build/uk_runtime/hmrc_replay.py +++ b/packages/microcosm-build/src/microcosm/build/uk_runtime/hmrc_replay.py @@ -641,10 +641,12 @@ class HMRCReplayReport: effective_mass_evidence: Mapping[str, object] diagnostic_aggregates: tuple[HMRCReplayDiagnosticAggregate, ...] = () schema_version: int = 1 + report_kind: str = "uk_hmrc_income_208_fact_replay" def __post_init__(self) -> None: if self.schema_version != 1: raise ValueError("HMRC replay report schema_version must be 1.") + _require_nonempty_string(self.report_kind, "report_kind") object.__setattr__(self, "facts", tuple(self.facts)) object.__setattr__(self, "fences", tuple(self.fences)) object.__setattr__( @@ -721,7 +723,7 @@ def summary(self) -> dict[str, object]: def to_payload(self) -> dict[str, object]: payload = { "schema_version": self.schema_version, - "report_kind": "uk_hmrc_income_208_fact_replay", + "report_kind": self.report_kind, "summary": self.summary, "source_evidence": _thaw_json(self.source_evidence), "build_evidence": _thaw_json(self.build_evidence), @@ -773,6 +775,7 @@ def build_conservative_hmrc_replay_report( qrf_evidence: Mapping[str, object], effective_mass_evidence: Mapping[str, object], diagnostic_aggregates: Sequence[HMRCReplayDiagnosticAggregate] = (), + report_kind: str = "uk_hmrc_income_208_fact_replay", ) -> HMRCReplayReport: """Build the reviewed 0 exact / 0 directional / 208 excluded report.""" @@ -784,6 +787,7 @@ def build_conservative_hmrc_replay_report( qrf_evidence=qrf_evidence, effective_mass_evidence=effective_mass_evidence, diagnostic_aggregates=tuple(diagnostic_aggregates), + report_kind=report_kind, ) 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 0e05f85a7..547c18d1e 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 @@ -62,6 +62,9 @@ def uk_stage_implementations( frs_person_draws_transform: Callable[[Frame], Frame] | None = None, frs_household_draws_transform: Callable[[Frame], Frame] | None = None, frs_brma_transform: Callable[[Frame], Frame] | None = None, + frs_hmrc_spine_leaves_transform: Callable[[Frame], Frame] | None = None, + spi_support_channel_transform: Callable[[Frame], Frame] | None = None, + hmrc_spi_income_spine_transform: Callable[[Frame], Frame] | None = None, ) -> dict[str, Callable[[Frame], Frame]]: """Return the whole-stage implementation map for the UK source plan.""" @@ -81,6 +84,9 @@ def uk_stage_implementations( "frs_person_draws": frs_person_draws_transform, "frs_household_draws": frs_household_draws_transform, "frs_brma": frs_brma_transform, + "frs_hmrc_spine_leaves": frs_hmrc_spine_leaves_transform, + "spi_support_channel": spi_support_channel_transform, + "hmrc_spi_income_spine": hmrc_spi_income_spine_transform, } implementations.update( { diff --git a/packages/microcosm-build/src/microcosm/build/uk_runtime/spi_income.py b/packages/microcosm-build/src/microcosm/build/uk_runtime/spi_income.py index 4b160b315..d0eefeab5 100644 --- a/packages/microcosm-build/src/microcosm/build/uk_runtime/spi_income.py +++ b/packages/microcosm-build/src/microcosm/build/uk_runtime/spi_income.py @@ -3,6 +3,7 @@ from __future__ import annotations import hashlib +from collections.abc import Mapping, Sequence from dataclasses import dataclass from functools import cache from importlib import import_module @@ -334,6 +335,8 @@ def impute_uk_spi_income_support( donor_sample_size: int | None = DEFAULT_SPI_DONOR_SAMPLE_SIZE, build_period: int | str = 2023, verified_donor: VerifiedSPIDonorIdentity | None = None, + initialize_frs_channel_columns: Mapping[str, float] | None = None, + stage1_base_redraw_columns: Sequence[str] = (), ) -> UKSPIIncomeImputationResult: """Run strict SPI-income and FRS-only QRFs on rebuilt positive support.""" @@ -403,6 +406,22 @@ def impute_uk_spi_income_support( spi_people = person[person_channel] == SPI_SYNTHETIC_SUPPORT_CHANNEL if not spi_people.any(): raise ValueError("SPI support has no person rows to impute.") + base_people = person[person_channel] == BASE_FRS_SUPPORT_CHANNEL + if not base_people.any(): + raise ValueError("SPI support has no FRS base rows.") + if initialize_frs_channel_columns is not None: + person = _initialize_frs_channel_columns( + person, + base_people=base_people, + columns=initialize_frs_channel_columns, + ) + base_redraw_columns = tuple(stage1_base_redraw_columns) + unknown_redraw = sorted(set(base_redraw_columns) - set(SPI_INCOME_QRF_OUTPUT_COLUMNS)) + if unknown_redraw: + raise ValueError( + "SPI stage-1 base redraw columns must be stage-1 QRF outputs; " + f"unknown column(s): {unknown_redraw}." + ) person = _seed_frs_hmrc_auxiliary_leaves(person, spi_people=spi_people) recipient_predictors = _person_predictors( @@ -506,6 +525,32 @@ def impute_uk_spi_income_support( for column in stage2_outputs: person.loc[spi_people, column] = stage2_draws[column].to_numpy() + if base_redraw_columns: + base_predictors = _person_predictors( + person.loc[base_people], + household, + income_predictors=(), + ) + _, encoded_base = _encode_predictor_pair( + donor[["age", "gender", "region"]], + base_predictors, + ) + base_draws = stage1.predict(encoded_base) + _validate_predictions( + base_draws, + expected=SPI_INCOME_QRF_OUTPUT_COLUMNS, + label="SPI stage-1 base redraw", + ) + if ( + base_draws[list(base_redraw_columns)] + .to_numpy(dtype=np.float64) + .min(initial=0.0) + < 0.0 + ): + raise ValueError("SPI stage-1 base redraw produced negative outputs.") + for column in base_redraw_columns: + person.loc[base_people, column] = base_draws[column].to_numpy() + tax_free = person.loc[spi_people, "tax_free_savings_income"].to_numpy( dtype=np.float64 ) @@ -540,6 +585,31 @@ def impute_uk_spi_income_support( ) +def _initialize_frs_channel_columns( + person: pd.DataFrame, + *, + base_people: pd.Series, + columns: Mapping[str, float], +) -> pd.DataFrame: + if not isinstance(columns, Mapping): + raise TypeError("initialize_frs_channel_columns must be a mapping.") + if not base_people.index.equals(person.index): + raise ValueError("FRS base mask must align to the person table.") + result = person.copy() + for column, value in columns.items(): + if not isinstance(column, str) or not column: + raise ValueError("initialize_frs_channel_columns keys must be strings.") + numeric = float(value) + if not np.isfinite(numeric): + raise ValueError( + f"initialize_frs_channel_columns[{column!r}] must be finite." + ) + if column not in result: + result[column] = np.nan + result.loc[base_people, column] = numeric + return result + + def _prepare_spi_donor(raw: pd.DataFrame, *, seed: int) -> pd.DataFrame: _require_columns(raw, SPI_DONOR_REQUIRED_COLUMNS, label="SPI 2022-23 donor") numeric = pd.DataFrame( diff --git a/packages/microcosm-build/src/microcosm/build/uk_runtime/spi_spine.py b/packages/microcosm-build/src/microcosm/build/uk_runtime/spi_spine.py new file mode 100644 index 000000000..4c3f4ce01 --- /dev/null +++ b/packages/microcosm-build/src/microcosm/build/uk_runtime/spi_spine.py @@ -0,0 +1,708 @@ +"""SPI income stages for the raw UK FRS spine build.""" + +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +import pandas as pd + +from microcosm.build.gates import FitWeightRecord +from microcosm.build.source_manifest import SourceStageSpec +from microcosm.build.uk_runtime.frs_hmrc_leaves import ( + FRS_HMRC_INCPBEN_COLUMN, + FRS_HMRC_PAY_COLUMN, + FRS_HMRC_RETAINED_LEAF_COLUMNS, + FRS_HMRC_UBISJA_COLUMN, + _materialize_source_leaves, + _read_raw_frs_table, +) +from microcosm.build.uk_runtime.hmrc_income import ( + HMRCIncomeTargetSet, + materialize_hmrc_spi_income_band_targets, + verify_hmrc_spi_collated_ods, +) +from microcosm.build.uk_runtime.hmrc_replay import ( + HMRCReplayReport, + build_conservative_hmrc_replay_report, +) +from microcosm.build.uk_runtime.hmrc_restoration import ( + _assert_post_draw_identity, + _distributional_mass_shares, +) +from microcosm.build.uk_runtime.national_frame import ( + uk_household_weight_kind, + uk_national_frame, + uk_time_period, + validate_uk_national_frame, +) +from microcosm.build.uk_runtime.release_input_coverage import ( + DEFAULT_MINIMUM_NONDEFAULT_MASS_SHARE, +) +from microcosm.build.uk_runtime.spi_income import ( + DEFAULT_SPI_DONOR_SAMPLE_SIZE, + SPI_SOURCE_TI_FORMULA, + UKSPIIncomeImputationResult, + assert_frs_hmrc_auxiliary_crosswalk_available, + impute_uk_spi_income_support, + verify_spi_donor_identity, +) +from microcosm.build.uk_runtime.spi_support import ( + DEFAULT_SPI_PRIOR_MASS_SHARE, + DEFAULT_SPI_SUPPORT_HOUSEHOLDS, + HOUSEHOLD_IS_SPI_SYNTHETIC_COLUMN, + SPI_HMRC_DERIVED_AUXILIARY_COLUMNS, + SPI_HMRC_EMPLOYMENT_BENEFITS_COLUMN, + SPI_HMRC_EMPLOYMENT_EXPENSES_COLUMN, + SPI_HMRC_MISCELLANEOUS_EMPLOYMENT_INCOME_COLUMN, + SPI_HMRC_OTHER_INCOME_COLUMN, + SPI_HMRC_OTHER_SOCIAL_SECURITY_INCOME_COLUMN, + SPI_HMRC_STATE_PENSION_INCOME_COLUMN, + SPI_HMRC_TAXABLE_TERMINATION_PAY_COLUMN, + SPI_SYNTHETIC_SUPPORT_CHANNEL, + UKSPISupportResult, + build_uk_spi_support_channel, + support_channel_column, + support_clone_index_column, + support_source_id_column, +) +from microcosm.build.uk_runtime.terminal_gates import ( + UKZeroWeightStratumDeclaration, +) +from microcosm.frame import Frame, WeightKind, engine_tables + +UK_FRS_HMRC_SPINE_LEAVES_STAGE_NAME = "frs_hmrc_spine_leaves" +UK_HMRC_SPI_INCOME_SPINE_STAGE_NAME = "hmrc_spi_income_spine" +EMPLOYER_PENSION_CONTRIBUTIONS_COLUMN = "employer_pension_contributions" +UK_HMRC_SPI_SPINE_REPLAY_REPORT_KIND = "uk_hmrc_spi_income_spine_208_fact_replay" + +UK_FRS_HMRC_SPINE_LEAF_OUTPUT_COLUMNS = ( + *FRS_HMRC_RETAINED_LEAF_COLUMNS, + EMPLOYER_PENSION_CONTRIBUTIONS_COLUMN, +) +UK_SPI_SUPPORT_CHANNEL_OUTPUT_COLUMNS = ( + HOUSEHOLD_IS_SPI_SYNTHETIC_COLUMN, + support_channel_column("person"), + support_clone_index_column("person"), + support_source_id_column("person"), + support_channel_column("benunit"), + support_clone_index_column("benunit"), + support_source_id_column("benunit"), + support_channel_column("household"), + support_clone_index_column("household"), + support_source_id_column("household"), + "source_household_id", + "source_year", + "source_household_key", +) +UK_SPI_INCOME_SPINE_OUTPUT_COLUMNS = ( + "other_investment_income", + "gift_aid", + "charitable_investment_gifts", + SPI_HMRC_EMPLOYMENT_BENEFITS_COLUMN, + SPI_HMRC_EMPLOYMENT_EXPENSES_COLUMN, + SPI_HMRC_OTHER_SOCIAL_SECURITY_INCOME_COLUMN, + SPI_HMRC_TAXABLE_TERMINATION_PAY_COLUMN, + SPI_HMRC_MISCELLANEOUS_EMPLOYMENT_INCOME_COLUMN, + SPI_HMRC_OTHER_INCOME_COLUMN, + SPI_HMRC_STATE_PENSION_INCOME_COLUMN, + *SPI_HMRC_DERIVED_AUXILIARY_COLUMNS, +) +UK_SPI_INCOME_SPINE_NONNEGATIVE_OUTPUT_COLUMNS = tuple( + column + for column in UK_SPI_INCOME_SPINE_OUTPUT_COLUMNS + if column != SPI_HMRC_MISCELLANEOUS_EMPLOYMENT_INCOME_COLUMN +) +UK_SPI_INCOME_SPINE_REWRITE_COLUMNS = ( + "employment_income", + "self_employment_income", + "savings_interest_income", + "dividend_income", + "private_pension_income", + "property_income", + "employee_pension_contributions", + EMPLOYER_PENSION_CONTRIBUTIONS_COLUMN, + "personal_pension_contributions", + "pension_contributions_via_salary_sacrifice", + "tax_free_savings_income", + "universal_credit_reported", + "pension_credit_reported", + "child_benefit_reported", + "housing_benefit_reported", + "income_support_reported", + "working_tax_credit_reported", + "child_tax_credit_reported", + "attendance_allowance_reported", + "state_pension_reported", + "dla_sc_reported", + "dla_m_reported", + "pip_m_reported", + "pip_dl_reported", + "sda_reported", + "carers_allowance_reported", + "iidb_reported", + "afcs_reported", + "bsp_reported", + "winter_fuel_allowance_reported", + "council_tax_benefit_reported", + "jsa_contrib_reported", + "jsa_income_reported", + "esa_contrib_reported", + "esa_income_reported", + FRS_HMRC_PAY_COLUMN, + FRS_HMRC_UBISJA_COLUMN, + FRS_HMRC_INCPBEN_COLUMN, +) + + +@dataclass(frozen=True) +class UKFRSHMRCSpineLeavesResult: + frame: Frame + source_signal_rows: dict[str, int] + structural_zero_columns: tuple[str, ...] + + def evidence(self) -> dict[str, object]: + return { + "stage": UK_FRS_HMRC_SPINE_LEAVES_STAGE_NAME, + "source_signal_rows": dict(self.source_signal_rows), + "structural_zero_columns": list(self.structural_zero_columns), + } + + +@dataclass(frozen=True) +class UKSPIIncomeSpineResult: + frame: Frame + support: UKSPISupportResult + imputation: UKSPIIncomeImputationResult + source_targets: HMRCIncomeTargetSet + replay_report: HMRCReplayReport + distributional_mass_shares: Mapping[str, float] + post_draw_identity_rows: int + + def evidence(self) -> dict[str, object]: + return { + "stage": UK_HMRC_SPI_INCOME_SPINE_STAGE_NAME, + "source_vintages": { + "spi_donor": "2022-23", + "hmrc_surface": self.source_targets.source.source_vintage, + "mapped_build_period": self.source_targets.source.build_period, + }, + "spi_prior": { + "households": self.support.n_spi_households, + "mass_share": self.support.spi_prior_mass_share, + "weight_kind": self.support.household_weight_kind.value, + "mass_change_reason": self.support.mass_log[-1].reason, + }, + "reviewed_absent_stage2_outputs": dict( + self.imputation.reviewed_absent_stage2_outputs + ), + "targets": { + "count": len(self.source_targets.targets), + "classification": dict(self.replay_report.summary), + }, + "effective_mass": { + "minimum_nondefault_mass_share": ( + DEFAULT_MINIMUM_NONDEFAULT_MASS_SHARE + ), + "columns": dict(self.distributional_mass_shares), + }, + "post_draw_identity": { + "formula": SPI_SOURCE_TI_FORMULA, + "rows_checked": self.post_draw_identity_rows, + "exact": True, + }, + } + + +@dataclass(frozen=True) +class UKFRSHMRCSpineLeavesStageTransform: + frs_raw_dir: Path + stage: SourceStageSpec + last_result: UKFRSHMRCSpineLeavesResult | None = field(default=None, init=False) + + def __init__(self, frs_raw_dir: str | Path, *, stage: SourceStageSpec) -> None: + object.__setattr__(self, "frs_raw_dir", Path(frs_raw_dir)) + object.__setattr__(self, "stage", stage) + object.__setattr__(self, "last_result", None) + + def __call__(self, frame: Frame) -> Frame: + validate_uk_national_frame(frame) + artifacts = _artifact_by_table(self.stage, expected=("adult", "benefits")) + adult, adult_identity = _read_raw_frs_table( + self.frs_raw_dir / str(artifacts["adult"]["locator"]), + expected_filename="adult.tab", + required_columns=("sernum", "person", "inearns"), + ) + benefits, benefits_identity = _read_raw_frs_table( + self.frs_raw_dir / str(artifacts["benefits"]["locator"]), + expected_filename="benefits.tab", + required_columns=("sernum", "person", "benefit", "benamt", "var2"), + ) + _assert_identity_matches_artifact(adult_identity.evidence(), artifacts["adult"]) + _assert_identity_matches_artifact( + benefits_identity.evidence(), artifacts["benefits"] + ) + source_leaves = _materialize_source_leaves(adult, benefits) + person = frame.table("person").copy() + missing_ids = sorted(set(source_leaves.index) - set(person["person_id"])) + if missing_ids: + raise ValueError( + "Raw FRS retained leaves contain person identity value(s) absent " + f"from the raw spine: {missing_ids[:5]}." + ) + aligned = source_leaves.reindex(person["person_id"], fill_value=0.0) + for column in FRS_HMRC_RETAINED_LEAF_COLUMNS: + person[column] = aligned[column].to_numpy(dtype=float) + if "employee_pension_contributions" not in person: + raise ValueError( + "FRS HMRC spine leaves require employee_pension_contributions " + "to derive employer pension contributions." + ) + employee = pd.to_numeric( + person["employee_pension_contributions"], errors="coerce" + ) + if employee.isna().any() or (employee < 0.0).any(): + raise ValueError( + "employee_pension_contributions must be finite and " + "non-negative." + ) + person[EMPLOYER_PENSION_CONTRIBUTIONS_COLUMN] = ( + 3.0 * employee.to_numpy(dtype=float) + ) + result_frame = uk_national_frame( + person=person, + benunit=frame.table("benunit"), + household=engine_tables(frame)["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_frame) + source_signal_rows = { + column: int((source_leaves[column] > 0.0).sum()) + for column in FRS_HMRC_RETAINED_LEAF_COLUMNS + } + result = UKFRSHMRCSpineLeavesResult( + frame=result_frame, + source_signal_rows=source_signal_rows, + structural_zero_columns=tuple( + column + for column, rows in source_signal_rows.items() + if rows == 0 + ), + ) + object.__setattr__(self, "last_result", result) + return result.frame + + @staticmethod + def output_columns() -> tuple[str, ...]: + return UK_FRS_HMRC_SPINE_LEAF_OUTPUT_COLUMNS + + def checkpoint_metadata(self) -> dict[str, object]: + if self.last_result is None: + raise RuntimeError("checkpoint metadata requires a completed stage run.") + return {"evidence": self.last_result.evidence()} + + +@dataclass(frozen=True) +class UKSPISupportChannelStageTransform: + stage: SourceStageSpec + seed: int = 42 + last_result: UKSPISupportResult | None = field(default=None, init=False) + + def __call__(self, frame: Frame) -> Frame: + validate_uk_national_frame(frame) + tables = engine_tables(frame) + count, share, strata, declarations = _support_stage_parameters( + self.stage, + seed=self.seed, + ) + result = build_uk_spi_support_channel( + tables["person"], + tables["benunit"], + tables["household"], + spi_household_count=count, + seed=self.seed, + source_year=int(uk_time_period(frame)), + spi_prior_mass_share=share, + strata_columns=strata, + input_weight_kind=uk_household_weight_kind(frame), + mass_log=frame.mass_log, + zero_weight_declarations=declarations, + ) + result_frame = uk_national_frame( + person=result.person, + benunit=result.benunit, + household=result.household, + time_period=uk_time_period(frame), + weight_kind=result.household_weight_kind or WeightKind.IMPORTANCE, + household_weights=result.household["household_weight"].to_numpy( + dtype=float + ), + mass_log=result.mass_log, + ) + validate_uk_national_frame(result_frame) + object.__setattr__(self, "last_result", result) + return result_frame + + @staticmethod + def output_columns() -> tuple[str, ...]: + return UK_SPI_SUPPORT_CHANNEL_OUTPUT_COLUMNS + + def checkpoint_metadata(self) -> dict[str, object]: + if self.last_result is None: + raise RuntimeError("checkpoint metadata requires a completed stage run.") + return { + "evidence": { + "stage": "spi_support_channel", + "spi_households": self.last_result.n_spi_households, + "spi_prior_mass_share": self.last_result.spi_prior_mass_share, + "household_weight_kind": ( + self.last_result.household_weight_kind.value + if self.last_result.household_weight_kind is not None + else None + ), + } + } + + +@dataclass(frozen=True) +class UKSPIIncomeSpineStageTransform: + spi_tab_path: Path + hmrc_ods_path: Path + stage: SourceStageSpec + seed: int = 42 + qrf_estimators: int = 100 + donor_sample_size: int | None = DEFAULT_SPI_DONOR_SAMPLE_SIZE + last_result: UKSPIIncomeSpineResult | None = field(default=None, init=False) + + def __init__( + self, + spi_tab_path: str | Path, + hmrc_ods_path: str | Path, + *, + stage: SourceStageSpec, + seed: int = 42, + qrf_estimators: int = 100, + donor_sample_size: int | None = DEFAULT_SPI_DONOR_SAMPLE_SIZE, + ) -> None: + object.__setattr__(self, "spi_tab_path", Path(spi_tab_path)) + object.__setattr__(self, "hmrc_ods_path", Path(hmrc_ods_path)) + object.__setattr__(self, "stage", stage) + object.__setattr__(self, "seed", seed) + object.__setattr__(self, "qrf_estimators", qrf_estimators) + object.__setattr__(self, "donor_sample_size", donor_sample_size) + object.__setattr__(self, "last_result", None) + + @property + def fit_weight_records(self) -> tuple[FitWeightRecord, ...]: + if self.last_result is None: + return () + return tuple(self.last_result.imputation.fit_weight_records) + + def __call__(self, frame: Frame) -> Frame: + validate_uk_national_frame(frame) + _assert_income_stage_parameters( + self.stage, + seed=self.seed, + qrf_estimators=self.qrf_estimators, + donor_sample_size=self.donor_sample_size, + ) + donor_identity = verify_spi_donor_identity(self.spi_tab_path) + ods_identity = verify_hmrc_spi_collated_ods(self.hmrc_ods_path) + tables = engine_tables(frame) + assert_frs_hmrc_auxiliary_crosswalk_available(tables["person"]) + support = _support_result_from_frame(frame, tables) + stage1_op = _operation(self.stage, "fit_weighted_qrf_stage1") + redraw_op = _operation(self.stage, "redraw_columns_from_fitted_qrf") + imputation = impute_uk_spi_income_support( + support, + self.spi_tab_path, + seed=self.seed, + n_estimators=self.qrf_estimators, + donor_sample_size=self.donor_sample_size, + build_period=uk_time_period(frame), + verified_donor=donor_identity, + initialize_frs_channel_columns=stage1_op.parameters[ + "initialize_frs_channel_columns" + ], + stage1_base_redraw_columns=redraw_op.parameters["columns"], + ) + result_frame = uk_national_frame( + person=imputation.person, + benunit=tables["benunit"], + household=tables["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_frame) + source_targets = materialize_hmrc_spi_income_band_targets( + ods_identity, + build_period=uk_time_period(frame), + ) + identity_rows = _assert_post_draw_identity(result_frame) + distributional_mass_shares = _distributional_mass_shares(result_frame) + insufficient = { + name: share + for name, share in distributional_mass_shares.items() + if share < DEFAULT_MINIMUM_NONDEFAULT_MASS_SHARE + } + if insufficient: + raise RuntimeError( + "Rebuilt SPI spine channel did not restore required " + f"effective-mass coverage: {insufficient}." + ) + report = _build_spine_replay_report( + source_targets=source_targets, + support=support, + imputation=imputation, + frame=result_frame, + distributional_mass_shares=distributional_mass_shares, + identity_rows=identity_rows, + ) + result = UKSPIIncomeSpineResult( + frame=result_frame, + support=support, + imputation=imputation, + source_targets=source_targets, + replay_report=report, + distributional_mass_shares=distributional_mass_shares, + post_draw_identity_rows=identity_rows, + ) + object.__setattr__(self, "last_result", result) + return result.frame + + @staticmethod + def output_columns() -> tuple[str, ...]: + return UK_SPI_INCOME_SPINE_OUTPUT_COLUMNS + + def checkpoint_metadata(self) -> dict[str, object]: + if self.last_result is None: + raise RuntimeError("checkpoint metadata requires a completed stage run.") + return { + "fit_weight_records": [ + { + "fit_name": record.fit_name, + "weight_kind": record.weight_kind, + } + for record in self.last_result.imputation.fit_weight_records + ], + "evidence": self.last_result.evidence(), + "replay_payload": self.last_result.replay_report.to_payload(), + } + + +def _support_result_from_frame( + frame: Frame, + tables: Mapping[str, pd.DataFrame], +) -> UKSPISupportResult: + household = tables["household"] + channel = support_channel_column("household") + if channel not in household: + raise ValueError("SPI income spine stage requires a support-channel frame.") + spi_households = household[channel].eq(SPI_SYNTHETIC_SUPPORT_CHANNEL) + if not spi_households.any(): + raise ValueError("SPI income spine stage found no SPI support households.") + return UKSPISupportResult( + person=tables["person"], + benunit=tables["benunit"], + household=household, + id_multiplier=1, + spi_household_ids=tuple(household.loc[spi_households, "household_id"]), + household_weight_kind=uk_household_weight_kind(frame), + mass_log=frame.mass_log, + spi_prior_mass_share=DEFAULT_SPI_PRIOR_MASS_SHARE, + ) + + +def _build_spine_replay_report( + *, + source_targets: HMRCIncomeTargetSet, + support: UKSPISupportResult, + imputation: UKSPIIncomeImputationResult, + frame: Frame, + distributional_mass_shares: Mapping[str, float], + identity_rows: int, +) -> HMRCReplayReport: + source_evidence = { + "spi_donor": { + "release": "2022-23", + "sha256": imputation.donor_sha256, + "size_bytes": imputation.donor_size_bytes, + "rows_used": imputation.donor_rows, + }, + "hmrc_surface": { + "vintage": source_targets.source.source_vintage, + "mapped_build_period": source_targets.source.build_period, + "sha256": source_targets.source.sha256, + "size_bytes": source_targets.source.size_bytes, + "mime_type": source_targets.source.mime_type, + "tables": list(source_targets.source.table_names), + }, + } + build_evidence = { + "stage": UK_HMRC_SPI_INCOME_SPINE_STAGE_NAME, + "output_weight_kind": uk_household_weight_kind(frame).value, + "calibration_performed": False, + "spi_prior_mass_share": support.spi_prior_mass_share, + "spi_households": support.n_spi_households, + "mass_change_reason": support.mass_log[-1].reason, + } + qrf_evidence = { + "fits": { + record.fit_name: {"weight_kind": record.weight_kind} + for record in imputation.fit_weight_records + }, + "donor_rows": imputation.donor_rows, + "stage2_training_rows": imputation.stage2_training_rows, + "spi_prediction_rows": imputation.spi_prediction_rows, + "post_draw_identity": { + "formula": SPI_SOURCE_TI_FORMULA, + "rows_checked": identity_rows, + "exact": True, + }, + } + effective_mass_evidence = { + "minimum_nondefault_mass_share": DEFAULT_MINIMUM_NONDEFAULT_MASS_SHARE, + "denominator": "all_person_effective_mass", + "required_support_channel": SPI_SYNTHETIC_SUPPORT_CHANNEL, + "columns": dict(distributional_mass_shares), + } + return build_conservative_hmrc_replay_report( + source_targets, + source_evidence=source_evidence, + build_evidence=build_evidence, + qrf_evidence=qrf_evidence, + effective_mass_evidence=effective_mass_evidence, + report_kind=UK_HMRC_SPI_SPINE_REPLAY_REPORT_KIND, + ) + + +def _support_stage_parameters( + stage: SourceStageSpec, + *, + seed: int, +) -> tuple[int, float, tuple[str, ...], tuple[UKZeroWeightStratumDeclaration, ...]]: + stack = _operation(stage, "stack_zero_weight_donors") + gate = _operation(stage, "gate_zero_weight_strata") + allocation = _operation(stage, "allocate_zero_weight_prior_mass") + if stack.parameters.get("count") != DEFAULT_SPI_SUPPORT_HOUSEHOLDS: + raise ValueError("SPI support manifest count drifted from the reviewed value.") + if stack.parameters.get("seed") != seed: + raise ValueError("SPI support manifest seed drifted from the reviewed value.") + if stack.parameters.get("draw") != "uniform_without_replacement": + raise ValueError("SPI support manifest must declare a uniform donor draw.") + if allocation.parameters.get("share") != DEFAULT_SPI_PRIOR_MASS_SHARE: + raise ValueError("SPI support prior-mass share drifted from 0.5.") + strata = tuple(allocation.parameters.get("strata", ())) + if strata != ("region",): + raise ValueError("SPI support spine allocation must use strata ['region'].") + return ( + int(stack.parameters["count"]), + float(allocation.parameters["share"]), + strata, + _coerce_declarations(gate.parameters.get("declarations", ())), + ) + + +def _assert_income_stage_parameters( + stage: SourceStageSpec, + *, + seed: int, + qrf_estimators: int, + donor_sample_size: int | None, +) -> None: + stage1 = _operation(stage, "fit_weighted_qrf_stage1") + stage2 = _operation(stage, "fit_weighted_qrf_stage2") + redraw = _operation(stage, "redraw_columns_from_fitted_qrf") + if stage1.parameters.get("seed") != seed: + raise ValueError("SPI income stage-1 seed drifted from the reviewed value.") + if stage2.parameters.get("seed") != seed + 1: + raise ValueError("SPI income stage-2 seed drifted from the reviewed value.") + if stage1.parameters.get("n_estimators") not in (None, qrf_estimators): + raise ValueError("SPI income stage-1 estimator count drifted.") + if stage2.parameters.get("n_estimators") not in (None, qrf_estimators): + raise ValueError("SPI income stage-2 estimator count drifted.") + if stage1.parameters.get("sample_size") != donor_sample_size: + raise ValueError("SPI income donor sample size drifted.") + if redraw.parameters.get("fit") != "stage1": + raise ValueError("SPI income base redraw must use the stage-1 fit.") + if redraw.parameters.get("rows") != "base_support_channel": + raise ValueError("SPI income base redraw must target the base support channel.") + + +def _operation(stage: SourceStageSpec, kind: str): + matches = [operation for operation in stage.operations if operation.kind == kind] + if len(matches) != 1: + raise ValueError( + f"Stage {stage.stage!r} must declare exactly one {kind!r} operation." + ) + return matches[0] + + +def _artifact_by_table( + stage: SourceStageSpec, + *, + expected: Sequence[str], +) -> dict[str, Mapping[str, Any]]: + artifacts = { + str(artifact["table"]): artifact + for artifact in stage.artifacts + if "table" in artifact + } + missing = sorted(set(expected) - set(artifacts)) + if missing: + raise ValueError(f"Stage {stage.stage!r} is missing tab artifact(s): {missing}.") + return artifacts + + +def _assert_identity_matches_artifact( + evidence: Mapping[str, object], + artifact: Mapping[str, Any], +) -> None: + if evidence["sha256"] != artifact["sha256"]: + raise ValueError( + f"FRS tab {artifact['locator']!r} hashes to {evidence['sha256']}, " + f"not the pinned {artifact['sha256']}." + ) + if evidence["size_bytes"] != artifact["size_bytes"]: + raise ValueError( + f"FRS tab {artifact['locator']!r} is {evidence['size_bytes']} bytes, " + f"not the pinned {artifact['size_bytes']}." + ) + + +def _coerce_declarations( + values: object, +) -> tuple[UKZeroWeightStratumDeclaration, ...]: + if not isinstance(values, Sequence) or isinstance(values, str | bytes): + raise ValueError("zero-weight declarations must be a sequence.") + declarations = [] + for value in values: + if not isinstance(value, Mapping): + raise ValueError("zero-weight declarations must be mappings.") + declarations.append(UKZeroWeightStratumDeclaration(**dict(value))) + return tuple(declarations) + + +__all__ = [ + "EMPLOYER_PENSION_CONTRIBUTIONS_COLUMN", + "UK_FRS_HMRC_SPINE_LEAF_OUTPUT_COLUMNS", + "UK_FRS_HMRC_SPINE_LEAVES_STAGE_NAME", + "UK_HMRC_SPI_INCOME_SPINE_STAGE_NAME", + "UK_HMRC_SPI_SPINE_REPLAY_REPORT_KIND", + "UK_SPI_INCOME_SPINE_NONNEGATIVE_OUTPUT_COLUMNS", + "UK_SPI_INCOME_SPINE_OUTPUT_COLUMNS", + "UK_SPI_INCOME_SPINE_REWRITE_COLUMNS", + "UK_SPI_SUPPORT_CHANNEL_OUTPUT_COLUMNS", + "UKFRSHMRCSpineLeavesResult", + "UKFRSHMRCSpineLeavesStageTransform", + "UKSPIIncomeSpineResult", + "UKSPIIncomeSpineStageTransform", + "UKSPISupportChannelStageTransform", +] diff --git a/packages/microcosm-build/src/microcosm/build/uk_runtime/spi_support.py b/packages/microcosm-build/src/microcosm/build/uk_runtime/spi_support.py index 3501d1797..d34949cbd 100644 --- a/packages/microcosm-build/src/microcosm/build/uk_runtime/spi_support.py +++ b/packages/microcosm-build/src/microcosm/build/uk_runtime/spi_support.py @@ -9,7 +9,7 @@ from __future__ import annotations -from collections.abc import Sequence +from collections.abc import Mapping, Sequence from dataclasses import dataclass from typing import Any @@ -20,6 +20,10 @@ HMRC_SPI_ASSESSABLE_INCOME_COLUMN, ) from microcosm.build.uk_runtime.rowwise_geography import id_multiplier_for_values +from microcosm.build.uk_runtime.terminal_gates import ( + UKZeroWeightStratumDeclaration, + uk_zero_weight_strata_gate, +) from microcosm.frame import ( MassChangeRecord, WeightKind, @@ -439,6 +443,67 @@ def replace_uk_spi_support_tables( ) +def build_uk_spi_support_channel( + person: pd.DataFrame, + benunit: pd.DataFrame, + household: pd.DataFrame, + *, + spi_household_count: int = DEFAULT_SPI_SUPPORT_HOUSEHOLDS, + seed: int = 42, + source_year: int | None = None, + spi_prior_mass_share: float = DEFAULT_SPI_PRIOR_MASS_SHARE, + strata_columns: Sequence[str] = ("region",), + input_weight_kind: WeightKind = WeightKind.DESIGN, + mass_log: tuple[MassChangeRecord, ...] = (), + zero_weight_declarations: ( + Sequence[UKZeroWeightStratumDeclaration] | Sequence[Mapping[str, Any]] + ) = (), +) -> UKSPISupportResult: + """Stack the E7 SPI support channel and allocate prior mass by stratum.""" + + if not isinstance(input_weight_kind, WeightKind): + raise TypeError("input_weight_kind must be a WeightKind.") + if not isinstance(mass_log, tuple) or any( + not isinstance(record, MassChangeRecord) for record in mass_log + ): + raise TypeError("mass_log must be a tuple of MassChangeRecord.") + if not isinstance(seed, int): + raise ValueError("seed must be an integer.") + share = float(spi_prior_mass_share) + if not np.isfinite(share) or not 0.0 < share < 1.0: + raise ValueError("spi_prior_mass_share must be finite and in (0, 1).") + stratum_columns = tuple(str(column) for column in strata_columns) + if not stratum_columns or any(not column for column in stratum_columns): + raise ValueError("strata_columns must contain non-empty column names.") + + stacked = create_uk_spi_support_tables( + person=person, + benunit=benunit, + household=household, + spi_household_count=spi_household_count, + seed=seed, + source_year=source_year, + ) + declarations = _coerce_zero_weight_declarations(zero_weight_declarations) + if declarations: + gate = uk_zero_weight_strata_gate( + stacked.household, + declarations=declarations, + ) + if not gate.passed: + raise ValueError( + "UK SPI support pre-clone zero-weight gate failed: " + f"{list(gate.failures)}." + ) + return _allocate_spi_prior_mass( + stacked, + spi_prior_mass_share=share, + input_weight_kind=input_weight_kind, + mass_log=mass_log, + strata_columns=stratum_columns, + ) + + def fill_support_channel_from_source( frame: pd.DataFrame, donor: pd.DataFrame, @@ -513,6 +578,23 @@ def fill_support_channel_from_source( return out +def _coerce_zero_weight_declarations( + declarations: Sequence[UKZeroWeightStratumDeclaration] | Sequence[Mapping[str, Any]], +) -> tuple[UKZeroWeightStratumDeclaration, ...]: + materialized: list[UKZeroWeightStratumDeclaration] = [] + for declaration in declarations: + if isinstance(declaration, UKZeroWeightStratumDeclaration): + materialized.append(declaration) + elif isinstance(declaration, Mapping): + materialized.append(UKZeroWeightStratumDeclaration(**dict(declaration))) + else: + raise TypeError( + "zero_weight_declarations must contain " + "UKZeroWeightStratumDeclaration instances or mappings." + ) + return tuple(materialized) + + def support_channel_column(entity: str) -> str: """Return the entity-prefixed support-channel column name.""" @@ -1099,6 +1181,7 @@ def _require_entity(entity: str) -> str: "SPI_REPLACEMENT_STRATA_COLUMNS", "UKSPISupportResult", "UK_SPI_SUPPORT_STAGE_NAME", + "build_uk_spi_support_channel", "create_uk_spi_support_tables", "fill_support_channel_from_source", "replace_uk_spi_support_tables", diff --git a/packages/microcosm-build/tests/test_country_spec.py b/packages/microcosm-build/tests/test_country_spec.py index 888d5247e..c39e58e2a 100644 --- a/packages/microcosm-build/tests/test_country_spec.py +++ b/packages/microcosm-build/tests/test_country_spec.py @@ -259,6 +259,36 @@ def test_explicit_stage_subset_refuses_empty_or_unknown_names(self) -> None: ) +class TestUKCountryPackage: + def test_spi_spine_adds_no_country_package_resources(self) -> None: + spec = load_country_spec("uk") + + assert spec.resources == ( + "cgt_source_stages.json", + "degenerate_reviewed_exclusions.json", + "efrs_parity_known_gaps.json", + "efrs_parity_reference.json", + "gates.json", + "brma_rent_counts.json", + "hmrc_income_release_gate_report.json", + "hmrc_income_replay_report.json", + "hmrc_income_source_stages.json", + "source_stages.json", + "take_up_contract.json", + "input_mass_reviewed_exclusions.json", + "national_staging_build_record.json", + "qrf_tail_reviewed_exclusions.json", + "release_input_coverage_manifest.json", + "uk_local_target_census.json", + ) + + def test_uk_source_manifest_loads_sixteen_stages(self) -> None: + spec = load_country_spec("uk") + + assert spec.sources is not None + assert len(spec.sources.stages) == 16 + + class TestExistingPackagesGeneralize: """The loader is country-neutral: the US and UK packages load unchanged.""" diff --git a/packages/microcosm-build/tests/test_uk_frs_spine.py b/packages/microcosm-build/tests/test_uk_frs_spine.py index 354cd119b..52e8183bf 100644 --- a/packages/microcosm-build/tests/test_uk_frs_spine.py +++ b/packages/microcosm-build/tests/test_uk_frs_spine.py @@ -30,10 +30,12 @@ ) from microcosm.build.uk_runtime.national_build import load_uk_national_frame 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 WeightKind +from microcosm.frame import Frame, WeightKind, engine_tables _TOOL_PATH = Path(__file__).resolve().parents[3] / "tools" / "build_uk_frs_spine.py" @@ -617,6 +619,109 @@ def source_stage( }, ), ), + source_stage( + "frs_hmrc_spine_leaves", + tables=("adult", "benefits"), + operations=[ + {"kind": "retain_adjudicated_frs_hmrc_leaves"}, + { + "kind": "derive", + "output": "employer_pension_contributions", + }, + ], + outputs=( + "hmrc_spi_pay", + "hmrc_spi_unemployment_benefit_income", + "hmrc_spi_incapacity_benefit_income", + "ossben_identifiable_subset", + "srp_regular_code5", + "employer_pension_contributions", + ), + ), + source_stage( + "spi_support_channel", + grain="household", + operations=[ + { + "kind": "stack_zero_weight_donors", + "count": 10000, + "seed": 42, + "draw": "uniform_without_replacement", + }, + { + "kind": "gate_zero_weight_strata", + "declarations": [ + { + "name": "e7_spi_synthetic_preclone", + "selector": { + "household_is_spi_synthetic": True + }, + "maximum_zero_weight_rows": 10000, + "reason": "synthetic driver fixture", + } + ], + }, + { + "kind": "allocate_zero_weight_prior_mass", + "share": 0.5, + "strata": ["region"], + }, + ], + outputs=( + "household_is_spi_synthetic", + "person_support_channel", + "person_support_clone_index", + "person_source_id", + "benunit_support_channel", + "benunit_support_clone_index", + "benunit_source_id", + "household_support_channel", + "household_support_clone_index", + "household_source_id", + "source_household_id", + "source_year", + "source_household_key", + ), + ), + source_stage( + "hmrc_spi_income_spine", + operations=[ + {"kind": "verify_pinned_hmrc_source_pair"}, + {"kind": "strict_read_private_table", "seed": 42}, + { + "kind": "fit_weighted_qrf_stage1", + "seed": 42, + "sample_size": 100000, + "initialize_frs_channel_columns": { + "gift_aid": 0.0, + "charitable_investment_gifts": 0.0, + }, + }, + {"kind": "fit_weighted_qrf_stage2", "seed": 43}, + { + "kind": "redraw_columns_from_fitted_qrf", + "fit": "stage1", + "columns": ["dividend_income"], + "rows": "base_support_channel", + }, + ], + outputs=( + "other_investment_income", + "gift_aid", + "charitable_investment_gifts", + "hmrc_spi_employment_benefits", + "hmrc_spi_employment_expenses", + "hmrc_spi_other_social_security_income", + "hmrc_spi_taxable_termination_pay", + "hmrc_spi_miscellaneous_employment_income", + "hmrc_spi_other_income", + "hmrc_spi_state_pension_income", + "hmrc_spi_employed_income", + "hmrc_spi_total_earned_income", + "hmrc_spi_total_investment_income", + "hmrc_spi_assessable_income", + ), + ), ), ), geography_spine=None, @@ -897,6 +1002,70 @@ def _stub_policy_readers(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setitem(sys.modules, "policyengine_uk", None) +def _patch_spi_spine_driver_runtime( + tool, + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> tuple[Path, Path]: + spi_tab = tmp_path / "put2223uk.tab" + hmrc_ods = tmp_path / "Collated_Tables_3_1_to_3_11_2324.ods" + spi_tab.write_text("synthetic\\n", encoding="utf-8") + hmrc_ods.write_text("synthetic\\n", encoding="utf-8") + + class _FakeStageTransform: + def __init__(self, *args, stage, **kwargs) -> None: + self.stage = stage + self.last_result = None + + def __call__(self, frame: Frame) -> Frame: + tables = engine_tables(frame) + person = tables["person"].copy() + benunit = tables["benunit"].copy() + household = tables["household"].copy() + for column in self.stage.outputs: + if column.startswith("household_") or column.startswith("source_"): + household[column] = _fake_value(column, len(household)) + elif column.startswith("benunit_"): + benunit[column] = _fake_value(column, len(benunit)) + else: + person[column] = _fake_value(column, len(person)) + result = uk_national_frame( + person=person, + benunit=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, + ) + self.last_result = SimpleNamespace(replay_report={"report_kind": "fake"}) + return result + + def _write_fake_replay(report, path): + output = Path(path) + output.write_text( + json.dumps({"report_kind": "fake_spine_replay"}) + "\n", + encoding="utf-8", + ) + return output + + monkeypatch.setattr(tool, "UKFRSHMRCSpineLeavesStageTransform", _FakeStageTransform) + monkeypatch.setattr(tool, "UKSPISupportChannelStageTransform", _FakeStageTransform) + monkeypatch.setattr(tool, "UKSPIIncomeSpineStageTransform", _FakeStageTransform) + monkeypatch.setattr(tool, "write_hmrc_replay_report", _write_fake_replay) + return spi_tab, hmrc_ods + + +def _fake_value(column: str, rows: int): + if column.endswith("_support_channel"): + return ["frs"] * rows + if column == "source_household_key": + return [f"2023:{index + 1}" for index in range(rows)] + if column == "household_is_spi_synthetic": + return [False] * rows + return np.arange(1, rows + 1, dtype=float) + + def test_driver_writes_spine_h5_sidecars_and_logbook( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: @@ -913,6 +1082,7 @@ def test_driver_writes_spine_h5_sidecars_and_logbook( ) monkeypatch.setattr(tool, "_rules_engine", lambda: _FakeUKEngine()) _stub_policy_readers(monkeypatch) + spi_tab, hmrc_ods = _patch_spi_spine_driver_runtime(tool, monkeypatch, tmp_path) monkeypatch.delenv("POPULACE_LEDGER_URL", raising=False) monkeypatch.delenv("POPULACE_LEDGER_KEY", raising=False) monkeypatch.delenv("POPULACE_LEDGER_API_KEY", raising=False) @@ -923,10 +1093,14 @@ def test_driver_writes_spine_h5_sidecars_and_logbook( [ "--frs-raw-dir", str(raw_dir), - "--spine-h5", - str(output), - "--emit-nonzero-shares", - str(shares), + "--spine-h5", + str(output), + "--spi-tab", + str(spi_tab), + "--hmrc-ods", + str(hmrc_ods), + "--emit-nonzero-shares", + str(shares), ] ) == 0 @@ -972,6 +1146,17 @@ def test_driver_writes_spine_h5_sidecars_and_logbook( assert set(sidecar["artifact_pins"]) == set(FRS_SPINE_TABLES) assert sidecar["declared_seeds"]["frs_take_up"]["would_claim_child_benefit"] == 0 assert sidecar["declared_seeds"]["frs_brma"] == {"brma": 0} + assert sidecar["declared_seeds"]["spi_support_channel"] == { + "stack_zero_weight_donors": 42 + } + assert sidecar["declared_seeds"]["hmrc_spi_income_spine"] == { + "donor_bootstrap": 42, + "stage1": 42, + "stage2": 43, + } + assert json.loads(output.with_suffix(".hmrc_replay.json").read_text()) == { + "report_kind": "fake_spine_replay" + } assert len(sidecar["stochastic_contract_sha256"]) == 64 assert sidecar["resource_pins"] == {"brma_rent_counts.json": "f" * 64} # Resolve the expected version the way the driver does, so the assertion @@ -1009,13 +1194,42 @@ def test_driver_writes_payload_identical_h5s( ) monkeypatch.setattr(tool, "_rules_engine", lambda: _FakeUKEngine()) _stub_policy_readers(monkeypatch) + spi_tab, hmrc_ods = _patch_spi_spine_driver_runtime(tool, monkeypatch, tmp_path) monkeypatch.delenv("POPULACE_LEDGER_URL", raising=False) monkeypatch.delenv("POPULACE_LEDGER_KEY", raising=False) monkeypatch.delenv("POPULACE_LEDGER_API_KEY", raising=False) - assert tool.main(["--frs-raw-dir", str(raw_dir), "--spine-h5", str(output)]) == 0 + assert ( + tool.main( + [ + "--frs-raw-dir", + str(raw_dir), + "--spine-h5", + str(output), + "--spi-tab", + str(spi_tab), + "--hmrc-ods", + str(hmrc_ods), + ] + ) + == 0 + ) first_frame, _ = load_uk_national_frame(output) - assert tool.main(["--frs-raw-dir", str(raw_dir), "--spine-h5", str(output)]) == 0 + assert ( + tool.main( + [ + "--frs-raw-dir", + str(raw_dir), + "--spine-h5", + str(output), + "--spi-tab", + str(spi_tab), + "--hmrc-ods", + str(hmrc_ods), + ] + ) + == 0 + ) second_frame, _ = load_uk_national_frame(output) for entity in ("person", "benunit", "household"): @@ -1025,6 +1239,122 @@ def test_driver_writes_payload_identical_h5s( ) +def test_driver_refuses_missing_spi_tab(tmp_path: Path) -> None: + tool = _load_tool() + raw_dir = tmp_path / "raw" + raw_dir.mkdir() + hmrc_ods = tmp_path / "Collated_Tables_3_1_to_3_11_2324.ods" + hmrc_ods.write_text("synthetic\n", encoding="utf-8") + args = tool._parse_args( + [ + "--frs-raw-dir", + str(raw_dir), + "--spine-h5", + str(tmp_path / "spine.h5"), + "--spi-tab", + str(tmp_path / "put2223uk.tab"), + "--hmrc-ods", + str(hmrc_ods), + ] + ) + + with pytest.raises(ValueError, match="--spi-tab must be an existing file"): + tool._validate_args(args) + + +def test_driver_refuses_misnamed_spi_tab(tmp_path: Path) -> None: + tool = _load_tool() + raw_dir = tmp_path / "raw" + raw_dir.mkdir() + spi_tab = tmp_path / "spi.tab" + spi_tab.write_text("synthetic\n", encoding="utf-8") + hmrc_ods = tmp_path / "Collated_Tables_3_1_to_3_11_2324.ods" + hmrc_ods.write_text("synthetic\n", encoding="utf-8") + args = tool._parse_args( + [ + "--frs-raw-dir", + str(raw_dir), + "--spine-h5", + str(tmp_path / "spine.h5"), + "--spi-tab", + str(spi_tab), + "--hmrc-ods", + str(hmrc_ods), + ] + ) + + with pytest.raises(ValueError, match="--spi-tab must name put2223uk.tab"): + tool._validate_args(args) + + +def test_driver_refuses_missing_hmrc_ods(tmp_path: Path) -> None: + tool = _load_tool() + raw_dir = tmp_path / "raw" + raw_dir.mkdir() + spi_tab = tmp_path / "put2223uk.tab" + spi_tab.write_text("synthetic\n", encoding="utf-8") + args = tool._parse_args( + [ + "--frs-raw-dir", + str(raw_dir), + "--spine-h5", + str(tmp_path / "spine.h5"), + "--spi-tab", + str(spi_tab), + "--hmrc-ods", + str(tmp_path / "hmrc.ods"), + ] + ) + + with pytest.raises(ValueError, match="--hmrc-ods must be an existing file"): + tool._validate_args(args) + + +def test_driver_refuses_misnamed_hmrc_ods(tmp_path: Path) -> None: + tool = _load_tool() + raw_dir = tmp_path / "raw" + raw_dir.mkdir() + spi_tab = tmp_path / "put2223uk.tab" + spi_tab.write_text("synthetic\n", encoding="utf-8") + hmrc_ods = tmp_path / "hmrc.txt" + hmrc_ods.write_text("synthetic\n", encoding="utf-8") + args = tool._parse_args( + [ + "--frs-raw-dir", + str(raw_dir), + "--spine-h5", + str(tmp_path / "spine.h5"), + "--spi-tab", + str(spi_tab), + "--hmrc-ods", + str(hmrc_ods), + ] + ) + + with pytest.raises(ValueError, match="--hmrc-ods must end with '.ods'"): + tool._validate_args(args) + + +def test_driver_has_no_sample_fraction_path(tmp_path: Path) -> None: + tool = _load_tool() + + with pytest.raises(SystemExit): + tool._parse_args( + [ + "--frs-raw-dir", + str(tmp_path), + "--spine-h5", + str(tmp_path / "spine.h5"), + "--spi-tab", + str(tmp_path / "put2223uk.tab"), + "--hmrc-ods", + str(tmp_path / "hmrc.ods"), + "--sample-fraction", + "0.5", + ] + ) + + def test_refuses_missing_tab(tmp_path: Path) -> None: stage = _write_fixture(tmp_path) (tmp_path / "adult.tab").unlink() diff --git a/packages/microcosm-build/tests/test_uk_hmrc_replay.py b/packages/microcosm-build/tests/test_uk_hmrc_replay.py index b41cd4170..04569c841 100644 --- a/packages/microcosm-build/tests/test_uk_hmrc_replay.py +++ b/packages/microcosm-build/tests/test_uk_hmrc_replay.py @@ -367,6 +367,14 @@ def test_atomic_writer_uses_exact_caller_json_path(tmp_path) -> None: assert not list(output.parent.glob(f".{output.name}.*.tmp")) +def test_conservative_report_accepts_spine_mode_report_kind(tmp_path) -> None: + report = _report(tmp_path, report_kind="uk_hmrc_spi_income_spine_208_fact_replay") + + assert report.to_payload()["report_kind"] == ( + "uk_hmrc_spi_income_spine_208_fact_replay" + ) + + def test_writer_rejects_non_json_and_symbolic_link_paths(tmp_path) -> None: report = _report(tmp_path) diff --git a/packages/microcosm-build/tests/test_uk_source_stages.py b/packages/microcosm-build/tests/test_uk_source_stages.py index 4a485647b..d7ca43bfe 100644 --- a/packages/microcosm-build/tests/test_uk_source_stages.py +++ b/packages/microcosm-build/tests/test_uk_source_stages.py @@ -32,13 +32,25 @@ "frs_household_draws", "frs_brma", ] +E7_STAGE_NAMES = [ + "frs_hmrc_spine_leaves", + "spi_support_channel", + "hmrc_spi_income_spine", +] UK_SOURCE_STAGE_NAMES = [ "frs_spine", *E3_STAGE_NAMES, *E4_STAGE_NAMES, + *E7_STAGE_NAMES, "frs_hmrc_retained_leaves", "hmrc_spi_income", ] +UK_FRS_SPI_SPINE_DRIVER_STAGE_NAMES = [ + "frs_spine", + *E3_STAGE_NAMES, + *E4_STAGE_NAMES, + *E7_STAGE_NAMES, +] FROZEN_SOURCE_STAGES_SHA256 = ( "c0341af7166ae3a85a3c1164e7d9e880c4b4aec122f1a8fa90c73b46c596e1ea" ) @@ -80,12 +92,19 @@ def test_source_stages_json_loads_as_shared_manifest(self) -> None: assert manifest.version == 1 assert [stage.stage for stage in manifest.stages] == UK_SOURCE_STAGE_NAMES - def test_country_spec_declares_three_uk_source_stages(self) -> None: + def test_country_spec_declares_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] == UK_SOURCE_STAGE_NAMES + def test_e7_block_is_contiguous_before_certified_pair(self) -> None: + canonical = _load_json(CANONICAL_SOURCE_STAGES) + names = [stage["stage"] for stage in canonical["stages"]] + + assert names[-5:-2] == E7_STAGE_NAMES + assert names[-2:] == ["frs_hmrc_retained_leaves", "hmrc_spi_income"] + def test_copy_is_lockstep_with_frozen_original_except_citation_rewrites( self, ) -> None: @@ -134,7 +153,9 @@ def test_frozen_original_bytes_are_pinned(self) -> None: assert digest == FROZEN_SOURCE_STAGES_SHA256 - def test_country_stage_plan_assembles_two_uk_national_stages(self) -> None: + def test_country_stage_plan_assembles_two_certified_uk_national_stages( + self, + ) -> None: spec = load_country_spec("uk") plan = country_stage_plan( spec, @@ -150,6 +171,19 @@ def test_country_stage_plan_assembles_two_uk_national_stages(self) -> None: "hmrc_spi_income", ] + def test_country_stage_plan_assembles_fourteen_stage_spine_plan(self) -> None: + spec = load_country_spec("uk") + implementations = {name: _identity for name in UK_SOURCE_STAGE_NAMES} + plan = country_stage_plan( + spec, + implementations, + stage_names=tuple(UK_FRS_SPI_SPINE_DRIVER_STAGE_NAMES), + ) + + assert [stage.name for stage in plan.stages] == ( + UK_FRS_SPI_SPINE_DRIVER_STAGE_NAMES + ) + @pytest.mark.parametrize( "implementations, match", [ @@ -167,6 +201,9 @@ def test_country_stage_plan_assembles_two_uk_national_stages(self) -> None: "frs_person_draws": _identity, "frs_household_draws": _identity, "frs_brma": _identity, + "frs_hmrc_spine_leaves": _identity, + "spi_support_channel": _identity, + "hmrc_spi_income_spine": _identity, "frs_hmrc_retained_leaves": _identity, "hmrc_spi_income": _identity, "hmrc_spi_income_fallback": _identity, @@ -271,6 +308,34 @@ def test_e3_outputs_are_backed_by_runtime_written_columns(self) -> None: ) assert stages["frs_brma"].outputs == FRS_BRMA_OUTPUT_COLUMNS + def test_e7_outputs_and_rewrites_are_backed_by_runtime_constants(self) -> None: + from microcosm.build.uk_runtime.spi_spine import ( + UK_FRS_HMRC_SPINE_LEAF_OUTPUT_COLUMNS, + UK_SPI_INCOME_SPINE_NONNEGATIVE_OUTPUT_COLUMNS, + UK_SPI_INCOME_SPINE_OUTPUT_COLUMNS, + UK_SPI_INCOME_SPINE_REWRITE_COLUMNS, + UK_SPI_SUPPORT_CHANNEL_OUTPUT_COLUMNS, + ) + + spec = load_country_spec("uk") + stages = {stage.stage: stage for stage in spec.sources.stages} + + assert ( + stages["frs_hmrc_spine_leaves"].outputs + == UK_FRS_HMRC_SPINE_LEAF_OUTPUT_COLUMNS + ) + assert ( + stages["spi_support_channel"].outputs + == UK_SPI_SUPPORT_CHANNEL_OUTPUT_COLUMNS + ) + income = stages["hmrc_spi_income_spine"] + assert income.outputs == UK_SPI_INCOME_SPINE_OUTPUT_COLUMNS + assert income.nonnegative_outputs == ( + UK_SPI_INCOME_SPINE_NONNEGATIVE_OUTPUT_COLUMNS + ) + assert income.rewrites == UK_SPI_INCOME_SPINE_REWRITE_COLUMNS + assert not (set(income.outputs) & set(income.rewrites)) + class TestE3ManifestLockstep: def test_e3_raw_tab_pins_match_spine_artifacts(self) -> None: @@ -350,6 +415,25 @@ def test_e3_operation_kinds_are_declared_in_order(self) -> None: "materialize_rules_engine_predictors", "sample_categorical_from_count_table", ] + assert [op.kind for op in stages["frs_hmrc_spine_leaves"].operations] == [ + "retain_adjudicated_frs_hmrc_leaves", + "derive", + ] + assert [op.kind for op in stages["spi_support_channel"].operations] == [ + "stack_zero_weight_donors", + "gate_zero_weight_strata", + "allocate_zero_weight_prior_mass", + ] + assert [op.kind for op in stages["hmrc_spi_income_spine"].operations] == [ + "verify_pinned_hmrc_source_pair", + "strict_read_private_table", + "fit_weighted_qrf_stage1", + "fit_weighted_qrf_stage2", + "redraw_columns_from_fitted_qrf", + "materialize_hmrc_income_bands_fail_closed", + "classify_hmrc_income_facts_with_reviewed_fences", + "gate_distributional_effective_mass", + ] def test_engine_predictor_and_rewrite_constants_match_manifest(self) -> None: from microcosm.build.uk_runtime.frs_brma import UK_BRMA_PREDICTORS @@ -432,11 +516,25 @@ def test_every_e4_stochastic_operation_declares_integer_seed(self) -> None: }: assert isinstance(operation.parameters.get("seed"), int) - def test_full_uk_source_stage_plan_compiles_with_e4_stages(self) -> None: + def test_e7_declared_seed_lockstep(self) -> None: + spec = load_country_spec("uk") + stages = {stage.stage: stage for stage in spec.sources.stages} + + assert ( + stages["spi_support_channel"].operations[0].parameters["seed"] == 42 + ) + assert ( + stages["hmrc_spi_income_spine"].operations[2].parameters["seed"] == 42 + ) + assert ( + stages["hmrc_spi_income_spine"].operations[3].parameters["seed"] == 43 + ) + + def test_full_uk_source_stage_plan_compiles_with_e7_stages(self) -> None: spec = load_country_spec("uk") implementations = {name: _identity for name in UK_SOURCE_STAGE_NAMES} - driver_stage_names = tuple(UK_SOURCE_STAGE_NAMES[:-1]) + driver_stage_names = tuple(UK_FRS_SPI_SPINE_DRIVER_STAGE_NAMES) plan = country_stage_plan(spec, implementations, stage_names=driver_stage_names) assert [stage.name for stage in plan.stages] == list(driver_stage_names) diff --git a/packages/microcosm-build/tests/test_uk_spi_spine.py b/packages/microcosm-build/tests/test_uk_spi_spine.py new file mode 100644 index 000000000..f69d17a17 --- /dev/null +++ b/packages/microcosm-build/tests/test_uk_spi_spine.py @@ -0,0 +1,423 @@ +from __future__ import annotations + +from pathlib import Path +from types import SimpleNamespace + +import numpy as np +import pandas as pd +import pytest + +from microcosm.build.source_manifest import SourceStageSpec +from microcosm.build.uk_runtime import spi_income +from microcosm.build.uk_runtime.national_frame import uk_national_frame +from microcosm.build.uk_runtime.spi_income import ( + FRS_ONLY_SPI_FILL_PERSON_COLUMNS, + SPI_DONOR_REQUIRED_COLUMNS, + SPI_INCOME_QRF_OUTPUT_COLUMNS, + impute_uk_spi_income_support, +) +from microcosm.build.uk_runtime.spi_spine import ( + EMPLOYER_PENSION_CONTRIBUTIONS_COLUMN, + UKFRSHMRCSpineLeavesStageTransform, + UKSPISupportChannelStageTransform, +) +from microcosm.build.uk_runtime.spi_support import ( + HOUSEHOLD_IS_SPI_SYNTHETIC_COLUMN, + SPI_SYNTHETIC_SUPPORT_CHANNEL, + build_uk_spi_support_channel, + support_channel_column, +) +from microcosm.build.uk_runtime.terminal_gates import UKZeroWeightStratumDeclaration +from microcosm.frame import WeightKind + + +def _base_frame() -> object: + person = pd.DataFrame( + { + "person_id": [2001, 1001], + "person_household_id": [2, 1], + "person_benunit_id": [201, 101], + "age": [44, 40], + "gender": ["FEMALE", "MALE"], + "employment_income": [20.0, 10.0], + "self_employment_income": [0.0, 0.0], + "savings_interest_income": [2.0, 1.0], + "dividend_income": [20.0, 10.0], + "private_pension_income": [0.0, 0.0], + "property_income": [0.0, 0.0], + "employee_pension_contributions": [3.0, 2.0], + "employer_pension_contributions": [9.0, 6.0], + "hmrc_spi_pay": [0.0, 0.0], + "hmrc_spi_unemployment_benefit_income": [0.0, 0.0], + "hmrc_spi_incapacity_benefit_income": [0.0, 0.0], + "ossben_identifiable_subset": [0.0, 0.0], + "srp_regular_code5": [0.0, 0.0], + **{ + column: [0.0, 0.0] + for column in FRS_ONLY_SPI_FILL_PERSON_COLUMNS + if column + not in { + "employee_pension_contributions", + "maternity_allowance_reported", + } + }, + } + ) + benunit = pd.DataFrame({"benunit_id": [101, 201]}) + household = pd.DataFrame( + { + "household_id": [1, 2], + "household_weight": [10.0, 20.0], + "region": ["LONDON", "SCOTLAND"], + } + ) + return uk_national_frame( + person=person, + benunit=benunit, + household=household, + time_period="2023", + weight_kind=WeightKind.DESIGN, + ) + + +def _leaves_stage( + tmp_path: Path, + *, + extra_adult_rows: tuple[dict[str, object], ...] = (), +) -> SourceStageSpec: + adult = pd.DataFrame( + [ + {"sernum": 1, "person": 1, "inearns": 4.0}, + {"sernum": 2, "person": 1, "inearns": 5.0}, + *extra_adult_rows, + ] + ) + benefits = pd.DataFrame( + { + "sernum": [1, 2, 2], + "person": [1, 1, 1], + "benefit": [14, 17, 5], + "benamt": [1.0, 2.0, 3.0], + "var2": [0, 0, 0], + } + ) + adult_path = tmp_path / "adult.tab" + benefits_path = tmp_path / "benefits.tab" + adult.to_csv(adult_path, sep="\t", index=False) + benefits.to_csv(benefits_path, sep="\t", index=False) + + def artifact(path: Path, table: str) -> dict[str, object]: + import hashlib + + return { + "role": "frs_table", + "table": table, + "kind": "licensed_microdata", + "format": "tab", + "locator": path.name, + "sha256": hashlib.sha256(path.read_bytes()).hexdigest(), + "size_bytes": path.stat().st_size, + "runtime_sha256_required": True, + } + + return SourceStageSpec.from_mapping( + { + "stage": "frs_hmrc_spine_leaves", + "survey": "Synthetic FRS", + "source": "local synthetic tabs", + "grain": "person", + "artifacts": [ + artifact(adult_path, "adult"), + artifact(benefits_path, "benefits"), + ], + "operations": [ + {"kind": "retain_adjudicated_frs_hmrc_leaves"}, + { + "kind": "derive", + "output": EMPLOYER_PENSION_CONTRIBUTIONS_COLUMN, + }, + ], + "outputs": [ + "hmrc_spi_pay", + "hmrc_spi_unemployment_benefit_income", + "hmrc_spi_incapacity_benefit_income", + "ossben_identifiable_subset", + "srp_regular_code5", + EMPLOYER_PENSION_CONTRIBUTIONS_COLUMN, + ], + } + ) + + +def test_spine_leaves_align_by_raw_person_id_not_position(tmp_path: Path) -> None: + transform = UKFRSHMRCSpineLeavesStageTransform( + tmp_path, + stage=_leaves_stage(tmp_path), + ) + + result = transform(_base_frame()) + person = result.table("person") + + assert person["person_id"].tolist() == [2001, 1001] + assert person["hmrc_spi_pay"].tolist() == pytest.approx( + [5.0 * 365.25 / 7.0, 4.0 * 365.25 / 7.0] + ) + assert person["hmrc_spi_incapacity_benefit_income"].tolist() == pytest.approx( + [2.0 * 365.25 / 7.0, 0.0] + ) + assert person[EMPLOYER_PENSION_CONTRIBUTIONS_COLUMN].tolist() == [9.0, 6.0] + + +def test_spine_leaves_fail_closed_on_unknown_raw_person(tmp_path: Path) -> None: + stage = _leaves_stage( + tmp_path, + extra_adult_rows=({"sernum": 9, "person": 1, "inearns": 1.0},), + ) + + with pytest.raises(ValueError, match="absent from the raw spine"): + UKFRSHMRCSpineLeavesStageTransform(tmp_path, stage=stage)(_base_frame()) + + +def _support_stage() -> SourceStageSpec: + return SourceStageSpec.from_mapping( + { + "stage": "spi_support_channel", + "survey": "Synthetic FRS", + "source": "local synthetic frame", + "grain": "household", + "artifacts": [], + "operations": [ + { + "kind": "stack_zero_weight_donors", + "count": 10000, + "seed": 42, + "draw": "uniform_without_replacement", + }, + { + "kind": "gate_zero_weight_strata", + "declarations": [ + { + "name": "e7_spi_synthetic_preclone", + "selector": {HOUSEHOLD_IS_SPI_SYNTHETIC_COLUMN: True}, + "maximum_zero_weight_rows": 10000, + "reason": "synthetic fixture", + } + ], + }, + { + "kind": "allocate_zero_weight_prior_mass", + "share": 0.5, + "strata": ["region"], + }, + ], + "outputs": ["household_is_spi_synthetic"], + } + ) + + +def test_support_transform_refuses_manifest_code_drift() -> None: + stage = SourceStageSpec.from_mapping( + { + **_support_stage().__dict__, + "operations": [ + {"kind": "stack_zero_weight_donors", "count": 9999, "seed": 42}, + {"kind": "gate_zero_weight_strata", "declarations": []}, + { + "kind": "allocate_zero_weight_prior_mass", + "share": 0.5, + "strata": ["region"], + }, + ], + } + ) + + with pytest.raises(ValueError, match="count drifted"): + UKSPISupportChannelStageTransform(stage=stage)(_base_frame()) + + +def test_preclone_support_gate_passes_at_limit_and_fails_above() -> None: + household = pd.DataFrame( + { + "household_id": [1, 2], + "household_weight": [10.0, 20.0], + "region": ["LONDON", "LONDON"], + } + ) + declaration = UKZeroWeightStratumDeclaration( + name="e7_spi_synthetic_preclone", + selector={HOUSEHOLD_IS_SPI_SYNTHETIC_COLUMN: True}, + maximum_zero_weight_rows=1, + reason="synthetic fixture", + ) + with pytest.raises(ValueError, match="exceed"): + build_uk_spi_support_channel( + person=_base_frame().table("person"), + benunit=_base_frame().table("benunit"), + household=household, + spi_household_count=2, + zero_weight_declarations=(declaration,), + ) + + +class _FakeQRF: + events: list[tuple[str, tuple[str, ...], list[float]]] = [] + + def __init__(self, n_estimators: int, seed: int) -> None: + self.seed = seed + self.targets: tuple[str, ...] = () + self.weight_kind = "unknown" + + def fit(self, frame, predictors, targets, weights): + self.targets = tuple(targets) + self.weight_kind = weights + label = "stage1" if "hmrc_spi_pay" in self.targets else "stage2" + dividend = ( + frame.table("person")["dividend_income"].tolist() + if "dividend_income" in frame.table("person") + else [] + ) + self.events.append((label, tuple(targets), dividend)) + return self + + def predict(self, predictors): + rows = len(predictors) + values: dict[str, np.ndarray] = {} + for column in self.targets: + if column == "hmrc_spi_miscellaneous_employment_income": + values[column] = np.zeros(rows) + elif column == "dividend_income": + values[column] = 100.0 + np.arange(rows, dtype=float) + elif column in {"gift_aid", "charitable_investment_gifts"}: + values[column] = np.ones(rows) + elif column == "savings_interest_income": + values[column] = np.full(rows, 5.0) + else: + values[column] = np.full(rows, 2.0) + return pd.DataFrame(values) + + +def _donor_file(tmp_path: Path) -> Path: + path = tmp_path / "put2223uk.tab" + row = {column: 0.0 for column in SPI_DONOR_REQUIRED_COLUMNS} + row.update( + { + "SEX": 1, + "FACT": 1.0, + "GORCODE": 8, + "AGERANGE": 1, + "PAY": 10.0, + "TEI": 10.0, + "TI": 10.0, + } + ) + pd.DataFrame([row]).to_csv(path, sep="\t", index=False) + return path + + +def test_spi_income_zero_initializes_frs_charity_and_redraws_dividends_after_stage2( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + _FakeQRF.events = [] + monkeypatch.setattr(spi_income, "QRF", _FakeQRF) + monkeypatch.setattr( + spi_income, + "_verify_spi_donor_identity", + lambda path: SimpleNamespace(path=path), + ) + monkeypatch.setattr( + spi_income, + "_refresh_disability_derived_inputs", + lambda person, spi_people, build_period: person, + ) + support = build_uk_spi_support_channel( + person=_base_frame().table("person"), + benunit=_base_frame().table("benunit"), + household=pd.DataFrame( + { + "household_id": [1, 2], + "household_weight": [10.0, 20.0], + "region": ["LONDON", "SCOTLAND"], + } + ), + spi_household_count=2, + zero_weight_declarations=( + UKZeroWeightStratumDeclaration( + name="e7_spi_synthetic_preclone", + selector={HOUSEHOLD_IS_SPI_SYNTHETIC_COLUMN: True}, + maximum_zero_weight_rows=2, + reason="synthetic fixture", + ), + ), + ) + + result = impute_uk_spi_income_support( + support, + _donor_file(tmp_path), + donor_sample_size=None, + initialize_frs_channel_columns={ + "gift_aid": 0.0, + "charitable_investment_gifts": 0.0, + }, + stage1_base_redraw_columns=("dividend_income",), + ) + person = result.person + base = person[support_channel_column("person")] != SPI_SYNTHETIC_SUPPORT_CHANNEL + spi = ~base + + assert person.loc[base, "gift_aid"].tolist() == [0.0, 0.0] + assert person.loc[base, "charitable_investment_gifts"].tolist() == [0.0, 0.0] + assert person.loc[base, "hmrc_spi_employment_benefits"].isna().all() + assert person.loc[base, "dividend_income"].tolist() == [100.0, 101.0] + assert person.loc[spi, "dividend_income"].tolist() == [100.0, 101.0] + assert _FakeQRF.events[1][0] == "stage2" + assert _FakeQRF.events[1][2] == [20.0, 10.0] + assert set(SPI_INCOME_QRF_OUTPUT_COLUMNS) <= set(person.columns) + + +def test_reviewed_absent_incapacity_signal_raises(tmp_path: Path) -> None: + _FakeQRF.events = [] + monkeypatch = pytest.MonkeyPatch() + monkeypatch.setattr(spi_income, "QRF", _FakeQRF) + monkeypatch.setattr( + spi_income, + "_verify_spi_donor_identity", + lambda path: SimpleNamespace(path=path), + ) + monkeypatch.setattr( + spi_income, + "_refresh_disability_derived_inputs", + lambda person, spi_people, build_period: person, + ) + support = build_uk_spi_support_channel( + person=_base_frame().table("person").assign( + incapacity_benefit_reported=[1.0, 0.0] + ), + benunit=_base_frame().table("benunit"), + household=pd.DataFrame( + { + "household_id": [1, 2], + "household_weight": [10.0, 20.0], + "region": ["LONDON", "SCOTLAND"], + } + ), + spi_household_count=2, + zero_weight_declarations=( + UKZeroWeightStratumDeclaration( + name="e7_spi_synthetic_preclone", + selector={HOUSEHOLD_IS_SPI_SYNTHETIC_COLUMN: True}, + maximum_zero_weight_rows=2, + reason="synthetic fixture", + ), + ), + ) + + with pytest.raises(ValueError, match="now carries non-default source signal"): + try: + impute_uk_spi_income_support( + support, + _donor_file(tmp_path), + donor_sample_size=None, + ) + finally: + monkeypatch.undo() diff --git a/packages/microcosm-build/tests/test_uk_spi_support.py b/packages/microcosm-build/tests/test_uk_spi_support.py index 062bba2e3..91d7b5b5a 100644 --- a/packages/microcosm-build/tests/test_uk_spi_support.py +++ b/packages/microcosm-build/tests/test_uk_spi_support.py @@ -14,6 +14,7 @@ SPI_INCOME_IMPUTATION_COLUMNS, SPI_PRIOR_MASS_CHANGE_REASON, SPI_SYNTHETIC_SUPPORT_CHANNEL, + build_uk_spi_support_channel, create_uk_spi_support_tables, fill_support_channel_from_source, replace_uk_spi_support_tables, @@ -22,6 +23,7 @@ support_clone_index_column, support_source_id_column, ) +from microcosm.build.uk_runtime.terminal_gates import UKZeroWeightStratumDeclaration from microcosm.frame import MassChangeRecord, WeightKind @@ -320,6 +322,88 @@ def test_replace_spi_support_preserves_quotas_and_allocates_real_mass() -> None: assert record.reason == SPI_PRIOR_MASS_CHANGE_REASON +def test_build_spi_support_channel_fresh_stack_allocates_region_mass() -> None: + prior_record = MassChangeRecord( + entity="household", + old_total=30.0, + new_total=60.0, + declared_factor=2.0, + reason="prior raw FRS calibration", + ) + + result = build_uk_spi_support_channel( + person=person_frame(), + benunit=benunit_frame(), + household=household_frame(), + spi_household_count=3, + seed=42, + source_year=2023, + strata_columns=("region",), + input_weight_kind=WeightKind.DESIGN, + mass_log=(prior_record,), + zero_weight_declarations=( + UKZeroWeightStratumDeclaration( + name="e7_spi_synthetic_preclone", + selector={HOUSEHOLD_IS_SPI_SYNTHETIC_COLUMN: True}, + maximum_zero_weight_rows=3, + reason="synthetic test declaration", + ), + ), + ) + + assert result.household_weight_kind is WeightKind.IMPORTANCE + assert result.mass_log[:-1] == (prior_record,) + assert result.mass_log[-1].old_total == 60.0 + assert result.mass_log[-1].new_total == 60.0 + channel = support_channel_column("household") + base = result.household[result.household[channel] == BASE_FRS_SUPPORT_CHANNEL] + spi = result.household[result.household[channel] == SPI_SYNTHETIC_SUPPORT_CHANNEL] + assert base["household_weight"].sum() == pytest.approx(30.0) + assert spi["household_weight"].sum() == pytest.approx(30.0) + pd.testing.assert_series_equal( + result.household.groupby("region")["household_weight"].sum(), + household_frame().groupby("region")["household_weight"].sum(), + ) + + +def test_build_spi_support_channel_fails_on_undeclared_zero_weight_row() -> None: + with pytest.raises(ValueError, match="match no declared stratum"): + build_uk_spi_support_channel( + person=person_frame(), + benunit=benunit_frame(), + household=household_frame(), + spi_household_count=3, + zero_weight_declarations=( + UKZeroWeightStratumDeclaration( + name="nonmatching", + selector={HOUSEHOLD_IS_SPI_SYNTHETIC_COLUMN: False}, + maximum_zero_weight_rows=0, + reason="synthetic test declaration", + ), + ), + ) + + +def test_build_spi_support_channel_fails_missing_positive_region_support() -> None: + with pytest.raises(ValueError, match="positive-mass UK base stratum"): + build_uk_spi_support_channel( + person=person_frame(), + benunit=benunit_frame(), + household=household_frame(), + spi_household_count=2, + seed=1, + strata_columns=("region",), + zero_weight_declarations=( + UKZeroWeightStratumDeclaration( + name="e7_spi_synthetic_preclone", + selector={HOUSEHOLD_IS_SPI_SYNTHETIC_COLUMN: True}, + maximum_zero_weight_rows=2, + reason="synthetic test declaration", + ), + ), + ) + + def test_replace_spi_support_builds_importance_pool_from_calibrated_base() -> None: dead = _certified_like_dead_spi_support() original_total = float(dead.household["household_weight"].sum()) diff --git a/tools/build_uk_frs_spine.py b/tools/build_uk_frs_spine.py index 90daa6d24..7ac805565 100644 --- a/tools/build_uk_frs_spine.py +++ b/tools/build_uk_frs_spine.py @@ -48,8 +48,14 @@ uk_frs_spine_seed_frame, ) from microcosm.build.uk_runtime.frs_take_up import UKFRSTakeUpStageTransform +from microcosm.build.uk_runtime.hmrc_replay import write_hmrc_replay_report 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.build.uk_runtime.spi_spine import ( + UKFRSHMRCSpineLeavesStageTransform, + UKSPIIncomeSpineStageTransform, + UKSPISupportChannelStageTransform, +) from microcosm.build.uk_runtime.take_up_contract import load_uk_take_up_contract from microcosm.frame.adapters.policyengine_uk import PolicyEngineUKEngine @@ -68,6 +74,9 @@ "frs_person_draws", "frs_household_draws", "frs_brma", + "frs_hmrc_spine_leaves", + "spi_support_channel", + "hmrc_spi_income_spine", ) @@ -91,6 +100,18 @@ def _parse_args(argv: list[str] | None = None) -> argparse.Namespace: required=True, help="Output H5 path for the raw FRS spine Frame.", ) + parser.add_argument( + "--spi-tab", + type=Path, + required=True, + help="Pinned local SPI 2022-23 put2223uk.tab path.", + ) + parser.add_argument( + "--hmrc-ods", + type=Path, + required=True, + help="Pinned local HMRC collated ODS path.", + ) parser.add_argument( "--checkpoint-dir", type=Path, @@ -116,9 +137,18 @@ def _validate_args(args: argparse.Namespace) -> None: ) if args.spine_h5.suffix != ".h5": raise ValueError("--spine-h5 must end with '.h5'.") + if not args.spi_tab.is_file(): + raise ValueError(f"--spi-tab must be an existing file: {args.spi_tab}") + if args.spi_tab.name != "put2223uk.tab": + raise ValueError("--spi-tab must name put2223uk.tab.") + if not args.hmrc_ods.is_file(): + raise ValueError(f"--hmrc-ods must be an existing file: {args.hmrc_ods}") + if args.hmrc_ods.suffix.lower() != ".ods": + raise ValueError("--hmrc-ods must end with '.ods'.") paths = { "spine_h5": args.spine_h5, "build_sidecar": args.spine_h5.with_suffix(".build.json"), + "hmrc_replay_sidecar": args.spine_h5.with_suffix(".hmrc_replay.json"), } if args.emit_nonzero_shares is not None: paths["emit_nonzero_shares"] = args.emit_nonzero_shares @@ -229,6 +259,15 @@ def _declared_seeds(stages) -> dict[str, dict[str, int]]: seed = operation.parameters.get("seed") if isinstance(output, str) and isinstance(seed, int): stage_seeds[output] = seed + elif isinstance(seed, int): + if operation.kind == "stack_zero_weight_donors": + stage_seeds["stack_zero_weight_donors"] = seed + elif operation.kind == "strict_read_private_table": + stage_seeds["donor_bootstrap"] = seed + elif operation.kind == "fit_weighted_qrf_stage1": + stage_seeds["stage1"] = seed + elif operation.kind == "fit_weighted_qrf_stage2": + stage_seeds["stage2"] = seed if stage_seeds: declared[stage.stage] = stage_seeds return declared @@ -364,6 +403,11 @@ def main(argv: list[str] | None = None) -> int: append_phase(state, "inputs_pinned") engine = _rules_engine() stochastic_contract = load_uk_take_up_contract() + hmrc_spine_transform = UKSPIIncomeSpineStageTransform( + args.spi_tab, + args.hmrc_ods, + stage=stages_by_name["hmrc_spi_income_spine"], + ) plan = country_stage_plan( spec, { @@ -413,6 +457,14 @@ def main(argv: list[str] | None = None) -> int: stage=stages_by_name["frs_brma"], engine=engine, ), + "frs_hmrc_spine_leaves": UKFRSHMRCSpineLeavesStageTransform( + args.frs_raw_dir, + stage=stages_by_name["frs_hmrc_spine_leaves"], + ), + "spi_support_channel": UKSPISupportChannelStageTransform( + stage=stages_by_name["spi_support_channel"], + ), + "hmrc_spi_income_spine": hmrc_spine_transform, }, stage_names=_STAGE_NAMES, ) @@ -425,6 +477,14 @@ def main(argv: list[str] | None = None) -> int: write_uk_national_frame(frame, args.checkpoint_dir / "frs_spine.h5") append_phase(state, "checkpoint_written") sidecar_path = output.with_suffix(".build.json") + replay_sidecar_path = output.with_suffix(".hmrc_replay.json") + if hmrc_spine_transform.last_result is None: + raise RuntimeError("HMRC SPI spine stage did not record replay evidence.") + write_hmrc_replay_report( + hmrc_spine_transform.last_result.replay_report, + replay_sidecar_path, + ) + append_phase(state, "hmrc_replay_sidecar_written") sidecar = _build_sidecar( frame=frame, stages=stages, From d53af94974f3125d99c1c7a5130bf5bedda65604 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mar=C3=ADa=20Juaristi?= <127882282+juaristi22@users.noreply.github.com> Date: Wed, 19 Aug 2026 10:13:34 +0200 Subject: [PATCH 2/2] Harden the spine build against the #717 adversarial-review findings Bind the SPI donor and HMRC ODS pins into the build sidecar and the Logbook input-pins digest (input_artifact_pins), bind the replay sidecar's hash and report_kind into the build receipt, clear stale outputs before a run so an interrupted build can never leave an old sidecar beside a fresh H5, and extend the manifest-drift asserts to every load-bearing SPI stage parameter (predictors, outputs, the FRS-channel initialization map, redraw columns, gate declaration, allocation weight kind/conservation, effective-mass floor). The rewrites-provenance finding is deferred to #721. Refs #683, #717. Co-Authored-By: Claude Fable 5 --- .../microcosm/build/uk_runtime/spi_spine.py | 89 ++++++++++++++- .../tests/test_uk_frs_spine.py | 42 +++++++- .../tests/test_uk_spi_spine.py | 102 ++++++++++++++++++ tools/build_uk_frs_spine.py | 78 +++++++++++++- 4 files changed, 307 insertions(+), 4 deletions(-) diff --git a/packages/microcosm-build/src/microcosm/build/uk_runtime/spi_spine.py b/packages/microcosm-build/src/microcosm/build/uk_runtime/spi_spine.py index 4c3f4ce01..68392488a 100644 --- a/packages/microcosm-build/src/microcosm/build/uk_runtime/spi_spine.py +++ b/packages/microcosm-build/src/microcosm/build/uk_runtime/spi_spine.py @@ -44,6 +44,7 @@ from microcosm.build.uk_runtime.spi_income import ( DEFAULT_SPI_DONOR_SAMPLE_SIZE, SPI_SOURCE_TI_FORMULA, + SPI_STAGE2_REVIEWED_ABSENT_OUTPUTS, UKSPIIncomeImputationResult, assert_frs_hmrc_auxiliary_crosswalk_available, impute_uk_spi_income_support, @@ -52,6 +53,7 @@ from microcosm.build.uk_runtime.spi_support import ( DEFAULT_SPI_PRIOR_MASS_SHARE, DEFAULT_SPI_SUPPORT_HOUSEHOLDS, + FRS_ONLY_SPI_FILL_INCOME_PREDICTOR_COLUMNS, HOUSEHOLD_IS_SPI_SYNTHETIC_COLUMN, SPI_HMRC_DERIVED_AUXILIARY_COLUMNS, SPI_HMRC_EMPLOYMENT_BENEFITS_COLUMN, @@ -61,6 +63,7 @@ SPI_HMRC_OTHER_SOCIAL_SECURITY_INCOME_COLUMN, SPI_HMRC_STATE_PENSION_INCOME_COLUMN, SPI_HMRC_TAXABLE_TERMINATION_PAY_COLUMN, + SPI_INCOME_QRF_OUTPUT_COLUMNS, SPI_SYNTHETIC_SUPPORT_CHANNEL, UKSPISupportResult, build_uk_spi_support_channel, @@ -156,6 +159,33 @@ FRS_HMRC_INCPBEN_COLUMN, ) +# Reviewed constants for every load-bearing manifest parameter the spine +# transforms consume. The drift asserts below fail closed on any manifest-only +# edit (adversarial-review finding on #717): a manifest change to these values +# requires a matching reviewed code change here. +SPI_SPINE_STAGE1_PREDICTORS = ("age", "gender", "region") +SPI_SPINE_STAGE2_PREDICTORS = ( + "age", + "gender", + "region", + *FRS_ONLY_SPI_FILL_INCOME_PREDICTOR_COLUMNS, +) +SPI_SPINE_STAGE2_OUTPUT_COLUMNS = tuple( + column + for column in UK_SPI_INCOME_SPINE_REWRITE_COLUMNS + if column not in FRS_ONLY_SPI_FILL_INCOME_PREDICTOR_COLUMNS + and column + not in (FRS_HMRC_PAY_COLUMN, FRS_HMRC_UBISJA_COLUMN, FRS_HMRC_INCPBEN_COLUMN) +) +SPI_SPINE_FRS_CHANNEL_INITIALIZATION = { + "gift_aid": 0.0, + "charitable_investment_gifts": 0.0, +} +SPI_SPINE_BASE_REDRAW_COLUMNS = ("dividend_income",) +SPI_SPINE_SUPPORT_CHANNELS = {"base": "frs", "synthetic": SPI_SYNTHETIC_SUPPORT_CHANNEL} +SPI_SPINE_PRECLONE_GATE_NAME = "e7_spi_synthetic_preclone" +SPI_SPINE_EFFECTIVE_MASS_COLUMNS = ("gift_aid", "charitable_investment_gifts") + @dataclass(frozen=True) class UKFRSHMRCSpineLeavesResult: @@ -597,16 +627,37 @@ def _support_stage_parameters( raise ValueError("SPI support manifest seed drifted from the reviewed value.") if stack.parameters.get("draw") != "uniform_without_replacement": raise ValueError("SPI support manifest must declare a uniform donor draw.") + if stack.parameters.get("flag_column") != HOUSEHOLD_IS_SPI_SYNTHETIC_COLUMN: + raise ValueError("SPI support manifest flag column drifted.") + if dict(stack.parameters.get("channels", {})) != SPI_SPINE_SUPPORT_CHANNELS: + raise ValueError("SPI support manifest channel labels drifted.") if allocation.parameters.get("share") != DEFAULT_SPI_PRIOR_MASS_SHARE: raise ValueError("SPI support prior-mass share drifted from 0.5.") + if allocation.parameters.get("weight_kind_out") != WeightKind.IMPORTANCE.value: + raise ValueError("SPI support allocation must advance to importance weights.") + if allocation.parameters.get("conservation") != "exact_total": + raise ValueError("SPI support allocation must declare exact-total conservation.") strata = tuple(allocation.parameters.get("strata", ())) if strata != ("region",): raise ValueError("SPI support spine allocation must use strata ['region'].") + declarations = _coerce_declarations(gate.parameters.get("declarations", ())) + if len(declarations) != 1: + raise ValueError("SPI support gate must declare exactly one pre-clone stratum.") + declaration = declarations[0] + if ( + declaration.name != SPI_SPINE_PRECLONE_GATE_NAME + or dict(declaration.selector) != {HOUSEHOLD_IS_SPI_SYNTHETIC_COLUMN: True} + or declaration.maximum_zero_weight_rows != DEFAULT_SPI_SUPPORT_HOUSEHOLDS + ): + raise ValueError( + "SPI support pre-clone gate declaration drifted from the reviewed " + "name/selector/maximum." + ) return ( int(stack.parameters["count"]), float(allocation.parameters["share"]), strata, - _coerce_declarations(gate.parameters.get("declarations", ())), + declarations, ) @@ -620,6 +671,7 @@ def _assert_income_stage_parameters( stage1 = _operation(stage, "fit_weighted_qrf_stage1") stage2 = _operation(stage, "fit_weighted_qrf_stage2") redraw = _operation(stage, "redraw_columns_from_fitted_qrf") + effective_mass = _operation(stage, "gate_distributional_effective_mass") if stage1.parameters.get("seed") != seed: raise ValueError("SPI income stage-1 seed drifted from the reviewed value.") if stage2.parameters.get("seed") != seed + 1: @@ -630,10 +682,45 @@ def _assert_income_stage_parameters( raise ValueError("SPI income stage-2 estimator count drifted.") if stage1.parameters.get("sample_size") != donor_sample_size: raise ValueError("SPI income donor sample size drifted.") + if stage1.parameters.get("post_sample_fit_weight") != "uniform": + raise ValueError("SPI income donor bootstrap must refit at uniform weight.") + if tuple(stage1.parameters.get("predictors", ())) != SPI_SPINE_STAGE1_PREDICTORS: + raise ValueError("SPI income stage-1 predictors drifted.") + if tuple(stage1.parameters.get("outputs", ())) != SPI_INCOME_QRF_OUTPUT_COLUMNS: + raise ValueError("SPI income stage-1 outputs drifted.") + initialization = dict(stage1.parameters.get("initialize_frs_channel_columns", {})) + if initialization != SPI_SPINE_FRS_CHANNEL_INITIALIZATION: + raise ValueError("SPI income FRS-channel initialization map drifted.") + if tuple(stage2.parameters.get("predictors", ())) != SPI_SPINE_STAGE2_PREDICTORS: + raise ValueError("SPI income stage-2 predictors drifted.") + if tuple(stage2.parameters.get("outputs", ())) != SPI_SPINE_STAGE2_OUTPUT_COLUMNS: + raise ValueError("SPI income stage-2 outputs drifted.") + reviewed_absent = stage2.parameters.get("reviewed_absent_outputs", {}) + if set(reviewed_absent) != set(SPI_STAGE2_REVIEWED_ABSENT_OUTPUTS): + raise ValueError("SPI income stage-2 reviewed-absent outputs drifted.") if redraw.parameters.get("fit") != "stage1": raise ValueError("SPI income base redraw must use the stage-1 fit.") if redraw.parameters.get("rows") != "base_support_channel": raise ValueError("SPI income base redraw must target the base support channel.") + if tuple(redraw.parameters.get("columns", ())) != SPI_SPINE_BASE_REDRAW_COLUMNS: + raise ValueError("SPI income base redraw columns drifted.") + if ( + tuple(effective_mass.parameters.get("columns", ())) + != SPI_SPINE_EFFECTIVE_MASS_COLUMNS + ): + raise ValueError("SPI income effective-mass columns drifted.") + if ( + effective_mass.parameters.get("minimum_nondefault_mass_share") + != DEFAULT_MINIMUM_NONDEFAULT_MASS_SHARE + ): + raise ValueError("SPI income effective-mass floor drifted.") + if effective_mass.parameters.get("fail_below_floor") is not True: + raise ValueError("SPI income effective-mass gate must fail below the floor.") + if ( + effective_mass.parameters.get("required_support_channel") + != SPI_SYNTHETIC_SUPPORT_CHANNEL + ): + raise ValueError("SPI income effective-mass support channel drifted.") def _operation(stage: SourceStageSpec, kind: str): diff --git a/packages/microcosm-build/tests/test_uk_frs_spine.py b/packages/microcosm-build/tests/test_uk_frs_spine.py index 52e8183bf..98a36e274 100644 --- a/packages/microcosm-build/tests/test_uk_frs_spine.py +++ b/packages/microcosm-build/tests/test_uk_frs_spine.py @@ -1088,6 +1088,13 @@ def test_driver_writes_spine_h5_sidecars_and_logbook( monkeypatch.delenv("POPULACE_LEDGER_API_KEY", raising=False) monkeypatch.delenv("POPULACE_LOGBOOK_PREV_ROW_DIGEST", raising=False) + # Stale sidecars from an earlier interrupted run must never survive + # beside a fresh H5 (adversarial-review finding on #717). + stale_replay = output.with_suffix(".hmrc_replay.json") + stale_build = output.with_suffix(".build.json") + stale_replay.write_text('{"report_kind": "stale_leftover"}') + stale_build.write_text('{"stale": true}') + assert ( tool.main( [ @@ -1154,8 +1161,16 @@ def test_driver_writes_spine_h5_sidecars_and_logbook( "stage1": 42, "stage2": 43, } - assert json.loads(output.with_suffix(".hmrc_replay.json").read_text()) == { - "report_kind": "fake_spine_replay" + replay_bytes = output.with_suffix(".hmrc_replay.json").read_bytes() + assert json.loads(replay_bytes) == {"report_kind": "fake_spine_replay"} + # The synthetic spec declares no non-table pinned artifacts, so the pin + # map is present but empty; the replay binding must match the file on + # disk byte-for-byte. + assert sidecar["input_artifact_pins"] == {} + assert sidecar["hmrc_replay"] == { + "filename": output.with_suffix(".hmrc_replay.json").name, + "report_kind": "fake_spine_replay", + "sha256": hashlib.sha256(replay_bytes).hexdigest(), } assert len(sidecar["stochastic_contract_sha256"]) == 64 assert sidecar["resource_pins"] == {"brma_rent_counts.json": "f" * 64} @@ -1395,3 +1410,26 @@ def test_refuses_nan_in_produced_weight_column(tmp_path: Path) -> None: with pytest.raises(ValueError, match="produced NaN"): build_uk_frs_spine_frame(tmp_path, stage=stage) + + +def test_input_artifact_pins_bind_spi_donor_and_ods() -> None: + tool = _load_tool() + spec = load_country_spec("uk") + assert spec.sources is not None + stage_map = spec.sources.stage_map() + stages = [stage_map[name] for name in tool._STAGE_NAMES] + + pins = tool._input_artifact_pins(stages) + + assert set(pins) == {"qrf_donor", "published_fact_surface"} + for pin in pins.values(): + assert len(str(pin["sha256"])) == 64 + assert int(pin["size_bytes"]) > 0 + assert str(pin["filename"]) + income_stage = stage_map["hmrc_spi_income_spine"] + declared = { + str(artifact["role"]): str(artifact["sha256"]) + for artifact in income_stage.artifacts + if "table" not in artifact and "resource" not in artifact + } + assert {role: pin["sha256"] for role, pin in pins.items()} == declared diff --git a/packages/microcosm-build/tests/test_uk_spi_spine.py b/packages/microcosm-build/tests/test_uk_spi_spine.py index f69d17a17..8b1f33e48 100644 --- a/packages/microcosm-build/tests/test_uk_spi_spine.py +++ b/packages/microcosm-build/tests/test_uk_spi_spine.py @@ -20,6 +20,8 @@ EMPLOYER_PENSION_CONTRIBUTIONS_COLUMN, UKFRSHMRCSpineLeavesStageTransform, UKSPISupportChannelStageTransform, + _assert_income_stage_parameters, + _support_stage_parameters, ) from microcosm.build.uk_runtime.spi_support import ( HOUSEHOLD_IS_SPI_SYNTHETIC_COLUMN, @@ -421,3 +423,103 @@ def test_reviewed_absent_incapacity_signal_raises(tmp_path: Path) -> None: ) finally: monkeypatch.undo() + + +def _committed_stage(name: str) -> SourceStageSpec: + from microcosm.build.country_spec import load_country_spec + + spec = load_country_spec("uk") + assert spec.sources is not None + return spec.sources.stage_map()[name] + + +def _with_mutated_operation( + stage: SourceStageSpec, kind: str, **overrides: object +) -> SourceStageSpec: + operations = [] + for operation in stage.operations: + payload: dict[str, object] = {"kind": operation.kind, **dict(operation.parameters)} + if operation.kind == kind: + payload.update(overrides) + operations.append(payload) + return SourceStageSpec.from_mapping({**stage.__dict__, "operations": operations}) + + +def test_income_stage_parameters_accept_the_committed_manifest() -> None: + _assert_income_stage_parameters( + _committed_stage("hmrc_spi_income_spine"), + seed=42, + qrf_estimators=100, + donor_sample_size=100_000, + ) + + +def test_income_stage_parameters_refuse_redraw_column_drift() -> None: + stage = _with_mutated_operation( + _committed_stage("hmrc_spi_income_spine"), + "redraw_columns_from_fitted_qrf", + columns=["savings_interest_income"], + ) + with pytest.raises(ValueError, match="base redraw columns drifted"): + _assert_income_stage_parameters( + stage, seed=42, qrf_estimators=100, donor_sample_size=100_000 + ) + + +def test_income_stage_parameters_refuse_frs_initialization_drift() -> None: + stage = _with_mutated_operation( + _committed_stage("hmrc_spi_income_spine"), + "fit_weighted_qrf_stage1", + initialize_frs_channel_columns={ + "gift_aid": 1.0, + "charitable_investment_gifts": 0.0, + }, + ) + with pytest.raises(ValueError, match="initialization map drifted"): + _assert_income_stage_parameters( + stage, seed=42, qrf_estimators=100, donor_sample_size=100_000 + ) + + +def test_income_stage_parameters_refuse_stage2_output_drift() -> None: + committed = _committed_stage("hmrc_spi_income_spine") + stage2 = next( + operation + for operation in committed.operations + if operation.kind == "fit_weighted_qrf_stage2" + ) + stage = _with_mutated_operation( + committed, + "fit_weighted_qrf_stage2", + outputs=list(stage2.parameters["outputs"])[:-1], + ) + with pytest.raises(ValueError, match="stage-2 outputs drifted"): + _assert_income_stage_parameters( + stage, seed=42, qrf_estimators=100, donor_sample_size=100_000 + ) + + +def test_support_stage_parameters_accept_the_committed_manifest() -> None: + count, share, strata, declarations = _support_stage_parameters( + _committed_stage("spi_support_channel"), seed=42 + ) + assert count == 10000 + assert share == 0.5 + assert strata == ("region",) + assert len(declarations) == 1 + + +def test_support_stage_parameters_refuse_gate_declaration_drift() -> None: + committed = _committed_stage("spi_support_channel") + gate = next( + operation + for operation in committed.operations + if operation.kind == "gate_zero_weight_strata" + ) + declaration = dict(gate.parameters["declarations"][0]) + declaration["maximum_zero_weight_rows"] = 20000 + stage = _with_mutated_operation( + committed, "gate_zero_weight_strata", declarations=[declaration] + ) + with pytest.raises(ValueError, match="gate declaration drifted"): + _support_stage_parameters(stage, seed=42) diff --git a/tools/build_uk_frs_spine.py b/tools/build_uk_frs_spine.py index 7ac805565..b81960bb4 100644 --- a/tools/build_uk_frs_spine.py +++ b/tools/build_uk_frs_spine.py @@ -4,6 +4,7 @@ import argparse import hashlib +import json import sys import time from datetime import UTC, datetime @@ -217,6 +218,46 @@ def _resource_pins(stages, spec) -> dict[str, str]: return dict(sorted(pins.items())) +def _input_artifact_pins(stages) -> dict[str, dict[str, object]]: + """Caller-supplied private input artifacts, pinned by role. + + Non-table, non-resource artifacts (the SPI donor tab and the HMRC ODS) + carry their own sha256/size pins in the manifest. Binding them here puts + the pins in the build sidecar and the Logbook input-pins digest, so two + runs with different high-impact source inputs can never share build-side + provenance (adversarial-review finding on #717). + """ + + pins: dict[str, dict[str, object]] = {} + for stage in stages: + for artifact in stage.artifacts: + if "table" in artifact or "resource" in artifact: + continue + if "sha256" not in artifact: + continue + role = str(artifact.get("role") or artifact.get("filename") or "") + if not role: + raise ValueError( + f"stage {stage.stage!r} declares a pinned input artifact " + "without a role or filename." + ) + pin = { + "filename": str( + artifact.get("filename") or artifact.get("locator") or "" + ), + "kind": str(artifact.get("kind", "")), + "sha256": str(artifact["sha256"]), + "size_bytes": int(artifact["size_bytes"]), + } + if role in pins and pins[role] != pin: + raise ValueError( + f"input artifact role {role!r} has inconsistent pins " + "across stages." + ) + pins[role] = pin + return dict(sorted(pins.items())) + + def _role_pins(pins: dict[str, dict[str, object]]) -> dict[str, dict[str, object]]: return { table: { @@ -280,6 +321,8 @@ def _build_sidecar( records, artifact_pins, resource_pins: dict[str, str], + input_artifact_pins: dict[str, dict[str, object]], + hmrc_replay: dict[str, object], stochastic_contract_sha256: str, ) -> dict[str, object]: household_weight = frame.weights_for("household") @@ -293,6 +336,8 @@ def _build_sidecar( "entity_row_counts": _entity_row_counts(frame), "artifact_pins": artifact_pins, "resource_pins": resource_pins, + "input_artifact_pins": input_artifact_pins, + "hmrc_replay": hmrc_replay, "stage_artifact_pins": { stage.stage: _stage_artifact_pins(stage) for stage in stages }, @@ -381,6 +426,20 @@ def main(argv: list[str] | None = None) -> int: spool_dir = args.spine_h5.parent / "logbook-spool" try: _validate_args(args) + # A crash between the H5 write and the sidecar writes must never + # leave a stale sidecar beside a fresh H5 (adversarial-review + # finding on #717): clear every output up front, and treat the + # build sidecar - written last, binding the replay hash - as the + # marker that the bundle is complete. + stale_outputs = [ + args.spine_h5, + args.spine_h5.with_suffix(".build.json"), + args.spine_h5.with_suffix(".hmrc_replay.json"), + ] + if args.emit_nonzero_shares is not None: + stale_outputs.append(args.emit_nonzero_shares) + for stale in stale_outputs: + stale.unlink(missing_ok=True) code_pin = git_code_pin(_REPOSITORY) append_phase(state, "configured") spec = load_country_spec("uk") @@ -390,7 +449,16 @@ def main(argv: list[str] | None = None) -> int: stages = [stages_by_name[name] for name in _STAGE_NAMES] artifact_pins = _artifact_pins(stages) resource_pins = _resource_pins(stages, spec) - state.input_pins_digest = role_pins_digest(_role_pins(artifact_pins)) + input_artifact_pins = _input_artifact_pins(stages) + overlapping_pin_roles = set(artifact_pins) & set(input_artifact_pins) + if overlapping_pin_roles: + raise ValueError( + "input artifact roles collide with FRS tab names: " + f"{sorted(overlapping_pin_roles)}." + ) + state.input_pins_digest = role_pins_digest( + _role_pins({**artifact_pins, **input_artifact_pins}) + ) run_config = { "pipeline": _PIPELINE, "stages": list(_STAGE_NAMES), @@ -485,12 +553,20 @@ def main(argv: list[str] | None = None) -> int: replay_sidecar_path, ) append_phase(state, "hmrc_replay_sidecar_written") + replay_bytes = replay_sidecar_path.read_bytes() + replay_binding = { + "filename": replay_sidecar_path.name, + "report_kind": str(json.loads(replay_bytes).get("report_kind", "")), + "sha256": hashlib.sha256(replay_bytes).hexdigest(), + } sidecar = _build_sidecar( frame=frame, stages=stages, records=records, artifact_pins=artifact_pins, resource_pins=resource_pins, + input_artifact_pins=input_artifact_pins, + hmrc_replay=replay_binding, stochastic_contract_sha256=stochastic_contract.resource_sha256, ) atomic_write_json(sidecar_path, sidecar)