src/analyzer.py - master - Str1ke/TheProject

src/analyzer.py
165 строк6 KB
AlekseyF7
Initial commit: Student Performance Analyzer project
26 дек 2025, 13:51
26 дек 2025, 13:5184047cc
"""
Модуль для анализа успеваемости студентов
"""

import pandas as pd
from typing import Dict, List, Any, Optional
from .data_loader import DataLoader


class PerformanceAnalyzer:
    """Класс для анализа успеваемости студентов"""

    def __init__(self, data_loader: DataLoader):
        """
        Инициализация анализатора

        Args:
            data_loader: Экземпляр DataLoader с загруженными данными
        """
        self.data_loader = data_loader
        if data_loader.data is None:
            raise ValueError("Данные не загружены в DataLoader")

    def calculate_overall_statistics(self) -> Dict[str, Any]:
        """
        Вычисляет общую статистику по всем студентам

        Returns:
            Словарь с общей статистикой
        """
        data = self.data_loader.data
        return {
            'total_students': data['student_id'].nunique(),
            'total_subjects': data['subject'].nunique(),
            'total_records': len(data),
            'overall_average': float(data['score'].mean()),
            'overall_median': float(data['score'].median()),
            'overall_std': float(data['score'].std()),
            'max_score': float(data['score'].max()),
            'min_score': float(data['score'].min())
        }

    def identify_at_risk_students(self, threshold: float = 3.0) -> List[Dict[str, Any]]:
        """
        Определяет студентов с низкой успеваемостью

        Args:
            threshold: Пороговое значение среднего балла

        Returns:
            Список словарей с информацией о студентах группы риска
        """
        summary = self.data_loader.get_all_students_summary()
        at_risk = summary[summary['average_score'] < threshold]

        result = []
        for _, row in at_risk.iterrows():
            student_stats = self.data_loader.get_student_stats(row['student_id'])
            result.append({
                'student_id': row['student_id'],
                'average_score': float(row['average_score']),
                'subjects_count': int(row['subjects_count']),
                'weak_subjects': student_stats['subjects']
            })

        return result

    def get_subject_statistics(self) -> pd.DataFrame:
        """
        Получает статистику по каждому предмету

        Returns:
            DataFrame со статистикой по предметам
        """
        data = self.data_loader.data
        subject_stats = data.groupby('subject').agg({
            'score': ['mean', 'median', 'std', 'count'],
            'student_id': 'nunique'
        }).reset_index()

        subject_stats.columns = [
            'subject', 'average_score', 'median_score',
            'std_score', 'total_records', 'students_count'
        ]

        return subject_stats

    def get_performance_trends(self, student_id: Optional[str] = None) -> Dict[str, Any]:
        """
        Анализирует тренды успеваемости

        Args:
            student_id: ID студента (если None, то по всем студентам)

        Returns:
            Словарь с информацией о трендах
        """
        data = self.data_loader.data.copy()

        if student_id:
            data = data[data['student_id'] == student_id]
            if data.empty:
                return {'student_id': student_id, 'trend': 'no_data'}

        # Если есть колонка с датой, можно анализировать тренды по времени
        # Для простоты анализируем по предметам
        if 'date' in data.columns:
            data['date'] = pd.to_datetime(data['date'])
            data = data.sort_values('date')
            trend_scores = data.groupby(data['date'].dt.to_period('M'))['score'].mean()
            trend_direction = 'improving' if trend_scores.iloc[-1] > trend_scores.iloc[0] else 'declining'
        else:
            # Простой анализ: сравниваем средние по группам предметов
            trend_direction = 'stable'

        return {
            'student_id': student_id if student_id else 'all',
            'trend': trend_direction,
            'current_average': float(data['score'].mean())
        }

    def generate_recommendations(self, student_id: str) -> List[str]:
        """
        Генерирует рекомендации для студента

        Args:
            student_id: ID студента

        Returns:
            Список рекомендаций
        """
        student_stats = self.data_loader.get_student_stats(student_id)
        recommendations = []

        if student_stats['average_score'] < 3.0:
            recommendations.append(
                f"Ваш средний балл ({student_stats['average_score']:.2f}) ниже среднего. "
                "Рекомендуется обратиться к преподавателям за дополнительной помощью."
            )

        if student_stats['max_score'] - student_stats['min_score'] > 2.0:
            recommendations.append(
                "Обнаружена большая разница между максимальным и минимальным баллами. "
                "Рекомендуется сосредоточиться на предметах с низкими оценками."
            )

        weak_subjects = [
            subj for subj in student_stats['subjects']
            if self.data_loader.data[
                (self.data_loader.data['student_id'] == student_id) &
                (self.data_loader.data['subject'] == subj)
            ]['score'].values[0] < 3.0
        ]

        if weak_subjects:
            recommendations.append(
                f"Предметы, требующие внимания: {', '.join(weak_subjects)}"
            )

        if not recommendations:
            recommendations.append(
                "Отличная работа! Продолжайте в том же духе."
            )

        return recommendations