"""Recompute registered paired statistics from the distributed reference scores.""" from pathlib import Path import argparse,contextlib,hashlib,io,json,tempfile,time import numpy as np import pandas as pd import torch from reproduce import ROOT,load,verify from finite_stats import sign_flip_pvalues from unionmind.decision_sufficiency.analysis import paired_interval from unionmind.continuous_decision.factor_study import MODELS def controlled(path): cfg=json.loads((ROOT/'protocols/decision_sufficiency_protocol_01.json').read_text()) frame=pd.read_csv(path) primary=frame[(frame.model==cfg['primary']['model'])&(frame.query_budget>0)] values=primary.groupby(['world','family','method']).utility.mean().unstack('method') expected=json.loads((ROOT/'results/controlled/decision.json').read_text()) rows=[] for control in cfg['primary']['compact_controls']+['identity']: delta=(values['query_joint']-values[control]).to_numpy() ci=paired_interval(delta,values.index.get_level_values('family').to_numpy(), cfg['primary']['bootstrap_seed'],cfg['primary']['bootstrap_draws'], .05/len(cfg['primary']['compact_controls']) if control!='identity' else .05) old=next(r for r in expected['compact_comparisons'] if r['control']==control) assert abs(delta.mean()-old['mean_gain'])<1e-10 assert np.max(np.abs(np.array(ci)-old['paired_interval']))<1e-10 rows.append(dict(control=control,gain=float(delta.mean()),interval=ci)) return dict(worlds=len(values),comparisons=rows) def hydraulic(device): records=load('results/hydraulic/predictions',device)['results'] old=json.loads((ROOT/'provenance/hydraulic/opened_hydraulic_summary.json').read_text()) rng=torch.Generator(device=device).manual_seed(20931);output=[] for n in [64,128]: rows=[r for r in records if r['samples']==n];mask=rows[0]['unseen'] groups,inverse=torch.unique(rows[0]['groups'][mask],return_inverse=True) counts=torch.bincount(inverse).double();values=counts.new_zeros((len(groups),len(MODELS))) for i,name in enumerate(MODELS): score=torch.stack([r['metrics']['top3'] for r in rows if r['model']==name]).mean(0) values[:,i].scatter_add_(0,inverse,score[mask]) sample=torch.randint(len(groups),(20000,len(groups)),device=device,generator=rng) weights=counts.new_zeros((20000,len(groups))).scatter_add_(1,sample,torch.ones_like(sample,dtype=counts.dtype)) boot=weights@values/(weights@counts)[:,None];point=values.sum(0)/counts.sum() for candidate in ['additive_full','factorized_neural']: ci=MODELS.index(candidate) for j,name in enumerate(MODELS): if ci==j:continue interval=torch.quantile(boot[:,ci]-boot[:,j],boot.new_tensor([.05/80,1-.05/80])).tolist() row=dict(samples=n,candidate=candidate,control=name,gain=float(point[ci]-point[j]),simultaneous_interval=interval) reference=next(r for r in old['comparisons'] if all(r[k]==row[k] for k in ['samples','candidate','control'])) assert abs(row['gain']-reference['gain'])<1e-10 assert np.max(np.abs(np.array(interval)-reference['simultaneous_interval']))<1e-9 output.append(row) return dict(exploratory=True,comparisons=output) def temporal(device): records=load('results/temperature/predictions',device)['results'] old=json.loads((ROOT/'provenance/temperature/summary.json').read_text()) rng=torch.Generator(device=device).manual_seed(20891);output=[] for n in [64,128]: rows=[r for r in records if r['samples']==n];mask=rows[0]['unseen'];day=rows[0]['days'][mask] matrix=torch.stack([torch.stack([r['metrics']['top3'] for r in rows if r['model']==name]).mean(0)[mask] for name in MODELS],-1) points=[];boots=[] for d in [9,10,11,12]: x=matrix[day==d];points.append(x.mean(0)) indices=torch.randint(len(x),(20000,len(x)),device=device,generator=rng) boots.append(x[indices].mean(1)) days=torch.stack(points);point=days.mean(0);boot=torch.stack(boots).mean(0) sampled=torch.randint(4,(20000,4),device=device,generator=rng);day_boot=days[sampled].mean(1) for candidate in ['additive_full','factorized_neural']: ci=MODELS.index(candidate) for j,name in enumerate(MODELS): if ci==j:continue interval=torch.quantile(boot[:,ci]-boot[:,j],boot.new_tensor([.05/80,1-.05/80])).tolist() cluster=torch.quantile(day_boot[:,ci]-day_boot[:,j],boot.new_tensor([.05/80,1-.05/80])).tolist() delta=days[:,ci]-days[:,j];p1,p2=sign_flip_pvalues(delta) row=dict(samples=n,candidate=candidate,control=name,gain=float(point[ci]-point[j]), fixed_day_simultaneous_interval=interval,day_cluster_simultaneous_interval=cluster, day_gains=delta.tolist(),exact_day_p_one_sided=float(p1),exact_day_p_two_sided=float(p2)) reference=next(r for r in old['comparisons'] if all(r[k]==row[k] for k in ['samples','candidate','control'])) assert abs(row['gain']-reference['gain'])<1e-10 assert np.max(np.abs(np.array(interval)-reference['fixed_day_simultaneous_interval']))<1e-9 output.append(row) return dict(secondary=True,complete_source_cohort=False,target_days=4,comparisons=output) def external(device): from unionmind.external_sufficiency import data,study,analysis with tempfile.TemporaryDirectory(prefix='external-statistics-') as tmp: out=Path(tmp);(out/'evaluation').mkdir() (out/'evaluation/metrics.csv.gz').symlink_to(ROOT/'results/external/all_metrics.csv.gz') cfg=json.loads((ROOT/'protocols/external_sufficiency_protocol_01.json').read_text());cfg['device']=device protocol=out/'protocol.json';protocol.write_text(json.dumps(cfg)) for module in [data,study,analysis]:module.ROOT=out;module.PROTOCOL=protocol paths=[protocol,ROOT/'results/external/all_metrics.csv.gz'] data.write(out/'preconfirmation_lock.json',dict(purpose='Reproduction of statistics, not a new confirmation lock', files={str(p):hashlib.sha256(p.read_bytes()).hexdigest() for p in paths})) with contextlib.redirect_stdout(io.StringIO()):analysis.run() report=json.loads((out/'analysis/paired_comparisons.json').read_text()) expected=json.loads((ROOT/'results/external/paired_comparisons.json').read_text()) for a,b in zip(report['comparisons'],expected['comparisons']): assert a['model']==b['model'] and a['representation']==b['representation'] assert abs(a['gain']-b['gain'])<1e-10 assert np.max(np.abs(np.array(a['simultaneous_95_interval'])-b['simultaneous_95_interval']))<1e-9 return report def main(): parser=argparse.ArgumentParser(description=__doc__);parser.add_argument('--device',default='cuda:0') parser.add_argument('--output',default='statistics.json');a=parser.parse_args() assert a.device.startswith('cuda') and torch.cuda.is_available() torch.cuda.set_device(a.device);torch.backends.cuda.matmul.allow_tf32=False started=time.perf_counter();verify() report=dict(controlled=controlled(ROOT/'results/controlled/primary_predictions.csv.gz'), hydraulic=hydraulic(a.device),temporal=temporal(a.device),external=external(a.device)) torch.cuda.synchronize();report['seconds']=time.perf_counter()-started report['all_archived_primary_intervals_reproduced']=True Path(a.output).write_text(json.dumps(report,indent=2)+'\n') print(json.dumps(dict(all_archived_primary_intervals_reproduced=True,seconds=report['seconds']),indent=2)) if __name__=='__main__':main()