src/analysis/activity_analyzer.py - master - CW3/The_Project

src/analysis/activity_analyzer.py
193 строки8 KB
CW3
Initial project setup with structure and CI/CD
28 дек 2025, 12:40
28 дек 2025, 12:405cacb7a
"""
Анализ активности студентов на основе логов LMS
"""
import pandas as pd
import numpy as np
from typing import Dict, List, Tuple
from datetime import datetime, timedelta
from collections import defaultdict


class ActivityAnalyzer:
    """Анализатор активности студентов"""
    
    def __init__(self, logs_df: pd.DataFrame, grades_df: pd.DataFrame = None):
        """
        Инициализация анализатора
        
        Args:
            logs_df: DataFrame с логами
            grades_df: DataFrame с оценками (опционально)
        """
        self.logs_df = logs_df
        self.grades_df = grades_df
        self.logs_df['timestamp'] = pd.to_datetime(self.logs_df['timestamp'])
        
    def calculate_activity_metrics(self) -> pd.DataFrame:
        """
        Расчет метрик активности для каждого студента
        
        Returns:
            DataFrame с метриками активности
        """
        metrics = []
        
        for student_id in self.logs_df['student_id'].unique():
            student_logs = self.logs_df[self.logs_df['student_id'] == student_id]
            
            # Основные метрики
            total_events = len(student_logs)
            unique_days = student_logs['timestamp'].dt.date.nunique()
            avg_events_per_day = total_events / max(unique_days, 1)
            
            # Распределение по типам событий
            event_distribution = student_logs['event_type'].value_counts().to_dict()
            
            # Время активности
            time_stats = self._calculate_time_stats(student_logs)
            
            # Активность по курсам
            course_activity = student_logs['course_id'].value_counts().to_dict()
            
            metrics.append({
                'student_id': student_id,
                'total_events': total_events,
                'active_days': unique_days,
                'avg_events_per_day': avg_events_per_day,
                'event_distribution': event_distribution,
                'preferred_course': max(course_activity, key=course_activity.get) if course_activity else None,
                'time_stats': time_stats
            })
        
        return pd.DataFrame(metrics)
    
    def _calculate_time_stats(self, student_logs: pd.DataFrame) -> Dict:
        """Расчет статистики по времени активности"""
        if len(student_logs) < 2:
            return {}
        
        # Сортируем по времени
        sorted_logs = student_logs.sort_values('timestamp')
        
        # Время первого и последнего события
        first_event = sorted_logs.iloc[0]['timestamp']
        last_event = sorted_logs.iloc[-1]['timestamp']
        
        # Среднее время между событиями
        time_diffs = []
        for i in range(1, len(sorted_logs)):
            diff = (sorted_logs.iloc[i]['timestamp'] - sorted_logs.iloc[i-1]['timestamp']).total_seconds()
            time_diffs.append(diff)
        
        avg_time_between_events = np.mean(time_diffs) if time_diffs else 0
        
        # Предпочтительное время дня
        hours = sorted_logs['timestamp'].dt.hour
        preferred_hour = hours.mode()[0] if not hours.mode().empty else 12
        
        return {
            'first_event': first_event,
            'last_event': last_event,
            'total_duration_days': (last_event - first_event).days,
            'avg_time_between_events_minutes': avg_time_between_events / 60,
            'preferred_hour': preferred_hour
        }
    
    def identify_learning_patterns(self) -> pd.DataFrame:
        """
        Идентификация паттернов обучения
        
        Returns:
            DataFrame с паттернами студентов
        """
        metrics_df = self.calculate_activity_metrics()
        
        patterns = []
        
        for _, row in metrics_df.iterrows():
            pattern = self._classify_learning_pattern(row)
            patterns.append(pattern)
        
        metrics_df['learning_pattern'] = patterns
        return metrics_df
    
    def _classify_learning_pattern(self, student_metrics: pd.Series) -> str:
        """Классификация паттерна обучения"""
        event_dist = student_metrics['event_distribution']
        
        # Определяем доли типов активности
        total = sum(event_dist.values())
        if total == 0:
            return 'inactive'
        
        # Процентное соотношение
        view_ratio = (event_dist.get('course_view', 0) + event_dist.get('resource_view', 0)) / total
        interact_ratio = (event_dist.get('assignment_submit', 0) + event_dist.get('quiz_attempt', 0)) / total
        social_ratio = (event_dist.get('forum_post', 0) + event_dist.get('message_sent', 0)) / total
        
        # Классификация
        if view_ratio > 0.6:
            return 'passive_learner'
        elif interact_ratio > 0.4:
            return 'active_learner'
        elif social_ratio > 0.3:
            return 'social_learner'
        elif student_metrics['avg_events_per_day'] > 20:
            return 'intensive_learner'
        elif student_metrics['avg_events_per_day'] < 5:
            return 'sporadic_learner'
        else:
            return 'balanced_learner'
    
    def correlate_with_grades(self) -> pd.DataFrame:
        """
        Корреляция активности с успеваемостью
        
        Returns:
            DataFrame с корреляцией
        """
        if self.grades_df is None:
            raise ValueError("DataFrame с оценками не предоставлен")
        
        # Расчет метрик активности
        activity_metrics = self.calculate_activity_metrics()
        
        # Объединение с оценками
        correlation_data = pd.merge(
            activity_metrics,
            self.grades_df,
            on='student_id',
            how='inner'
        )
        
        # Расчет корреляций
        correlations = {}
        
        # Простые корреляции
        numeric_cols = ['total_events', 'active_days', 'avg_events_per_day']
        
        for col in numeric_cols:
            if col in correlation_data.columns and 'final_grade' in correlation_data.columns:
                corr = correlation_data[col].corr(correlation_data['final_grade'])
                correlations[col] = corr
        
        # Корреляция типов событий с оценками
        event_types = ['course_view', 'assignment_submit', 'quiz_attempt', 'forum_post']
        
        for event_type in event_types:
            # Подсчет событий этого типа для каждого студента
            event_counts = []
            for _, row in correlation_data.iterrows():
                count = row['event_distribution'].get(event_type, 0)
                event_counts.append(count)
            
            correlation_data[f'{event_type}_count'] = event_counts
            
            # Расчет корреляции
            if 'final_grade' in correlation_data.columns:
                corr = correlation_data[f'{event_type}_count'].corr(correlation_data['final_grade'])
                correlations[f'{event_type}_correlation'] = corr
        
        correlation_data['correlation_summary'] = str(correlations)
        
        return correlation_data