Accept a restated Ledger filter bound only where it agrees with the compiled constraint

changelog.d/us-labelled-restated-filters.added.md

@@ -0,0 +1 @@
+The US fiscal materializer now accepts a Ledger filter key that restates a constraint it already applies — a labelled feed's `ledger_filter_us:statutes/26/62#adjusted_gross_income_lower_bound` beside the compiled `agi_lower_bound`, or a qualifying-child lower bound or exact count beside `ledger_filter_eitc_child_count` — but only on specs where the restatement and the applied filter select the same population. A restated qualifying-child upper bound is refused outright, whatever it would agree with, until the Ledger confirms whether it means `<` or `<=`. A restatement that disagrees, or that has no compiled counterpart, keeps the existing refusal and now names both values, and the SOI slice's silent skip honours the same rule so an accepted restatement cannot drop a target without an error.

experiments/us-labelled-filter-support/census_compiled_ledger_filters.py

@@ -0,0 +1,180 @@
+"""Measure the Ledger filter guards against a compiled US fiscal registry.
+
+Compiles the US fiscal target registry from a Chronicle consumer-facts feed
+and reports, for the whole registry and for the state-level surface
+``tools/build_us_acs_local_release.py`` selects in both SOI modes:
+
+* target counts by family and ``target_role``;
+* every ``ledger_filter_*`` key carried, with its value histogram;
+* what ``_unsupported_ledger_filter_metadata`` refuses (the fatal guard) and
+  what ``_unsupported_soi_ledger_filters`` would skip silently;
+* a sample of compiled metadata mappings, one per (family, ``target_role``)
+  plus IRS SOI rows carrying an AGI band or an EITC child-count filter, which
+  regenerates ``tests/fixtures/us_compiled_ledger_filter_specs.json``.
+
+Usage::
+
+    uv run python experiments/us-labelled-filter-support/\\
+        census_compiled_ledger_filters.py <consumer_facts.jsonl> [out.json]
+
+Needs no engine: the compile path reads the feed and the congressional-district
+vintage crosswalk only. It does read the whole feed, so it is slow (order ten
+minutes for the 39,158-fact US feed) and is a receipt, not a test.
+"""
+
+from __future__ import annotations
+
+import collections
+import importlib.util
+import json
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[2]
+
+
+def _load_tool(name: str):
+    spec = importlib.util.spec_from_file_location(name, ROOT / "tools" / f"{name}.py")
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[name] = module
+    assert spec.loader is not None
+    spec.loader.exec_module(module)
+    return module
+
+
+def _survey(builder, specs: list, label: str) -> dict[str, object]:
+    unsupported = builder._unsupported_ledger_filter_metadata(specs)
+    refusals = collections.Counter(
+        entry for entries in unsupported.values() for entry in entries
+    )
+    skipped = collections.Counter()
+    skipped_targets = 0
+    for spec in specs:
+        if getattr(spec, "family", "") != "irs_soi":
+            continue
+        keys = builder._unsupported_soi_ledger_filters(spec.metadata)
+        if keys:
+            skipped_targets += 1
+            skipped.update(keys)
+    keys = collections.Counter()
+    values: dict[str, collections.Counter] = collections.defaultdict(
+        collections.Counter
+    )
+    for spec in specs:
+        for key, value in spec.metadata.items():
+            if str(key).startswith("ledger_filter"):
+                keys[str(key)] += 1
+                values[str(key)][str(value)] += 1
+    roles = collections.Counter(
+        (spec.family, spec.metadata.get("target_role", "")) for spec in specs
+    )
+    return {
+        "label": label,
+        "targets": len(specs),
+        "by_family": dict(collections.Counter(spec.family for spec in specs)),
+        "by_family_role": {
+            f"{family}|{role}": count for (family, role), count in sorted(roles.items())
+        },
+        "ledger_filter_key_census": {
+            key: {"targets": count, "values": dict(values[key].most_common())}
+            for key, count in keys.most_common()
+        },
+        "refused_targets": len(unsupported),
+        "refusals": dict(refusals),
+        "soi_silently_skipped_targets": skipped_targets,
+        "soi_skipped_keys": dict(skipped),
+    }
+
+
+def _fixture_rows(specs: list) -> list[dict[str, object]]:
+    rows: dict[str, dict[str, object]] = {}
+    for spec in specs:
+        role = (spec.family, spec.metadata.get("target_role", ""))
+        if role in {
+            (row["family"], row["metadata"].get("target_role", ""))
+            for row in rows.values()
+        }:
+            continue
+        rows[spec.name] = {
+            "name": spec.name,
+            "family": spec.family,
+            "metadata": dict(sorted(spec.metadata.items())),
+        }
+    banded = 0
+    for spec in specs:
+        if spec.family != "irs_soi" or spec.name in rows:
+            continue
+        metadata = spec.metadata
+        interesting = metadata.get("ledger_filter_eitc_child_count") or (
+            metadata.get("agi_lower_bound") not in (None, "-inf")
+            and metadata.get("agi_upper_bound") not in (None, "inf")
+        )
+        if not interesting:
+            continue
+        rows[spec.name] = {
+            "name": spec.name,
+            "family": spec.family,
+            "metadata": dict(sorted(metadata.items())),
+        }
+        banded += 1
+        if banded >= 4:
+            break
+    return sorted(rows.values(), key=lambda row: row["name"])
+
+
+def main(argv: list[str]) -> int:
+    if not argv:
+        print(__doc__)
+        return 2
+    feed = argv[0]
+    out_path = Path(argv[1]) if len(argv) > 1 else None
+
+    builder = _load_tool("build_us_fiscal_refresh_release")
+    acs = _load_tool("build_us_acs_local_release")
+
+    from microcosm.build.ledger_artifact import load_ledger_consumer_artifact
+    from microcosm.build.us_runtime import (
+        default_congressional_district_vintage_crosswalk_path,
+        load_congressional_district_vintage_crosswalk,
+    )
+    from microcosm.build.us_runtime.fiscal_targets import (
+        compile_us_fiscal_target_registry,
+    )
+
+    crosswalk = load_congressional_district_vintage_crosswalk(
+        default_congressional_district_vintage_crosswalk_path()
+    )
+    artifact = load_ledger_consumer_artifact(feed)
+    registry = compile_us_fiscal_target_registry(
+        artifact.facts,
+        target_period=acs.PERIOD,
+        congressional_district_vintage_crosswalk=crosswalk,
+        age_targets=True,
+    )
+    whole = list(registry.specs)
+
+    surveys = [_survey(builder, whole, "whole_registry")]
+    for soi_mode in ("full", "totals"):
+        selected, _substitutions = acs.state_admin_specs(
+            feed, ["snap", "medicaid", "soi"], soi_mode=soi_mode
+        )
+        surveys.append(
+            _survey(builder, list(selected.specs), f"state_surface_{soi_mode}")
+        )
+
+    payload = {
+        "feed": feed,
+        "target_period": acs.PERIOD,
+        "surveys": surveys,
+        "fixture_specs": _fixture_rows(whole),
+    }
+    text = json.dumps(payload, indent=2, default=str)
+    if out_path is None:
+        print(text)
+    else:
+        out_path.write_text(text + "\n")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main(sys.argv[1:]))

