02 Screening

This notebook reads the title and abstract decisions and the full-text decisions exported from the screening tool, checks them against the deduplicated records, and counts the sources at each stage. It adds the screening, included, and by_block sections to data/processed/screening_counts.yaml, and writes the included reports to data/processed/included_reports.csv. 04 Synthesis draws the flow diagram from that file.

The export formats are in the screening guide. A record is a title and abstract. A report is a full-text document. A source of evidence can have several reports.

The checks come first. If any check finds an error, the notebook stops and writes nothing, so the counts never rest on data that do not add up. A warning is printed and does not stop the run. When reviewers export their own decisions as well, the notebook reports their agreement.

import re
from pathlib import Path

import pandas as pd
import yaml

ROOT = next(p for p in [Path.cwd(), *Path.cwd().parents] if (p / "_quarto.yml").exists())
PROC = ROOT / "data" / "processed"
EXP = ROOT / "data" / "raw" / "screening_exports"
dedup_path = PROC / "deduplicated_records.csv"
ta_path = EXP / "title_abstract.csv"
ft_path = EXP / "full_text.csv"
counts_path = PROC / "screening_counts.yaml"
REASONS = [r["code"] for r in yaml.safe_load((ROOT / "codebook.yml").read_text(encoding="utf-8"))["exclusion_reasons"]]

missing = [str(p.relative_to(ROOT)) for p in (dedup_path, ta_path) if not p.exists()]
have_data = not missing
if not have_data:
    print("Not run yet. Missing: " + ", ".join(missing))


def read(path, required, optional=()):
    df = pd.read_csv(path, dtype=str).fillna("")
    df.columns = [c.strip().lower() for c in df.columns]
    absent = [c for c in required if c not in df.columns]
    if absent:
        raise ValueError(f"{path.name} is missing column(s): {', '.join(absent)}")
    for c in optional:
        if c not in df.columns:
            df[c] = ""
    return df.apply(lambda s: s.str.strip())


def sample(values, n=5):
    values = sorted(set(values))
    return ", ".join(values[:n]) + (f" and {len(values) - n} more" if len(values) > n else "")

Not run yet. Missing: data/processed/deduplicated_records.csv, data/raw/screening_exports/title_abstract.csv

if have_data:
    dedup = read(dedup_path, ["record_id"], ["blocks"])
    ta = read(ta_path, ["record_id", "decision"])
    ta["decision"] = ta["decision"].str.lower()
    ft_cols = ["record_id", "source_id", "retrieved", "decision", "reason"]
    if ft_path.exists():
        ft = read(ft_path, ["record_id", "retrieved", "decision", "reason"], ["source_id"])
    else:
        ft = pd.DataFrame({c: pd.Series(dtype=str) for c in ft_cols})
    for col in ["retrieved", "decision", "reason"]:
        ft[col] = ft[col].str.lower()
    ft["source_id"] = ft["source_id"].where(ft["source_id"] != "", ft["record_id"])

    # Optional files with each reviewer's own decisions, for example title_abstract_r1.csv
    reviewers = {"title_abstract": {}, "full_text": {}}
    for p in sorted(EXP.glob("*.csv")):
        m = re.fullmatch(r"(title_abstract|full_text)_(.+)", p.stem)
        if m:
            r = read(p, ["record_id", "decision"])
            r["decision"] = r["decision"].str.lower()
            reviewers[m[1]][m[2]] = r[["record_id", "decision"]]
    print(f"{len(dedup)} deduplicated record(s), {len(ta)} title/abstract decision(s), {len(ft)} full-text row(s).")
    for stage, found in reviewers.items():
        if found:
            print(f"Reviewer files for {stage}: {', '.join(found)}")

Checks

