ml_testing_framework.py - master - dievavar/lab12_ml_testing

ml_testing_framework.py
596 строк23 KB
dievavar
upload files
04 дек 2025, 20:59
04 дек 2025, 20:59d25d61d
import pytest
import pandas as pd
import numpy as np
import requests
import json
from datetime import datetime, timedelta
from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score
import warnings

warnings.filterwarnings('ignore')

# Проверяем наличие ml_pipeline
try:
    from ml_pipeline import MLPipeline

    MLPipeline_available = True
except ImportError:
    print("⚠️  MLPipeline не найден, создаем заглушку")
    MLPipeline_available = False


    class MLPipeline:
        def __init__(self):
            self.model = None
            self.feature_columns = []
            self.target_column = 'churn'

        def generate_sample_data(self, n_samples=1000):
            """Генерируем тестовые данные"""
            np.random.seed(42)

            data = pd.DataFrame({
                'customer_id': [f'CUST_{i:04d}' for i in range(n_samples)],
                'age': np.random.randint(18, 80, n_samples),
                'tenure': np.random.randint(1, 72, n_samples),
                'monthly_charges': np.random.uniform(20, 120, n_samples),
                'total_charges': np.random.uniform(50, 5000, n_samples),
                'contract_type': np.random.choice(['Monthly', 'Yearly', 'Two-Year'], n_samples),
                'payment_method': np.random.choice(['Credit Card', 'Electronic Check', 'Bank Transfer'], n_samples),
                'paperless_billing': np.random.randint(0, 2, n_samples),
                'dependents': np.random.randint(0, 2, n_samples),
                'partner': np.random.randint(0, 2, n_samples),
                'online_security': np.random.randint(0, 2, n_samples),
                'tech_support': np.random.randint(0, 2, n_samples),
                'monthly_usage_gb': np.random.randint(50, 500, n_samples),
                'customer_service_calls': np.random.randint(0, 10, n_samples),
                'churn': np.random.randint(0, 2, n_samples)
            })
            return data

        def preprocess_data(self, data):
            """Предобработка данных (заглушка)"""
            X = data.drop(['customer_id', 'churn'], axis=1, errors='ignore')
            y = data['churn']
            return X, y

        def train_model(self, X, y):
            """Обучение модели (заглушка)"""
            # Создаем тестовые предсказания
            n = len(y)
            y_pred = np.random.randint(0, 2, n)

            # Разделяем на train/test (заглушка)
            split_idx = int(n * 0.8)
            X_test = X.iloc[split_idx:]
            y_test = y.iloc[split_idx:]
            y_pred_test = y_pred[split_idx:]

            return X_test, y_test, y_pred_test

        def load_model(self, path):
            """Загрузка модели (заглушка)"""
            return True


