src/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},
                "item_performance": {},  # Changed from 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, item_id: str = None, topic: str = None):
        """Update performance based on quiz results or flashcard interaction"""
        # Record overall performance history
        performance_entry = {
            "timestamp": datetime.now().isoformat(),
            "correct": is_correct,
            "difficulty": difficulty,
            "item_id": item_id, # Store item_id if provided
            "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 individual item performance (for spaced repetition)
        if item_id:
            if item_id not in self.data["item_performance"]:
                self.data["item_performance"][item_id] = {
                    "attempts": 0,
                    "correct": 0,
                    "last_seen": None,
                    "topic": topic, # Store topic with item for context
                }

            self.data["item_performance"][item_id]["attempts"] += 1
            if is_correct:
                self.data["item_performance"][item_id]["correct"] += 1
            self.data["item_performance"][item_id]["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 (based on item performance)"""
        weak_topics = {} # Use dict to aggregate performance by topic

        for item_id, performance in self.data["item_performance"].items():
            topic = performance.get("topic")
            if topic:
                if topic not in weak_topics:
                    weak_topics[topic] = {"attempts": 0, "correct": 0}
                weak_topics[topic]["attempts"] += performance["attempts"]
                weak_topics[topic]["correct"] += performance["correct"]
        
        sorted_weak_topics = []
        for topic, agg_performance in weak_topics.items():
            if agg_performance["attempts"] > 0:
                success_rate = agg_performance["correct"] / agg_performance["attempts"]
                if success_rate < 0.6:
                    sorted_weak_topics.append((topic, success_rate))

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

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

    def get_strong_topics(self, limit: int = 5) -> List[str]:
        """Get topics where user excels (based on item performance)"""
        strong_topics = {} # Use dict to aggregate performance by topic

        for item_id, performance in self.data["item_performance"].items():
            topic = performance.get("topic")
            if topic:
                if topic not in strong_topics:
                    strong_topics[topic] = {"attempts": 0, "correct": 0}
                strong_topics[topic]["attempts"] += performance["attempts"]
                strong_topics[topic]["correct"] += performance["correct"]
        
        sorted_strong_topics = []
        for topic, agg_performance in strong_topics.items():
            if agg_performance["attempts"] >= 3:  # Minimum attempts for strong topic
                success_rate = agg_performance["correct"] / agg_performance["attempts"]
                if success_rate > 0.8:
                    sorted_strong_topics.append((topic, success_rate))

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

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

    def should_review_item(self, item_id: str) -> bool:
        """Determine if an item (flashcard) needs review based on spaced repetition"""
        if item_id not in self.data["item_performance"]:
            return True # New item, should be reviewed

        performance = self.data["item_performance"][item_id]

        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 # Should be reviewed if no last_seen date

    def get_items_due_for_review(self, topic: str = None, limit: int = 5) -> List[str]:
        """Get item_ids that are due for review for a given topic or all topics"""
        review_items = []
        for item_id, performance in self.data["item_performance"].items():
            if (topic is None or performance.get("topic") == topic) and self.should_review_item(item_id):
                review_items.append(item_id)
        
        # Prioritize items with lower success rates
        review_items.sort(key=lambda item_id: self.data["item_performance"][item_id]["correct"] / self.data["item_performance"][item_id]["attempts"] if self.data["item_performance"][item_id]["attempts"] > 0 else 0)
        
        return review_items[:limit]


    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"],
            "items_studied": len(self.data["item_performance"]), # Changed from topics_studied
            "recommended_difficulty": self.get_recommended_difficulty(),
            "weak_topics": self.get_weak_topics(3),
            "strong_topics": self.get_strong_topics(3),
        }

        return summary