File size: 7,852 Bytes
5775a1b
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
# 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