class MLTestingFramework:
    def __init__(self, api_url="http://localhost:5001"):
        self.pipeline = MLPipeline()
        self.api_url = api_url
        self.test_results = []

    def log_test(self, test_name, description, success, details=None):
        """Логируем результаты теста"""
        test_result = {
            'test_name': test_name,
            'description': description,
            'success': success,
            'details': details,
            'timestamp': datetime.now().isoformat()
        }
        self.test_results.append(test_result)

        status = "✅ УСПЕХ" if success else "❌ ПРОВАЛ"
        print(f"{status} {test_name}: {description}")

        if details and not success:
            print(f"   📝 Детали: {details}")

    def test_data_quality(self, data):
        """Тестируем качество данных"""
        print("\n🔍 ТЕСТИРУЕМ КАЧЕСТВО ДАННЫХ")

        # Проверяем наличие обязательных колонок
        required_columns = ['age', 'tenure', 'monthly_charges', 'contract_type', 'churn']
        missing_columns = [col for col in required_columns if col not in data.columns]

        if missing_columns:
            self.log_test(
                "DATA_COMPLETENESS",
                "Проверка обязательных колонок",
                False,
                f"Отсутствуют колонки: {missing_columns}"
            )
        else:
            self.log_test(
                "DATA_COMPLETENESS",
                "Проверка обязательных колонок",
                True
            )

        # Проверяем пропущенные значения
        missing_values = data.isnull().sum()
        high_missing = missing_values[missing_values > 0]

        if not high_missing.empty:
            self.log_test(
                "MISSING_VALUES",
                "Проверка пропущенных значений",
                False,
                f"Пропущенные значения: {dict(high_missing)}"
            )
        else:
            self.log_test(
                "MISSING_VALUES",
                "Проверка пропущенных значений",
                True
            )

        # Проверяем выбросы в числовых колонках
        numerical_columns = data.select_dtypes(include=[np.number]).columns
        outlier_tests = []

        for col in numerical_columns:
            if col == 'churn':  # Пропускаем целевую переменную
                continue

            Q1 = data[col].quantile(0.25)
            Q3 = data[col].quantile(0.75)
            IQR = Q3 - Q1
            lower_bound = Q1 - 1.5 * IQR
            upper_bound = Q3 + 1.5 * IQR

            outliers = data[(data[col] < lower_bound) | (data[col] > upper_bound)]
            outlier_percentage = len(outliers) / len(data) * 100

            if outlier_percentage > 5:  # Больше 5% выбросов
                outlier_tests.append(f"{col}: {outlier_percentage:.1f}%")

        if outlier_tests:
            self.log_test(
                "OUTLIERS",
                "Проверка выбросов",
                False,
                f"Высокий процент выбросов: {', '.join(outlier_tests)}"
            )
        else:
            self.log_test(
                "OUTLIERS",
                "Проверка выбросов",
                True
            )

        # Проверяем распределение целевой переменной
        if 'churn' in data.columns:
            churn_distribution = data['churn'].value_counts(normalize=True)
            minority_class = churn_distribution.min()

            if minority_class < 0.2:  # Меньше 20% в миноритарном классе
                self.log_test(
                    "TARGET_DISTRIBUTION",
                    "Проверка распределения целевой переменной",
                    False,
                    f"Дисбаланс классов: {dict(churn_distribution)}"
                )
            else:
                self.log_test(
                    "TARGET_DISTRIBUTION",
                    "Проверка распределения целевой переменной",
                    True,
                    f"Распределение: {dict(churn_distribution)}"
                )
        else:
            self.log_test(
                "TARGET_DISTRIBUTION",
                "Проверка распределения целевой переменной",
                False,
                "Колонка 'churn' не найдена"
            )

    def test_model_performance(self, X_test, y_test, y_pred):
        """Тестируем производительность модели"""
        print("\n🎯 ТЕСТИРУЕМ ПРОИЗВОДИТЕЛЬНОСТЬ МОДЕЛИ")

        try:
            accuracy = accuracy_score(y_test, y_pred)
            precision = precision_score(y_test, y_pred, zero_division=0)
            recall = recall_score(y_test, y_pred, zero_division=0)
            f1 = f1_score(y_test, y_pred, zero_division=0)

            # Проверяем точность
            if accuracy >= 0.7:
                self.log_test(
                    "MODEL_ACCURACY",
                    "Проверка точности модели",
                    True,
                    f"Accuracy: {accuracy:.3f}"
                )
            else:
                self.log_test(
                    "MODEL_ACCURACY",
                    "Проверка точности модели",
                    False,
                    f"Accuracy: {accuracy:.3f} (ниже порога 0.7)"
                )

            # Проверяем precision
            if precision >= 0.6:
                self.log_test(
                    "MODEL_PRECISION",
                    "Проверка precision модели",
                    True,
                    f"Precision: {precision:.3f}"
                )
            else:
                self.log_test(
                    "MODEL_PRECISION",
                    "Проверка precision модели",
                    False,
                    f"Precision: {precision:.3f} (ниже порога 0.6)"
                )

            # Проверяем recall
            if recall >= 0.5:
                self.log_test(
                    "MODEL_RECALL",
                    "Проверка recall модели",
                    True,
                    f"Recall: {recall:.3f}"
                )
            else:
                self.log_test(
                    "MODEL_RECALL",
                    "Проверка recall модели",
                    False,
                    f"Recall: {recall:.3f} (ниже порога 0.5)"
                )

            # Проверяем F1-score
            if f1 >= 0.6:
                self.log_test(
                    "MODEL_F1",
                    "Проверка F1-score модели",
                    True,
                    f"F1-score: {f1:.3f}"
                )
            else:
                self.log_test(
                    "MODEL_F1",
                    "Проверка F1-score модели",
                    False,
                    f"F1-score: {f1:.3f} (ниже порога 0.6)"
                )

        except Exception as e:
            self.log_test(
                "MODEL_PERFORMANCE",
                "Проверка производительности модели",
                False,
                f"Ошибка при расчете метрик: {e}"
            )

    def test_data_drift(self, reference_data, current_data):
        """Тестируем дрифт данных"""
        print("\n📊 ТЕСТИРУЕМ ДРИФТ ДАННЫХ")

        try:
            # Упрощенная проверка дрифта
            numeric_columns = reference_data.select_dtypes(include=[np.number]).columns

            drift_detected = False
            drift_details = []

            for col in numeric_columns:
                if col in current_data.columns:
                    ref_mean = reference_data[col].mean()
                    curr_mean = current_data[col].mean()

                    # Простая проверка: если среднее отличается более чем на 20%
                    if ref_mean != 0 and abs(curr_mean - ref_mean) / abs(ref_mean) > 0.2:
                        drift_detected = True
                        drift_details.append(f"{col}: {ref_mean:.2f} → {curr_mean:.2f}")

            if not drift_detected:
                self.log_test(
                    "DATA_DRIFT",
                    "Проверка дрифта данных",
                    True,
                    "Значительный дрифт не обнаружен"
                )
            else:
                self.log_test(
                    "DATA_DRIFT",
                    "Проверка дрифта данных",
                    False,
                    f"Обнаружен дрифт: {', '.join(drift_details[:3])}"
                )

        except Exception as e:
            self.log_test(
                "DATA_DRIFT",
                "Проверка дрифта данных",
                False,
                f"Ошибка при проверке дрифта: {e}"
            )

    def test_api_functionality(self):
        """Тестируем функциональность API"""
        print("\n🌐 ТЕСТИРУЕМ API ФУНКЦИОНАЛЬНОСТЬ")

        # Тестируем health check
        try:
            response = requests.get(f"{self.api_url}/health", timeout=5)
            if response.status_code == 200:
                health_data = response.json()
                if health_data.get('model_loaded'):
                    self.log_test(
                        "API_HEALTH",
                        "Проверка health check API",
                        True
                    )
                else:
                    self.log_test(
                        "API_HEALTH",
                        "Проверка health check API",
                        False,
                        "Модель не загружена в API"
                    )
            else:
                self.log_test(
                    "API_HEALTH",
                    "Проверка health check API",
                    False,
                    f"Status code: {response.status_code}"
                )
        except requests.exceptions.ConnectionError:
            self.log_test(
                "API_HEALTH",
                "Проверка health check API",
                False,
                f"API недоступен по адресу {self.api_url}. Запустите ml_api.py сначала."
            )
        except Exception as e:
            self.log_test(
                "API_HEALTH",
                "Проверка health check API",
                False,
                f"Ошибка подключения: {e}"
            )

        # Тестируем предсказание (только если API доступен)
        try:
            test_customer = {
                'customer_id': 'API_TEST_001',
                'age': 45,
                'tenure': 36,
                'monthly_charges': 89.99,
                'total_charges': 3239.64,
                'contract_type': 'Yearly',
                'payment_method': 'Credit Card',
                'paperless_billing': 1,
                'dependents': 0,
                'partner': 1,
                'online_security': 1,
                'tech_support': 1,
                'monthly_usage_gb': 350,
                'customer_service_calls': 1
            }

            response = requests.post(f"{self.api_url}/predict", json=test_customer, timeout=5)

            if response.status_code == 200:
                prediction_data = response.json()
                if 'predictions' in prediction_data:
                    self.log_test(
                        "API_PREDICTION",
                        "Проверка предсказания через API",
                        True,
                        f"Предсказание: {prediction_data['predictions']}"
                    )
                else:
                    self.log_test(
                        "API_PREDICTION",
                        "Проверка предсказания через API",
                        False,
                        "Некорректный ответ от API"
                    )
            else:
                self.log_test(
                    "API_PREDICTION",
                    "Проверка предсказания через API",
                    False,
                    f"Status code: {response.status_code}, Response: {response.text}"
                )

        except Exception as e:
            self.log_test(
                "API_PREDICTION",
                "Проверка предсказания через API",
                False,
                f"Ошибка: {e}"
            )

    def test_model_fairness(self, data, predictions, test_indices=None):
        """Тестируем справедливость модели"""
        print("\n⚖️ ТЕСТИРУЕМ СПРАВЕДЛИВОСТЬ МОДЕЛИ")

        try:
            # Создаем копию данных и добавляем предсказания
            data_with_preds = data.copy()

            if test_indices is None:
                # Если не указаны индексы, предполагаем что predictions соответствуют всему датасету
                data_with_preds['prediction'] = predictions
            else:
                # Если указаны индексы тестовой выборки
                data_with_preds['prediction'] = np.nan  # Инициализируем NaN
                # Заполняем только тестовые индексы
                data_with_preds.loc[test_indices, 'prediction'] = predictions

            # Удаляем строки с NaN в предсказаниях
            data_with_preds = data_with_preds.dropna(subset=['prediction'])

            # Группируем по возрастным группам
            data_with_preds['age_group'] = pd.cut(data_with_preds['age'],
                                                  bins=[0, 30, 50, 100],
                                                  labels=['young', 'middle', 'senior'])

            churn_rates = data_with_preds.groupby('age_group')['prediction'].mean()

            max_difference = churn_rates.max() - churn_rates.min()

            if max_difference < 0.2:  # Разница менее 20%
                self.log_test(
                    "MODEL_FAIRNESS_AGE",
                    "Проверка справедливости по возрасту",
                    True,
                    f"Разница в предсказаниях: {max_difference:.3f}"
                )
            else:
                self.log_test(
                    "MODEL_FAIRNESS_AGE",
                    "Проверка справедливости по возрасту",
                    False,
                    f"Большая разница в предсказаниях по возрастным группам: {max_difference:.3f}"
                )

        except Exception as e:
            self.log_test(
                "MODEL_FAIRNESS_AGE",
                "Проверка справедливости по возрасту",
                False,
                f"Ошибка: {e}"
            )

    def run_complete_test_suite(self):
        """Запускаем полный набор тестов"""
        print("🎯 ЗАПУСКАЕМ ПОЛНЫЙ ТЕСТ ML PIPELINE")
        print("=" * 60)

        # Генерируем данные
        data = self.pipeline.generate_sample_data(1000)

        # Обучаем модель
        X, y = self.pipeline.preprocess_data(data)
        X_test, y_test, y_pred = self.pipeline.train_model(X, y)

        # Запускаем все тесты
        self.test_data_quality(data)
        self.test_model_performance(X_test, y_test, y_pred)

        # Тестируем справедливость только на тестовых данных
        # Получаем индексы тестовых данных
        test_indices = X_test.index  # Индексы тестовой выборки
        self.test_model_fairness(data, y_pred, test_indices)

        self.test_api_functionality()

        # Тестируем дрифт (создаем "текущие" данные с небольшими изменениями)
        current_data = self.pipeline.generate_sample_data(200)
        self.test_data_drift(data, current_data)

        # Генерируем отчет
        return self.generate_test_report()

    def generate_test_report(self):
        """Генерируем отчет по тестированию"""
        print("\n📊 ГЕНЕРИРУЕМ ОТЧЕТ ПО ТЕСТИРОВАНИЮ")
        print("=" * 50)

        if not self.test_results:
            print("❌ Нет результатов тестирования")
            return None

        df = pd.DataFrame(self.test_results)

        # Статистика
        total_tests = len(df)
        passed_tests = df['success'].sum()
        success_rate = (passed_tests / total_tests) * 100

        print(f"🎯 ОБЩАЯ СТАТИСТИКА:")
        print(f"Всего тестов: {total_tests}")
        print(f"Пройдено: {passed_tests}")
        print(f"Успешность: {success_rate:.1f}%")

        # Детали по категориям тестов
        test_categories = df['test_name'].str.split('_').str[0].value_counts()
        print(f"\n📈 ТЕСТЫ ПО КАТЕГОРИЯМ:")
        for category, count in test_categories.items():
            category_success = df[df['test_name'].str.startswith(category)]['success'].sum()
            category_rate = (category_success / count) * 100
            print(f"  {category}: {category_success}/{count} ({category_rate:.1f}%)")

        # Неудачные тесты
        failed_tests = df[~df['success']]
        if not failed_tests.empty:
            print(f"\n🚨 НЕУДАЧНЫЕ ТЕСТЫ:")
            for _, test in failed_tests.iterrows():
                print(f"  ❌ {test['test_name']}: {test['description']}")
                if test['details']:
                    print(f"     📝 {test['details']}")

        # Сохраняем отчет
        report = {
            'timestamp': datetime.now().isoformat(),
            'total_tests': int(total_tests),
            'passed_tests': int(passed_tests),
            'success_rate': float(success_rate),
            'test_details': df.to_dict('records')
        }

        with open('ml_testing_report.json', 'w') as f:
            json.dump(report, f, indent=2)

        print(f"\n✅ Отчет сохранен в ml_testing_report.json")

        # Рекомендации
        if success_rate >= 80:
            print("🏆 ОТЛИЧНЫЙ РЕЗУЛЬТАТ: ML pipeline готов к продакшену!")
        elif success_rate >= 60:
            print("⚠️  ХОРОШИЙ РЕЗУЛЬТАТ: ML pipeline работает, но нужны улучшения")
        else:
            print("🚨 КРИТИЧЕСКИЙ РЕЗУЛЬТАТ: ML pipeline требует серьезной доработки")

        return report


# Пример использования
if __name__ == "__main__":
    print("🚀 Запуск ML Testing Framework")
    print("=" * 50)

    # Проверяем наличие зависимостей
    try:
        import pandas as pd

        print("✅ pandas установлен")
    except ImportError:
        print("❌ pandas не установлен")

    try:
        import requests

        print("✅ requests установлен")
    except ImportError:
        print("❌ requests не установлен")

    # Создаем тестер
    tester = MLTestingFramework(api_url="http://localhost:5001")

    # Запускаем тесты
    try:
        report = tester.run_complete_test_suite()
        print(f"\n🎉 Тестирование завершено!")
    except Exception as e:
        print(f"\n❌ Ошибка при выполнении тестов: {e}")