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

src/analysis/recommendation.py
184 строки9 KB
CW3
Initial project setup with structure and CI/CD
28 дек 2025, 12:40
28 дек 2025, 12:405cacb7a
"""
Модуль для генерации рекомендаций по оптимизации пути обучения
"""
import pandas as pd
import numpy as np
from typing import Dict, List, Any
from sklearn.ensemble import RandomForestRegressor


class LearningPathRecommender:
    """Генератор рекомендаций для оптимизации обучения"""
    
    RECOMMENDATION_TEMPLATES = {
        'passive_learner': [
            "Увеличьте активность по сдаче заданий",
            "Попробуйте пройти дополнительные тесты",
            "Рекомендуется чаще входить в систему",
            "Участвуйте в форумных обсуждениях"
        ],
        'active_learner': [
            "Продолжайте активную работу с заданиями",
            "Попробуйте больше участвовать в форумных обсуждениях",
            "Рекомендуется просмотреть дополнительные материалы курса",
            "Помогайте другим студентам на форуме"
        ],
        'social_learner': [
            "Используйте свои социальные навыки для групповой работы",
            "Попробуйте вести блог о своем обучении",
            "Организуйте учебную группу",
            "Делитесь полезными материалами с другими"
        ],
        'intensive_learner': [
            "Делайте перерывы между учебными сессиями",
            "Попробуйте разные форматы материалов",
            "Рекомендуется планировать учебное время",
            "Обращайте внимание на качество, а не только количество"
        ],
        'sporadic_learner': [
            "Создайте регулярное расписание занятий",
            "Установите напоминания о дедлайнах",
            "Начинайте с небольших ежедневных целей",
            "Используйте мобильное приложение LMS для постоянного доступа"
        ],
        'balanced_learner': [
            "Продолжайте текущий подход к обучению",
            "Попробуйте новые типы учебных активностей",
            "Экспериментируйте с разным временем занятий",
            "Делитесь своим успешным опытом с другими"
        ]
    }
    
    def __init__(self, activity_metrics: pd.DataFrame, 
                 correlation_data: pd.DataFrame = None):
        """
        Инициализация рекомендательной системы
        
        Args:
            activity_metrics: DataFrame с метриками активности
            correlation_data: DataFrame с корреляциями (опционально)
        """
        self.activity_metrics = activity_metrics
        self.correlation_data = correlation_data
        self.model = None
    
    def generate_recommendations(self) -> Dict[int, Dict[str, Any]]:
        """
        Генерация персонализированных рекомендаций
        
        Returns:
            Словарь с рекомендациями для каждого студента
        """
        recommendations = {}
        
        for _, student in self.activity_metrics.iterrows():
            student_id = student['student_id']
            
            # Определяем паттерн обучения
            pattern = student.get('learning_pattern', 'balanced_learner')
            
            # Выбираем рекомендации по паттерну
            pattern_recs = self.RECOMMENDATION_TEMPLATES.get(
                pattern, 
                self.RECOMMENDATION_TEMPLATES['balanced_learner']
            )
            
            # Персонализируем рекомендации на основе метрик
            personalized_recs = self._personalize_recommendations(
                pattern_recs, student
            )
            
            # Предсказываем оценку (если есть данные для обучения)
            predicted_grade = None
            if self.correlation_data is not None and 'final_grade' in self.correlation_data.columns:
                predicted_grade = self._predict_grade(student)
            
            recommendations[student_id] = {
                'pattern': pattern,
                'predicted_grade': predicted_grade,
                'recommendations': personalized_recs,
                'metrics': {
                    'total_events': student.get('total_events'),
                    'active_days': student.get('active_days'),
                    'avg_events_per_day': student.get('avg_events_per_day')
                }
            }
        
        return recommendations
    
    def _personalize_recommendations(self, base_recommendations: List[str], 
                                   student_data: pd.Series) -> List[str]:
        """Персонализация рекомендаций на основе данных студента"""
        personalized = []
        
        # Анализируем распределение событий
        event_dist = student_data.get('event_distribution', {})
        
        # Если мало событий определенного типа, рекомендуем увеличить
        if event_dist.get('assignment_submit', 0) < 5:
            personalized.append("Увеличьте количество сдаваемых заданий")
        
        if event_dist.get('forum_post', 0) < 2:
            personalized.append("Попробуйте участвовать в обсуждениях на форуме")
        
        if event_dist.get('resource_view', 0) < 10:
            personalized.append("Изучайте больше учебных материалов")
        
        # Добавляем базовые рекомендации
        personalized.extend(base_recommendations[:3])  # Берем 3 основные
        
        return personalized[:5]  # Ограничиваем 5 рекомендациями
    
    def _predict_grade(self, student_data: pd.Series) -> float:
        """Предсказание оценки на основе активности"""
        # Простая модель на основе корреляций
        if self.correlation_data is None:
            return None
        
        # Используем среднюю оценку студентов с похожим паттерном
        pattern = student_data.get('learning_pattern', 'balanced_learner')
        
        if 'learning_pattern' in self.correlation_data.columns:
            pattern_students = self.correlation_data[
                self.correlation_data['learning_pattern'] == pattern
            ]
            
            if len(pattern_students) > 0 and 'final_grade' in pattern_students.columns:
                return pattern_students['final_grade'].mean()
        
        return None
    
    def train_prediction_model(self):
        """Обучение модели для предсказания успеваемости"""
        if self.correlation_data is None:
            return None
        
        # Подготовка признаков
        X = []
        y = []
        
        for _, student in self.correlation_data.iterrows():
            # Извлекаем признаки из метрик активности
            features = [
                student.get('total_events', 0),
                student.get('active_days', 0),
                student.get('avg_events_per_day', 0),
                student.get('course_view_count', 0),
                student.get('assignment_submit_count', 0),
                student.get('quiz_attempt_count', 0),
                student.get('forum_post_count', 0)
            ]
            
            X.append(features)
            
            # Целевая переменная - оценка
            if 'final_grade' in student:
                y.append(student['final_grade'])
        
        if len(X) > 10 and len(y) > 10:
            self.model = RandomForestRegressor(n_estimators=100, random_state=42)
            self.model.fit(X, y)
            
            return self.model
        
        return None