if have_data:
    errors, warnings = [], []
    dedup_ids = set(dedup["record_id"])
    ok = ["include", "exclude"]

    # Title and abstract
    if ta["record_id"].duplicated().any():
        errors.append(f"title_abstract.csv has record_id values more than once: {sample(ta.loc[ta['record_id'].duplicated(), 'record_id'])}.")
    unknown = set(ta["record_id"]) - dedup_ids
    if unknown:
        errors.append(f"title_abstract.csv has {len(unknown)} record_id value(s) that are not in deduplicated_records.csv: {sample(unknown)}.")
    bad = ta[~ta["decision"].isin(ok)]
    if len(bad):
        errors.append(f"title_abstract.csv has {len(bad)} row(s) with a decision other than include or exclude.")
    unscreened = dedup_ids - set(ta["record_id"])
    if unscreened:
        errors.append(f"{len(unscreened)} deduplicated record(s) have no title/abstract decision: {sample(unscreened)}.")
    sought_ids = set(ta.loc[ta["decision"] == "include", "record_id"])

    # Full text
    if ft["record_id"].duplicated().any():
        errors.append(f"full_text.csv has record_id values more than once: {sample(ft.loc[ft['record_id'].duplicated(), 'record_id'])}.")
    stray = set(ft["record_id"]) - sought_ids
    if stray:
        errors.append(f"full_text.csv has {len(stray)} record_id value(s) that were not included at title and abstract: {sample(stray)}.")
    bad = ft[~ft["retrieved"].isin(["yes", "no"])]
    if len(bad):
        errors.append(f"full_text.csv has {len(bad)} row(s) where retrieved is not yes or no.")
    bad = ft[(ft["retrieved"] == "yes") & ~ft["decision"].isin(ok)]
    if len(bad):
        errors.append(f"full_text.csv has {len(bad)} retrieved report(s) without an include or exclude decision.")
    bad = ft[(ft["retrieved"] == "no") & (ft["decision"] != "")]
    if len(bad):
        errors.append(f"full_text.csv has {len(bad)} report(s) that were not retrieved but have a decision.")
    excluded = ft[(ft["retrieved"] == "yes") & (ft["decision"] == "exclude")]
    if (excluded["reason"] == "").any():
        errors.append(f"full_text.csv has {int((excluded['reason'] == '').sum())} excluded report(s) without a reason.")
    odd = set(excluded["reason"]) - set(REASONS) - {""}
    if odd:
        errors.append(f"full_text.csv has exclusion reasons that are not in codebook.yml: {sample(odd)}.")
    pending = sought_ids - set(ft["record_id"])
    if pending:
        errors.append(f"{len(pending)} record(s) included at title and abstract have no full-text row: {sample(pending)}.")

    # Reviewer files
    for stage, found in reviewers.items():
        for name, r in found.items():
            label = f"{stage}_{name}.csv"
            if r["record_id"].duplicated().any():
                errors.append(f"{label} has record_id values more than once.")
            unknown = set(r["record_id"]) - dedup_ids
            if unknown:
                errors.append(f"{label} has {len(unknown)} record_id value(s) that are not in deduplicated_records.csv: {sample(unknown)}.")
            bad = r[~r["decision"].isin(ok)]
            if len(bad):
                errors.append(f"{label} has {len(bad)} row(s) with a decision other than include or exclude.")

    for w in warnings:
        print("Warning:", w)
    if errors:
        raise ValueError("Fix these problems in the screening exports, then run this notebook again:\n- " + "\n- ".join(errors))

if have_data:
    screened = len(ta)
    ta_excluded = int((ta["decision"] == "exclude").sum())
    sought = len(sought_ids)
    not_retrieved = int((ft["retrieved"] == "no").sum())
    ft_ok = ft[ft["retrieved"] == "yes"]
    assessed = len(ft_ok)
    ft_excl = ft_ok[ft_ok["decision"] == "exclude"]
    ft_incl = ft_ok[ft_ok["decision"] == "include"]
    reasons = ft_excl["reason"].value_counts()
    included_reports = len(ft_incl)
    included_sources = int(ft_incl["source_id"].nunique())

    # These hold by construction once the checks pass. They guard against a later edit breaking that.
    assert screened == len(dedup), "Every deduplicated record must be screened."
    assert screened == ta_excluded + sought
    assert sought == not_retrieved + assessed
    assert assessed == len(ft_excl) + included_reports
    assert included_sources <= included_reports
    print("No problems found.")

Agreement between reviewers

With two or more reviewer files for a stage, each pair of reviewers is compared on the records they both decided. Cohen’s kappa corrects the percent agreement for the agreement expected by chance. A reviewer file may cover only a sample, such as a calibration set. The final decisions in title_abstract.csv and full_text.csv are what the counts use.

