Modelle versionieren: Warum Dateinamen und Metadaten wichtig sind

from datetime import date

def model_filename(dataset, algorithm, metric_name, metric_value,
                   data_version='v1', ext='joblib'):
    today = date.today().strftime('%Y%m%d')
    metric_str = f'{metric_name}{int(metric_value * 100)}'
    return f'{dataset}__{algorithm}__{today}__dv{data_version}__{metric_str}.{ext}'

# Examples
print(model_filename('titanic', 'rf', 'acc', 0.834, data_version='2'))
print(model_filename('fraud', 'xgb', 'auc', 0.971))
print(model_filename('cancer', 'logreg', 'f1', 0.955))

import json
import sklearn, sys
from datetime import datetime

def save_metadata(path, dataset, algorithm, params, metrics, features, notes=''):
    meta = {
        'model_path': path,
        'dataset': dataset,
        'algorithm': algorithm,
        'hyperparameters': params,
        'metrics': metrics,
        'features': features,
        'sklearn_version': sklearn.__version__,
        'python_version': sys.version.split()[0],
        'trained_at': datetime.utcnow().isoformat(),
        'notes': notes
    }
    meta_path = path.replace('.joblib', '_metadata.json')
    with open(meta_path, 'w') as f:
        json.dump(meta, f, indent=2)
    print('Metadata saved to:', meta_path)
    return meta

# Usage example
save_metadata(
    '/tmp/cancer__logreg__20260620__dv1__acc97.joblib',
    dataset='breast_cancer', algorithm='LogisticRegression',
    params={'C': 1.0, 'max_iter': 300},
    metrics={'accuracy': 0.9789, 'roc_auc': 0.9941, 'f1': 0.9831},
    features=['mean radius', 'mean texture', '... 30 total'],
    notes='Trained on full UCI breast cancer dataset'
)

import hashlib

def file_hash(path, algo='sha256'):
    h = hashlib.new(algo)
    with open(path, 'rb') as f:
        for chunk in iter(lambda: f.read(65536), b''):
            h.update(chunk)
    return h.hexdigest()[:12]  # first 12 hex chars as short ID

# Example: hash the model file itself as a unique ID
model_path = '/tmp/cancer_model.joblib'
model_hash = file_hash(model_path)
print('Model hash (short):', model_hash)

model_registry = [
    {'version': '1.0.0', 'algorithm': 'LogisticRegression', 'auc': 0.921,
     'note': 'Initial production model'},
    {'version': '1.1.0', 'algorithm': 'LogisticRegression', 'auc': 0.935,
     'note': 'Retrained on 3 months more data'},
    {'version': '2.0.0', 'algorithm': 'XGBoost', 'auc': 0.971,
     'note': 'New algorithm; feature set changed — incompatible schema'}
]

print('Model Registry:')
for entry in model_registry:
    print(f"  v{entry['version']}  AUC={entry['auc']}  {entry['note']}")

import json
import os

REGISTRY_PATH = '/tmp/model_registry.json'

def register_model(filename, version, metrics, is_production=False):
    try:
        with open(REGISTRY_PATH) as f:
            registry = json.load(f)
    except FileNotFoundError:
        registry = []

    # Mark all as not-production if this one is production
    if is_production:
        for entry in registry:
            entry['is_production'] = False

    registry.append({
        'filename': filename,
        'version': version,
        'metrics': metrics,
        'is_production': is_production
    })

    with open(REGISTRY_PATH, 'w') as f:
        json.dump(registry, f, indent=2)
    print(f'Registered v{version} (production={is_production})')

register_model('cancer__logreg__v1.0.0.joblib', '1.0.0',
               {'accuracy': 0.979, 'auc': 0.994}, is_production=True)

import json
import joblib

def load_production_model(registry_path, model_dir='/tmp'):
    with open(registry_path) as f:
        registry = json.load(f)
    prod = next((r for r in registry if r['is_production']), None)
    if prod is None:
        raise RuntimeError('No production model registered!')
    path = f'{model_dir}/{prod["filename"]}'
    print(f'Loading production model: {prod["filename"]} (v{prod["version"]})')
    print(f'Metrics: {prod["metrics"]}')
    # return joblib.load(path)  # would load for real
    return None  # demo

load_production_model('/tmp/model_registry.json')

import mlflow
from sklearn.linear_model import LogisticRegression
from sklearn.datasets import load_iris
from sklearn.model_selection import cross_val_score
import numpy as np

X, y = load_iris(return_X_y=True)

with mlflow.start_run(run_name='logreg_iris_v1'):
    model = LogisticRegression(C=1.0, max_iter=200)
    cv_score = cross_val_score(model, X, y, cv=5).mean()

    mlflow.log_param('C', 1.0)
    mlflow.log_param('max_iter', 200)
    mlflow.log_metric('cv_accuracy', cv_score)

    model.fit(X, y)
    mlflow.sklearn.log_model(model, 'model')

    print(f'CV Accuracy: {cv_score:.4f}')
    print('Run logged to MLflow')

# Simulated registry entry with tags (no MLflow required)
model_entry = {
    'version': '2.1.0',
    'filename': 'fraud__xgb__2.1.0.joblib',
    'tags': {
        'analyst': 'data-science-team',
        'task': 'binary-classification',
        'fairness_audit': 'passed',
        'retrain_trigger': 'monthly-schedule',
        'deployment_region': 'eu-west-1'
    },
    'metrics': {'roc_auc': 0.971, 'precision': 0.83, 'recall': 0.79}
}

print('Model entry:')
print(json.dumps(model_entry, indent=2))

def should_promote(new_metrics, baseline_metrics, min_auc_improvement=0.005):
    if new_metrics['auc'] < baseline_metrics['auc'] + min_auc_improvement:
        return False, 'AUC improvement too small'
    if new_metrics.get('fairness_delta', 0) > 0.05:
        return False, 'Fairness constraint violated'
    if new_metrics.get('latency_ms', 0) > 100:
        return False, 'Latency too high'
    return True, 'All criteria met'

baseline = {'auc': 0.921}
candidate = {'auc': 0.937, 'fairness_delta': 0.02, 'latency_ms': 45}

promote, reason = should_promote(candidate, baseline)
print(f'Promote: {promote} — {reason}')