Monitorare le previsioni: registrare input e output

import json
from datetime import datetime
from fastapi import FastAPI
from pydantic import BaseModel
import numpy as np

app = FastAPI()
LOG_FILE = '/tmp/prediction_log.jsonl'

def log_prediction(features: dict, prediction: int, confidence: float):
    entry = {
        'timestamp': datetime.utcnow().isoformat(),
        'features': features,
        'predicted_class': prediction,
        'confidence': confidence
    }
    with open(LOG_FILE, 'a') as f:
        f.write(json.dumps(entry) + '\n')

# Call inside the predict route:
# log_prediction(features.dict(), pred, confidence)
print('Logging enabled — entries written to', LOG_FILE)

import json
import pandas as pd
from datetime import datetime
import random

# Simulate log entries
LOG_FILE = '/tmp/prediction_log.jsonl'
with open(LOG_FILE, 'w') as f:
    for i in range(100):
        entry = {
            'timestamp': datetime.utcnow().isoformat(),
            'features': {'sepal_length': round(4.5 + random.gauss(1, 0.5), 2),
                         'sepal_width': round(3.0 + random.gauss(0, 0.3), 2),
                         'petal_length': round(1.0 + random.gauss(2, 1), 2),
                         'petal_width': round(0.2 + random.gauss(0.5, 0.2), 2)},
            'predicted_class': random.choice([0, 1, 2]),
            'confidence': round(random.uniform(0.6, 1.0), 4)
        }
        f.write(json.dumps(entry) + '\n')

# Read back as DataFrame
rows = [json.loads(line) for line in open(LOG_FILE)]
df = pd.json_normalize(rows)
print(df.shape)
print(df[['confidence', 'predicted_class']].describe())

import pandas as pd
import matplotlib.pyplot as plt
import json

df = pd.json_normalize([json.loads(l) for l in open('/tmp/prediction_log.jsonl')])
df['timestamp'] = pd.to_datetime(df['timestamp'])
df = df.sort_values('timestamp').reset_index(drop=True)

# Rolling window of 20 predictions
df['rolling_confidence'] = df['confidence'].rolling(20).mean()

plt.figure(figsize=(10, 4))
plt.plot(df.index, df['confidence'], alpha=0.3, label='Per-prediction')
plt.plot(df.index, df['rolling_confidence'], linewidth=2, label='Rolling mean (n=20)')
plt.axhline(0.7, color='red', linestyle='--', label='Alert threshold')
plt.xlabel('Prediction number')
plt.ylabel('Confidence')
plt.title('Prediction Confidence Over Time')
plt.legend()
plt.show()

import numpy as np
from scipy import stats
import pandas as pd
import json

df = pd.json_normalize([json.loads(l) for l in open('/tmp/prediction_log.jsonl')])

# Simulated training distribution for sepal_length
train_sepal = np.random.normal(5.8, 0.8, 500)  # reference
recent_sepal = df['features.sepal_length'].values

# KS test: tests if two samples come from same distribution
stat, p_value = stats.ks_2samp(train_sepal, recent_sepal)
print(f'KS statistic: {stat:.4f}  p-value: {p_value:.4f}')
if p_value < 0.05:
    print('WARNING: sepal_length distribution has drifted!')
else:
    print('No significant drift detected in sepal_length')

import pandas as pd
import json

df = pd.json_normalize([json.loads(l) for l in open('/tmp/prediction_log.jsonl')])

# Compute rolling class frequencies in windows of 20
window = 20
rolling_freq = pd.get_dummies(df['predicted_class']).rolling(window).mean()
rolling_freq.columns = [f'class_{c}' for c in rolling_freq.columns]

print('Recent class frequencies (last 20 predictions):')
print(rolling_freq.tail(1).to_string())

print('\nTraining class frequency (expected):')
print('class_0: 0.33  class_1: 0.33  class_2: 0.33')

import logging
import json
from datetime import datetime

# Configure structured JSON logging
class JSONFormatter(logging.Formatter):
    def format(self, record):
        log_data = {
            'time': datetime.utcnow().isoformat(),
            'level': record.levelname,
            'message': record.getMessage()
        }
        if hasattr(record, 'prediction_data'):
            log_data.update(record.prediction_data)
        return json.dumps(log_data)

logger = logging.getLogger('ml_predictor')
handler = logging.StreamHandler()
handler.setFormatter(JSONFormatter())
logger.addHandler(handler)
logger.setLevel(logging.INFO)

# Usage in prediction route:
extra = {'prediction_data': {'features': {'x': 1.5}, 'pred': 0, 'conf': 0.95}}
logger.info('prediction', extra=extra)

from datetime import datetime, timedelta

class RetrainingPolicy:
    def __init__(self, last_trained: datetime, max_age_days=30,
                 min_auc=0.85, max_drift_psi=0.2):
        self.last_trained = last_trained
        self.max_age_days = max_age_days
        self.min_auc = min_auc
        self.max_drift_psi = max_drift_psi

    def should_retrain(self, current_auc=None, max_psi=None):
        age = (datetime.utcnow() - self.last_trained).days
        if age >= self.max_age_days:
            return True, f'Model is {age} days old (max {self.max_age_days})'
        if current_auc and current_auc < self.min_auc:
            return True, f'AUC {current_auc:.3f} below threshold {self.min_auc}'
        if max_psi and max_psi > self.max_drift_psi:
            return True, f'PSI {max_psi:.3f} above threshold {self.max_drift_psi}'
        return False, 'No trigger met — model healthy'

policy = RetrainingPolicy(last_trained=datetime.utcnow() - timedelta(days=35))
print(policy.should_retrain(current_auc=0.90))

# Evidently AI example (pip install evidently)
from evidently.report import Report
from evidently.metric_preset import DataDriftPreset
import pandas as pd
import numpy as np

# Reference: training feature distributions
reference = pd.DataFrame({
    'sepal_length': np.random.normal(5.8, 0.8, 300),
    'petal_length': np.random.normal(3.7, 1.7, 300)
})

# Current: recent predictions features
current = pd.DataFrame({
    'sepal_length': np.random.normal(6.5, 1.2, 100),  # drifted!
    'petal_length': np.random.normal(3.7, 1.7, 100)
})

report = Report(metrics=[DataDriftPreset()])
report.run(reference_data=reference, current_data=current)
report.save_html('/tmp/drift_report.html')
print('Drift report saved to /tmp/drift_report.html')