claim5_authors.py

"""Claim 5 by independent reanalysis of the AUTHORS' released raw outputs.

The paper links https://github.com/AngelReyero/loss_based_KO, which ships the
per-seed p-value tables behind Figures 4 and 5:

    results/res_csv/p_values_<setting>_<model>_seed<k>.csv

Each row is one method; columns are `tr_V{j}` (1.0 = truly non-null, 0.0 = null)
and `pval{j}` for j = 0..49. So type-I error and power can be recomputed from
the raw p-values without re-running anything, which is the cleanest way to test
"Semi-knockoffs maintains type-I error control while achieving higher power than
HRT".

Semi-knockoffs appears as `S-CPI_Wilcox` (Algorithm 1, Wilcoxon) and
`S-CPI_ST` (sign test); the baseline is `HRT`.
"""
import csv
import io
import json
import os
import subprocess

import numpy as np

RAW = ("https://raw.githubusercontent.com/AngelReyero/loss_based_KO/master/"
       "results/res_csv/{name}")
ALPHA = 0.05
# NAMING, resolved from the released tables: the paper's Semi-knockoffs is the
# *knockoff* CPI variant, i.e. CPI_KO_Wilcox (Algorithm 1, Wilcoxon) and
# CPI_KO_ST (sign test).  The S-CPI_* rows are the SPLIT variants the paper
# contrasts against, not the proposed method — reading those as "Semi-knockoffs"
# inverts the comparison.
METHODS = ["CPI_KO_Wilcox", "CPI_KO_ST", "HRT", "dCRT", "CPI", "LOCO",
           "S-CPI_Wilcox"]
CACHE = "authors_csv"
os.makedirs(CACHE, exist_ok=True)


def fetch(name):
    p = os.path.join(CACHE, name)
    if os.path.exists(p) and os.path.getsize(p) > 0:
        return open(p).read()
    # urllib fails with CERTIFICATE_VERIFY_FAILED on this machine; curl works.
    r = subprocess.run(["curl", "-sL", "--max-time", "45", "-o", p,
                        RAW.format(name=name)], capture_output=True)
    if r.returncode != 0 or not os.path.exists(p) or os.path.getsize(p) == 0:
        return None
    txt = open(p).read()
    if txt.lstrip().startswith("404") or "Not Found" in txt[:80]:
        os.remove(p)
        return None
    return txt


def analyse(setting, model, seeds):
    acc = {m: {"rej_null": 0, "n_null": 0, "rej_alt": 0, "n_alt": 0}
           for m in METHODS}
    got = 0
    for s in seeds:
        txt = fetch(f"p_values_{setting}_{model}_seed{s}.csv")
        if txt is None:
            continue
        got += 1
        for row in csv.DictReader(io.StringIO(txt)):
            m = row["method"]
            if m not in acc:
                continue
            for j in range(50):
                tv = row.get(f"tr_V{j}")
                pv = row.get(f"pval{j}")
                if tv is None or pv in (None, ""):
                    continue
                try:
                    truth = float(tv); p = float(pv)
                except ValueError:
                    continue
                if truth > 0.5:
                    acc[m]["n_alt"] += 1
                    acc[m]["rej_alt"] += (p <= ALPHA)
                else:
                    acc[m]["n_null"] += 1
                    acc[m]["rej_null"] += (p <= ALPHA)
    out = {}
    for m, a in acc.items():
        if a["n_null"] == 0:
            continue
        out[m] = {
            "type_I": a["rej_null"] / a["n_null"],
            "power": a["rej_alt"] / max(a["n_alt"], 1),
            "n_null_tests": a["n_null"], "n_alt_tests": a["n_alt"],
        }
    return out, got


def main():
    seeds = list(range(1, 111))
    res = {"alpha": ALPHA, "source": "AngelReyero/loss_based_KO @ master",
           "settings": {}}
    for setting in ("adjacent", "spaced"):
        for model in ("GB", "RF", "NN"):
            r, got = analyse(setting, model, seeds)
            if not r:
                print(f"  {setting}/{model}: no data"); continue
            res["settings"][f"{setting}_{model}"] = {"seeds_found": got, "methods": r}
            sko = r.get("CPI_KO_Wilcox"); hrt = r.get("HRT")
            line = f"  {setting:<9}/{model:<3} seeds={got:<4}"
            if sko and hrt:
                line += (f" SKO(CPI_KO_Wcx) power {sko['power']:.3f} (t1 {sko['type_I']:.3f}) | "
                         f"HRT power {hrt['power']:.3f} (t1 {hrt['type_I']:.3f}) | "
                         f"gap {sko['power']-hrt['power']:+.3f}")
            print(line, flush=True)
    json.dump(res, open("outputs/claim5_authors.json", "w"), indent=2)
    print("\nsaved outputs/claim5_authors.json")


if __name__ == "__main__":
    main()