ml_monitoring.py - master - burning_hell/lab12_ml_testing

ml_monitoring.py
397 строк18 KB
burning-hel
uploaded
13 дек 2025, 01:14
13 дек 2025, 01:14659f38d
import time
from datetime import datetime, timedelta
import pandas as pd
import numpy as np
from ml_pipeline import MLPipeline
import matplotlib.pyplot as plt
import json
import os

# Кастомный JSON энкодер для numpy типов
class NumpyEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, (np.integer, np.int64, np.int32, np.int16, np.int8)):
            return int(obj)
        elif isinstance(obj, (np.floating, np.float64, np.float32, np.float16)):
            return float(obj)
        elif isinstance(obj, np.ndarray):
            return obj.tolist()
        elif isinstance(obj, pd.Series):
            return obj.tolist()
        elif isinstance(obj, pd.DataFrame):
            return obj.to_dict('records')
        return super().default(obj)

class MLMonitoring:
    def __init__(self):
        self.pipeline = MLPipeline()
        self.monitoring_data = []
        self.alert_threshold = 0.7  # Порог для алертов
    
    def collect_monitoring_data(self, days=7, interval_hours=6):
        """Собираем данные мониторинга"""
        print(f"📊 Собираем данные мониторинга за {days} дней...")
        
        end_date = datetime.now()
        start_date = end_date - timedelta(days=days)
        
        current_date = start_date
        iteration = 0
        
        while current_date <= end_date:
            iteration += 1
            print(f"\n📅 Итерация {iteration}: {current_date.strftime('%Y-%m-%d %H:%M')}")
            
            # Генерируем "текущие" данные
            try:
                current_data = self.pipeline.generate_sample_data(200)
            except Exception as e:
                print(f"❌ Ошибка генерации данных: {e}")
                # Создаем простые данные вручную
                current_data = pd.DataFrame({
                    'age': np.random.randint(18, 70, 200),
                    'monthly_charges': np.random.uniform(20, 100, 200),
                    'tenure': np.random.randint(1, 60, 200),
                    'churn': np.random.randint(0, 2, 200)
                })
            
            # Если у нас уже есть модель, тестируем на новых данных
            if self.pipeline.model is not None:
                try:
                    # Предобрабатываем данные
                    X_current, y_current = self.pipeline.preprocess_data(current_data)
                    
                    # Предсказываем на новых данных
                    current_predictions = self.pipeline.model.predict(X_current)
                    
                    # Считаем метрики
                    from sklearn.metrics import accuracy_score
                    current_accuracy = accuracy_score(y_current, current_predictions)
                    
                    # Считаем дрифт фичей
                    feature_drift = self.calculate_feature_drift(current_data)
                    
                    # Собираем мониторинговые данные
                    monitoring_point = {
                        'timestamp': current_date.isoformat(),
                        'data_size': int(len(current_data)),
                        'accuracy': float(current_accuracy),
                        'churn_rate': float(current_data['churn'].mean() if 'churn' in current_data.columns else 0.5),
                        'feature_drift': float(feature_drift),
                        'alerts': []
                    }
                    
                    # Проверяем алерты
                    if current_accuracy < self.alert_threshold:
                        monitoring_point['alerts'].append(f"Низкая точность: {current_accuracy:.3f}")
                    
                    if feature_drift > 0.1:
                        monitoring_point['alerts'].append(f"Высокий дрифт фич: {feature_drift:.3f}")
                    
                    self.monitoring_data.append(monitoring_point)
                    
                    print(f"   📈 Accuracy: {current_accuracy:.3f}")
                    print(f"   📊 Churn rate: {monitoring_point['churn_rate']:.3f}")
                    print(f"   📉 Feature drift: {feature_drift:.3f}")
                    
                    if monitoring_point['alerts']:
                        print(f"   🚨 Alerts: {', '.join(monitoring_point['alerts'])}")
                        
                except Exception as e:
                    print(f"❌ Ошибка обработки данных: {e}")
                    # Добавляем точку с ошибкой
                    monitoring_point = {
                        'timestamp': current_date.isoformat(),
                        'data_size': int(len(current_data)),
                        'accuracy': 0.5,  # Значение по умолчанию
                        'churn_rate': 0.5,
                        'feature_drift': 0.0,
                        'alerts': [f"Ошибка обработки: {str(e)[:50]}"]
                    }
                    self.monitoring_data.append(monitoring_point)
            else:
                print("   ⚠️  Модель не загружена, пропускаем")
            
            # "Перемещаемся" вперед во времени
            current_date += timedelta(hours=interval_hours)
            time.sleep(0.1)  # Небольшая задержка для имитации времени
        
        print(f"\n✅ Собрано {len(self.monitoring_data)} точек мониторинга")
        return self.monitoring_data
    
    def calculate_feature_drift(self, current_data, reference_data=None):
        """Рассчитываем дрифт фичей"""
        if reference_data is None:
            # Используем сгенерированные данные как референс
            try:
                reference_data = self.pipeline.generate_sample_data(500)
            except:
                # Создаем простые данные вручную
                reference_data = pd.DataFrame({
                    'age': np.random.randint(18, 70, 500),
                    'monthly_charges': np.random.uniform(20, 100, 500),
                    'tenure': np.random.randint(1, 60, 500)
                })
        
        # Сравниваем распределения ключевых фичей
        drift_score = 0.0
        key_features = ['age', 'monthly_charges', 'tenure']
        features_found = 0
        
        for feature in key_features:
            if feature in reference_data.columns and feature in current_data.columns:
                try:
                    # Простой расчет дрифта: разница в средних значениях
                    ref_mean = reference_data[feature].mean()
                    curr_mean = current_data[feature].mean()
                    ref_std = reference_data[feature].std()
                    
                    if ref_std > 0:
                        drift = abs(ref_mean - curr_mean) / ref_std
                        drift_score += min(drift, 1.0)  # Ограничиваем максимальный дрифт
                        features_found += 1
                except:
                    continue
        
        if features_found > 0:
            return float(drift_score / features_found)
        else:
            return 0.0
    
    def create_monitoring_dashboard(self):
        """Создаем дашборд мониторинга"""
        if not self.monitoring_data:
            print("❌ Нет данных для дашборда")
            return
        
        try:
            df = pd.DataFrame(self.monitoring_data)
            df['timestamp'] = pd.to_datetime(df['timestamp'])
            
            print("\n📊 СОЗДАЕМ ДАШБОРД МОНИТОРИНГА")
            
            # Создаем графики
            plt.figure(figsize=(15, 10))
            
            # 1. Точность модели во времени
            plt.subplot(2, 2, 1)
            plt.plot(df['timestamp'], df['accuracy'], marker='o', linewidth=2)
            plt.axhline(y=self.alert_threshold, color='red', linestyle='--', label='Порог алерта')
            plt.title('Точность модели во времени')
            plt.xlabel('Время')
            plt.ylabel('Accuracy')
            plt.legend()
            plt.grid(True, alpha=0.3)
            plt.xticks(rotation=45)
            
            # 2. Дрифт фичей во времени
            plt.subplot(2, 2, 2)
            plt.plot(df['timestamp'], df['feature_drift'], marker='s', color='orange', linewidth=2)
            plt.axhline(y=0.1, color='red', linestyle='--', label='Порог дрифта')
            plt.title('Дрифт фичей во времени')
            plt.xlabel('Время')
            plt.ylabel('Feature Drift Score')
            plt.legend()
            plt.grid(True, alpha=0.3)
            plt.xticks(rotation=45)
            
            # 3. Распределение churn rate
            plt.subplot(2, 2, 3)
            plt.hist(df['churn_rate'], bins=10, alpha=0.7, color='green', edgecolor='black')
            plt.title('Распределение Churn Rate')
            plt.xlabel('Churn Rate')
            plt.ylabel('Частота')
            plt.grid(True, alpha=0.3)
            
            # 4. Количество алертов по времени
            plt.subplot(2, 2, 4)
            alert_counts = df['alerts'].apply(len)
            dates = df['timestamp'].dt.strftime('%m-%d %H:%M')
            plt.bar(range(len(dates)), alert_counts, color='red', alpha=0.7, edgecolor='black')
            plt.title('Количество алертов по времени')
            plt.xlabel('Точки мониторинга')
            plt.ylabel('Количество алертов')
            plt.xticks(range(len(dates)), dates, rotation=45, ha='right')
            plt.grid(True, alpha=0.3)
            
            plt.tight_layout()
            plt.savefig('ml_monitoring_dashboard.png', dpi=300, bbox_inches='tight')
            plt.close()
            
            print("✅ Дашборд сохранен в ml_monitoring_dashboard.png")
            
        except Exception as e:
            print(f"❌ Ошибка создания дашборда: {e}")
    
    def generate_monitoring_report(self):
        """Генерируем отчет мониторинга"""
        if not self.monitoring_data:
            print("❌ Нет данных для отчета")
            return None
        
        try:
            df = pd.DataFrame(self.monitoring_data)
            
            # Статистика
            total_alerts = df['alerts'].apply(len).sum()
            avg_accuracy = df['accuracy'].mean()
            max_drift = df['feature_drift'].max()
            
            print("\n📈 ОТЧЕТ МОНИТОРИНГА ML PIPELINE")
            print("=" * 50)
            print(f"📅 Период мониторинга: {len(df)} точек")
            print(f"🎯 Средняя точность: {avg_accuracy:.3f}")
            print(f"📉 Максимальный дрифт: {max_drift:.3f}")
            print(f"🚨 Всего алертов: {int(total_alerts)}")
            
            # Детали алертов
            if total_alerts > 0:
                print(f"\n🔍 ДЕТАЛИ АЛЕРТОВ:")
                all_alerts = []
                for alerts in df['alerts']:
                    all_alerts.extend(alerts)
                
                if all_alerts:
                    alert_counts = pd.Series(all_alerts).value_counts()
                    for alert, count in alert_counts.items():
                        print(f"  • {alert}: {count} раз")
            
            # Сохраняем отчет
            report = {
                'timestamp': datetime.now().isoformat(),
                'monitoring_period': int(len(df)),
                'average_accuracy': float(avg_accuracy),
                'max_feature_drift': float(max_drift),
                'total_alerts': int(total_alerts),
                'stability_score': float(self.calculate_stability_score(df)),
                'monitoring_data': [
                    {
                        'timestamp': str(row['timestamp']),
                        'data_size': int(row['data_size']),
                        'accuracy': float(row['accuracy']),
                        'churn_rate': float(row['churn_rate']),
                        'feature_drift': float(row['feature_drift']),
                        'alerts': list(row['alerts'])
                    }
                    for _, row in df.iterrows()
                ]
            }
            
            with open('ml_monitoring_report.json', 'w', encoding='utf-8') as f:
                json.dump(report, f, indent=2, cls=NumpyEncoder, ensure_ascii=False)
            
            print(f"\n✅ Отчет сохранен в ml_monitoring_report.json")
            
            # Оценка стабильности
            stability = report['stability_score']
            if stability >= 0.8:
                print("🏆 ВЫСОКАЯ СТАБИЛЬНОСТЬ: ML pipeline работает стабильно")
            elif stability >= 0.6:
                print("⚠️  СРЕДНЯЯ СТАБИЛЬНОСТЬ: ML pipeline требует наблюдения")
            else:
                print("🚨 НИЗКАЯ СТАБИЛЬНОСТЬ: ML pipeline нестабилен, нужны действия")
            
            return report
            
        except Exception as e:
            print(f"❌ Ошибка генерации отчета: {e}")
            return None
    
    def calculate_stability_score(self, df):
        """Рассчитываем оценку стабильности pipeline"""
        try:
            # Основано на точности, дрифте и количестве алертов
            accuracy_score = df['accuracy'].mean()
            drift_penalty = min(df['feature_drift'].max() * 2, 0.3)  # Штраф за дрифт
            alert_penalty = min(len(df[df['alerts'].apply(len) > 0]) / len(df), 0.3)  # Штраф за алерты
            
            stability = accuracy_score - drift_penalty - alert_penalty
            return float(max(stability, 0))  # Не ниже 0
            
        except:
            return 0.5

