modules/adaptive_engine.py

# modules/adaptive_engine.py
"""Adaptive difficulty engine for personalized learning"""

import json
import os
from typing import Dict, List, Optional
from datetime import datetime, timedelta

# Constants for adaptive logic
PERFORMANCE_HISTORY_LENGTH = 100
RECENT_HISTORY_WINDOW = 10
LEARNING_RATE_ALPHA = 0.2

# Thresholds for difficulty adjustment
DIFFICULTY_INCREASE_THRESHOLD = 0.8  # Recent success rate to increase difficulty
DIFFICULTY_DECREASE_THRESHOLD = 0.4  # Recent success rate to decrease difficulty
DIFFICULTY_SCORE_THRESHOLD_HARD = 0.7  # Medium score to recommend Hard
DIFFICULTY_SCORE_THRESHOLD_MEDIUM = 0.6  # Medium score to recommend Medium


class AdaptiveEngine:
    def __init__(self, user_id: str):
        self.user_id = user_id
        self.performance_file = f"data/performance_{user_id}.json"
        self.load_performance_data()

    def load_performance_data(self):
        """Load user performance data"""
        if os.path.exists(self.performance_file):
            with open(self.performance_file, "r") as f:
                self.data = json.load(f)
        else:
            self.data = {
                "performance_history": [],
                "difficulty_scores": {"Easy": 0.8, "Medium": 0.5, "Hard": 0.2},
                "topic_performance": {},
                "last_updated": datetime.now().isoformat(),
            }

    def save_performance_data(self):
        """Save performance data"""
        os.makedirs("data", exist_ok=True)
        self.data["last_updated"] = datetime.now().isoformat()
        with open(self.performance_file, "w") as f:
            json.dump(self.data, f, indent=2)

    def update_performance(self, is_correct: bool, difficulty: str, topic: str = None):
        """Update performance based on quiz results"""
        # Record performance
        performance_entry = {
            "timestamp": datetime.now().isoformat(),
            "correct": is_correct,
            "difficulty": difficulty,
            "topic": topic,
        }

        self.data["performance_history"].append(performance_entry)

        # Update difficulty scores using exponential moving average
        alpha = LEARNING_RATE_ALPHA
        score = 1.0 if is_correct else 0.0

        old_score = self.data["difficulty_scores"].get(difficulty, 0.5)
        new_score = alpha * score + (1 - alpha) * old_score
        self.data["difficulty_scores"][difficulty] = new_score

        # Update topic performance
        if topic:
            if topic not in self.data["topic_performance"]:
                self.data["topic_performance"][topic] = {
                    "attempts": 0,
                    "correct": 0,
                    "last_seen": None,
                }

            self.data["topic_performance"][topic]["attempts"] += 1
            if is_correct:
                self.data["topic_performance"][topic]["correct"] += 1
            self.data["topic_performance"][topic]["last_seen"] = (
                datetime.now().isoformat()
            )

        # Keep only recent history (last 100 entries)
        if len(self.data["performance_history"]) > PERFORMANCE_HISTORY_LENGTH:
            self.data["performance_history"] = self.data["performance_history"][
                -PERFORMANCE_HISTORY_LENGTH:
            ]

        self.save_performance_data()

    def get_recommended_difficulty(self) -> str:
        """Get recommended difficulty based on performance"""
        scores = self.data["difficulty_scores"]

        # Calculate recent performance (last 10 attempts)
        recent_history = self.data["performance_history"][-RECENT_HISTORY_WINDOW:]
        if recent_history:
            recent_success_rate = sum(1 for h in recent_history if h["correct"]) / len(
                recent_history
            )
        else:
            return "Easy"  # Default to Easy if no history

        # Decision logic
        if recent_success_rate > DIFFICULTY_INCREASE_THRESHOLD:
            # Doing great, increase difficulty
            if scores["Medium"] > DIFFICULTY_SCORE_THRESHOLD_HARD:
                return "Hard"
            else:
                return "Medium"
        elif recent_success_rate < DIFFICULTY_DECREASE_THRESHOLD:
            # Struggling, decrease difficulty
            return "Easy"
        else:
            # Balanced performance
            if scores["Medium"] > DIFFICULTY_SCORE_THRESHOLD_MEDIUM:
                return "Medium"
            else:
                return "Easy"

    def get_weak_topics(self, limit: int = 5) -> List[str]:
        """Get topics where user needs more practice"""
        weak_topics = []

        for topic, performance in self.data["topic_performance"].items():
            if performance["attempts"] > 0:
                success_rate = performance["correct"] / performance["attempts"]
                if success_rate < 0.6:
                    weak_topics.append((topic, success_rate))

        # Sort by success rate (ascending)
        weak_topics.sort(key=lambda x: x[1])

        return [topic for topic, _ in weak_topics[:limit]]

    def get_strong_topics(self, limit: int = 5) -> List[str]:
        """Get topics where user excels"""
        strong_topics = []

        for topic, performance in self.data["topic_performance"].items():
            if performance["attempts"] >= 3:  # Minimum attempts
                success_rate = performance["correct"] / performance["attempts"]
                if success_rate > 0.8:
                    strong_topics.append((topic, success_rate))

        # Sort by success rate (descending)
        strong_topics.sort(key=lambda x: x[1], reverse=True)

        return [topic for topic, _ in strong_topics[:limit]]

    def should_review_topic(self, topic: str) -> bool:
        """Determine if a topic needs review based on spaced repetition"""
        if topic not in self.data["topic_performance"]:
            return False

        performance = self.data["topic_performance"][topic]

        # Check last seen date
        if performance["last_seen"]:
            last_seen = datetime.fromisoformat(performance["last_seen"])
            days_since = (datetime.now() - last_seen).days

            # Spaced repetition intervals based on performance
            success_rate = (
                performance["correct"] / performance["attempts"]
                if performance["attempts"] > 0
                else 0
            )

            if success_rate < 0.5:
                review_interval = 1  # Review daily
            elif success_rate < 0.7:
                review_interval = 3  # Review every 3 days
            elif success_rate < 0.9:
                review_interval = 7  # Review weekly
            else:
                review_interval = 14  # Review bi-weekly

            return days_since >= review_interval

        return True

    def get_performance_summary(self) -> Dict:
        """Get overall performance summary"""
        total_attempts = len(self.data["performance_history"])
        total_correct = sum(1 for h in self.data["performance_history"] if h["correct"])

        summary = {
            "total_attempts": total_attempts,
            "total_correct": total_correct,
            "overall_success_rate": total_correct / total_attempts
            if total_attempts > 0
            else 0,
            "difficulty_mastery": self.data["difficulty_scores"],
            "topics_studied": len(self.data["topic_performance"]),
            "recommended_difficulty": self.get_recommended_difficulty(),
            "weak_topics": self.get_weak_topics(3),
            "strong_topics": self.get_strong_topics(3),
        }

        return summary