feat(v3): restructure project + add complete frontend
- Restructure: move backend from new_project/ to backend/ - Add full React/TypeScript frontend (37 pages, 17 services, 16 type defs, 11 query hooks) - Add docs/ with SRS specs, user stories, and workflow documentation - Update .gitignore for new directory layout Workflows implemented: WF1 User Signup, WF2 Placement Test, WF3 Exam Configuration, WF4 General English Exam, WF5 Course Generation, WF6 Entity Student Onboarding, AI Course Generation, Adaptive Learning Engine UI, White-Label Branding, Score Release Made-with: Cursor
This commit is contained in:
159
custom_addons/encoach_placement/services/cat_engine.py
Normal file
159
custom_addons/encoach_placement/services/cat_engine.py
Normal file
@@ -0,0 +1,159 @@
|
||||
import math
|
||||
import logging
|
||||
|
||||
_logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class CatEngine:
|
||||
"""Computerized Adaptive Testing engine using IRT 3PL model."""
|
||||
|
||||
SEM_THRESHOLD = 0.3
|
||||
MAX_QUESTIONS = 40
|
||||
MIN_QUESTIONS = 10
|
||||
THETA_MIN = -4.0
|
||||
THETA_MAX = 4.0
|
||||
|
||||
@staticmethod
|
||||
def irt_probability(theta, a, b, c):
|
||||
"""3PL IRT probability of correct response."""
|
||||
exponent = -a * (theta - b)
|
||||
exponent = max(min(exponent, 20), -20)
|
||||
return c + (1 - c) / (1 + math.exp(exponent))
|
||||
|
||||
@staticmethod
|
||||
def fisher_information(theta, a, b, c):
|
||||
"""Fisher information for a single item at given theta."""
|
||||
p = CatEngine.irt_probability(theta, a, b, c)
|
||||
q = 1 - p
|
||||
if p <= c or q <= 0:
|
||||
return 0.0
|
||||
numerator = a**2 * (p - c)**2 * q
|
||||
denominator = (1 - c)**2 * p
|
||||
if denominator == 0:
|
||||
return 0.0
|
||||
return numerator / denominator
|
||||
|
||||
@staticmethod
|
||||
def select_next_question(theta, available_questions):
|
||||
"""Select the question with maximum Fisher information at current theta.
|
||||
|
||||
available_questions: list of dicts with keys: id, irt_a, irt_b, irt_c
|
||||
Returns: the selected question dict, or None if empty.
|
||||
"""
|
||||
if not available_questions:
|
||||
return None
|
||||
|
||||
best_question = None
|
||||
best_info = -1.0
|
||||
|
||||
for q in available_questions:
|
||||
a = q.get('irt_a', 1.0)
|
||||
b = q.get('irt_b', 0.0)
|
||||
c = q.get('irt_c', 0.25)
|
||||
info = CatEngine.fisher_information(theta, a, b, c)
|
||||
if info > best_info:
|
||||
best_info = info
|
||||
best_question = q
|
||||
|
||||
return best_question
|
||||
|
||||
@staticmethod
|
||||
def update_theta(theta, responses, questions):
|
||||
"""Update theta using Newton-Raphson MLE.
|
||||
|
||||
responses: list of 0/1 (incorrect/correct)
|
||||
questions: list of dicts with irt_a, irt_b, irt_c
|
||||
Returns: updated theta, standard error of measurement
|
||||
"""
|
||||
if not responses:
|
||||
return theta, 1.0
|
||||
|
||||
current_theta = theta
|
||||
for _iteration in range(30):
|
||||
numerator = 0.0
|
||||
denominator = 0.0
|
||||
|
||||
for resp, q in zip(responses, questions):
|
||||
a = q.get('irt_a', 1.0)
|
||||
b = q.get('irt_b', 0.0)
|
||||
c = q.get('irt_c', 0.25)
|
||||
|
||||
p = CatEngine.irt_probability(current_theta, a, b, c)
|
||||
q_val = 1 - p
|
||||
|
||||
if p <= c:
|
||||
continue
|
||||
|
||||
w = a * (p - c) / ((1 - c) * p)
|
||||
numerator += w * (resp - p)
|
||||
denominator += w**2 * p * q_val
|
||||
|
||||
if abs(denominator) < 1e-10:
|
||||
break
|
||||
|
||||
delta = numerator / denominator
|
||||
current_theta += delta
|
||||
current_theta = max(CatEngine.THETA_MIN, min(CatEngine.THETA_MAX, current_theta))
|
||||
|
||||
if abs(delta) < 0.001:
|
||||
break
|
||||
|
||||
total_info = 0.0
|
||||
for q in questions:
|
||||
a = q.get('irt_a', 1.0)
|
||||
b = q.get('irt_b', 0.0)
|
||||
c = q.get('irt_c', 0.25)
|
||||
total_info += CatEngine.fisher_information(current_theta, a, b, c)
|
||||
|
||||
sem = 1.0 / math.sqrt(total_info) if total_info > 0 else 1.0
|
||||
|
||||
return current_theta, sem
|
||||
|
||||
@staticmethod
|
||||
def check_stopping(sem, questions_answered):
|
||||
"""Check if CAT should stop.
|
||||
Returns: (should_stop, reason)
|
||||
"""
|
||||
if questions_answered >= CatEngine.MAX_QUESTIONS:
|
||||
return True, 'max_questions_reached'
|
||||
if sem < CatEngine.SEM_THRESHOLD and questions_answered >= CatEngine.MIN_QUESTIONS:
|
||||
return True, 'sem_converged'
|
||||
return False, None
|
||||
|
||||
@staticmethod
|
||||
def estimate_ability_eap(responses, questions, prior_mean=0.0, prior_sd=1.0, num_points=61):
|
||||
"""Expected A Posteriori (EAP) ability estimation.
|
||||
More robust than MLE for short tests.
|
||||
"""
|
||||
theta_range = [
|
||||
CatEngine.THETA_MIN + i * (CatEngine.THETA_MAX - CatEngine.THETA_MIN) / (num_points - 1)
|
||||
for i in range(num_points)
|
||||
]
|
||||
|
||||
log_posteriors = []
|
||||
for theta in theta_range:
|
||||
log_likelihood = 0.0
|
||||
for resp, q in zip(responses, questions):
|
||||
a = q.get('irt_a', 1.0)
|
||||
b = q.get('irt_b', 0.0)
|
||||
c = q.get('irt_c', 0.25)
|
||||
p = CatEngine.irt_probability(theta, a, b, c)
|
||||
p = max(min(p, 0.9999), 0.0001)
|
||||
if resp == 1:
|
||||
log_likelihood += math.log(p)
|
||||
else:
|
||||
log_likelihood += math.log(1 - p)
|
||||
|
||||
log_prior = -0.5 * ((theta - prior_mean) / prior_sd) ** 2
|
||||
log_posteriors.append(log_likelihood + log_prior)
|
||||
|
||||
max_log = max(log_posteriors)
|
||||
posteriors = [math.exp(lp - max_log) for lp in log_posteriors]
|
||||
total = sum(posteriors)
|
||||
posteriors = [p / total for p in posteriors]
|
||||
|
||||
eap = sum(t * p for t, p in zip(theta_range, posteriors))
|
||||
eap_var = sum(p * (t - eap) ** 2 for t, p in zip(theta_range, posteriors))
|
||||
sem = math.sqrt(eap_var)
|
||||
|
||||
return eap, sem
|
||||
Reference in New Issue
Block a user