test_project.py - master - dayekb/vibecoding_ML

test_project.py
302 строки11 KB
Your Name
Initial commit: Complete ML course project with EDA, pipelines, modeling, and advanced techniques
26 авг 2025, 20:23
26 авг 2025, 20:23774682e
#!/usr/bin/env python3
"""Тестовый скрипт для проверки работы всех компонентов ML проекта.

Этот скрипт тестирует:
1. Импорт всех модулей
2. Создание синтетических данных
3. EDA анализ
4. Пайплайны обработки данных
5. Моделирование
6. Продвинутые техники
"""

import sys
import os
import numpy as np
import pandas as pd
from sklearn.datasets import make_classification, make_regression
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score, mean_squared_error

# Добавляем src в путь
sys.path.append(os.path.join(os.path.dirname(__file__), 'src'))

def test_imports():
    """Тестируем импорт всех модулей."""
    print("🔍 Тестируем импорты...")

    try:
        # Основные модули
        from src.data.eda import DataExplorer
        from src.pipelines.data_pipeline import DataPipeline, ModelPipeline
        from src.models.model_training import ModelFactory, ModelComparator
        from src.utils.mlflow_tracker import MLflowTracker, ExperimentManager
        from src.utils.advanced_ml import (
            ImbalancedDataHandler, CustomMetrics,
            ProbabilityCalibrator, EnsembleBuilder
        )
        from configs.config import TEST_SIZE, RANDOM_STATE

        print("✅ Все импорты успешны!")
        return True

    except ImportError as e:
        print(f"❌ Ошибка импорта: {e}")
        return False

def test_data_generation():
    """Тестируем генерацию данных."""
    print("\n📊 Генерируем тестовые данные...")

    try:
        # Создаем данные для классификации
        X_cls, y_cls = make_classification(
            n_samples=1000, n_features=20, n_informative=15,
            n_redundant=5, random_state=RANDOM_STATE
        )

        # Создаем данные для регрессии
        X_reg, y_reg = make_regression(
            n_samples=1000, n_features=20, n_informative=15,
            noise=0.1, random_state=RANDOM_STATE
        )

        # Преобразуем в DataFrame
        feature_names = [f'feature_{i}' for i in range(X_cls.shape[1])]

        df_cls = pd.DataFrame(X_cls, columns=feature_names)
        df_cls['target'] = y_cls

        df_reg = pd.DataFrame(X_reg, columns=feature_names)
        df_reg['target'] = y_reg

        print(f"✅ Данные классификации: {df_cls.shape}")
        print(f"✅ Данные регрессии: {df_reg.shape}")

        return df_cls, df_reg

    except Exception as e:
        print(f"❌ Ошибка генерации данных: {e}")
        return None, None

def test_eda(df_cls):
    """Тестируем EDA функциональность."""
    print("\n🔍 Тестируем EDA...")

    try:
        from src.data.eda import DataExplorer

        explorer = DataExplorer(df_cls, target_column='target')

        # Получаем обзор данных
        overview = explorer.get_data_overview()
        print(f"✅ Обзор данных: {overview['shape']} строк и столбцов")

        # Анализ пропусков
        missing_stats = explorer.analyze_missing_values()
        print(f"✅ Анализ пропусков: {len(missing_stats)} признаков")

        # Корреляционная матрица
        corr_matrix = explorer.plot_correlation_matrix()
        if corr_matrix is not None:
            print("✅ Корреляционная матрица построена")
        else:
            print("⚠️  Корреляционная матрица не построена (мало числовых признаков)")

        return True

    except Exception as e:
        print(f"❌ Ошибка EDA: {e}")
        return False

def test_pipeline(df_cls):
    """Тестируем пайплайны обработки данных."""
    print("\n🔧 Тестируем пайплайны...")

    try:
        from src.pipelines.data_pipeline import DataPipeline, ModelPipeline
        from sklearn.ensemble import RandomForestClassifier

        # Разделяем данные
        X = df_cls.drop('target', axis=1)
        y = df_cls['target']
        X_train, X_test, y_train, y_test = train_test_split(
            X, y, test_size=0.2, random_state=RANDOM_STATE
        )

        # Создаем пайплайн
        data_pipeline = DataPipeline()
        model = RandomForestClassifier(n_estimators=10, random_state=RANDOM_STATE)
        model_pipeline = ModelPipeline(model)

        # Кросс-валидация
        cv_results = model_pipeline.train_with_cv(X_train, y_train, cv_folds=3)
        print(f"✅ CV Accuracy: {cv_results['mean_score']:.3f} (+/- {cv_results['std_score']:.3f})")

        # Финальное обучение
        final_results = model_pipeline.train_final_model(X_train, y_train, X_test, y_test)
        print(f"✅ Test Accuracy: {final_results['test_score']:.3f}")
        return True

    except Exception as e:
        print(f"❌ Ошибка пайплайна: {e}")
        return False

