Campaign-UK staging producer: archive -> resolved claim-id joins + 14 reckoner attaches

scorecard_db/export_populations.py

@@ -2,13 +2,15 @@
 
 The Urban SotSN population reaches the app through the file-based
 pipeline/build_comparison.py export (data/comparison.json). Everything
-else — today, exactly the reform-validation registry (issue #20): its 205
-minted claims plus the 36 harvested JCX-35-25 provision claims its OBBBA
-results attach to — lives only in scorecard.db. This module exports every
-non-Urban claim that has at least one pe_result, carrying the dimension
-the Urban export doesn't have: the full per-release result history
-(one row per certified release, engine pins and OBBBA scoring mode in the
-construction), so cross-release drift is visible.
+else lives only in scorecard.db: the reform-validation registry (issue
+#20, its 205 minted claims plus the 36 harvested JCX-35-25 provision
+claims its OBBBA results attach to), the US campaign attaches, and —
+since the campaign-UK producer — the 14 uk_hmrc reckoner claims with
+campaign results. This module exports every non-Urban claim that has at
+least one pe_result, carrying the dimension the Urban export doesn't
+have: the full per-release result history (one row per certified
+release, engine pins and OBBBA scoring mode in the construction), so
+cross-release drift is visible.
 
 Doctrine (issues #1/#9): descriptive only. Statuses and calibration
 relationships are exported verbatim; ratios are raw pe/external with no
@@ -193,7 +195,8 @@ def export(
         "note": (
             "Non-Urban populations exported from scorecard.db: the populace"
             " reform-validation registry (issue #20) plus the compute"
-            " campaign's attached comparisons (TPC/CPSP/PWBM/CBO/JCT)."
+            " campaign's attached comparisons (US: TPC/CPSP/PWBM/CBO/JCT;"
+            " UK: the HMRC ready-reckoner family)."
             " Statuses and calibration relationships are verbatim; nothing"
             " here is a pass/fail grade."
         ),

scorecard_db/ingest_campaign.py

@@ -9,11 +9,15 @@
 family-vocabulary descriptor (translated below) or, for claims already in
 the DB (the Urban subgroup joins), the claim_id directly.
 
-US families attach here. The UK families (free_joins, hmrc_reckoner_t2,
-obr_measures, uprating_april2026, two_child) are vendored alongside but
-NOT ingested: their claims live in the UK harvest, which has no DB ingest
-yet — they attach when it lands, and this module fails loudly if pointed
-at them early.
+US families attach from sources/campaign-20260802/us. UK families
+attach from sources/campaign-20260802/uk_resolved — the DERIVED staging
+produce_campaign_uk builds from the frozen uk/ archive by resolving
+each row to a claim_id against the ingested UK claims (today:
+hmrc_reckoner_t2's 14 rows; the other archived families stay blocked
+with per-row reasons in that module until their target sources are
+staged). Pointing this module at the frozen uk/ archive directly still
+fails loudly (its descriptors under-specify by design — resolution is
+the producer's job).
 
 Match contract: descriptors were verified by the campaign against the
 harvest STAGING files; the DB's per-source adapters normalized vocabulary
@@ -263,4 +267,5 @@ def ingest(db_path: Path, staged_dir: Path | None = None) -> dict:
     import sys
 
     out = Path(sys.argv[1] if len(sys.argv) > 1 else "data/scorecard.db")
-    print(json.dumps(ingest(out), indent=1))
+    staged = Path(sys.argv[2]) if len(sys.argv) > 2 else None
+    print(json.dumps(ingest(out, staged), indent=1))

scorecard_db/produce_campaign_uk.py

@@ -0,0 +1,204 @@
+"""Resolve the archived UK campaign staging against the ingested UK claims.
+
+The 2026-08-02 compute campaign staged five UK families before any UK
+claims existed in the DB; the archive is frozen under
+sources/campaign-20260802/uk (PR #70). Measured against the ingested
+vocabulary (PR #48) the archived descriptors under-specify — the
+adjudication's finding: two personal-allowance reckoner rows share
+identical descriptor conditions, and the descriptor periods key the PE
+calendar year, not the claim convention (FY END year). This producer
+DERIVES a resolved staging deterministically instead of loosening the
+match contract:
+
+    archive row.pe_construction (its " | " prefix, an exact string)
+      -> uk_runs/<key>_2026.json pe_construction   (the executed run)
+      -> t2_collation.csv hmrc_hint            (HMRC's verbatim change
+                                                 label for that run)
+      -> exactly ONE ingested claim: source uk_hmrc, metric
+         revenue_change, conditions.option == hint, conditions.fy ==
+         the archived row's fy                            [fail-loud]
+      -> {"claim_id": ...} (ingest_campaign's strict direct form)
+
+Every step is a closed lookup that raises on 0 or 2+ — a drifted
+archive, collation, or claim re-ingest fails loudly, never mis-joins.
+
+Family disposition (this module's summary reports it; the per-row
+compositions below are pinned in tests so a re-frozen archive cannot
+silently outgrow its stated reason):
+    hmrc_reckoner_t2   RESOLVED (14 rows) -> uk_resolved/
+    free_joins         NOT RESOLVED (16 rows = 7 obr revenue_level +
+        7 uk_dwp benefit_cost + 2 metaless exhibits): the OBR rows
+        target EFO receipts forecast lines and the uk_dwp rows DWP
+        benefit-expenditure forecast lines — NEITHER is staged as
+        claims, and pe-uk-data consumes both publication families as
+        calibration surfaces, so staging them is relationship-evidence
+        work (two source lanes), never a quick join
+    obr_measures       NOT RESOLVED (10 rows = 9 obr revenue_change
+        fiscal-event costings + 1 metaless exhibit): targets the OBR
+        policy-measures costings database (long-tail source, held on
+        the DB-storage decision)
+    two_child          NOT RESOLVED (4 rows = 2 resolution_foundation
+        + 1 ukmod poverty_count_change + 1 metaless exhibit): RF is a
+        long-tail source (held), and the ukmod row targets a REFORM
+        claim — the staged UKMOD family is baseline validation
+        statistics only, so a 2CL-reform claim needs its own staging
+        decision
+    uprating_april2026 NOT RESOLVED (4 rows = 3 resolution_foundation
+        benefit_uprating_pct + 1 metaless exhibit): RF long-tail, held
+
+The metaless exhibits (5 rows across the families — the free_joins
+pair is one construction duplicated) carry exhibit_context but no
+exhibit_meta; ingest_campaign would only ever defer them, so they
+block with their families rather than shipping as noise.
+
+Usage:
+    PYTHONPATH=. python -m scorecard_db.produce_campaign_uk
+    PYTHONPATH=. python -m scorecard_db.ingest_campaign \
+        data/scorecard.db sources/campaign-20260802/uk_resolved
+"""
+
+from __future__ import annotations
+
+import csv
+import json
+from pathlib import Path
+
+from .db import ScorecardDB
+from .harvest import REPO
+
+CAMPAIGN = REPO / "sources" / "campaign-20260802"
+ARCHIVE = CAMPAIGN / "uk"
+RUNS = CAMPAIGN / "uk_runs"
+RESOLVED = CAMPAIGN / "uk_resolved"
+
+# Archived families this producer deliberately does NOT resolve, with
+# the reason a future lane must clear first. Every archive family must
+# appear either here or in the resolve/copy sets — a new family in the
+# archive fails loudly rather than being silently skipped.
+BLOCKED = {
+    "free_joins": (
+        "16 rows = 7 obr revenue_level (EFO receipts forecast lines) + "
+        "7 uk_dwp benefit_cost (DWP benefit-expenditure forecast "
+        "lines) + 2 metaless exhibits; neither publication family is "
+        "staged as claims, and pe-uk-data consumes both as calibration "
+        "surfaces — staging needs the relationship evidence read at "
+        "the pin (two source lanes)"
+    ),
+    "obr_measures": (
+        "10 rows = 9 obr revenue_change fiscal-event costings + 1 "
+        "metaless exhibit; targets the OBR policy-measures costings "
+        "database — long-tail source held on the DB-storage decision"
+    ),
+    "two_child": (
+        "4 rows = 2 resolution_foundation + 1 ukmod "
+        "poverty_count_change + 1 metaless exhibit; RF is long-tail "
+        "(held), and the ukmod row targets a REFORM claim — the staged "
+        "UKMOD family is baseline statistics only, so a 2CL-reform "
+        "claim needs its own staging decision"
+    ),
+    "uprating_april2026": (
+        "4 rows = 3 resolution_foundation benefit_uprating_pct + 1 "
+        "metaless exhibit; RF long-tail, held"
+    ),
+}
+RESOLVED_FAMILIES = {"hmrc_reckoner_t2"}
+
+
+def _runs_by_construction() -> dict[str, str]:
+    """pe_construction -> reform_key over the archived executed runs."""
+    out: dict[str, str] = {}
+    for f in sorted(RUNS.glob("*_2026.json")):
+        d = json.loads(f.read_text())
+        if isinstance(d, dict) and "pe_construction" in d and "reform_key" in d:
+            if d["pe_construction"] in out:
+                raise ValueError(
+                    f"two runs share a construction: {d['pe_construction']!r}"
+                )
+            out[d["pe_construction"]] = d["reform_key"]
+    if not out:
+        raise FileNotFoundError(f"no run files under {RUNS}")
+    return out
+
+
+def _hints_by_key() -> dict[str, str]:
+    """reform_key -> HMRC's verbatim change label (t2 collation)."""
+    with open(RUNS / "t2_collation.csv") as f:
+        rows = list(csv.DictReader(f))
+    out = {r["key"]: r["hmrc_hint"] for r in rows}
+    if len(out) != len(rows):
+        raise ValueError("t2_collation.csv has duplicate keys")
+    return out
+
+
+def _resolve_reckoner(db: ScorecardDB, rows: list[dict]) -> list[dict]:
+    constructions = _runs_by_construction()
+    hints = _hints_by_key()
+    resolved = []
+    for row in rows:
+        prefix = row["pe_construction"].split(" | ")[0]
+        if prefix not in constructions:
+            raise ValueError(
+                f"hmrc_reckoner_t2: construction {prefix!r} matches no archived run"
+            )
+        key = constructions[prefix]
+        if key not in hints:
+            raise ValueError(f"hmrc_reckoner_t2: run {key!r} not in t2 collation")
+        option = hints[key]
+        fy = row["external_claim_match"]["conditions"]["fy"]
+        hits = [
+            r[0]
+            for r in db.conn.execute(
+                "SELECT claim_id FROM external_scores"
+                " WHERE source='uk_hmrc' AND metric='revenue_change'"
+                " AND json_extract(conditions, '$.option') = ?"
+                " AND json_extract(conditions, '$.fy') = ?",
+                (option, fy),
+            )
+        ]
+        if len(hits) != 1:
+            raise ValueError(
+                f"hmrc_reckoner_t2: {len(hits)} claims for option "
+                f"{option!r} fy {fy!r} — need exactly one"
+            )
+        out = dict(row)
+        out["external_claim_match"] = {"claim_id": hits[0]}
+        resolved.append(out)
+    if len({r["external_claim_match"]["claim_id"] for r in resolved}) != len(resolved):
+        raise ValueError("hmrc_reckoner_t2: two rows resolved to one claim")
+    return resolved
+
+
+def produce(db_path: Path, out_dir: Path | None = None) -> dict:
+    """Write the resolved staging; returns the disposition summary."""
+    out_dir = out_dir or RESOLVED
+    db = ScorecardDB(db_path)
+    families = {p.stem: p for p in sorted(ARCHIVE.glob("*.jsonl"))}
+    unaccounted = (set(families) - set(BLOCKED) - RESOLVED_FAMILIES) | (
+        RESOLVED_FAMILIES - set(families)
+    )
+    if unaccounted:
+        raise ValueError(
+            f"archive families without a disposition: {sorted(unaccounted)}"
+        )
+    out_dir.mkdir(parents=True, exist_ok=True)
+    summary: dict = {"resolved": {}, "blocked": BLOCKED}
+    for name, path in families.items():
+        rows = [
+            json.loads(line) for line in path.read_text().splitlines() if line.strip()
+        ]
+        if name in BLOCKED:
+            continue
+        resolved = _resolve_reckoner(db, rows)
+        (out_dir / path.name).write_text(
+            "\n".join(json.dumps(r, sort_keys=True) for r in resolved) + "\n"
+        )
+        summary["resolved"][name] = len(resolved)
+    db.close()
+    return summary
+
+
+if __name__ == "__main__":
+    import sys
+
+    db = Path(sys.argv[1] if len(sys.argv) > 1 else "data/scorecard.db")
+    print(json.dumps(produce(db), indent=1))