Distinguish schema, freshness and distribution signals

"""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 available<feature_at or event['contract']!=CONTRACT:raise ValueError('schema')
            values=event['features']
            if not isinstance(values,dict) or set(values)!=set(baseline.FEATURES):raise ValueError('schema')
            if any(type(v) not in (int,float) or not math.isfinite(v) for v in values.values()):raise ValueError('schema')
        except (ValueError,TypeError,OverflowError):counts['schema_invalid']+=1;continue
        if feature_at>now 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 ready<p['predicted_at']+5:raise ValueError('label not finalized after horizon')
    eligible=[p for p in predictions if p['predicted_at']<=as_of]
    ready=[p for p in eligible if p['id'] in by_id and by_id[p['id']]['ready_at']<=as_of]
    correct=sum(p['label']==by_id[p['id']]['label'] for p in ready)
    return {'as_of':as_of,'eligible_predictions':len(eligible),'ready_labels':len(ready),
            'pending':sum(p['id'] in by_id and by_id[p['id']]['ready_at']>as_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'