def test_modeling(df_cls):
    """Тестируем моделирование."""
    print("\n🤖 Тестируем моделирование...")

    try:
        from src.models.model_training import ModelFactory, ModelComparator

        # Разделяем данные
        X = df_cls.drop('target', axis=1)
        y = df_cls['target']
        X_train, X_test, y_train, y_test = train_test_split(
            X, y, test_size=0.2, random_state=RANDOM_STATE
        )

        # Создаем модели
        models = [
            ModelFactory.create_logistic_regression(),
            ModelFactory.create_random_forest('classification')
        ]

        # Сравниваем модели
        comparator = ModelComparator(models, task_type='classification')
        comparison_results = comparator.compare_models(X_train, y_train, X_test, y_test)

        print(f"✅ Сравнение моделей: {len(comparison_results)} моделей")
        for model_name in comparison_results.keys():
            accuracy = comparison_results[model_name]['test_metrics']['accuracy']
            print(f"  - {model_name}: {accuracy:.3f}")
        return True

    except Exception as e:
        print(f"❌ Ошибка моделирования: {e}")
        return False

def test_advanced_techniques(df_cls):
    """Тестируем продвинутые техники."""
    print("\n🚀 Тестируем продвинутые техники...")

    try:
        from src.utils.advanced_ml import ImbalancedDataHandler, CustomMetrics
        from sklearn.linear_model import LogisticRegression

        # Разделяем данные
        X = df_cls.drop('target', axis=1)
        y = df_cls['target']
        X_train, X_test, y_train, y_test = train_test_split(
            X, y, test_size=0.2, random_state=RANDOM_STATE
        )

        # Создаем искусственный дисбаланс
        np.random.seed(RANDOM_STATE)
        mask = np.random.choice(len(y_train), size=int(len(y_train) * 0.9), replace=False)
        y_train.iloc[mask] = 0  # Делаем 90% класс 0

        print(f"Распределение классов: {y_train.value_counts().to_dict()}")

        # Обработка несбалансированных данных
        imbalance_handler = ImbalancedDataHandler(strategy='smote')
        X_resampled, y_resampled = imbalance_handler.apply_sampling(X_train, y_train)
        print(f"После SMOTE: {len(X_resampled)} сэмплов")

        # Кастомные метрики
        custom_metrics = CustomMetrics()

        # Обучаем модель на сбалансированных данных
        model = LogisticRegression(random_state=RANDOM_STATE)
        model.fit(X_resampled, y_resampled)
        y_pred = model.predict(X_test)

        cost_accuracy = custom_metrics.cost_sensitive_accuracy(y_test, y_pred)
        print(f"✅ Cost-sensitive accuracy: {cost_accuracy:.3f}")
        return True

    except Exception as e:
        print(f"❌ Ошибка продвинутых техник: {e}")
        return False

def test_mlflow_integration(df_cls):
    """Тестируем интеграцию с MLflow."""
    print("\n📊 Тестируем MLflow интеграцию...")

    try:
        from src.utils.mlflow_tracker import MLflowTracker, ExperimentManager
        from sklearn.ensemble import RandomForestClassifier

        # Разделяем данные
        X = df_cls.drop('target', axis=1)
        y = df_cls['target']
        X_train, X_test, y_train, y_test = train_test_split(
            X, y, test_size=0.2, random_state=RANDOM_STATE
        )

        # Инициализируем MLflow
        tracker = MLflowTracker(experiment_name="test_experiment")
        experiment_manager = ExperimentManager(tracker)

        # Обучаем модель
        model = RandomForestClassifier(n_estimators=10, random_state=RANDOM_STATE)
        model.fit(X_train, y_train)
        y_pred = model.predict(X_test)
        accuracy = accuracy_score(y_test, y_pred)

        # Логируем эксперимент
        run_id = experiment_manager.log_experiment(
            model_name="RandomForest_Test",
            model=model,
            X_train=X_train, y_train=y_train,
            X_test=X_test, y_test=y_test,
            params={'n_estimators': 10, 'random_state': RANDOM_STATE},
            metrics={'accuracy': accuracy},
            task_type='classification'
        )

        print(f"✅ Эксперимент залогирован: {run_id}")
        return True

    except Exception as e:
        print(f"❌ Ошибка MLflow: {e}")
        return False

def main():
    """Основная функция тестирования."""
    print("🚀 Начинаем тестирование ML проекта...\n")

    # Тест 1: Импорты
    if not test_imports():
        print("❌ Критическая ошибка: не удалось импортировать модули")
        return

    # Тест 2: Генерация данных
    df_cls, df_reg = test_data_generation()
    if df_cls is None or df_reg is None:
        print("❌ Критическая ошибка: не удалось сгенерировать данные")
        return

    # Тест 3: EDA
    test_eda(df_cls)

    # Тест 4: Пайплайны
    test_pipeline(df_cls)

    # Тест 5: Моделирование
    test_modeling(df_cls)

    # Тест 6: Продвинутые техники
    test_advanced_techniques(df_cls)

    # Тест 7: MLflow (опционально, может требовать установки)
    try:
        test_mlflow_integration(df_cls)
    except ImportError:
        print("\n⚠️  MLflow не установлен, пропускаем тест интеграции")

    print("\n🎉 Тестирование завершено!")
    print("📝 Проверьте результаты выше для каждого компонента")

if __name__ == "__main__":
    main()