From ad7f78e52d60133db63fd2e35127b9b16d145eb8 Mon Sep 17 00:00:00 2001 From: kert Date: Sat, 31 Oct 2026 20:47:15 -0400 Subject: [PATCH] feat(lead): claims, GRADE start, and the precursor catalog --- src/lead/ledger/claims.py | 148 +++++++++++++++++++++++++++++++ tests/lead/test_ledger_claims.py | 109 +++++++++++++++++++++++ 2 files changed, 257 insertions(+) create mode 100644 src/lead/ledger/claims.py create mode 100644 tests/lead/test_ledger_claims.py diff --git a/src/lead/ledger/claims.py b/src/lead/ledger/claims.py new file mode 100644 index 0000000..d5df5e3 --- /dev/null +++ b/src/lead/ledger/claims.py @@ -0,0 +1,148 @@ +"""Turn an extraction into claims, grade them, and list the precursor catalog. + +Cohort, case-control, and quasi-experimental designs start at high. Every +other design starts at low. Each downgrade steps high, moderate, low, very +low, and stops at very low. A claim's content hash is the extraction, so +recording the same extraction again inserts nothing. +""" + +from __future__ import annotations + +import hashlib +from collections.abc import Mapping, Sequence +from typing import TYPE_CHECKING + +from lead.targets import by_id + +if TYPE_CHECKING: + import duckdb + +_LEVELS = ("high", "moderate", "low", "very low") +_HIGH_DESIGNS = frozenset({"cohort", "case-control", "quasi-experimental"}) +_KIND = { + "lead-precursors": "precursor", + "lead-limits": "limit", + "lead-history": "lineage", + "lead-scan": "context", + "lead-external": "context", +} +_DOWNGRADE_FIELDS = ( + "risk_of_bias", + "inconsistency", + "indirectness", + "imprecision", + "publication_bias", +) +_DOWNGRADE_LABELS = { + "risk_of_bias": "risk of bias", + "publication_bias": "publication bias", +} + + +def grade(design: str, downgrades: Sequence[str]) -> str: + """GRADE start for a design, then one step per downgrade.""" + start = 0 if design.casefold() in _HIGH_DESIGNS else 2 + return _LEVELS[min(start + len(downgrades), len(_LEVELS) - 1)] + + +def may_pool(measures: Sequence[str]) -> bool: + """True when at least three studies share one effect measure.""" + return len(measures) >= 3 and len(set(measures)) == 1 + + +def _target_id(project: str, extraction: Mapping[str, str]) -> str: + if project == "lead-precursors": + return extraction.get("outcome_target_id") or extraction.get("target_id") or "" + return extraction.get("target_id") or extraction.get("outcome_target_id") or "" + + +def _downgrades(extraction: Mapping[str, str]) -> tuple[str, ...]: + labels: list[str] = [] + for field in _DOWNGRADE_FIELDS: + if extraction.get(field, "").strip(): + labels.append(_DOWNGRADE_LABELS.get(field, field)) + return tuple(labels) + + +def _hash(project: str, kind: str, extraction: Mapping[str, str]) -> str: + parts = [project, kind, *[f"{key}={extraction[key]}" for key in sorted(extraction)]] + return hashlib.sha256("\n".join(parts).encode()).hexdigest() + + +def claims_from_extraction( + extraction: Mapping[str, str], *, project: str, actor: str +) -> tuple[Mapping[str, object], ...]: + """One claim for this extraction. An unknown target raises ``KeyError``.""" + target_id = _target_id(project, extraction) + by_id(target_id) + kind = _KIND.get(project, "context") + content_hash = _hash(project, kind, extraction) + claim: dict[str, object] = { + "id": content_hash, + "content_hash": content_hash, + "actor": actor, + "project": project, + "kind": kind, + "target_id": target_id, + "tiers": "literature", + "certainty": grade(extraction.get("design", ""), _downgrades(extraction)), + } + for column in ( + "grain", + "lead_time", + "design", + "direction", + "effect_measure", + "population", + "setting", + "risk_of_bias", + "subject", + ): + value = extraction.get(column, "").strip() + if value: + claim[column] = value + if extraction.get("zotero_key") or extraction.get("extraction_version"): + claim["extraction_ids"] = " ".join( + part + for part in ( + extraction.get("zotero_key", ""), + extraction.get("extraction_version", ""), + ) + if part + ) + return (claim,) + + +def precursor_catalog(con: duckdb.DuckDBPyConnection) -> list[Mapping[str, object]]: + """One row per target: precursor lead time, grain, certainty, and any limit.""" + rows = con.execute( + """ + SELECT id, kind, target_id, lead_time, grain, certainty + FROM lead_evidence.claim + ORDER BY target_id, id + """ + ).fetchall() + grouped: dict[str, dict[str, object]] = {} + for claim_id, kind, target_id, lead_time, grain, certainty in rows: + entry = grouped.setdefault( + target_id, + { + "target_id": target_id, + "lead_time": "", + "grain": "", + "certainty": "", + "limit_claim_id": None, + "_precursor": False, + }, + ) + if kind == "precursor" and not entry["_precursor"]: + entry["lead_time"] = lead_time or "" + entry["grain"] = grain or "" + entry["certainty"] = certainty or "" + entry["_precursor"] = True + if kind == "limit" and entry["limit_claim_id"] is None: + entry["limit_claim_id"] = claim_id + return [ + {key: value for key, value in entry.items() if key != "_precursor"} + for entry in grouped.values() + ] diff --git a/tests/lead/test_ledger_claims.py b/tests/lead/test_ledger_claims.py new file mode 100644 index 0000000..aecbb42 --- /dev/null +++ b/tests/lead/test_ledger_claims.py @@ -0,0 +1,109 @@ +"""Claims from extractions, GRADE start, the pool gate, and the catalog.""" + +from __future__ import annotations + +import pytest + +from lead.ledger.claims import ( + claims_from_extraction, + grade, + may_pool, + precursor_catalog, +) +from lead.ledger.store import append, open_ledger + + +def test_precursor_and_limit_extractions_become_claims(): + precursor = claims_from_extraction( + { + "zotero_key": "ABCD2345", + "extraction_version": "v1", + "outcome_target_id": "Q.ACR", + "grain": "beneficiary", + "lead_time": "12 months", + "design": "cohort", + }, + project="lead-precursors", + actor="tester", + ) + assert len(precursor) == 1 + claim = precursor[0] + assert claim["kind"] == "precursor" + assert claim["tiers"] == "literature" + assert claim["grain"] == "beneficiary" + assert claim["target_id"] == "Q.ACR" + assert claim["lead_time"] == "12 months" + assert claim["certainty"] == "high" + + limit = claims_from_extraction( + {"target_id": "Q.ACR", "grain": "ACO"}, + project="lead-limits", + actor="tester", + ) + assert len(limit) == 1 + assert limit[0]["kind"] == "limit" + assert limit[0]["target_id"] == "Q.ACR" + assert limit[0]["grain"] == "ACO" + + +def test_grade_starts_high_for_three_designs_and_stops_at_very_low(): + assert grade("cohort", ()) == "high" + assert grade("case-control", ()) == "high" + assert grade("quasi-experimental", ()) == "high" + assert grade("cohort", ("risk of bias",)) == "moderate" + assert grade("cross-sectional", ()) == "low" + assert grade("cohort", ("a", "b", "c", "d")) == "very low" + assert grade("cohort", ("a", "b", "c", "d", "e")) == "very low" + + +def test_may_pool_needs_three_identical_measures(): + assert may_pool(["OR", "OR", "OR"]) is True + assert may_pool(["OR", "OR"]) is False + assert may_pool(["OR", "OR", "RR"]) is False + + +def test_same_extraction_inserts_one_claim_and_the_catalog_reads_it(tmp_path): + extraction = { + "zotero_key": "ABCD2345", + "extraction_version": "v1", + "outcome_target_id": "Q.ACR", + "grain": "beneficiary", + "lead_time": "12 months", + "design": "cohort", + } + first = claims_from_extraction( + extraction, project="lead-precursors", actor="tester" + )[0] + second = claims_from_extraction( + extraction, project="lead-precursors", actor="tester" + )[0] + limit = claims_from_extraction( + {"target_id": "Q.ACR", "grain": "ACO", "zotero_key": "EFGH5678"}, + project="lead-limits", + actor="tester", + )[0] + con = open_ledger(str(tmp_path / "ledger.duckdb")) + try: + assert append(con, "claim", first) is True + assert append(con, "claim", second) is False + assert ( + con.execute("select count(*) from lead_evidence.claim").fetchone()[0] == 1 + ) + assert append(con, "claim", limit) is True + catalog = precursor_catalog(con) + row = next(item for item in catalog if item["target_id"] == "Q.ACR") + assert row["lead_time"] == first["lead_time"] + assert row["certainty"] == first["certainty"] + assert row["grain"] == first["grain"] + assert row["limit_claim_id"] == limit["content_hash"] + finally: + con.close() + + +def test_unknown_target_raises_key_error(): + with pytest.raises(KeyError): + claims_from_extraction( + {"outcome_target_id": "nope", "grain": "beneficiary"}, + project="lead-precursors", + actor="tester", + )