Pin non_sch_d to the aged SOI fact (SS-repair precedent); wire the tail-gate exclusion register

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",

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: