# 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