"""Compute independent signals and delayed-label coverage from authored fixtures.""" import math from predictor import CONTRACT,baseline def integer(v): if type(v) is not int or v<0:raise ValueError('nonnegative integer clock required') return v def unique(rows): if not isinstance(rows,list) or any(not isinstance(r,dict) or not isinstance(r.get('id'),str) or not 1<=len(r['id'])<=80 for r in rows): raise ValueError('records need bounded string identities') ids=[r['id'] for r in rows] if len(ids)!=len(set(ids)):raise ValueError('duplicate event identity') def signals(events,reference,now,max_age=15,cutpoint=12500): integer(now);integer(max_age);unique(events) if not reference or any(type(v) not in (int,float) or not math.isfinite(v) for v in reference): raise ValueError('finite nonempty reference required') counts={'schema_invalid':0,'unavailable':0,'stale':0,'fresh':0};areas=[] for event in events: try: if set(event)!={'id','feature_at','available_at','contract','features'}:raise ValueError('schema') feature_at=integer(event['feature_at']);available=integer(event['available_at']) if availablenow or available>now:counts['unavailable']+=1 elif now-feature_at>max_age:counts['stale']+=1 else:counts['fresh']+=1;areas.append(values['area']) reference_share=sum(v<=cutpoint for v in reference)/len(reference) distance=abs(sum(v<=cutpoint for v in areas)/len(areas)-reference_share) if len(areas)>=4 else None return {'events':len(events),'counts':counts,'distribution':{'feature':'area','cutpoint':cutpoint, 'reference_rows':len(reference),'eligible_rows':len(areas),'minimum_rows':4, 'reference_low_share':reference_share,'absolute_share_difference':distance, 'threshold':0.5,'alert':distance>=0.5 if distance is not None else None}} def outcomes(predictions,labels,as_of): integer(as_of);unique(predictions);unique(labels) expected={p['id']:p for p in predictions};by_id={r['id']:r for r in labels} if not set(by_id)<=set(expected):raise ValueError('unknown label identity') for p in predictions: if set(p)!={'id','predicted_at','model','contract','label'} or not isinstance(p['model'],str) or not p['model']:raise ValueError('prediction schema') integer(p['predicted_at']) if p['label'] not in baseline.LABELS or p['contract']!=CONTRACT:raise ValueError('prediction contract') if len({p['model'] for p in predictions})>1:raise ValueError('select one model cohort before joining outcomes') for row in labels: if set(row)!={'id','ready_at','label'}:raise ValueError('label schema') ready=integer(row['ready_at']);p=expected[row['id']] if row['label'] not in baseline.LABELS or readyas_of for p in eligible), 'missing':sum(p['id'] not in by_id for p in eligible),'correct':correct, 'coverage':len(ready)/len(eligible) if eligible else None, 'accuracy_on_ready':correct/len(ready) if ready else None, 'ready_prediction_ids':[p['id'] for p in ready]} def action(signal,quality,review): # This explicit teaching policy selects a review action; it never starts training. if any(signal['counts'][key] for key in ('schema_invalid','stale','unavailable')):return 'investigate_data' if quality['coverage'] is None or quality['coverage']<0.75:return 'wait_for_labels' if quality['accuracy_on_ready'] is not None and quality['accuracy_on_ready']<0.75: if review=={'data_issue_ruled_out':True,'persistent_failure_reviewed':True,'new_labels_reviewed':True}: return 'prepare_retraining_review' return 'investigate_quality' return 'monitor_distribution' if signal['distribution']['alert'] else 'continue_monitoring'