experiments/us-labelled-filter-support/measure_restated_filter_arms.py

@@ -0,0 +1,181 @@
+"""Count the restated-filter guard's refusals with the rule and with it reverted.
+
+For each Chronicle consumer-facts feed given, compiles the US fiscal target
+registry and surveys three surfaces — the whole compiled registry and the
+state-level surface ``tools/build_us_acs_local_release.py::state_admin_specs``
+selects in both SOI modes — under two arms:
+
+* ``rule``: the tree as it stands, ``_restated_ledger_filter_refusal`` at both
+  call sites;
+* ``reverted``: ``_restated_ledger_filter_refusal`` replaced by a function that
+  returns the bare key for every otherwise-unsupported ``ledger_filter_*`` key.
+  That is the pre-rule behaviour of both ``_unsupported_ledger_filter_metadata``
+  and ``_unsupported_soi_ledger_filters``, and so ``main``'s before this branch.
+
+Per surface and arm it reports refused targets, refusal entries by key,
+refused targets by (family, ``target_role``), and ``irs_soi`` targets the SOI
+loop would skip silently. Output is aggregate target-spec counts only; no
+microdata is read. It regenerates ``restated_filter_arms_receipt.json``::
+
+    uv run python experiments/us-labelled-filter-support/\\
+        measure_restated_filter_arms.py <consumer_facts.jsonl> [...] > receipt.json
+
+Needs no engine. Each surface is compiled through the production path, so a
+feed is compiled three times; the receipt records the wall time per feed
+(``compile_and_select_seconds``) and the process's peak RSS
+(``max_rss_bytes``, bytes on macOS). It also records the SHA-256 of this
+script and the ``git status --porcelain`` of the tree it ran in, so a receipt
+can be matched to the committed script that produced it.
+"""
+
+from __future__ import annotations
+
+import collections
+import hashlib
+import importlib.util
+import json
+import resource
+import subprocess
+import sys
+import time
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[2]
+
+
+def _load_tool(name: str):
+    spec = importlib.util.spec_from_file_location(name, ROOT / "tools" / f"{name}.py")
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[name] = module
+    assert spec.loader is not None
+    spec.loader.exec_module(module)
+    return module
+
+
+def _sha256(path: str) -> str:
+    digest = hashlib.sha256()
+    with open(path, "rb") as handle:
+        for block in iter(lambda: handle.read(1 << 20), b""):
+            digest.update(block)
+    return digest.hexdigest()
+
+
+def _survey(builder, specs: list, label: str) -> dict[str, object]:
+    unsupported = builder._unsupported_ledger_filter_metadata(specs)
+    entries: collections.Counter[str] = collections.Counter()
+    for spec_entries in unsupported.values():
+        for entry in spec_entries:
+            entries[entry.split("=", 1)[0]] += 1
+    by_name = {spec.name: spec for spec in specs}
+    by_role: collections.Counter[str] = collections.Counter()
+    for name in unsupported:
+        spec = by_name[name]
+        by_role[f"{spec.family}|{spec.metadata.get('target_role', '')}"] += 1
+    skipped = sum(
+        1
+        for spec in specs
+        if getattr(spec, "family", "") == "irs_soi"
+        and builder._unsupported_soi_ledger_filters(spec.metadata)
+    )
+    return {
+        "label": label,
+        "targets": len(specs),
+        "refused_targets": len(unsupported),
+        "refusal_entries": sum(entries.values()),
+        "refusal_entries_by_key": dict(entries.most_common()),
+        "refused_by_family_role": dict(by_role.most_common()),
+        "soi_silently_skipped_targets": skipped,
+    }
+
+
+def main(argv: list[str]) -> int:
+    if not argv:
+        print(__doc__)
+        return 2
+
+    builder = _load_tool("build_us_fiscal_refresh_release")
+    acs = _load_tool("build_us_acs_local_release")
+
+    import microcosm.build.ledger_targets as ledger_targets
+    from microcosm.build.ledger_artifact import load_ledger_consumer_artifact
+    from microcosm.build.us_runtime import (
+        default_congressional_district_vintage_crosswalk_path,
+        load_congressional_district_vintage_crosswalk,
+    )
+    from microcosm.build.us_runtime.fiscal_targets import (
+        compile_us_fiscal_target_registry,
+    )
+
+    # The compile must be this checkout's, not another installed tree's.
+    assert Path(ledger_targets.__file__).resolve().is_relative_to(ROOT), (
+        ledger_targets.__file__
+    )
+
+    rule = builder._restated_ledger_filter_refusal
+
+    def reverted(key: str, value: str, metadata: object) -> str:
+        return key
+
+    results = []
+    for feed in argv:
+        started = time.time()
+        artifact = load_ledger_consumer_artifact(feed)
+        registry = compile_us_fiscal_target_registry(
+            artifact.facts,
+            target_period=acs.PERIOD,
+            congressional_district_vintage_crosswalk=(
+                load_congressional_district_vintage_crosswalk(
+                    default_congressional_district_vintage_crosswalk_path()
+                )
+            ),
+            age_targets=True,
+        )
+        surfaces = {"whole_registry": list(registry.specs)}
+        del artifact, registry
+        for soi_mode in ("full", "totals"):
+            selected, _substitutions = acs.state_admin_specs(
+                feed, ["snap", "medicaid", "soi"], soi_mode=soi_mode
+            )
+            surfaces[f"state_surface_{soi_mode}"] = list(selected.specs)
+        result: dict[str, object] = {
+            "feed": feed,
+            "feed_sha256": _sha256(feed),
+            "compile_and_select_seconds": round(time.time() - started, 1),
+        }
+        try:
+            for arm, function in (("rule", rule), ("reverted", reverted)):
+                builder._restated_ledger_filter_refusal = function
+                result[arm] = [
+                    _survey(builder, specs, label) for label, specs in surfaces.items()
+                ]
+        finally:
+            builder._restated_ledger_filter_refusal = rule
+        results.append(result)
+        print(f"done {feed}", file=sys.stderr, flush=True)
+
+    head = subprocess.run(
+        ["git", "-C", str(ROOT), "rev-parse", "HEAD"],
+        capture_output=True,
+        text=True,
+        check=False,
+    ).stdout.strip()
+    status = subprocess.run(
+        ["git", "-C", str(ROOT), "status", "--porcelain"],
+        capture_output=True,
+        text=True,
+        check=False,
+    ).stdout.splitlines()
+    payload = {
+        "head": head,
+        "worktree_status": status,
+        "script_sha256": _sha256(__file__),
+        "target_period": acs.PERIOD,
+        "max_rss_bytes": resource.getrusage(resource.RUSAGE_SELF).ru_maxrss,
+        "results": results,
+    }
+    print(json.dumps(payload, indent=2))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main(sys.argv[1:]))