# Упрощенная версия для быстрого запуска
def run_basic_monitoring():
    """Запуск упрощенного мониторинга"""
    print("🎯 ЗАПУСКАЕМ МОНИТОРИНГ ML PIPELINE")
    print("=" * 50)
    
    monitor = MLMonitoring()
    
    # Быстро собираем данные (5 точек вместо многих дней)
    print("\n📊 Быстрый сбор данных мониторинга...")
    
    for i in range(5):
        print(f"\n📅 Точка {i+1}/5...")
        
        # Генерируем данные
        current_data = pd.DataFrame({
            'age': np.random.randint(18, 70, 100),
            'monthly_charges': np.random.uniform(20, 100, 100),
            'tenure': np.random.randint(1, 60, 100),
            'churn': np.random.randint(0, 2, 100)
        })
        
        # Случайные метрики
        accuracy = 0.7 + np.random.normal(0, 0.05)
        feature_drift = np.random.uniform(0, 0.15)
        
        monitoring_point = {
            'timestamp': (datetime.now() - timedelta(hours=4*(4-i))).isoformat(),
            'data_size': int(len(current_data)),
            'accuracy': float(accuracy),
            'churn_rate': float(current_data['churn'].mean()),
            'feature_drift': float(feature_drift),
            'alerts': []
        }
        
        if accuracy < 0.7:
            monitoring_point['alerts'].append(f"Низкая точность: {accuracy:.3f}")
        
        if feature_drift > 0.1:
            monitoring_point['alerts'].append(f"Высокий дрифт фич: {feature_drift:.3f}")
        
        monitor.monitoring_data.append(monitoring_point)
        
        print(f"   📈 Accuracy: {accuracy:.3f}")
        print(f"   📊 Churn rate: {monitoring_point['churn_rate']:.3f}")
        print(f"   📉 Feature drift: {feature_drift:.3f}")
        
        if monitoring_point['alerts']:
            print(f"   🚨 Alerts: {', '.join(monitoring_point['alerts'])}")
    
    # Создаем дашборд и отчет
    monitor.create_monitoring_dashboard()
    report = monitor.generate_monitoring_report()
    
    return report

# Пример использования
if __name__ == "__main__":
    # Выбираем режим: полный или быстрый
    mode = "quick"  # Измени на "full" для полного мониторинга
    
    if mode == "full":
        monitor = MLMonitoring()
        
        # Сначала обучаем и сохраняем модель
        print("🤖 Обучаем модель...")
        pipeline = MLPipeline()
        data = pipeline.generate_sample_data(500)
        X, y = pipeline.preprocess_data(data)
        pipeline.train_model(X, y)
        pipeline.save_model()
        
        # Загружаем модель в мониторинг
        monitor.pipeline.load_model()
        
        # Собираем данные мониторинга
        monitoring_data = monitor.collect_monitoring_data(days=2, interval_hours=6)
        
    else:
        # Быстрый режим для тестирования
        report = run_basic_monitoring()
        print(f"\n✅ Мониторинг завершен! Проверьте файлы:")
        print("   • ml_monitoring_dashboard.png")
        print("   • ml_monitoring_report.json")