| """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() |
|
|