def kappa(a, b):
    po = float((a == b).mean())
    pe = sum(float((a == label).mean() * (b == label).mean()) for label in ("include", "exclude"))
    return float("nan") if pe == 1 else (po - pe) / (1 - pe)


if have_data:
    final = {"title_abstract": ta.set_index("record_id")["decision"], "full_text": ft[ft["decision"] != ""].set_index("record_id")["decision"]}
    agreement = []
    for stage, found in reviewers.items():
        names = sorted(found)
        for k, a in enumerate(names):
            for b in names[k + 1:]:
                both = found[a].merge(found[b], on="record_id", suffixes=("_a", "_b"))
                if both.empty:
                    continue
                agreement.append({
                    "stage": stage, "reviewer_a": a, "reviewer_b": b,
                    "n_compared": len(both),
                    "percent_agreement": round(100 * float((both["decision_a"] == both["decision_b"]).mean()), 1),
                    "kappa": round(kappa(both["decision_a"], both["decision_b"]), 3),
                })
        if names:
            wide = pd.concat({n: found[n].set_index("record_id")["decision"] for n in names}, axis=1)
            disagree = wide[wide.nunique(axis=1) > 1]
            split = disagree.assign(final=final[stage].reindex(disagree.index))
            same = wide[(wide.nunique(axis=1) == 1) & wide.notna().all(axis=1)]
            overridden = int((final[stage].reindex(same.index) != same.iloc[:, 0]).sum()) if len(same) else 0
            print(f"{stage}: {len(split)} record(s) with a disagreement. {len(split[split['final'].isna()])} of them have no final decision.")
            if overridden:
                print(f"Warning: {overridden} record(s) have a final decision that differs from the decision all reviewers agreed on.")
            if len(split):
                display(split.head(50))
    if agreement:
        display(pd.DataFrame(agreement))
    else:
        print("No pair of reviewer files to compare.")

Counts

if have_data:
    summary = pd.DataFrame(
        [
            ("Records screened", screened),
            ("Records excluded at title and abstract", ta_excluded),
            ("Reports sought for retrieval", sought),
            ("Reports not retrieved", not_retrieved),
            ("Reports assessed for eligibility", assessed),
            ("Reports excluded at full text", len(ft_excl)),
            ("Reports included", included_reports),
            ("Sources of evidence included", included_sources),
        ],
        columns=["Stage", "n"],
    )
    display(summary)
    if len(reasons):
        display(reasons.rename_axis("Exclusion reason").reset_index(name="n"))

if have_data:
    block_ids = {}
    for rid, blocks in zip(dedup["record_id"], dedup["blocks"]):
        for b in [x.strip() for x in blocks.split(";") if x.strip()]:
            block_ids.setdefault(b, set()).add(rid)
    ta_excl_ids = ta.loc[ta["decision"] == "exclude", "record_id"]
    by_block = {}
    for b, ids in sorted(block_ids.items()):
        by_block[b] = {
            "records_identified": len(ids),
            "records_screened": int(ta["record_id"].isin(ids).sum()),
            "records_excluded": int(ta_excl_ids.isin(ids).sum()),
            "reports_assessed": int(ft_ok["record_id"].isin(ids).sum()),
            "reports_excluded": int(ft_excl["record_id"].isin(ids).sum()),
            "reports_included": int(ft_incl["record_id"].isin(ids).sum()),
        }

    counts = (yaml.safe_load(counts_path.read_text()) or {}) if counts_path.exists() else {}
    counts["screening"] = {
        "records_screened": screened,
        "records_excluded": ta_excluded,
        "reports_sought": sought,
        "reports_not_retrieved": not_retrieved,
        "reports_assessed": assessed,
        "reports_excluded": [{"reason": r, "n": int(n)} for r, n in reasons.items()],
    }
    counts["included"] = {"sources": included_sources, "reports": included_reports}
    counts["by_block"] = by_block
    if agreement:
        counts["agreement"] = agreement
    else:
        counts.pop("agreement", None)
    counts_path.write_text(yaml.safe_dump(counts, sort_keys=False))
    ft_incl[["source_id", "record_id"]].sort_values(["source_id", "record_id"]).to_csv(PROC / "included_reports.csv", index=False)
    print("Wrote data/processed/screening_counts.yaml and included_reports.csv")
    if by_block:
        display(pd.DataFrame(by_block).T.rename_axis("block"))