diff --git a/README.md b/README.md index b5b4adac00..6f859a224e 100644 --- a/README.md +++ b/README.md @@ -203,7 +203,11 @@ AT-RISK only, `0` clean): 3. **Export-mass parity risk** — each export-mass column's pool mass at *base* weights against its reference band, honoring the release tool's `US_EXPORT_INPUT_MASS_REVIEWED_EXCLUSIONS` register (reused, never - re-declared). A column out of band pre-solve is flagged for review. + re-declared). A column out of band pre-solve is flagged for review. Against + a reference where no record holds `s_corp_income`, the pool's + `partnership_income` counts both columns when a pool record with nonzero + weight holds it, as the export gate's candidate does, and the check's + details then name both columns. 4. **Smoke-probe support audit** — every reform-coverage probe leaf's pool vs *selected* nonzero support and pool sign-leg decomposition. A leaf with pool support but zero selected support **fails** (the input the frozen selection @@ -307,7 +311,16 @@ never as a release refusal. The checks: - **`export_input_mass`** runs the export input-mass gate with the staged frame standing in for the export. It classifies the `US_EXPORT_INPUT_MASS_REVIEWED_EXCLUSIONS` register as the gate does: `used`, - `below_reference_floor` or `unused`. + `below_reference_floor` or `unused`. The gate and this check read a reference + where no record holds `s_corp_income` as one from before microcosm#1261 split + Schedule E line 32: they count `partnership_income` plus `s_corp_income` as + the candidate's `partnership_income`, and the gate's details name both + columns. Any reference record that holds `s_corp_income` marks the split, even + when those amounts cancel, and the gate then compares each column alone. + Without a record of nonzero weight that holds `partnership_income`, the + candidate's column counts alone, so S corporation income can't stand in for a + lost column. A reference without the split can't show a wrong division of the + income between the columns. - **`degenerate_input_register`**, **`ecps_parity_register`**, **`input_coverage_register`**, **`stored_inputs`**, **`spm_composition`**, **`zero_support_preview`** and **`pre_solve_battery`** grade the remaining diff --git a/changelog.d/us-input-mass-line-32-pre-split-reference.fixed.md b/changelog.d/us-input-mass-line-32-pre-split-reference.fixed.md new file mode 100644 index 0000000000..04efdd8660 --- /dev/null +++ b/changelog.d/us-input-mass-line-32-pre-split-reference.fixed.md @@ -0,0 +1 @@ +The US input-mass gate now counts `partnership_income` plus `s_corp_income` when no reference record holds `s_corp_income`, and the release preflight and dry run read the line the same way. A reference from before #1261 holds partnership and S corporation income together in `partnership_income`, so a build that kept that income still failed the gate once the split left too little of it in that column. Any reference record that holds `s_corp_income` marks the split, even when those amounts cancel, and the gate then compares each column alone. The candidate's column counts alone unless a record with nonzero weight holds `partnership_income`, so S corporation income can't stand in for a lost column. The gate's and the preflight's details name both columns whenever they count them together. diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/input_mass.py b/packages/microcosm-build/src/microcosm/build/us_runtime/input_mass.py index 2aa0ba66ff..77cb4931d6 100644 --- a/packages/microcosm-build/src/microcosm/build/us_runtime/input_mass.py +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/input_mass.py @@ -3,13 +3,201 @@ The implementation was promoted to :mod:`microcosm.build.input_mass` when the UK terminal battery adopted the same #278 input-mass parity gate (#609); the computation is schema-driven and never carried US-specific logic. This module -remains so existing US call sites keep their import path. +remains so existing US call sites keep their import path, and it holds the one +US comparison rule the shared gate can't know: how to read Schedule E line 32 +against a reference built before the entity split. """ from __future__ import annotations +from collections.abc import Mapping +from dataclasses import dataclass, replace +from typing import Any + +import numpy as np +import pandas as pd + +from microcosm.build.gates import GateResult, input_mass_parity_gate from microcosm.build.input_mass import input_mass_totals +from microcosm.frame import Frame -__all__ = ["us_input_mass_totals"] +__all__ = [ + "US_LINE_32_COLUMNS", + "UsLine32Carriers", + "us_input_mass_parity_gate", + "us_input_mass_totals", + "us_line_32_carriers", + "us_line_32_counts_both", + "us_line_32_on_reference_terms", +] us_input_mass_totals = input_mass_totals + +#: The two columns microcosm#1261 split Schedule E line 32 into. +US_LINE_32_COLUMNS: tuple[str, str] = ("partnership_income", "s_corp_income") + + +@dataclass(frozen=True) +class UsLine32Carriers: + """Which line 32 columns have records that hold an amount. + + A weighted total can't tell records that cancel from no records at all, so + the line 32 rule reads these instead. :func:`us_line_32_carriers` reads them + from the frames whose totals the gate compares. + + Attributes: + candidate_partnership_income: A candidate record with nonzero weight + holds a nonzero ``partnership_income``. + reference_s_corp_income: A reference record, at any weight, holds a + nonzero ``s_corp_income``. + """ + + candidate_partnership_income: bool + reference_s_corp_income: bool + + +def _carries(frame: Frame, column: str, *, weighted: bool) -> bool: + """Whether a record of ``frame`` holds a nonzero ``column``. + + Values convert as :func:`microcosm.build.input_mass.input_mass_totals` + converts them, so a column whose total differs from zero always has a + carrier. With ``weighted``, only a record under a nonzero weight counts. + """ + for entity in frame.entities: + table = frame.table(entity) + if column not in table.columns: + continue + values = table[column] + if pd.api.types.is_bool_dtype(values): + numeric = values.fillna(False).to_numpy(dtype=np.float64) + elif pd.api.types.is_numeric_dtype(values): + numeric = pd.to_numeric(values, errors="coerce") + numeric = numeric.fillna(0.0).to_numpy(dtype=np.float64) + else: + return False + carriers = numeric != 0.0 + if weighted: + weights = np.asarray(frame.resolve_weights(entity).values, dtype=np.float64) + carriers &= weights != 0.0 + return bool(carriers.any()) + return False + + +def us_line_32_carriers(candidate: Frame, reference: Frame) -> UsLine32Carriers: + """Read which line 32 columns have carriers in a candidate and its reference. + + The candidate's ``partnership_income`` counts only records with nonzero + weight, because the gate guards the mass the weighted build keeps. The + reference's ``s_corp_income`` counts every record, because one that holds an + amount shows the build split the line, whatever its weight. + """ + return UsLine32Carriers( + candidate_partnership_income=_carries( + candidate, "partnership_income", weighted=True + ), + reference_s_corp_income=_carries(reference, "s_corp_income", weighted=False), + ) + + +def _nonzero(total: float) -> bool: + return bool(np.isfinite(total)) and total != 0.0 + + +def us_line_32_counts_both( + candidate_totals: Mapping[str, float], + reference_totals: Mapping[str, float], + carriers: UsLine32Carriers, +) -> bool: + """Whether the candidate's ``partnership_income`` counts both line 32 columns. + + It does when the candidate measures ``partnership_income`` and a record with + nonzero weight carries it, and no reference record carries ``s_corp_income``. + + Raises: + ValueError: If a nonzero total has no carrier. A record carries every + nonzero total, so the carriers then came from frames other than the + ones the totals came from. + """ + partnership = candidate_totals.get("partnership_income", 0.0) + if _nonzero(partnership) and not carriers.candidate_partnership_income: + raise ValueError( + f"partnership_income: the candidate totals {partnership:.6g} but no " + "candidate record with nonzero weight carries it; read the carriers " + "from the frames the totals came from." + ) + s_corp = reference_totals.get("s_corp_income", 0.0) + if _nonzero(s_corp) and not carriers.reference_s_corp_income: + raise ValueError( + f"s_corp_income: the reference totals {s_corp:.6g} but no reference " + "record carries it; read the carriers from the frames the totals " + "came from." + ) + return ( + "partnership_income" in candidate_totals + and carriers.candidate_partnership_income + and not carriers.reference_s_corp_income + ) + + +def us_line_32_on_reference_terms( + candidate_totals: Mapping[str, float], + reference_totals: Mapping[str, float], + carriers: UsLine32Carriers, +) -> dict[str, float]: + """Return candidate totals that read Schedule E line 32 as the reference does. + + A reference built before microcosm#1261 holds partnership and S corporation + income together in ``partnership_income`` and no ``s_corp_income`` record. + One built before microcosm#1251 holds only the nonpassive part of that + income. A candidate built after #1261 splits the income between the two + columns, so its ``partnership_income`` alone can fall short of that + reference however well the build keeps the income. Against a reference with + no record that carries ``s_corp_income``, or no such column, the candidate's + ``partnership_income`` counts both columns. + + Any reference record with an ``s_corp_income`` amount means the reference + carries the split, even when those amounts cancel, and the gate compares + each column alone. So the rule stops once a reference carries the split, and + a frame compared with itself never drifts. + + A candidate with no ``partnership_income`` column, or none on a record with + nonzero weight, stays as it is, and the input-mass gate refuses it. S + corporation income can't stand in for a lost column, even when it alone + lands inside the band. Records that carry partnership income count however + their amounts net, so when they cancel, the gate compares the combined line. + + Against a reference without the split, the gate reads only the total, so it + can't catch a wrong division of the income between the two columns. + """ + totals = dict(candidate_totals) + if us_line_32_counts_both(candidate_totals, reference_totals, carriers): + totals["partnership_income"] += totals.get("s_corp_income", 0.0) + return totals + + +def us_input_mass_parity_gate( + candidate_totals: Mapping[str, float], + reference_totals: Mapping[str, float], + *, + carriers: UsLine32Carriers, + **gate_terms: Any, +) -> GateResult: + """Run the input-mass gate with Schedule E line 32 on the reference's terms. + + When the rule counts both columns, the gate's details say so, because the + mass and drift under ``partnership_income`` then measure both. + """ + gate = input_mass_parity_gate( + us_line_32_on_reference_terms(candidate_totals, reference_totals, carriers), + reference_totals, + **gate_terms, + ) + if not us_line_32_counts_both(candidate_totals, reference_totals, carriers): + return gate + return replace( + gate, + details={ + **gate.details, + "partnership_income_counts": list(US_LINE_32_COLUMNS), + }, + ) diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/l0_refit_export.py b/packages/microcosm-build/src/microcosm/build/us_runtime/l0_refit_export.py index 10cd209e3d..c35fe8d9b6 100644 --- a/packages/microcosm-build/src/microcosm/build/us_runtime/l0_refit_export.py +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/l0_refit_export.py @@ -12,7 +12,6 @@ import numpy as np -from microcosm.build.gates import input_mass_parity_gate from microcosm.build.us_runtime.adult_care import ( US_ADULT_CARE_OUTPUT_COLUMNS, ) @@ -76,7 +75,11 @@ from microcosm.build.us_runtime.immigration import ( US_IMMIGRATION_NONCONSTANT_PERSON_COLUMNS, ) -from microcosm.build.us_runtime.input_mass import us_input_mass_totals +from microcosm.build.us_runtime.input_mass import ( + us_input_mass_parity_gate, + us_input_mass_totals, + us_line_32_carriers, +) from microcosm.build.us_runtime.medicare_take_up import ( US_MEDICARE_TAKE_UP_NONCONSTANT_PERSON_COLUMNS, ) @@ -607,9 +610,10 @@ def export_us_l0_refit_h5( reference_frame = ( load_us_frame(reference_h5) if reference_h5 is not None else base_frame ) - input_mass_gate = input_mass_parity_gate( + input_mass_gate = us_input_mass_parity_gate( us_input_mass_totals(export_frame), us_input_mass_totals(reference_frame), + carriers=us_line_32_carriers(export_frame, reference_frame), candidate_name="l0_refit_export", reference_name="reference_h5" if reference_h5 is not None else "base_h5", relative_tolerance=input_mass_relative_tolerance, diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/release_gate_preflight.py b/packages/microcosm-build/src/microcosm/build/us_runtime/release_gate_preflight.py index f1e2978a42..2544d4c711 100644 --- a/packages/microcosm-build/src/microcosm/build/us_runtime/release_gate_preflight.py +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/release_gate_preflight.py @@ -33,7 +33,11 @@ weights is an AT-RISK signal: the solve optimizes the target surface, not this untargeted column, so a pre-solve out-of-band mass usually stays out. The band math is reused verbatim via - :func:`microcosm.build.gates.input_mass_parity_gate`. + :func:`microcosm.build.gates.input_mass_parity_gate`, and the pool reads + Schedule E line 32 as the gate's candidate does + (:func:`microcosm.build.us_runtime.input_mass.us_line_32_on_reference_terms`). + When the pool's ``partnership_income`` counts both columns, the check's + details name them, as the gate's do. 4. **Smoke-probe support audit** — for every reform-coverage probe, for every ``binding_inputs`` leaf: pool nonzero support, *selected* nonzero support @@ -86,7 +90,7 @@ import hashlib import sys from collections.abc import Iterable, Mapping, Sequence -from dataclasses import dataclass, field +from dataclasses import dataclass, field, replace from pathlib import Path from typing import Any @@ -100,7 +104,13 @@ require_authenticated_us_multispine_pool_h5, us_multispine_pool_release_receipt, ) -from microcosm.build.us_runtime.input_mass import us_input_mass_totals +from microcosm.build.us_runtime.input_mass import ( + US_LINE_32_COLUMNS, + us_input_mass_totals, + us_line_32_carriers, + us_line_32_counts_both, + us_line_32_on_reference_terms, +) from microcosm.build.us_runtime.puf_capital_gains_tail import ( assert_puf_capital_gains_tail_survives_selection, ) @@ -1086,19 +1096,27 @@ def run_preflight( release.US_EXPORT_INPUT_MASS_REVIEWED_EXCLUSIONS ) reference_frame = load_us_frame(Path(export_input_mass_reference_h5)) - pool_totals = us_input_mass_totals(base_frame, columns=engine_input_variables) reference_totals = us_input_mass_totals( reference_frame, columns=engine_input_variables ) - checks.append( - check_export_mass_parity_risk( - pool_totals, - reference_totals, - relative_tolerance=relative_tolerance, - minimum_reference_total=minimum_reference_total, - reviewed_exclusions=export_mass_reviewed_exclusions, - ) + base_totals = us_input_mass_totals(base_frame, columns=engine_input_variables) + carriers = us_line_32_carriers(base_frame, reference_frame) + mass_check = check_export_mass_parity_risk( + us_line_32_on_reference_terms(base_totals, reference_totals, carriers), + reference_totals, + relative_tolerance=relative_tolerance, + minimum_reference_total=minimum_reference_total, + reviewed_exclusions=export_mass_reviewed_exclusions, ) + if us_line_32_counts_both(base_totals, reference_totals, carriers): + mass_check = replace( + mass_check, + details={ + **mass_check.details, + "partnership_income_counts": list(US_LINE_32_COLUMNS), + }, + ) + checks.append(mass_check) else: checks.append( CheckResult( diff --git a/packages/microcosm-build/tests/engine_free/us/test_us_fiscal_refresh_builder.py b/packages/microcosm-build/tests/engine_free/us/test_us_fiscal_refresh_builder.py index 537298e02e..e3431268d9 100644 --- a/packages/microcosm-build/tests/engine_free/us/test_us_fiscal_refresh_builder.py +++ b/packages/microcosm-build/tests/engine_free/us/test_us_fiscal_refresh_builder.py @@ -16,6 +16,7 @@ import pytest from microcosm.build.us_runtime import h5_io as builder_h5_io +from microcosm.build.us_runtime.input_mass import UsLine32Carriers from microcosm.calibrate import ( CalibrationHierarchy, HierarchyCategory, @@ -14928,13 +14929,32 @@ def _sentinel_frame(label: str): """A stand-in frame whose only role is object identity. The export-gate tests patch ``us_input_mass_totals`` to key off ``id(frame)`` - and never touch the schema, so a bare tagged object suffices — building a - real US-schema Frame (which needs the engine + every entity table) is + and the line 32 carriers reader to check which frames it gets, and never + touch the schema, so a bare tagged object suffices — building a real + US-schema Frame (which needs the engine + every entity table) is unnecessary to exercise the #327 reference-selection logic. """ return SimpleNamespace(label=label) +def _expect_line_32_carriers_from(monkeypatch, builder, candidate, reference): + """Patch the line 32 carriers reader to check the frames the gate passes. + + The gate reads carriers from the frames whose totals it compares, so the + reference it passes follows the same #327 selection as the totals. These + tests total no line 32 column, so the carriers don't change the verdict. + """ + + def carriers(candidate_frame, reference_frame): + assert candidate_frame is candidate + assert reference_frame is reference + return UsLine32Carriers( + candidate_partnership_income=False, reference_s_corp_income=False + ) + + monkeypatch.setattr(builder, "us_line_32_carriers", carriers) + + def test_export_input_mass_gate_defaults_to_base_reference(monkeypatch) -> None: """#327: with no reference_frame, the export gate compares vs the raw base. @@ -14957,6 +14977,7 @@ def test_export_input_mass_gate_defaults_to_base_reference(monkeypatch) -> None: "us_input_mass_totals", lambda frame, columns=None: totals[id(frame)], ) + _expect_line_32_carriers_from(monkeypatch, builder, export, base) gate = builder._export_input_mass_gate( export, @@ -14996,6 +15017,7 @@ def test_export_input_mass_gate_passes_against_certified_reference(monkeypatch) "us_input_mass_totals", lambda frame, columns=None: totals[id(frame)], ) + _expect_line_32_carriers_from(monkeypatch, builder, export, reference) gate = builder._export_input_mass_gate( export, @@ -15045,6 +15067,7 @@ def test_export_input_mass_gate_still_fails_genuine_drift_vs_reference( "us_input_mass_totals", lambda frame, columns=None: totals[id(frame)], ) + _expect_line_32_carriers_from(monkeypatch, builder, export, reference) gate = builder._export_input_mass_gate( export, diff --git a/packages/microcosm-build/tests/engine_free/us/test_us_input_mass_line_32.py b/packages/microcosm-build/tests/engine_free/us/test_us_input_mass_line_32.py new file mode 100644 index 0000000000..3aa82714cb --- /dev/null +++ b/packages/microcosm-build/tests/engine_free/us/test_us_input_mass_line_32.py @@ -0,0 +1,655 @@ +"""Schedule E line 32 against a reference built before the entity split. + +microcosm#1261 split line 32 between ``partnership_income`` and +``s_corp_income``. A reference built before it holds partnership and S +corporation income together in ``partnership_income`` and no ``s_corp_income`` +record, so the input-mass gate counts both candidate columns against that one +until any reference record carries ``s_corp_income``. The rule reads which +records carry each column, because a weighted total can't tell records that +cancel from no records. +""" + +from __future__ import annotations + +import importlib.util +import math +from types import SimpleNamespace + +import numpy as np +import pandas as pd +import pytest +from hypothesis import assume, given, settings +from hypothesis import strategies as st + +from microcosm.build.gates import input_mass_parity_gate +from microcosm.build.us_runtime import release_gate_dry_run as dry +from microcosm.build.us_runtime.input_mass import ( + US_LINE_32_COLUMNS, + UsLine32Carriers, + us_input_mass_parity_gate, + us_input_mass_totals, + us_line_32_carriers, + us_line_32_counts_both, + us_line_32_on_reference_terms, +) +from microcosm.frame import US_SCHEMA, Frame, WeightKind, Weights +from test_support.paths import paths_for + +_TEST_PATHS = paths_for("microcosm-build") + +FLOOR = 1e9 +TOLERANCE = 0.5 +BOTH_COLUMNS = ["partnership_income", "s_corp_income"] + +# Weighted totals from release attempt +# populace-us-2024-65671e1-9434c504ba2a-20261103T095117Z: its export and its +# reference, populace_us_2024.h5, which carries no s_corp_income record. +REFERENCE = {"partnership_income": 847.756e9, "s_corp_income": 0.0} +EXPORT = {"partnership_income": 391.947e9, "s_corp_income": 664.150e9} + +# That pair's carriers: weighted export records hold partnership income, and no +# reference record holds S corporation income. +PRE_SPLIT = UsLine32Carriers( + candidate_partnership_income=True, reference_s_corp_income=False +) +# A reference record holds S corporation income. +SPLIT = UsLine32Carriers( + candidate_partnership_income=True, reference_s_corp_income=True +) +# No candidate record with weight holds partnership income. +LOST = UsLine32Carriers( + candidate_partnership_income=False, reference_s_corp_income=False +) + +# The rule tests for exact zeros, so every draw can land on one. +_mass = st.floats(min_value=-2e12, max_value=2e12) | st.just(0.0) +_REFUSING_VERDICTS = {"zero_mass", "zero_mass_by_cancellation", "out_of_band"} + +# Small frames: amounts that cancel in pairs or go missing, and weights that +# include zero. +_amount = st.sampled_from([0.0, 1.0, -1.0, 2.5, -2.5, 3e9, -3e9, math.nan]) +_weight = st.sampled_from([0.0, 0.5, 1.0, 2.0]) + + +def _frame(partnership, s_corp, weights): + """A US frame of one-person households; ``None`` leaves a column out.""" + ids = np.arange(1, len(weights) + 1, dtype=np.int64) + person = {"person_id": ids} + for entity in US_SCHEMA.group_entities: + person[US_SCHEMA.membership_column(entity)] = ids + for column, values in zip(US_LINE_32_COLUMNS, (partnership, s_corp), strict=True): + if values is not None: + person[column] = np.asarray(values, dtype=np.float64) + tables = {"person": pd.DataFrame(person)} + for entity in US_SCHEMA.group_entities: + tables[entity] = pd.DataFrame({US_SCHEMA.id_column(entity): ids}) + household = Weights(np.asarray(weights, dtype=np.float64), WeightKind.DESIGN) + return Frame(tables, US_SCHEMA, {"household": household}) + + +@st.composite +def _frames(draw): + size = draw(st.integers(min_value=1, max_value=4)) + column = st.none() | st.lists(_amount, min_size=size, max_size=size) + # A frame refuses weights that are all zero. + weights = st.lists(_weight, min_size=size, max_size=size).filter(any) + return draw(column), draw(column), draw(weights) + + +def _gate(candidate, reference, carriers=PRE_SPLIT, floor=FLOOR): + return us_input_mass_parity_gate( + candidate, + reference, + carriers=carriers, + relative_tolerance=TOLERANCE, + minimum_reference_total=floor, + ) + + +def _frame_gate(candidate, reference, floor=0.0): + return _gate( + us_input_mass_totals(candidate), + us_input_mass_totals(reference), + us_line_32_carriers(candidate, reference), + floor=floor, + ) + + +def _preview( + candidate, + reference, + carriers=PRE_SPLIT, + *, + gate_failures=None, + partnership_share=0.02, +): + gate = _gate(candidate, reference, carriers) + return dry.export_input_mass_check( + gate_failures=gate.failures if gate_failures is None else gate_failures, + gate_details=gate.details, + candidate_totals=us_line_32_on_reference_terms(candidate, reference, carriers), + reference_totals=reference, + candidate_nonzero_shares={ + "partnership_income": partnership_share, + "s_corp_income": 0.02, + }, + reviewed_exclusions={}, + relative_tolerance=TOLERANCE, + minimum_reference_total=FLOOR, + margin=0.1, + allow_drift=False, + reference_is_candidate=False, + reference_label="reference.h5", + ) + + +def _partnership_row(check): + return {row["column"]: row for row in check.rows}["partnership_income"] + + +@pytest.fixture(scope="module") +def builder(): + path = _TEST_PATHS.repository / "tools" / "build_us_fiscal_refresh_release.py" + spec = importlib.util.spec_from_file_location( + "build_us_fiscal_refresh_release_line_32", path + ) + module = importlib.util.module_from_spec(spec) + assert spec.loader is not None + spec.loader.exec_module(module) + return module + + +def test_the_export_that_keeps_line_32_passes_a_pre_split_reference() -> None: + alone = input_mass_parity_gate( + EXPORT, + REFERENCE, + relative_tolerance=TOLERANCE, + minimum_reference_total=FLOOR, + ) + assert not alone.passed + assert alone.failures[0].startswith("partnership_income:") + + gate = _gate(EXPORT, REFERENCE) + assert gate.passed, gate.failures + assert gate.details["worst_drifts"]["partnership_income"] == pytest.approx( + (391.947 + 664.150 - 847.756) / 847.756 + ) + assert gate.details["partnership_income_counts"] == BOTH_COLUMNS + + +def test_a_pre_split_reference_still_refuses_lost_line_32_mass() -> None: + gate = _gate({"partnership_income": 2e11, "s_corp_income": 1e11}, REFERENCE) + assert not gate.passed + assert gate.failures[0].startswith("partnership_income:") + + +def test_a_zeroed_partnership_column_stays_refused() -> None: + # S corporation income alone lands inside the reference's band, so counting + # it here would pass a build that lost the whole column. + alone = EXPORT["s_corp_income"] / REFERENCE["partnership_income"] - 1 + assert abs(alone) <= TOLERANCE + + candidate = {"partnership_income": 0.0, "s_corp_income": EXPORT["s_corp_income"]} + gate = _gate(candidate, REFERENCE, LOST) + assert not gate.passed + assert "mass is zero" in gate.failures[0] + assert "partnership_income_counts" not in gate.details + + +def test_a_wrong_split_passes_a_pre_split_reference() -> None: + # Intended limit: a reference without the split can't show how a candidate + # divides the income, so the gate reads only the total. + assert _gate( + {"partnership_income": -1e12, "s_corp_income": 1.9e12}, REFERENCE + ).passed + assert _gate({"partnership_income": 1.0, "s_corp_income": 8e11}, REFERENCE).passed + + +def test_a_reference_with_the_split_compares_each_column_alone() -> None: + reference = {"partnership_income": 450e9, "s_corp_income": 600e9} + gate = _gate( + {"partnership_income": 100e9, "s_corp_income": 400e9}, reference, SPLIT + ) + assert not gate.passed + assert [line.split(":", 1)[0] for line in gate.failures] == [ + "partnership_income", + ] + assert "partnership_income_counts" not in gate.details + + +@pytest.mark.parametrize( + "reference_s_corp", [1.0, -1.0, FLOOR, -FLOOR, 0.9 * FLOOR, 0.0] +) +def test_any_s_corp_record_in_the_reference_ends_the_rule( + reference_s_corp: float, +) -> None: + # The gate skips a total at or under its floor, and records that cancel + # leave a zero total; neither says what the reference's columns hold. + reference = {"partnership_income": 847.756e9, "s_corp_income": reference_s_corp} + gate = _gate(EXPORT, reference, SPLIT) + assert not gate.passed + assert gate.failures[0].startswith("partnership_income:") + + +@pytest.mark.parametrize( + ("partnership", "s_corp"), + [(1.5e9, 1e9), (1.1e9, 0.9e9), (1.5e9, 1.00001e9), (1.5e9, -1e9)], +) +def test_a_frame_with_small_s_corp_income_passes_against_itself( + partnership: float, s_corp: float +) -> None: + totals = {"partnership_income": partnership, "s_corp_income": s_corp} + gate = _gate(totals, totals, SPLIT) + assert gate.passed, gate.failures + + +def test_a_candidate_without_partnership_income_stays_refused() -> None: + # Totals that don't measure the column keep it out, whatever the carriers say. + candidate = {"s_corp_income": 5e11} + assert "partnership_income" not in us_line_32_on_reference_terms( + candidate, REFERENCE, PRE_SPLIT + ) + gate = _gate(candidate, REFERENCE, LOST) + assert not gate.passed + assert "absent" in gate.failures[0] + + +def test_a_candidate_without_s_corp_income_keeps_its_partnership_income() -> None: + assert us_line_32_on_reference_terms( + {"partnership_income": 8e11}, REFERENCE, PRE_SPLIT + ) == {"partnership_income": 8e11} + + +def test_carriers_that_contradict_the_totals_raise() -> None: + # A record carries every nonzero total, so a mismatch means the carriers + # came from other frames than the totals did. + with pytest.raises(ValueError, match="no candidate record with nonzero weight"): + _gate(EXPORT, REFERENCE, LOST) + split = {"partnership_income": 450e9, "s_corp_income": 600e9} + with pytest.raises(ValueError, match="no reference record carries it"): + us_line_32_on_reference_terms(EXPORT, split, PRE_SPLIT) + + +def test_a_non_finite_total_reaches_the_gates_own_refusal() -> None: + candidate = {"partnership_income": math.nan, "s_corp_income": 1.0} + with pytest.raises(ValueError, match="must be finite"): + _gate(candidate, REFERENCE, LOST) + + +def test_carriers_read_records_rather_than_totals() -> None: + # A split reference whose S corporation records cancel to a zero total. + split = _frame([1, 1], [2, -2], [1, 1]) + assert us_input_mass_totals(split)["s_corp_income"] == 0.0 + assert us_line_32_carriers(split, split) == SPLIT + # Pre-split references: every s_corp_income amount is zero, or no column. + for reference in (_frame([2, 0], [0, 0], [1, 1]), _frame([2], None, [1])): + assert us_line_32_carriers(split, reference) == PRE_SPLIT + + +def test_a_reference_s_corp_record_counts_at_any_weight() -> None: + reference = _frame([2, 0], [0, 3], [1, 0]) + assert us_input_mass_totals(reference)["s_corp_income"] == 0.0 + assert us_line_32_carriers(reference, reference).reference_s_corp_income + + +def test_a_candidate_partnership_record_counts_only_under_weight() -> None: + def carries(candidate): + return us_line_32_carriers(candidate, candidate).candidate_partnership_income + + assert carries(_frame([1, -1], [0.5, 0.5], [1, 1])) + assert not carries(_frame([1, 0], [0, 0.8], [0, 1])) + assert not carries(_frame(None, [1], [1])) + + +@pytest.mark.parametrize("floor", [0.0, 1.0]) +@pytest.mark.parametrize("scale", [1.0, 1e9]) +def test_a_reference_whose_s_corp_records_cancel_compares_each_column( + scale: float, floor: float +) -> None: + # Its totals match a pre-split reference's: the S corporation records net + # to zero. + reference = _frame([scale, scale], [2 * scale, -2 * scale], [1, 1]) + + # Its own records reweighted keep partnership income, which counting both + # columns would read as +100%. + reweighted = _frame([scale, scale], [2 * scale, -2 * scale], [1.5, 0.5]) + assert us_input_mass_totals(reweighted) == { + "partnership_income": 2 * scale, + "s_corp_income": 2 * scale, + } + gate = _frame_gate(reweighted, reference, floor=floor * scale) + assert gate.passed, gate.failures + assert "partnership_income_counts" not in gate.details + + # A 75% partnership loss, which counting both columns would pass. + lost = _frame([0.25 * scale, 0.25 * scale], [3 * scale, -1.5 * scale], [1, 1]) + gate = _frame_gate(lost, reference, floor=floor * scale) + assert [line.split(":", 1)[0] for line in gate.failures] == [ + "partnership_income", + ] + assert gate.details["worst_drifts"]["partnership_income"] == pytest.approx(-0.75) + + +@pytest.mark.parametrize(("scale", "floor"), [(1.0, 0.0), (2e9, FLOOR)]) +def test_cancelling_partnership_records_count_both_columns( + scale: float, floor: float +) -> None: + reference = _frame([scale], [0], [1]) + candidate = _frame([scale, -scale], [0.5 * scale, 0.5 * scale], [1, 1]) + assert us_input_mass_totals(candidate)["partnership_income"] == 0.0 + gate = _frame_gate(candidate, reference, floor=floor) + assert gate.passed, gate.failures + assert gate.details["partnership_income_counts"] == BOTH_COLUMNS + assert gate.details["worst_drifts"]["partnership_income"] == 0.0 + + +def test_partnership_records_without_weight_stay_refused() -> None: + # S corporation income alone reads -20%, inside the band. + reference = _frame([1], [0], [1]) + candidate = _frame([1, 0], [0, 0.8], [0, 1]) + gate = _frame_gate(candidate, reference) + assert not gate.passed + assert "mass is zero" in gate.failures[0] + assert "partnership_income_counts" not in gate.details + + +@given( + partnership=_mass, + s_corp=_mass, + other=_mass, + reference_s_corp=_mass, + candidate_carries=st.booleans(), + reference_carries=st.booleans(), +) +def test_only_partnership_income_moves_and_only_on_a_pre_split_reference( + partnership: float, + s_corp: float, + other: float, + reference_s_corp: float, + candidate_carries: bool, + reference_carries: bool, +) -> None: + # A record carries every nonzero total. + carriers = UsLine32Carriers( + candidate_partnership_income=candidate_carries or partnership != 0.0, + reference_s_corp_income=reference_carries or reference_s_corp != 0.0, + ) + candidate = { + "partnership_income": partnership, + "s_corp_income": s_corp, + "other": other, + } + reference = {"s_corp_income": reference_s_corp} + before = dict(candidate) + totals = us_line_32_on_reference_terms(candidate, reference, carriers) + assert candidate == before + assert totals["s_corp_income"] == s_corp + assert totals["other"] == other + counts_both = ( + carriers.candidate_partnership_income and not carriers.reference_s_corp_income + ) + assert us_line_32_counts_both(candidate, reference, carriers) is counts_both + expected = partnership + s_corp if counts_both else partnership + assert totals["partnership_income"] == expected + + +@given( + partnership=_mass, + s_corp=_mass, + floor=st.sampled_from([0.0, FLOOR]), + partnership_carries=st.booleans(), + s_corp_carries=st.booleans(), +) +def test_totals_compared_with_themselves_never_drift( + partnership: float, + s_corp: float, + floor: float, + partnership_carries: bool, + s_corp_carries: bool, +) -> None: + carriers = UsLine32Carriers( + candidate_partnership_income=partnership_carries or partnership != 0.0, + reference_s_corp_income=s_corp_carries or s_corp != 0.0, + ) + totals = {"partnership_income": partnership, "s_corp_income": s_corp} + gate = _gate(totals, totals, carriers, floor=floor) + assert gate.passed, gate.failures + assert all(drift == 0.0 for drift in gate.details["worst_drifts"].values()) + + +@settings(deadline=None) +@given(frame=_frames(), floor=st.sampled_from([0.0, 1.0])) +def test_a_frame_compared_with_itself_never_drifts(frame, floor: float) -> None: + built = _frame(*frame) + gate = _frame_gate(built, built, floor=floor) + assert gate.passed, gate.failures + assert all(drift == 0.0 for drift in gate.details["worst_drifts"].values()) + + +@settings(deadline=None) +@given(candidate=_frames(), reference=_frames()) +def test_carriers_match_a_record_by_record_reading(candidate, reference) -> None: + # A missing amount totals as zero, so it holds nothing. + def holds(amount): + return not math.isnan(amount) and amount != 0 + + partnership, _, weights = candidate + _, s_corp, _ = reference + carriers = us_line_32_carriers(_frame(*candidate), _frame(*reference)) + assert carriers.candidate_partnership_income is ( + partnership is not None + and any(holds(p) and w != 0 for p, w in zip(partnership, weights, strict=True)) + ) + assert carriers.reference_s_corp_income is ( + s_corp is not None and any(holds(s) for s in s_corp) + ) + + +@settings(deadline=None) +@given(candidate=_frames(), reference=_frames()) +def test_carriers_from_the_frames_never_contradict_their_totals( + candidate, reference +) -> None: + # The rule raises on a nonzero total without a carrier; real frames never + # trip it. + candidate_frame, reference_frame = _frame(*candidate), _frame(*reference) + us_line_32_counts_both( + us_input_mass_totals(candidate_frame), + us_input_mass_totals(reference_frame), + us_line_32_carriers(candidate_frame, reference_frame), + ) + + +@given(line_32=st.floats(min_value=1e9, max_value=2e12), partnership=_mass) +def test_the_split_never_changes_the_verdict_on_a_pre_split_reference( + line_32: float, partnership: float +) -> None: + # Moving signed income between the entity columns keeps the total to + # rounding, so the gate's verdict tracks the total alone while a weighted + # record carries partnership income, even one whose amounts net to zero. + s_corp = line_32 - partnership + reference = REFERENCE["partnership_income"] + drift = abs(partnership + s_corp - reference) / abs(reference) + assume(abs(drift - TOLERANCE) >= 1e-9) + candidate = {"partnership_income": partnership, "s_corp_income": s_corp} + assert _gate(candidate, REFERENCE).passed is (drift <= TOLERANCE) + + +@given( + partnership=_mass, + s_corp=_mass, + reference_s_corp=_mass, + candidate_carries=st.booleans(), + reference_carries=st.booleans(), + unweighted_share=st.sampled_from([0.0, 0.02]), +) +def test_the_dry_run_preview_agrees_with_the_gate( + partnership: float, + s_corp: float, + reference_s_corp: float, + candidate_carries: bool, + reference_carries: bool, + unweighted_share: float, +) -> None: + carriers = UsLine32Carriers( + candidate_partnership_income=candidate_carries or partnership != 0.0, + reference_s_corp_income=reference_carries or reference_s_corp != 0.0, + ) + # A weighted carrier is a record, and the record share also counts records + # at zero weight. + share = 0.02 if carriers.candidate_partnership_income else unweighted_share + reference = {"partnership_income": 8e11, "s_corp_income": reference_s_corp} + candidate = {"partnership_income": partnership, "s_corp_income": s_corp} + gate = _gate(candidate, reference, carriers) + refused = any(line.startswith("partnership_income:") for line in gate.failures) + row = _partnership_row( + _preview(candidate, reference, carriers, partnership_share=share) + ) + assert (row["verdict"] in _REFUSING_VERDICTS) is refused + + +def test_the_preview_raises_when_the_gate_passes_what_it_reads_as_failing() -> None: + lost = {"partnership_income": 2e11, "s_corp_income": 1e11} + with pytest.raises(ValueError, match="the release's gate passes it"): + _preview(lost, REFERENCE, gate_failures=()) + + +def test_the_preview_raises_when_the_gate_fails_what_it_reads_in_band() -> None: + with pytest.raises(ValueError, match="the preview reads it in band"): + _preview(EXPORT, REFERENCE, gate_failures=("partnership_income: refused",)) + + +def test_the_preview_refuses_a_column_with_no_partnership_record_for_certain() -> None: + # No weights can give these records partnership income, so the rule never + # counts their S corporation income. + candidate = {"partnership_income": 0.0, "s_corp_income": EXPORT["s_corp_income"]} + row = _partnership_row(_preview(candidate, REFERENCE, LOST, partnership_share=0.0)) + assert (row["verdict"], row["status"], row["certain"]) == ( + "zero_mass", + "FAIL", + True, + ) + + +def test_the_preview_counts_both_columns_for_cancelling_partnership_records() -> None: + # Weighted records carry partnership income that nets to zero, so the gate + # compares the combined line, which keeps 78% of the reference. + candidate = {"partnership_income": 0.0, "s_corp_income": EXPORT["s_corp_income"]} + assert _gate(candidate, REFERENCE).passed + row = _partnership_row(_preview(candidate, REFERENCE)) + assert row["mass_at_base_weights"] == EXPORT["s_corp_income"] + assert (row["verdict"], row["status"]) == ("in_band", "PASS") + + +def test_the_preview_reads_partnership_records_without_weight_as_at_risk() -> None: + # Records at zero base weight can gain weight in the solve, and the rule + # then counts both columns, so the gate's refusal here isn't certain. + candidate = {"partnership_income": 0.0, "s_corp_income": EXPORT["s_corp_income"]} + assert "mass is zero" in _gate(candidate, REFERENCE, LOST).failures[0] + row = _partnership_row(_preview(candidate, REFERENCE, LOST)) + assert (row["verdict"], row["status"], row["certain"]) == ( + "zero_mass_by_cancellation", + "AT_RISK", + False, + ) + + +def test_the_preview_reads_records_that_cancel_across_the_columns_as_at_risk() -> None: + # Intended: if every partnership record has an opposite S corporation record, + # no weights can rescue the total, so FAIL would hold too. The preview sees + # one share per column and can't tell, and the gate still refuses at export. + candidate = {"partnership_income": 5e11, "s_corp_income": -5e11} + assert "mass is zero" in _gate(candidate, REFERENCE).failures[0] + row = _partnership_row(_preview(candidate, REFERENCE)) + assert (row["verdict"], row["status"], row["certain"]) == ( + "zero_mass_by_cancellation", + "AT_RISK", + False, + ) + + +def _sentinel_frames(monkeypatch, builder, *, carriers_from, **totals_by_label): + frames = {label: SimpleNamespace(label=label) for label in totals_by_label} + monkeypatch.setattr(builder, "_engine_input_variables", lambda: ()) + monkeypatch.setattr( + builder, + "us_input_mass_totals", + lambda frame, columns=None: totals_by_label[frame.label], + ) + + def carriers(candidate, reference): + # The gate must read carriers from the frames whose totals it compares. + assert (candidate.label, reference.label) == carriers_from + return PRE_SPLIT + + monkeypatch.setattr(builder, "us_line_32_carriers", carriers) + return frames + + +def test_the_release_export_gate_reads_line_32_on_the_reference_terms( + builder, monkeypatch +) -> None: + frames = _sentinel_frames( + monkeypatch, + builder, + carriers_from=("export", "reference"), + export=EXPORT, + base=EXPORT, + reference=REFERENCE, + ) + gate = builder._export_input_mass_gate( + frames["export"], + frames["base"], + relative_tolerance=TOLERANCE, + minimum_reference_total=FLOOR, + reference_frame=frames["reference"], + reference_name="populace_us_2024.h5", + ) + assert gate.passed, gate.failures + assert gate.details["partnership_income_counts"] == BOTH_COLUMNS + + +def test_the_release_export_gate_refuses_lost_line_32_mass( + builder, monkeypatch +) -> None: + lost = {"partnership_income": 2e11, "s_corp_income": 1e11} + frames = _sentinel_frames( + monkeypatch, + builder, + carriers_from=("export", "reference"), + export=lost, + base=EXPORT, + reference=REFERENCE, + ) + gate = builder._export_input_mass_gate( + frames["export"], + frames["base"], + relative_tolerance=TOLERANCE, + minimum_reference_total=FLOOR, + reference_frame=frames["reference"], + reference_name="populace_us_2024.h5", + ) + assert not gate.passed + assert gate.failures[0].startswith("partnership_income:") + + +def test_the_release_base_gate_reads_line_32_on_the_reference_terms( + builder, monkeypatch, tmp_path +) -> None: + frames = _sentinel_frames( + monkeypatch, + builder, + carriers_from=("base", "reference"), + base=EXPORT, + reference=REFERENCE, + ) + monkeypatch.setattr(builder, "load_us_frame", lambda path: frames["reference"]) + gate = builder._input_mass_reference_gate( + frames["base"], + reference_h5=tmp_path / "populace_us_2024.h5", + relative_tolerance=TOLERANCE, + minimum_reference_total=FLOOR, + ) + assert gate is not None + assert gate.passed, gate.failures + assert gate.details["partnership_income_counts"] == BOTH_COLUMNS diff --git a/packages/microcosm-build/tests/engine_free/us/test_us_l0_refit_export.py b/packages/microcosm-build/tests/engine_free/us/test_us_l0_refit_export.py index 5aa49ccc9b..903ad93f31 100644 --- a/packages/microcosm-build/tests/engine_free/us/test_us_l0_refit_export.py +++ b/packages/microcosm-build/tests/engine_free/us/test_us_l0_refit_export.py @@ -7,6 +7,7 @@ import pytest from microcosm.build.us_runtime import l0_refit_export +from microcosm.build.us_runtime.input_mass import UsLine32Carriers from microcosm.build.us_runtime.l0_refit_export import ( assert_required_us_release_source_columns, attach_l0_refit_entity_weights, @@ -421,6 +422,62 @@ def test_export_us_l0_refit_h5_fails_when_selection_zeroes_input_mass( assert not output.exists() +def test_export_us_l0_refit_h5_reads_line_32_on_the_reference_terms( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + frame = _us_frame() + base_h5 = tmp_path / "base.h5" + base_h5.write_text("base") + output = tmp_path / "populace_us_2024.h5" + + monkeypatch.setattr(l0_refit_export, "load_us_frame", lambda path: frame) + monkeypatch.setattr( + l0_refit_export, + "copy_microcosm_root_attrs", + lambda source, destination: (), + ) + # A reference built before microcosm#1261 holds partnership and S + # corporation income together in partnership_income; the export splits the + # income between the two columns. + monkeypatch.setattr( + l0_refit_export, + "us_input_mass_totals", + lambda bundle: ( + {"partnership_income": 8.5e11, "s_corp_income": 0.0} + if bundle is frame + else {"partnership_income": 3.9e11, "s_corp_income": 6.6e11} + ), + ) + + def carriers(candidate, reference): + # The gate reads carriers from the frames whose totals it compares: the + # reweighted export, then the base frame that stands in as reference. + assert candidate is not frame + assert reference is frame + return UsLine32Carriers( + candidate_partnership_income=True, reference_s_corp_income=False + ) + + monkeypatch.setattr(l0_refit_export, "us_line_32_carriers", carriers) + + class FakeEngine: + def write_dataset(self, bundle, path, period): + Path(path).write_text("sentinel") + + monkeypatch.setattr(l0_refit_export, "PolicyEngineUSEngine", FakeEngine) + + summary = export_us_l0_refit_h5( + base_h5=base_h5, + weights_npz=_l0_refit_npz(tmp_path), + output_h5=output, + input_mass_relative_tolerance=0.5, + require_geography_ladder=False, + ) + + assert summary["input_mass_parity"]["passed"] is True + + def test_export_us_l0_refit_h5_records_input_mass_drift_when_allowed( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, diff --git a/packages/microcosm-build/tests/engine_free/us/test_us_release_gate_dry_run.py b/packages/microcosm-build/tests/engine_free/us/test_us_release_gate_dry_run.py index c0c8a2c27c..c52fbe2575 100644 --- a/packages/microcosm-build/tests/engine_free/us/test_us_release_gate_dry_run.py +++ b/packages/microcosm-build/tests/engine_free/us/test_us_release_gate_dry_run.py @@ -45,6 +45,7 @@ parity_gate, ) from microcosm.build.us_runtime import release_gate_dry_run as dry +from microcosm.build.us_runtime.input_mass import UsLine32Carriers from microcosm.build.us_runtime.release_gate_preflight import PreflightReport from microcosm.build.us_runtime.spm_composition import CheckResult from microcosm.calibrate.registry import TargetSpec @@ -1435,9 +1436,7 @@ def test_checkpoint_status_reads_only_the_identity(builder, tmp_path) -> None: assert not builder._dry_run_checkpoint_status(path, "def")["identity_matches"] -def test_release_dry_run_checks_run_the_release_gates( - builder, tail_frame, monkeypatch -) -> None: +def _release_dry_run_checks(builder, tail_frame, monkeypatch, *extra_args): """The stop point's check assembly, on a real frame. Engine-backed pieces (input defaults, input coverage) are stubbed because @@ -1461,6 +1460,7 @@ def test_release_dry_run_checks_run_the_release_gates( "--dense-default-dataset", "--dry-run-gates-report", "r.json", + *extra_args, ] ) early = ["Retirement-distribution signal failed: fixture early failure"] @@ -1498,6 +1498,13 @@ def test_release_dry_run_checks_run_the_release_gates( ), ecps_parity_gate=GateResult(name="parity", passed=True), ) + return early, checks + + +def test_release_dry_run_checks_run_the_release_gates( + builder, tail_frame, monkeypatch +) -> None: + early, checks = _release_dry_run_checks(builder, tail_frame, monkeypatch) by_name = {check.name: check for check in checks} assert list(by_name) == [ "pre_solve_battery", @@ -1534,6 +1541,56 @@ def test_release_dry_run_checks_run_the_release_gates( ) +def test_release_dry_run_reads_line_32_as_the_gate_does( + builder, tail_frame, monkeypatch +) -> None: + """The preview raises if it reads partnership_income alone here. + + A reference built before microcosm#1261 holds partnership and S corporation + income together in partnership_income, so the gate counts both candidate + columns against it. + """ + reference = SimpleNamespace(label="reference") + monkeypatch.setattr(builder, "load_us_frame", lambda path: reference) + + def carriers(candidate, reference_frame): + # The gate and the preview read carriers from the frames whose totals + # they compare. + assert candidate is tail_frame + assert reference_frame is reference + return UsLine32Carriers( + candidate_partnership_income=True, reference_s_corp_income=False + ) + + monkeypatch.setattr(builder, "us_line_32_carriers", carriers) + monkeypatch.setattr( + builder, + "us_input_mass_totals", + lambda frame, columns=None: ( + {"partnership_income": 8.5e11, "s_corp_income": 0.0} + if frame is reference + else {"partnership_income": 3.9e11, "s_corp_income": 6.6e11} + ), + ) + monkeypatch.setattr( + builder, + "us_nonzero_shares", + lambda frame, columns=None: {"partnership_income": 0.02, "s_corp_income": 0.02}, + ) + _, checks = _release_dry_run_checks( + builder, + tail_frame, + monkeypatch, + "--export-input-mass-reference-h5", + "populace_us_2024.h5", + ) + check = {item.name: item for item in checks}["export_input_mass"] + assert not check.details.get("evaluation_error") + row = {row["column"]: row for row in check.rows}["partnership_income"] + assert row["mass_at_base_weights"] == pytest.approx(1.05e12) + assert row["verdict"] == "in_band" + + def test_release_dry_run_checks_evidence_tier_missing_register_and_l0( builder, tail_frame, monkeypatch, tmp_path ) -> None: diff --git a/packages/microcosm-build/tests/engine_free/us/test_us_release_gate_preflight.py b/packages/microcosm-build/tests/engine_free/us/test_us_release_gate_preflight.py index 3ae7135822..8d1f80f561 100644 --- a/packages/microcosm-build/tests/engine_free/us/test_us_release_gate_preflight.py +++ b/packages/microcosm-build/tests/engine_free/us/test_us_release_gate_preflight.py @@ -33,6 +33,7 @@ load_congressional_district_vintage_crosswalk, ) from microcosm.build.us_runtime.h5_io import AuthenticatedPoolH5 +from microcosm.build.us_runtime.input_mass import UsLine32Carriers from microcosm.build.us_runtime.release_gate_preflight import ( MAX_REPORTED_SPM_UNITS_HARD_CAP, PreflightReport, @@ -1939,7 +1940,9 @@ def _whole_pool_selection(pool: list[dict[str, object]]) -> SelectionSource: ) -def _run_synthetic_preflight(monkeypatch, tmp_path, *, frame: Frame, **mode): +def _run_synthetic_preflight( + monkeypatch, tmp_path, *, frame: Frame, reference_frame: Frame | None = None, **mode +): """``run_preflight`` over a synthetic base, with every optional input given. The base loader, the Ledger target compiler and the reference-H5 loader are @@ -1964,7 +1967,11 @@ def _run_synthetic_preflight(monkeypatch, tmp_path, *, frame: Frame, **mode): "_load_ledger_target_specs", lambda path, *, target_period, ledger_facts_sha256, **kwargs: (spec,), ) - monkeypatch.setattr(l0_refit_export, "load_us_frame", lambda path: frame) + monkeypatch.setattr( + l0_refit_export, + "load_us_frame", + lambda path: frame if reference_frame is None else reference_frame, + ) return preflight_module.run_preflight( base_h5=tmp_path / "base.h5", export_input_mass_reference_h5=tmp_path / "reference.h5", @@ -1977,6 +1984,73 @@ def _run_synthetic_preflight(monkeypatch, tmp_path, *, frame: Frame, **mode): ) +def _line_32_preflight_check(monkeypatch, tmp_path, *, reference_s_corp_income): + """The preflight's mass check on line 32 totals rounded from an 11/03 export. + + The reference's ``s_corp_income`` total is zero either way; the carriers + say whether a reference record holds an amount, as records that cancel do. + """ + base = _new_lineage_frame() + reference = _new_lineage_frame() + monkeypatch.setattr( + preflight_module, + "us_input_mass_totals", + lambda frame, columns=None: ( + {"partnership_income": 8.5e11, "s_corp_income": 0.0} + if frame is reference + else {"partnership_income": 3.9e11, "s_corp_income": 6.6e11} + ), + ) + + def carriers(candidate, reference_frame): + # The check reads carriers from the frames whose totals it compares. + assert candidate is base + assert reference_frame is reference + return UsLine32Carriers( + candidate_partnership_income=True, + reference_s_corp_income=reference_s_corp_income, + ) + + monkeypatch.setattr(preflight_module, "us_line_32_carriers", carriers) + report = _run_synthetic_preflight( + monkeypatch, tmp_path, frame=base, reference_frame=reference, new_lineage=True + ) + return next(c for c in report.checks if c.name == "export_mass_parity_risk") + + +def test__run_preflight__reads_line_32_on_the_reference_terms( + monkeypatch, tmp_path +) -> None: + check = _line_32_preflight_check( + monkeypatch, tmp_path, reference_s_corp_income=False + ) + + assert check.status == "PASS" + row = next(r for r in check.rows if r["column"] == "partnership_income") + assert row["pool_mass_at_base_weights"] == pytest.approx(1.05e12) + assert row["verdict"] == "in_band" + # The row's mass measures both columns, so the details say so, as the + # gate's do. + assert check.details["partnership_income_counts"] == [ + "partnership_income", + "s_corp_income", + ] + + +def test__run_preflight__compares_each_line_32_column_against_a_split_reference( + monkeypatch, tmp_path +) -> None: + check = _line_32_preflight_check( + monkeypatch, tmp_path, reference_s_corp_income=True + ) + + assert check.status == "AT_RISK" + row = next(r for r in check.rows if r["column"] == "partnership_income") + assert row["pool_mass_at_base_weights"] == pytest.approx(3.9e11) + assert row["verdict"] == "out_of_band_at_base_weights" + assert "partnership_income_counts" not in check.details + + def test__run_preflight__new_lineage_skips_only_carryover_and_records_why( monkeypatch, tmp_path ) -> None: diff --git a/tools/build_us_fiscal_refresh_release.py b/tools/build_us_fiscal_refresh_release.py index 4a52f0c5fa..f0b951e455 100644 --- a/tools/build_us_fiscal_refresh_release.py +++ b/tools/build_us_fiscal_refresh_release.py @@ -63,7 +63,6 @@ GateResult, TargetFitRequirement, default_valued_columns_gate, - input_mass_parity_gate, nonconstant_columns_gate, observed_value_counts, parity_gate, @@ -274,7 +273,12 @@ require_authenticated_us_multispine_pool_h5, us_multispine_pool_release_receipt, ) -from microcosm.build.us_runtime.input_mass import us_input_mass_totals +from microcosm.build.us_runtime.input_mass import ( + us_input_mass_parity_gate, + us_input_mass_totals, + us_line_32_carriers, + us_line_32_on_reference_terms, +) from microcosm.build.us_runtime.l0_refit_export import ( attach_l0_refit_entity_weights, load_us_frame, @@ -7618,9 +7622,11 @@ def _input_mass_reference_gate( if reference_h5 is None: return None input_variables = _engine_input_variables() - return input_mass_parity_gate( + reference_frame = load_us_frame(reference_h5) + return us_input_mass_parity_gate( us_input_mass_totals(base_frame, columns=input_variables), - us_input_mass_totals(load_us_frame(reference_h5), columns=input_variables), + us_input_mass_totals(reference_frame, columns=input_variables), + carriers=us_line_32_carriers(base_frame, reference_frame), candidate_name="base_frame", reference_name=reference_h5.name, relative_tolerance=relative_tolerance, @@ -7991,9 +7997,10 @@ def _export_input_mass_gate( """ input_variables = _engine_input_variables() reference = reference_frame if reference_frame is not None else base_frame - return input_mass_parity_gate( + return us_input_mass_parity_gate( us_input_mass_totals(export_frame, columns=input_variables), us_input_mass_totals(reference, columns=input_variables), + carriers=us_line_32_carriers(export_frame, reference), candidate_name="export_frame", reference_name=reference_name, relative_tolerance=relative_tolerance, @@ -12137,15 +12144,23 @@ def export_input_mass() -> CheckResult: ) input_variables = _engine_input_variables() candidate_totals = us_input_mass_totals(base_frame, columns=input_variables) + reference_totals = ( + candidate_totals + if reference_frame is None + else us_input_mass_totals(reference_frame, columns=input_variables) + ) check = export_input_mass_check( gate_failures=gate.failures, gate_details=gate.details, - candidate_totals=candidate_totals, - reference_totals=( - candidate_totals - if reference_frame is None - else us_input_mass_totals(reference_frame, columns=input_variables) + candidate_totals=us_line_32_on_reference_terms( + candidate_totals, + reference_totals, + us_line_32_carriers( + base_frame, + base_frame if reference_frame is None else reference_frame, + ), ), + reference_totals=reference_totals, candidate_nonzero_shares=us_nonzero_shares( base_frame, columns=input_variables ),