ml_testing_framework.py - master - maryllinn/lab12_ml_testing

ml_testing_framework.py
323 строки15 KB
Elvira Khamadalieva
first_commit
20 дек 2025, 13:23
20 дек 2025, 13:238472623
\
import json
from datetime import datetime
from typing import Any, Dict, List, Optional

import numpy as np
import pandas as pd
import requests

from ml_pipeline import MLPipeline

# Evidently может быть капризным по версиям — делаем импорт безопасным
try:
    from evidently.test_suite import TestSuite
    from evidently.tests import (
        TestNumberOfRows,
        TestNumberOfColumns,
        TestColumnsType,
        TestAllColumnsShareOfMissingValues,
        TestNumColumnsMean,
        TestCatColumnsShareOfMissingValues,
    )

    EVIDENTLY_AVAILABLE = True
except Exception:
    EVIDENTLY_AVAILABLE = False


class MLTestingFramework:
    def __init__(self, api_url: str = "http://localhost:5000") -> None:
        self.pipeline = MLPipeline()
        self.api_url = api_url
        self.test_results: List[Dict[str, Any]] = []

    def log_test(self, test_name: str, description: str, success: bool, details: Optional[str] = None) -> None:
        test_result = {
            "test_name": test_name,
            "description": description,
            "success": bool(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: pd.DataFrame) -> None:
        print("\n🔍 ТЕСТИРУЕМ КАЧЕСТВО ДАННЫХ")

        required_columns = ["age", "tenure", "monthly_charges", "contract_type"]
        missing_columns = [c for c in required_columns if c 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)

        # выбросы (IQR) для числовых колонок
        numerical_columns = data.select_dtypes(include=[np.number]).columns
        outlier_issues = []
        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 = Q1 - 1.5 * IQR
            upper = Q3 + 1.5 * IQR
            outliers = data[(data[col] < lower) | (data[col] > upper)]
            pct = len(outliers) / len(data) * 100
            if pct > 5:
                outlier_issues.append(f"{col}: {pct:.1f}%")

        if outlier_issues:
            self.log_test("OUTLIERS", "Проверка выбросов", False, ", ".join(outlier_issues))
        else:
            self.log_test("OUTLIERS", "Проверка выбросов", True)

        if "churn" in data.columns:
            churn_dist = data["churn"].value_counts(normalize=True)
            minority = float(churn_dist.min())
            if minority < 0.2:
                self.log_test("TARGET_DISTRIBUTION", "Проверка распределения целевой переменной", False, str(dict(churn_dist)))
            else:
                self.log_test("TARGET_DISTRIBUTION", "Проверка распределения целевой переменной", True, str(dict(churn_dist)))
        else:
            self.log_test("TARGET_DISTRIBUTION", "Проверка распределения целевой переменной", False, "Колонка churn отсутствует")

    def test_model_performance(self, y_true: pd.Series, y_pred: np.ndarray) -> None:
        print("\n🎯 ТЕСТИРУЕМ ПРОИЗВОДИТЕЛЬНОСТЬ МОДЕЛИ")
        from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score

        accuracy = float(accuracy_score(y_true, y_pred))
        precision = float(precision_score(y_true, y_pred, zero_division=0))
        recall = float(recall_score(y_true, y_pred, zero_division=0))
        f1 = float(f1_score(y_true, y_pred, zero_division=0))

        self.log_test("MODEL_ACCURACY", "Проверка точности модели", accuracy >= 0.7, f"Accuracy: {accuracy:.3f}")
        self.log_test("MODEL_PRECISION", "Проверка precision модели", precision >= 0.6, f"Precision: {precision:.3f}")
        self.log_test("MODEL_RECALL", "Проверка recall модели", recall >= 0.5, f"Recall: {recall:.3f}")
        self.log_test("MODEL_F1", "Проверка F1-score модели", f1 >= 0.6, f"F1-score: {f1:.3f}")

    def test_data_drift(self, reference_data: pd.DataFrame, current_data: pd.DataFrame) -> None:
        print("\n📊 ТЕСТИРУЕМ ДРИФТ ДАННЫХ")

        # вариант 1: Evidently
        if EVIDENTLY_AVAILABLE:
            try:
                suite = TestSuite(
                    tests=[
                        TestNumberOfRows(),
                        TestNumberOfColumns(),
                        TestColumnsType(),
                        TestAllColumnsShareOfMissingValues(),
                        TestNumColumnsMean(),
                        TestCatColumnsShareOfMissingValues(),
                    ]
                )
                suite.run(reference_data=reference_data, current_data=current_data)
                all_passed = bool(suite.as_dict()["summary"]["all_passed"])
                if all_passed:
                    self.log_test("DATA_DRIFT", "Проверка дрифта данных (Evidently)", True, "Дрифт не обнаружен")
                else:
                    failed = [t["name"] for t in suite.as_dict()["tests"] if t.get("status") == "FAIL"]
                    self.log_test("DATA_DRIFT", "Проверка дрифта данных (Evidently)", False, f"FAIL: {failed}")
                return
            except Exception as e:
                # упадем в fallback
                self.log_test("DATA_DRIFT", "Проверка дрифта данных (Evidently)", False, f"Ошибка Evidently: {e}")

        # вариант 2: простой статистический дрейф (без scipy)
        try:
            num_cols = [c for c in reference_data.columns if pd.api.types.is_numeric_dtype(reference_data[c]) and c != "churn"]
            if not num_cols:
                self.log_test("DATA_DRIFT", "Проверка дрифта данных (fallback)", True, "Нет числовых колонок для сравнения")
                return

            scores = []
            for c in num_cols:
                ref_mean, cur_mean = float(reference_data[c].mean()), float(current_data[c].mean())
                ref_std = float(reference_data[c].std() + 1e-9)
                scores.append(abs(cur_mean - ref_mean) / ref_std)

            drift_score = float(np.mean(scores))
            # условный порог
            ok = drift_score < 0.25
            self.log_test("DATA_DRIFT", "Проверка дрифта данных (fallback)", ok, f"Drift score: {drift_score:.3f}")
        except Exception as e:
            self.log_test("DATA_DRIFT", "Проверка дрифта данных (fallback)", False, f"Ошибка: {e}")

    def test_api_functionality(self) -> None:
        print("\n🌐 ТЕСТИРУЕМ API ФУНКЦИОНАЛЬНОСТЬ")

        # health
        try:
            resp = requests.get(f"{self.api_url}/health", timeout=3)
            if resp.status_code == 200 and resp.json().get("status") == "healthy":
                self.log_test("API_HEALTH", "Проверка health check API", True, None)
            else:
                self.log_test("API_HEALTH", "Проверка health check API", False, f"{resp.status_code}: {resp.text}")
        except Exception as e:
            self.log_test("API_HEALTH", "Проверка health check API", False, f"Ошибка подключения: {e}")
            # если health не прошел — остальное скорее всего бессмысленно
            return

        # single predict
        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,
        }
        try:
            resp = requests.post(f"{self.api_url}/predict", json=test_customer, timeout=5)
            if resp.status_code == 200 and "predictions" in resp.json():
                self.log_test("API_PREDICTION", "Проверка предсказания через API", True, str(resp.json()["predictions"]))
            else:
                self.log_test("API_PREDICTION", "Проверка предсказания через API", False, f"{resp.status_code}: {resp.text}")
        except Exception as e:
            self.log_test("API_PREDICTION", "Проверка предсказания через API", False, f"Ошибка: {e}")

        # batch predict
        payload = {
            "customers": [
                {
                    "customer_id": "BATCH_TEST_001",
                    "age": 30,
                    "tenure": 12,
                    "monthly_charges": 65.50,
                    "total_charges": 786.00,
                    "contract_type": "Monthly",
                    "payment_method": "Electronic Check",
                    "paperless_billing": 1,
                    "dependents": 0,
                    "partner": 0,
                    "online_security": 0,
                    "tech_support": 0,
                    "monthly_usage_gb": 150,
                    "customer_service_calls": 5,
                },
                {
                    "customer_id": "BATCH_TEST_002",
                    "age": 55,
                    "tenure": 48,
                    "monthly_charges": 45.00,
                    "total_charges": 2160.00,
                    "contract_type": "Two-Year",
                    "payment_method": "Bank Transfer",
                    "paperless_billing": 0,
                    "dependents": 1,
                    "partner": 1,
                    "online_security": 1,
                    "tech_support": 1,
                    "monthly_usage_gb": 200,
                    "customer_service_calls": 0,
                },
            ]
        }
        try:
            resp = requests.post(f"{self.api_url}/batch_predict", json=payload, timeout=5)
            if resp.status_code == 200 and "predictions" in resp.json() and len(resp.json()["predictions"]) == 2:
                self.log_test("API_BATCH_PREDICTION", "Проверка батчевого предсказания через API", True, f"OK: {resp.json().get('total_customers')}")
            else:
                self.log_test("API_BATCH_PREDICTION", "Проверка батчевого предсказания через API", False, f"{resp.status_code}: {resp.text}")
        except Exception as e:
            self.log_test("API_BATCH_PREDICTION", "Проверка батчевого предсказания через API", False, f"Ошибка: {e}")

    def test_model_fairness_age(self, data: pd.DataFrame) -> None:
        """
        Простейшая проверка fairness:
        сравниваем средний churn-pred по age-группам.
        """
        print("\n⚖️ ТЕСТИРУЕМ СПРАВЕДЛИВОСТЬ МОДЕЛИ")

        try:
            if self.pipeline.model is None:
                # обучим quick модель на этих же данных
                X, y = self.pipeline.preprocess_data(data)
                self.pipeline.train_model(X, y)

            X_all, _ = self.pipeline.preprocess_data(data)
            preds = self.pipeline.model.predict(X_all)

            df = data.copy()
            df["prediction"] = preds
            df["age_group"] = pd.cut(df["age"], bins=[0, 30, 50, 100], labels=["young", "middle", "senior"])

            churn_rates = df.groupby("age_group")["prediction"].mean()
            max_diff = float(churn_rates.max() - churn_rates.min())

            ok = max_diff < 0.2
            self.log_test("MODEL_FAIRNESS_AGE", "Проверка справедливости по возрасту", ok, f"Δ={max_diff:.3f}; rates={dict(churn_rates)}")
        except Exception as e:
            self.log_test("MODEL_FAIRNESS_AGE", "Проверка справедливости по возрасту", False, f"Ошибка: {e}")

    def generate_test_report(self, save_path: str = "ml_testing_report.json") -> Dict[str, Any]:
        print("\n📊 ГЕНЕРИРУЕМ ОТЧЕТ ПО ТЕСТИРОВАНИЮ")

        df = pd.DataFrame(self.test_results)
        total = int(len(df))
        passed = int(df["success"].sum()) if total else 0
        success_rate = float((passed / total * 100) if total else 0)

        report = {
            "timestamp": datetime.now().isoformat(),
            "total_tests": total,
            "passed_tests": passed,
            "success_rate": success_rate,
            "test_details": self.test_results,
        }

        with open(save_path, "w", encoding="utf-8") as f:
            json.dump(report, f, ensure_ascii=False, indent=2)

        print(f"✅ Отчет сохранен в {save_path}")
        print(f"🎯 Успешность: {success_rate:.1f}% ({passed}/{total})")
        return report

    def run_complete_test_suite(self) -> Dict[str, Any]:
        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(y_test, y_pred)
        self.test_model_fairness_age(data)

        # API тесты (могут упасть, если API не запущен — это нормально)
        self.test_api_functionality()

        # drift: сравним с "текущими" данными
        current_data = self.pipeline.generate_sample_data(200)
        self.test_data_drift(data, current_data)

        return self.generate_test_report()


if __name__ == "__main__":
    tester = MLTestingFramework()
    tester.run_complete_test_suite()