"""Evidence-based mastery engine — the single place mastery math lives. Pure, deterministic functions only (no DB, no AI, no randomness) so every rule is unit-testable and inspectable. Persistence stays in the EXISTING ``topic_mastery`` row: the numeric columns plus a versioned dict stored in its ``evidence`` JSON column (no schema migration needed). Callers (``adaptive_engine``) load/store rows; this module only computes. Documented policies ------------------- Evidence strength (0..1). Different actions are different proof: - finishing a lesson/video is weak evidence (0.15-0.20) - answering a recall prompt correctly is stronger (0.30) - a correct MCQ checkpoint is stronger (0.45) - a correct short/board answer is stronger (0.50-0.60) - solving a numerical is strongest (0.70) - a hint halves the strength of a correct answer - re-answering a question already answered correctly earns 25% strength (memorising one question must not prove mastery) Score: exponentially weighted toward the new result, where the learning rate scales with evidence strength (strong evidence moves the score more, weak evidence barely moves it). Wrong answers weigh 1.25x their strength so a recent failure genuinely reduces standing. Revision intervals (deterministic, not a memory-science claim): - needs_repair -> 1 day, developing -> 2 days - secure with confidence < 0.6 -> 4 days, secure otherwise -> 7 days - each consecutive successful recall beyond 2 multiplies the interval by 1.5, capped at 21 days; always clamped to finish at least 1 day before the exam. """ from __future__ import annotations from dataclasses import dataclass, field from datetime import date, datetime, timedelta from typing import Any, Literal MASTERY_ENGINE_VERSION = 2 EvidenceKind = Literal[ "lesson_completed", "video_completed", "recall_correct", "checkpoint_mcq", "checkpoint_short", "checkpoint_numerical", "checkpoint_diagram", "board_answer", "transfer_check", "revision_recall", "quiz_aggregate", ] EVIDENCE_STRENGTH: dict[str, float] = { "lesson_completed": 0.20, "video_completed": 0.15, "recall_correct": 0.30, "checkpoint_mcq": 0.45, "checkpoint_short": 0.50, "checkpoint_numerical": 0.70, "checkpoint_diagram": 0.50, "board_answer": 0.60, "transfer_check": 0.55, "revision_recall": 0.50, "quiz_aggregate": 0.50, } # Evidence kinds that count as an actual understanding check (a completed # lesson is exposure, not proof). CHECK_EVIDENCE_KINDS: frozenset[str] = frozenset( kind for kind in EVIDENCE_STRENGTH if kind not in {"lesson_completed", "video_completed"} ) HINT_STRENGTH_MULTIPLIER = 0.5 REPEAT_QUESTION_MULTIPLIER = 0.25 WRONG_ANSWER_WEIGHT = 1.25 BASE_LEARNING_RATE = 0.6 MIN_LEARNING_RATE = 0.05 MAX_LEARNING_RATE = 0.5 SECURE_SCORE = 75.0 SECURE_CONFIDENCE = 0.5 SECURE_CONSECUTIVE = 2 AT_RISK_OVERDUE_DAYS = 4 AT_RISK_EXAM_WINDOW_DAYS = 14 AT_RISK_SCORE = 60.0 MAX_LOG_ENTRIES = 20 MAX_SEEN_QUESTIONS = 40 MasteryState = Literal[ "not_started", "introduced", "developing", "needs_repair", "secure", "revision_due", "at_risk", ] ErrorCategory = Literal[ "concept_misunderstanding", "formula_selection", "unit_error", "sign_or_direction_error", "calculation_error", "diagram_interpretation", "incomplete_explanation", "missing_exam_keyword", "memorized_answer", ] @dataclass class EvidenceEvent: kind: str correct: bool at: datetime question_id: str | None = None hint_used: bool = False error_category: str | None = None time_spent_seconds: int | None = None source: str = "guided_class" @dataclass class MasterySnapshot: """The mutable mastery values for one (student, concept).""" score: float = 0.0 confidence: float = 0.0 attempts_count: int = 0 state: str = "not_started" next_review_at: datetime | None = None evidence: dict[str, Any] = field(default_factory=dict) @dataclass class MasteryUpdate: before_score: float after_score: float before_state: str after_state: str confidence: float consecutive_success: int applied_strength: float next_review_at: datetime | None evidence: dict[str, Any] attempts_count: int def _fresh_evidence() -> dict[str, Any]: return { "version": MASTERY_ENGINE_VERSION, "consecutive_success": 0, "hint_count": 0, "first_attempt_correct": 0, "first_attempt_total": 0, "error_categories": {}, "seen_question_correct": [], "log": [], "last_correct_at": None, "last_wrong_at": None, } def _normalized_evidence(raw: dict[str, Any] | None) -> dict[str, Any]: data = _fresh_evidence() if isinstance(raw, dict): for key in data: if key in raw and isinstance(raw[key], type(data[key]) if data[key] is not None else object): data[key] = raw[key] # tolerate legacy v1 payloads ({"last_score_percent": ...}) silently return data def effective_strength(event: EvidenceEvent, evidence: dict[str, Any]) -> float: """Strength of THIS event given history (hints, repeated questions).""" strength = EVIDENCE_STRENGTH.get(event.kind, 0.3) if event.hint_used and event.correct: strength *= HINT_STRENGTH_MULTIPLIER if ( event.correct and event.question_id and event.question_id in evidence.get("seen_question_correct", []) ): strength *= REPEAT_QUESTION_MULTIPLIER return round(strength, 4) def compute_state( *, score: float, confidence: float, consecutive_success: int, attempts_count: int, has_check_evidence: bool, has_open_repair: bool, next_review_at: datetime | None, now: datetime, exam_date: date | None, ) -> MasteryState: """Deterministic state resolution, priority top-down.""" if attempts_count == 0: return "not_started" if has_open_repair: return "needs_repair" if next_review_at is not None and now >= next_review_at: overdue_days = (now - next_review_at).days exam_close = ( exam_date is not None and (exam_date - now.date()).days <= AT_RISK_EXAM_WINDOW_DAYS ) if overdue_days > AT_RISK_OVERDUE_DAYS or (exam_close and score < AT_RISK_SCORE): return "at_risk" return "revision_due" if not has_check_evidence: return "introduced" if score >= SECURE_SCORE and confidence >= SECURE_CONFIDENCE and consecutive_success >= SECURE_CONSECUTIVE: return "secure" return "developing" def review_interval_days(state: str, confidence: float, consecutive_success: int) -> int: if state == "needs_repair": return 1 # repairs revise tomorrow, no streak credit if state in {"developing", "introduced"}: return 2 base = 4 if confidence < 0.6 else 7 extra_recalls = max(0, consecutive_success - SECURE_CONSECUTIVE) interval = base * (1.5 ** extra_recalls) return min(21, max(1, round(interval))) def clamp_review_to_exam(review_at: datetime, exam_date: date | None, now: datetime) -> datetime: if exam_date is None: return review_at latest = datetime.combine(exam_date - timedelta(days=1), datetime.min.time(), tzinfo=review_at.tzinfo) if review_at > latest: return max(now + timedelta(days=1), latest) return review_at def apply_evidence( snapshot: MasterySnapshot, event: EvidenceEvent, *, exam_date: date | None, has_open_repair: bool, ) -> MasteryUpdate: """Apply one evidence event to a mastery snapshot (mutates the snapshot).""" evidence = _normalized_evidence(snapshot.evidence) before_score = snapshot.score before_state = snapshot.state or "not_started" strength = effective_strength(event, evidence) target = 100.0 if event.correct else 0.0 weight = strength * (1.0 if event.correct else WRONG_ANSWER_WEIGHT) learning_rate = max(MIN_LEARNING_RATE, min(MAX_LEARNING_RATE, weight * BASE_LEARNING_RATE)) if snapshot.attempts_count == 0: # First evidence anchors the score instead of averaging against 0. after_score = target * max(strength, 0.2) else: after_score = snapshot.score * (1 - learning_rate) + target * learning_rate after_score = round(max(0.0, min(100.0, after_score)), 2) if event.correct: snapshot.confidence = round(min(1.0, snapshot.confidence + strength * 0.15), 4) else: snapshot.confidence = round(max(0.05, snapshot.confidence - 0.2), 4) is_check = event.kind in CHECK_EVIDENCE_KINDS if is_check: if event.correct: evidence["consecutive_success"] = int(evidence.get("consecutive_success", 0)) + 1 evidence["last_correct_at"] = event.at.isoformat() else: evidence["consecutive_success"] = 0 evidence["last_wrong_at"] = event.at.isoformat() evidence["first_attempt_total"] = int(evidence.get("first_attempt_total", 0)) + 1 if event.correct and not event.hint_used: evidence["first_attempt_correct"] = int(evidence.get("first_attempt_correct", 0)) + 1 if event.hint_used: evidence["hint_count"] = int(evidence.get("hint_count", 0)) + 1 if not event.correct and event.error_category: categories = dict(evidence.get("error_categories", {})) categories[event.error_category] = int(categories.get(event.error_category, 0)) + 1 evidence["error_categories"] = categories if event.correct and event.question_id: seen = list(evidence.get("seen_question_correct", [])) if event.question_id not in seen: seen.append(event.question_id) evidence["seen_question_correct"] = seen[-MAX_SEEN_QUESTIONS:] log = list(evidence.get("log", [])) log.append( { "kind": event.kind, "correct": event.correct, "strength": strength, "at": event.at.isoformat(), "question_id": event.question_id, "error_category": event.error_category if not event.correct else None, "hint_used": event.hint_used, "source": event.source, } ) evidence["log"] = log[-MAX_LOG_ENTRIES:] snapshot.score = after_score snapshot.attempts_count += 1 consecutive = int(evidence.get("consecutive_success", 0)) has_check_evidence = int(evidence.get("first_attempt_total", 0)) > 0 # Successful evidence schedules the next recall; failures schedule sooner. provisional_state = compute_state( score=snapshot.score, confidence=snapshot.confidence, consecutive_success=consecutive, attempts_count=snapshot.attempts_count, has_check_evidence=has_check_evidence, has_open_repair=has_open_repair, next_review_at=None, # review scheduling below uses the fresh state now=event.at, exam_date=exam_date, ) interval = review_interval_days(provisional_state, snapshot.confidence, consecutive) snapshot.next_review_at = clamp_review_to_exam( event.at + timedelta(days=interval), exam_date, event.at ) after_state = provisional_state snapshot.state = after_state evidence["version"] = MASTERY_ENGINE_VERSION snapshot.evidence = evidence return MasteryUpdate( before_score=round(before_score, 2), after_score=snapshot.score, before_state=before_state, after_state=after_state, confidence=snapshot.confidence, consecutive_success=consecutive, applied_strength=strength, next_review_at=snapshot.next_review_at, evidence=evidence, attempts_count=snapshot.attempts_count, ) # --------------------------------------------------------------------------- # Deterministic error classification # --------------------------------------------------------------------------- _NUMBER_PATTERN = r"-?\d+(?:[.,]\d+)?" _UNIT_TOKENS = ( "m/s", "ms-1", "hz", "khz", "hertz", "metre", "meter", "second", "cm", "km", "mm", "s", "m", ) _FORMULA_TOKENS = ("v=f", "v = f", "f=1/t", "f = 1/t", "t=1/f", "v=2d", "d=vt") def _extract_numbers(text: str) -> list[float]: import re values: list[float] = [] for raw in re.findall(_NUMBER_PATTERN, text.replace(",", ".")): try: values.append(float(raw)) except ValueError: continue return values def _contains_unit(text: str) -> bool: lowered = f" {text.casefold()} " return any(f" {unit}" in lowered or f"{unit} " in lowered for unit in _UNIT_TOKENS) def _keyword_coverage(answer: str, keywords: list[str]) -> tuple[int, int]: lowered = answer.casefold() matched = sum(1 for keyword in keywords if keyword.casefold() in lowered) return matched, len(keywords) def classify_error( *, question_type: str, student_answer: str, correct_answer: str, expected_keywords: list[str] | None = None, misconception: str | None = None, ) -> ErrorCategory: """Deterministic first-pass error classification. Not an AI judgement — a small, inspectable rule set. AI may LATER refine a category, but the stored category never comes from an unvalidated model. """ answer = (student_answer or "").strip() keywords = expected_keywords or [] if question_type == "diagram": return "diagram_interpretation" if question_type == "numerical": expected_numbers = _extract_numbers(correct_answer) student_numbers = _extract_numbers(answer) if expected_numbers and student_numbers: # Same magnitude somewhere in the answer -> the maths happened; look at units. if any(abs(sn - en) < 1e-6 for sn in student_numbers for en in expected_numbers): if _contains_unit(correct_answer) and not _contains_unit(answer): return "unit_error" return "unit_error" if not _contains_unit(answer) else "calculation_error" # Off by a power of ten -> almost always a unit conversion slip. for sn in student_numbers: for en in expected_numbers: if en != 0 and abs(sn / en) in (10.0, 100.0, 1000.0, 0.1, 0.01, 0.001): return "unit_error" lowered = answer.casefold().replace(" ", "") if any(token.replace(" ", "") in lowered for token in _FORMULA_TOKENS): return "calculation_error" return "formula_selection" if not student_numbers: return "formula_selection" return "calculation_error" if question_type == "mcq": return "concept_misunderstanding" # short / board-style answers matched, total = _keyword_coverage(answer, keywords) if total > 0 and matched == 0: return "concept_misunderstanding" if total > 0 and matched < total: return "missing_exam_keyword" if len(answer.split()) < 6: return "incomplete_explanation" return "incomplete_explanation" # --------------------------------------------------------------------------- # Student-facing copy per error category (deterministic diagnosis + activity) # --------------------------------------------------------------------------- ERROR_CATEGORY_COPY: dict[str, dict[str, str]] = { "concept_misunderstanding": { "diagnosis": "The core idea itself is not settled yet — this was not a careless slip.", "activity": "re_explain", "activity_prompt": "Reread the one explanation block for this idea, then answer one fresh check question in your own words.", }, "formula_selection": { "diagnosis": "The wrong formula (or no formula) was chosen for this situation.", "activity": "worked_solution", "activity_prompt": "Compare the formulas you know for this chapter and match each to the situation it serves, then redo one worked example.", }, "unit_error": { "diagnosis": "The physics was right but the units were not converted or written.", "activity": "scaffolded_numerical", "activity_prompt": "Solve one scaffolded numerical where the first step is converting every value to SI units before substituting.", }, "sign_or_direction_error": { "diagnosis": "A direction or sign flipped somewhere in the working.", "activity": "worked_solution", "activity_prompt": "Correct one worked solution line-by-line, marking where the direction changed.", }, "calculation_error": { "diagnosis": "Setup was right; the arithmetic slipped.", "activity": "scaffolded_numerical", "activity_prompt": "Redo the same style of numerical slowly, writing every substitution step.", }, "diagram_interpretation": { "diagnosis": "The diagram's labels or meaning were misread.", "activity": "label_diagram", "activity_prompt": "Label the key parts of this concept's diagram from memory, then check against the board.", }, "incomplete_explanation": { "diagnosis": "The idea is present but the answer stops before the marks are earned.", "activity": "board_answer_rewrite", "activity_prompt": "Rewrite the answer using the full answer structure — definition, reason, and conclusion.", }, "missing_exam_keyword": { "diagnosis": "The answer misses the exact keywords the board awards marks for.", "activity": "board_answer_rewrite", "activity_prompt": "Rewrite one board answer including every underlined keyword.", }, "memorized_answer": { "diagnosis": "The same memorised line is repeating without transfer to new questions.", "activity": "transfer_question", "activity_prompt": "Answer one unseen question on the same idea, phrased differently.", }, }