MLOps: data contracts, CI gates and monitoring

# Step 1 — Prepare the explicit contract: The fixture gives every observation an ID, a source group and a split.
import hashlib, json, math

# Step 2 — Run and inspect the controlled experiment: The input mix changes from [0.9,0.1] to [0.5,0.5] across two...
def digest(value):
    return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(',', ':'), allow_nan=False).encode()).hexdigest()

def validate_rows(rows):
    """An observation has a stable identity, split, group, input and target."""
    # Guard input contract (`not rows`) and fail fast if violated.
    if not rows:
        raise ValueError('empty dataset')
    ids = set()
    groups = {}
    for row in rows:
        # Guard input contract (`set(row) != {'id', 'group', 'split', 'x', 'y'}`) and fail fast if violated.
        if set(row) != {'id', 'group', 'split', 'x', 'y'}:
            raise ValueError('schema mismatch')
        # Guard input contract (`not isinstance(row['id'], str) or not row['id'] or row['id'] in ids`) and fail fast if violated.
        if not isinstance(row['id'], str) or not row['id'] or row['id'] in ids:
            raise ValueError('duplicate or empty ID')
        # Guard input contract (`not isinstance(row['group'], str) or not row['group']`) and fail fast if violated.
        if not isinstance(row['group'], str) or not row['group']:
            raise ValueError('missing source group')
        # Guard input contract (`row['split'] not in {'train', 'validation', 'test'}`) and fail fast if violated.
        if row['split'] not in {'train', 'validation', 'test'}:
            raise ValueError('unknown split')
        for name in ['x', 'y']:
            # Guard input contract (`isinstance(row[name], bool) or not isinstance(row[name], (float, int)) or (not math.isfinite(row[name]))`) and fail fast if violated.
            if isinstance(row[name], bool) or not isinstance(row[name], (float, int)) or not math.isfinite(row[name]):
                raise ValueError('nonfinite or nonnumeric observation')
        # Guard input contract (`row['group'] in groups and groups[row['group']] != row['split']`) and fail fast if violated.
        if row['group'] in groups and groups[row['group']] != row['split']:
            raise ValueError('source group leaks between splits')
        # Execute `ids.add(row['id'])`
        ids.add(row['id'])
        groups[row['group']] = row['split']
    return digest(rows)

def gate_metrics(errors, slices, limit):
    """Count-weighted MSE plus every required slice; no empty-slice pass."""
    # Guard input contract (`len(errors) != len(slices) or not errors or (not math.isfinite(limit)) or (limit < 0)`) and fail fast if violated.
    if len(errors) != len(slices) or not errors or not math.isfinite(limit) or limit < 0:
        raise ValueError('invalid metric contract')
    # Guard input contract (`any((not math.isfinite(e) or e < 0 for e in errors))`) and fail fast if violated.
    if any(not math.isfinite(e) or e < 0 for e in errors):
        raise ValueError('invalid squared error')
    # Guard input contract (`not set(slices) <= {'low', 'high'}`) and fail fast if violated.
    if not set(slices) <= {'low', 'high'}:
        raise ValueError('unknown slice')
    result = {name: {'count': slices.count(name), 'mse': None} for name in ['low', 'high']}
    for name, item in result.items():
        selected = [e for e, s in zip(errors, slices) if s == name]
        if selected: item['mse'] = math.fsum(selected) / len(selected)
    result['overall'] = {'count': len(errors), 'mse': math.fsum(errors) / len(errors)}
    result['passed'] = all(v['count'] > 0 and v['mse'] <= limit for v in result.values())
    return result

rows = [dict(id=f'r{i}', group=f'g{i}', split='train' if i<4 else 'validation', x=float(i), y=2.*i+1.) for i in range(6)]
fingerprint = validate_rows(rows)
# Accumulate the next contribution into `changed[-1]['x']`.
changed = [dict(r) for r in rows]
changed[-1]['x'] += 1
assert validate_rows(changed) != fingerprint
leaked = [dict(r) for r in rows]
leaked[-1]['group'] = rows[0]['group']
# Run the boundary check and catch the expected exception:
try:
    validate_rows(leaked)
except ValueError:
    pass
else:
    raise AssertionError('cross-split source leakage accepted')
errors = [.01] * 9 + [4.]
slices = ['low'] * 9 + ['high']
metrics = gate_metrics(errors, slices, .5)
assert math.isclose(metrics['overall']['mse'], .409)
assert metrics['overall']['mse'] < .5 and not metrics['passed']
assert metrics['high']['count'] == 1 and metrics['high']['mse'] == 4.
assert not gate_metrics([.01]*9, ['low']*9, .5)['passed']
# Total variation on two declared bins: changed input mix is a diagnostic, not proof of degraded accuracy.
reference_mix = [.9, .1]
current_mix = [.5, .5]
shift = .5 * sum(abs(a-b) for a,b in zip(reference_mix,current_mix))
assert math.isclose(shift, .4)
print('Overall MSE:',metrics['overall']['mse'],'high-slice MSE:',metrics['high']['mse'],'release:',metrics['passed'])
print('Two-bin input total variation:',shift,'(requires investigation; not an automatic retraining order)')

shift = .5 * sum(abs(a-b) for a,b in zip(reference_mix,current_mix))
assert math.isclose(shift, .4)