#!/usr/bin/env python3 """ Model Evaluation & Reporting Module Für umfassende ML-Model-Evaluation """ import numpy as np import json import os from datetime import datetime from collections import defaultdict try: from sklearn.model_selection import cross_val_score from sklearn.metrics import r2_score, mean_absolute_error, mean_squared_error SKLEARN_AVAILABLE = True except ImportError: SKLEARN_AVAILABLE = False class ModelEvaluator: """Evaluiert und dokumentiert ML-Modelle.""" def __init__(self, cache_path=None): self.cache_path = cache_path self.evaluation_results = {} self.report_file = os.path.join(cache_path, 'model_evaluation.json') if cache_path else None def evaluate_model(self, model, X_train, X_test, y_train, y_test, model_name, number): """ Umfassende Model-Evaluation. Returns: dict with metrics """ if not SKLEARN_AVAILABLE: return {'error': 'scikit-learn not available'} results = { 'model_name': model_name, 'number': number, 'timestamp': datetime.now().isoformat(), 'data_size': { 'train': len(X_train), 'test': len(X_test) } } try: # 1. Training Score y_train_pred = model.predict(X_train) results['train_r2'] = r2_score(y_train, y_train_pred) results['train_mae'] = mean_absolute_error(y_train, y_train_pred) results['train_rmse'] = np.sqrt(mean_squared_error(y_train, y_train_pred)) # 2. Test Score y_test_pred = model.predict(X_test) results['test_r2'] = r2_score(y_test, y_test_pred) results['test_mae'] = mean_absolute_error(y_test, y_test_pred) results['test_rmse'] = np.sqrt(mean_squared_error(y_test, y_test_pred)) # 3. Overfit Detection results['overfitting'] = results['train_r2'] - results['test_r2'] results['is_overfit'] = results['overfitting'] > 0.2 # 4. Cross-Validation (3-fold for speed) try: cv_scores = cross_val_score(model, X_train, y_train, cv=3, scoring='r2') results['cv_mean'] = float(np.mean(cv_scores)) results['cv_std'] = float(np.std(cv_scores)) results['cv_scores'] = [float(s) for s in cv_scores] except Exception as e: results['cv_error'] = str(e) # 5. Feature Importance (if available) if hasattr(model, 'feature_importances_'): importances = model.feature_importances_ results['top_features'] = { f'feature_{i}': float(imp) for i, imp in enumerate(importances[:10]) # Top 10 } results['feature_importance_sum'] = float(np.sum(importances)) # 6. Prediction Distribution results['pred_distribution'] = { 'min': float(np.min(y_test_pred)), 'max': float(np.max(y_test_pred)), 'mean': float(np.mean(y_test_pred)), 'std': float(np.std(y_test_pred)) } # 7. Quality Rating test_r2 = results['test_r2'] if test_r2 > 0.7: results['quality'] = 'Excellent' elif test_r2 > 0.5: results['quality'] = 'Good' elif test_r2 > 0.3: results['quality'] = 'Fair' elif test_r2 > 0.1: results['quality'] = 'Poor' else: results['quality'] = 'Very Poor' except Exception as e: results['error'] = str(e) return results def add_evaluation(self, number, model_name, results): """Fügt Evaluation-Result hinzu.""" key = f"{number}_{model_name}" self.evaluation_results[key] = results def generate_summary_report(self): """Generiert Zusammenfassungs-Report.""" if not self.evaluation_results: return "Keine Evaluation-Daten vorhanden" report = [] report.append("=" * 80) report.append("MODEL EVALUATION SUMMARY") report.append("=" * 80) report.append(f"Timestamp: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") report.append(f"Total Evaluations: {len(self.evaluation_results)}") report.append("") # Aggregate stats all_test_r2 = [] all_cv_mean = [] overfit_count = 0 quality_dist = defaultdict(int) for key, results in self.evaluation_results.items(): if 'test_r2' in results: all_test_r2.append(results['test_r2']) if 'cv_mean' in results: all_cv_mean.append(results['cv_mean']) if results.get('is_overfit'): overfit_count += 1 if 'quality' in results: quality_dist[results['quality']] += 1 # Overall Stats report.append("OVERALL STATISTICS") report.append("-" * 80) if all_test_r2: report.append(f"Test R² Score:") report.append(f" Mean: {np.mean(all_test_r2):.4f}") report.append(f" Median: {np.median(all_test_r2):.4f}") report.append(f" Std: {np.std(all_test_r2):.4f}") report.append(f" Min: {np.min(all_test_r2):.4f}") report.append(f" Max: {np.max(all_test_r2):.4f}") report.append("") if all_cv_mean: report.append(f"Cross-Validation R² Score:") report.append(f" Mean: {np.mean(all_cv_mean):.4f}") report.append(f" Std: {np.std(all_cv_mean):.4f}") report.append("") report.append(f"Overfitting Detection:") report.append(f" Overfit Models: {overfit_count}/{len(self.evaluation_results)}") report.append("") report.append(f"Quality Distribution:") for quality in ['Excellent', 'Good', 'Fair', 'Poor', 'Very Poor']: count = quality_dist.get(quality, 0) pct = (count / len(self.evaluation_results)) * 100 if self.evaluation_results else 0 report.append(f" {quality:12}: {count:3} ({pct:5.1f}%)") report.append("") # Top 10 Best Models sorted_results = sorted( [(k, v) for k, v in self.evaluation_results.items() if 'test_r2' in v], key=lambda x: x[1]['test_r2'], reverse=True )[:10] report.append("TOP 10 MODELS (by Test R²)") report.append("-" * 80) report.append(f"{'Number':<10} {'Model':<20} {'Test R²':<12} {'CV Mean':<12} {'Quality':<15}") report.append("-" * 80) for key, results in sorted_results: number = results.get('number', 'N/A') model = results.get('model_name', 'N/A') test_r2 = results.get('test_r2', 0) cv_mean = results.get('cv_mean', 0) quality = results.get('quality', 'N/A') report.append(f"{number:<10} {model:<20} {test_r2:<12.4f} {cv_mean:<12.4f} {quality:<15}") report.append("") report.append("=" * 80) return "\n".join(report) def save_evaluation(self): """Speichert Evaluation persistent.""" if not self.report_file: return try: os.makedirs(os.path.dirname(self.report_file), exist_ok=True) data = { 'timestamp': datetime.now().isoformat(), 'total_evaluations': len(self.evaluation_results), 'results': self.evaluation_results } with open(self.report_file, 'w') as f: json.dump(data, f, indent=2) print(f" 💾 Evaluation saved: {self.report_file}") except Exception as e: print(f" ⚠️ Could not save evaluation: {e}") def load_evaluation(self): """Lädt gespeicherte Evaluation.""" if not self.report_file or not os.path.exists(self.report_file): return try: with open(self.report_file, 'r') as f: data = json.load(f) self.evaluation_results = data.get('results', {}) timestamp = data.get('timestamp', 'Unknown') print(f" 📂 Evaluation loaded: {len(self.evaluation_results)} results (from {timestamp[:10]})") except Exception as e: print(f" ⚠️ Could not load evaluation: {e}") def print_summary(self): """Druckt Summary auf Console.""" summary = self.generate_summary_report() print("\n" + summary)