src/analyzer.py - master - mssnle/learning-path-analyzer

src/analyzer.py
255 строк11 KB
mssnle
create src/analyzer.py
13 дек 2025, 17:10
13 дек 2025, 17:108355022
"""
Модуль анализа данных обучения
"""

import pandas as pd
import numpy as np
from typing import Dict, List, Any, Optional
from datetime import datetime, timedelta
from sklearn.linear_model import LinearRegression
from sklearn.ensemble import RandomForestRegressor
from sklearn.preprocessing import LabelEncoder
import warnings
warnings.filterwarnings('ignore')


class LearningPathAnalyzer:
    """Анализатор пути обучения"""
    
    def __init__(self, data: pd.DataFrame):
        """
        Инициализация анализатора
        
        Args:
            data: DataFrame с логами LMS
        """
        self.data = data.copy()
        self.results_cache = {}
        
    def analyze_student_performance(self, student_id: str) -> Dict[str, Any]:
        """
        Анализ успеваемости конкретного студента
        
        Args:
            student_id: ID студента
            
        Returns:
            Словарь с результатами анализа
        """
        student_data = self.data[self.data['student_id'] == student_id].copy()
        
        if student_data.empty:
            raise ValueError(f"Студент {student_id} не найден в данных")
        
        # Базовые метрики
        total_activities = len(student_data)
        unique_days = student_data['date'].nunique()
        
        # Расчет успеваемости
        if 'score' in student_data.columns:
            overall_score = student_data['score'].mean()
            score_std = student_data['score'].std()
            score_trend = self._calculate_score_trend(student_data)
        else:
            overall_score = np.nan
            score_std = np.nan
            score_trend = 0
        
        # Анализ активностей
        activity_distribution = student_data['activity_type'].value_counts().to_dict()
        
        # Расчет корреляций
        correlations = self._calculate_activity_correlations(student_data)
        
        # Определение эффективных активностей
        effective_activities = self._identify_effective_activities(correlations)
        
        # Предсказание риска отчисления
        dropout_risk = self._predict_dropout_risk(student_data)
        
        # Формирование результатов
        results = {
            'student_id': student_id,
            'overall_score': float(overall_score),
            'score_std': float(score_std),
            'score_trend': float(score_trend),
            'total_activities': int(total_activities),
            'unique_days': int(unique_days),
            'activity_distribution': activity_distribution,
            'correlations': correlations,
            'effective_activities': effective_activities,
            'dropout_risk': float(dropout_risk),
            'analysis_date': datetime.now().isoformat(),
        }
        
        self.results_cache[student_id] = results
        return results
    
    def _calculate_score_trend(self, student_data: pd.DataFrame) -> float:
        """Расчет тренда успеваемости"""
        if 'date' not in student_data.columns or 'score' not in student_data.columns:
            return 0
            
        student_data = student_data.sort_values('date')
        
        if len(student_data) < 2:
            return 0
            
        # Линейная регрессия для определения тренда
        dates_numeric = pd.to_numeric(pd.to_datetime(student_data['date']))
        X = dates_numeric.values.reshape(-1, 1)
        y = student_data['score'].values
        
        model = LinearRegression()
        model.fit(X, y)
        
        return model.coef_[0] * 1e9  # Масштабирование для читаемости
    
    def _calculate_activity_correlations(self, student_data: pd.DataFrame) -> Dict[str, float]:
        """Расчет корреляции между активностями и успеваемостью"""
        correlations = {}
        
        if 'score' not in student_data.columns:
            return correlations
        
        activity_types = student_data['activity_type'].unique()
        
        for activity in activity_types:
            activity_scores = student_data[student_data['activity_type'] == activity]['score']
            
            if len(activity_scores) > 1:
                # Для упрощения считаем корреляцию со средним баллом
                all_scores = student_data['score']
                activity_mask = student_data['activity_type'] == activity
                
                # Расчет корреляции Пирсона
                if len(activity_scores) > 1 and activity_scores.std() > 0:
                    correlation = np.corrcoef(
                        activity_mask.astype(int),
                        student_data['score']
                    )[0, 1]
                    correlations[activity] = float(correlation)
        
        return correlations
    
    def _identify_effective_activities(self, correlations: Dict[str, float]) -> List[Dict[str, Any]]:
        """Определение наиболее эффективных активностей"""
        effective = []
        
        for activity, correlation in correlations.items():
            if correlation > 0.3:  # Порог значимости
                effectiveness = "высокая" if correlation > 0.6 else "средняя"
                effective.append({
                    'name': activity,
                    'correlation': correlation,
                    'effectiveness': effectiveness,
                    'recommendation': self._generate_activity_recommendation(activity, correlation)
                })
        
        # Сортировка по корреляции
        effective.sort(key=lambda x: x['correlation'], reverse=True)
        return effective
    
    def _predict_dropout_risk(self, student_data: pd.DataFrame) -> float:
        """Предсказание риска отчисления"""
        # Упрощенная модель на основе признаков
        features = {
            'recent_activity': self._calculate_recent_activity(student_data),
            'score_consistency': self._calculate_score_consistency(student_data),
            'activity_variety': self._calculate_activity_variety(student_data),
        }
        
        # Веса признаков (можно заменить на ML модель)
        weights = {
            'recent_activity': 0.4,
            'score_consistency': 0.4,
            'activity_variety': 0.2,
        }
        
        # Нормализация признаков
        normalized_features = {
            'recent_activity': max(0, min(1, features['recent_activity'])),
            'score_consistency': max(0, min(1, features['score_consistency'])),
            'activity_variety': max(0, min(1, features['activity_variety'])),
        }
        
        # Расчет риска (чем ниже значения, тем выше риск)
        risk_score = sum(
            (1 - normalized_features[feature]) * weight
            for feature, weight in weights.items()
        )
        
        return risk_score
    
    def _calculate_recent_activity(self, student_data: pd.DataFrame) -> float:
        """Расчет недавней активности"""
        if 'date' not in student_data.columns:
            return 0.5
            
        student_data['date'] = pd.to_datetime(student_data['date'])
        latest_date = student_data['date'].max()
        
        # Активность за последние 7 дней
        week_ago = latest_date - timedelta(days=7)
        recent_activities = student_data[student_data['date'] >= week_ago]
        
        if len(student_data) == 0:
            return 0
            
        return len(recent_activities) / len(student_data)
    
    def _calculate_score_consistency(self, student_data: pd.DataFrame) -> float:
        """Расчет стабильности оценок"""
        if 'score' not in student_data.columns or len(student_data) < 2:
            return 0.5
            
        scores = student_data['score']
        consistency = 1 - (scores.std() / scores.mean() if scores.mean() > 0 else 1)
        
        return max(0, min(1, consistency))
    
    def _calculate_activity_variety(self, student_data: pd.DataFrame) -> float:
        """Расчет разнообразия активностей"""
        if 'activity_type' not in student_data.columns:
            return 0.5
            
        total_activities = len(student_data)
        unique_activities = student_data['activity_type'].nunique()
        
        if total_activities == 0:
            return 0
            
        variety = unique_activities / total_activities
        return min(1, variety * 3)  # Масштабирование
    
    def _generate_activity_recommendation(self, activity: str, correlation: float) -> str:
        """Генерация рекомендации для активности"""
        recommendations = {
            'video_lecture': {
                'high': "Продолжайте смотреть видео-лекции, они хорошо влияют на вашу успеваемость",
                'medium': "Видео-лекции помогают, но попробуйте комбинировать с другими активностями",
                'low': "Рассмотрите другие форматы обучения"
            },
            'quiz': {
                'high': "Тесты отлично работают для вас, продолжайте в том же духе",
                'medium': "Регулярные тесты помогают закрепить материал",
                'low': "Попробуйте больше практиковаться перед тестами"
            },
            'assignment': {
                'high': "Задания очень эффективны для вашего обучения",
                'medium': "Задания помогают, но обратите внимание на обратную связь",
                'low': "Проанализируйте ошибки в заданиях для улучшения"
            }
        }
        
        if activity in recommendations:
            if correlation > 0.6:
                level = 'high'
            elif correlation > 0.3:
                level = 'medium'
            else:
                level = 'low'
            
            return recommendations[activity].get(level, "Нет конкретной рекомендации")
        
        return f"Активность '{activity}' имеет корреляцию {correlation:.2f} с успеваемостью"