diff --git a/packages/populace-build/tests/test_us_fiscal_refresh_builder.py b/packages/populace-build/tests/test_us_fiscal_refresh_builder.py index b28dc6bc..c170da89 100644 --- a/packages/populace-build/tests/test_us_fiscal_refresh_builder.py +++ b/packages/populace-build/tests/test_us_fiscal_refresh_builder.py @@ -1380,6 +1380,94 @@ def test_social_security_component_value_repair_uses_registry_targets( ) +def test_non_sch_d_cgd_value_repair_pins_the_aged_soi_fact(small_frame) -> None: + builder = _load_builder_module() + person = small_frame.table("person").copy() + person["non_sch_d_capital_gains"] = [100.0, 0.0, 300.0, 0.0] + frame = Frame( + { + "person": person, + "household": small_frame.table("household").copy(), + }, + small_frame.schema, + {"household": small_frame.weights_for("household")}, + ) + # The REAL compiled spec name is the unsuffixed ledger source_record_id + # (verified against a live v9.2 compile; PR #486 review finding 1). + spec = TargetSpec( + name="irs_soi.ty2023.table_1_4.all.capital_gain_distributions_amount", + entity="household", + value=500.0, + measure="unused", + period=builder.PERIOD, + source="IRS SOI", + metadata={"source_measure_id": "payment_amount", "aged_to": "2024"}, + ) + returns_decoy = TargetSpec( + name="irs_soi.ty2023.table_1_4.all.capital_gain_distributions_returns", + entity="household", + value=3_209_131.0, + measure="unused", + period=builder.PERIOD, + source="IRS SOI", + metadata={"source_measure_id": "return_count"}, + ) + state_decoy = TargetSpec( + name="irs_soi.ty2023.table_1_4.all.capital_gain_distributions_amount", + entity="household", + value=9.0, + measure="unused", + period=builder.PERIOD, + source="IRS SOI", + metadata={"source_measure_id": "payment_amount", "state_fips": "06"}, + ) + + repaired, repair = builder._with_non_sch_d_cgd_value_repair( + frame, (returns_decoy, spec, state_decoy) + ) + + assert repair["applied"] + weights = pd.Series(repaired.resolve_weights("person").values) + total = float((repaired.table("person")["non_sch_d_capital_gains"] * weights).sum()) + assert np.isclose(total, 500.0) + assert np.isclose(repair["repaired_estimate"], 500.0) + assert np.isclose(repair["factor"], repair["target"] / repair["initial_estimate"]) + assert repair["target_aged_to"] == "2024" + assert "mean-reverting" in repair["reason"] + + with pytest.raises(RuntimeError, match="exactly one aged Table 1.4"): + builder._with_non_sch_d_cgd_value_repair(frame, ()) + person_missing = small_frame.table("person").copy() + frame_missing = Frame( + { + "person": person_missing, + "household": small_frame.table("household").copy(), + }, + small_frame.schema, + {"household": small_frame.weights_for("household")}, + ) + with pytest.raises(RuntimeError, match="requires person column"): + builder._with_non_sch_d_cgd_value_repair(frame_missing, (spec,)) + + +def test_load_qrf_tail_concentration_exclusions(tmp_path) -> None: + builder = _load_builder_module() + assert builder._load_qrf_tail_concentration_exclusions(None) == {} + good = tmp_path / "tail.json" + good.write_text('{"estate_income": "tracked defect populace#481"}') + assert builder._load_qrf_tail_concentration_exclusions(good) == { + "estate_income": "tracked defect populace#481" + } + bad = tmp_path / "bad.json" + bad.write_text('{"estate_income": " "}') + with pytest.raises(ValueError, match="non-empty reason"): + builder._load_qrf_tail_concentration_exclusions(bad) + notdict = tmp_path / "list.json" + notdict.write_text("[1]") + with pytest.raises(ValueError, match="JSON object"): + builder._load_qrf_tail_concentration_exclusions(notdict) + + def test_release_gate_failures_reject_positive_zero_support_targets() -> None: builder = _load_builder_module() result = SimpleNamespace( @@ -2208,6 +2296,15 @@ def fake_base_population_mass_repair(frame): "_with_social_security_component_value_repair", lambda frame, specs: (frame, ss_repair_payload), ) + cgd_repair_payload = { + "method": "rescale_non_sch_d_capital_gains_to_soi_table_1_4_fact", + "applied": True, + } + monkeypatch.setattr( + builder, + "_with_non_sch_d_cgd_value_repair", + lambda frame, specs: (frame, cgd_repair_payload), + ) monkeypatch.setattr( builder, "_base_population_scale_gate", @@ -3201,7 +3298,8 @@ def fake_final_medicaid_diagnostics( == repair_payload ) assert captured["diagnostics"]["support_value_repairs"] == { - "social_security_components": ss_repair_payload + "social_security_components": ss_repair_payload, + "non_sch_d_capital_gains": cgd_repair_payload, } assert captured["diagnostics"]["default_dataset"] == { "method": "l0_refit", diff --git a/tools/build_us_fiscal_refresh_release.py b/tools/build_us_fiscal_refresh_release.py index 0bc367f5..306e0ece 100644 --- a/tools/build_us_fiscal_refresh_release.py +++ b/tools/build_us_fiscal_refresh_release.py @@ -802,6 +802,17 @@ def _parse_args() -> argparse.Namespace: "release manifest. The module registry is never mutated." ), ) + parser.add_argument( + "--qrf-tail-concentration-exclusions", + type=Path, + help=( + "Optional JSON object of export column -> reason for sparse " + "QRF-imputed columns allowed past the tail-concentration " + "top-share threshold (populace#464 gate). Stale entries fail the " + "gate; the file sha and entries are recorded in the release " + "diagnostics." + ), + ) parser.add_argument( "--selection-mass-protection", action="append", @@ -4799,6 +4810,8 @@ def _qrf_imputed_source_outputs() -> frozenset[str]: def _qrf_tail_concentration_gate( export_frame: Frame, + *, + reviewed_exclusions: Mapping[str, str] | None = None, ) -> tuple[GateResult, dict[str, object]]: """Tail-concentration gate over the sparse QRF-imputed export columns. @@ -4844,6 +4857,7 @@ def _qrf_tail_concentration_gate( top_k=US_QRF_TAIL_CONCENTRATION_TOP_K, max_top_share=US_QRF_TAIL_CONCENTRATION_MAX_TOP_SHARE, min_nonzero_records=US_QRF_TAIL_CONCENTRATION_MIN_NONZERO_RECORDS, + reviewed_exclusions=reviewed_exclusions, ) surface: dict[str, object] = { "qrf_imputed_outputs": len(qrf_outputs), @@ -4987,6 +5001,95 @@ def _with_social_security_component_value_repair( } +US_NON_SCH_D_CGD_REPAIR_REASON = ( + "The PUF E01100-lineage donor carries $24.31B across 4.67M weighted " + "carriers (weighted mean $5,206) against the SOI Pub 1304 Table 1.4 " + "TY2023 direct-route concept of $10.16B across 3.21M returns (mean " + "$3,165) - 2.39x on mass, measured on the sha-pinned puf_2024.h5 donor " + "via puf_tax_unit_donor_from_arrays. The eCPS-era pipeline produced " + "$13.69B from the same lineage, so the current 2024-level uprating " + "overstates a mean-reverting distribution series. Until the donor " + "uprating is variable-specific (root issue filed on the #462 thread), " + "the level is pinned to the ledger-fed Table 1.4 dollar fact (aging " + "provenance in target_aged_to) - the " + "same repair class as the Social Security component rescale above; the " + "returns-count row is an indicator and is unaffected." +) + + +def _with_non_sch_d_cgd_value_repair( + frame: Frame, + target_specs: Iterable[object], +) -> tuple[Frame, dict[str, object]]: + """Rescale non_sch_d_capital_gains to the aged SOI Table 1.4 dollar fact.""" + + column = "non_sch_d_capital_gains" + # TargetSpec names are the unsuffixed ledger source_record_ids (the + # @period suffix exists only on diagnostic names); the national row is + # the .all. filing-status segment with no state fips (PR #486 review + # finding 1 — the suffixed matcher matched nothing on a real compile). + matching = [ + spec + for spec in target_specs + if str(getattr(spec, "name", "")).startswith("irs_soi.") + and ".table_1_4.all." in str(getattr(spec, "name", "")) + and str(getattr(spec, "name", "")).endswith("capital_gain_distributions_amount") + and not spec.metadata.get("state_fips") + ] + if len(matching) != 1: + raise RuntimeError( + "non_sch_d capital-gain-distributions repair requires exactly one " + f"aged Table 1.4 dollar target; found {len(matching)}." + ) + target = float(matching[0].value) + if not math.isfinite(target) or target <= 0.0: + raise RuntimeError( + "non_sch_d capital-gain-distributions repair target must be " + f"finite and positive; got {target!r}." + ) + + person = frame.table("person").copy() + if column not in person.columns: + raise RuntimeError( + f"non_sch_d capital-gain-distributions repair requires person " + f"column {column!r}." + ) + person_weights = pd.Series( + frame.resolve_weights("person").values, index=person.index + ) + values = pd.to_numeric(person[column], errors="coerce").fillna(0.0) + initial = float((values * person_weights).sum()) + if not math.isfinite(initial) or initial <= 0.0: + raise RuntimeError( + "non_sch_d capital-gain-distributions repair requires positive " + f"finite support; got {initial!r}." + ) + factor = target / initial + applied = not math.isclose(factor, 1.0, rel_tol=1e-12, abs_tol=0.0) + if applied: + person[column] = values.to_numpy(dtype=np.float64) * factor + + tables_out = {entity: frame.table(entity).copy() for entity in frame.entities} + tables_out["person"] = person + repaired = Frame( + tables_out, + frame.schema, + {entity: frame.weights_for(entity) for entity in frame.weighted_entities}, + frame.strata, + mass_log=frame.mass_log, + ) + return repaired, { + "method": "rescale_non_sch_d_capital_gains_to_soi_table_1_4_fact", + "applied": applied, + "reason": US_NON_SCH_D_CGD_REPAIR_REASON, + "target": target, + "target_aged_to": matching[0].metadata.get("aged_to"), + "initial_estimate": initial, + "factor": factor, + "repaired_estimate": initial * factor, + } + + def _base_population_scale_gate( frame: Frame, *, @@ -6542,6 +6645,36 @@ def _load_zero_support_exclusions(path: Path | None) -> dict[str, str]: return exclusions +def _load_qrf_tail_concentration_exclusions(path: Path | None) -> dict[str, str]: + """Load a per-run QRF tail-concentration exclusion mapping. + + JSON object of ``column -> reason``: sparse QRF-imputed export columns + allowed to stay concentrated past the #464 top-share threshold, each with + a non-empty reason naming the tracked defect (the #481 weighted-leaf-draw + root fix) or the genuinely concentrated instrument. The gate itself + reports dormant entries and FAILS stale ones (a column now under the + threshold), so the register cannot rot. Returns an empty mapping when no + path is given. + """ + if path is None: + return {} + payload = json.loads(path.read_text()) + if not isinstance(payload, dict): + raise ValueError( + f"QRF tail-concentration exclusions file {path} must be a JSON " + "object of column -> reason." + ) + exclusions: dict[str, str] = {} + for column, reason in payload.items(): + if not isinstance(reason, str) or not reason.strip(): + raise ValueError( + "Every QRF tail-concentration exclusion needs a non-empty " + f"reason; {column!r} in {path} has {reason!r}." + ) + exclusions[str(column)] = reason + return exclusions + + def _reviewed_exclusions(active_aliases: Iterable[str]) -> dict[str, str]: active = set(active_aliases) hard = set(hard_target_package_aliases()) @@ -6936,6 +7069,9 @@ def main() -> None: base_frame, social_security_component_repair = ( _with_social_security_component_value_repair(base_frame, target_specs) ) + base_frame, non_sch_d_cgd_repair = _with_non_sch_d_cgd_value_repair( + base_frame, target_specs + ) if telemetry is not None: telemetry.stage( "base_population_repair", @@ -6951,6 +7087,19 @@ def main() -> None: applied=social_security_component_repair.get("applied"), components=social_security_component_repair.get("components"), ) + telemetry.stage( + "non_sch_d_cgd_repair", + message=( + "Pinned non_sch_d_capital_gains to the registry's SOI Table " + "1.4 dollar fact (populace#462 donor-uprating interim " + "repair; aging recorded in target_aged_to)." + ), + applied=non_sch_d_cgd_repair.get("applied"), + target_aged_to=non_sch_d_cgd_repair.get("target_aged_to"), + factor=non_sch_d_cgd_repair.get("factor"), + target=non_sch_d_cgd_repair.get("target"), + initial_estimate=non_sch_d_cgd_repair.get("initial_estimate"), + ) base_population_gate = _base_population_scale_gate( base_frame, mass_repair=base_population_repair, @@ -8553,7 +8702,8 @@ def main() -> None: pregnancy_gate=pregnancy_gate, snap_discretionary_exemption_gate=snap_discretionary_exemption_gate, support_value_repairs={ - "social_security_components": social_security_component_repair + "social_security_components": social_security_component_repair, + "non_sch_d_capital_gains": non_sch_d_cgd_repair, }, warm_start_calibration=warm_start_calibration, selection_source=selection_source_payload, @@ -8751,7 +8901,39 @@ def main() -> None: # parity (column excluded from the reference band), but is unmistakable as # top-k weighted-mass share. try: - qrf_tail_gate, qrf_tail_surface = _qrf_tail_concentration_gate(export_frame) + qrf_tail_exclusions = _load_qrf_tail_concentration_exclusions( + args.qrf_tail_concentration_exclusions + ) + qrf_tail_gate, qrf_tail_surface = _qrf_tail_concentration_gate( + export_frame, + reviewed_exclusions=qrf_tail_exclusions, + ) + register_dormant = sorted( + set(qrf_tail_exclusions) + - set(qrf_tail_gate.details.get("reviewed_exclusions", ())) + ) + if register_dormant: + raise RuntimeError( + "QRF tail-concentration exclusion register carries entries " + "the checked surface did not use (column dense, thin, " + f"absent, or below threshold): {register_dormant}. The " + "per-run register must exactly match the concentrated " + "columns — remove the stale entries." + ) + qrf_tail_surface = { + **qrf_tail_surface, + "reviewed_exclusions_file": ( + str(args.qrf_tail_concentration_exclusions) + if args.qrf_tail_concentration_exclusions is not None + else None + ), + "reviewed_exclusions_sha256": ( + _sha256(args.qrf_tail_concentration_exclusions) + if args.qrf_tail_concentration_exclusions is not None + else None + ), + "reviewed_exclusions": dict(qrf_tail_exclusions), + } except Exception as exc: # Same degraded-mode contract as the coverage gate above. if not terminal_gate_failures: