encoach_backend_v4/custom_addons/encoach_placement/services/cat_engine.py

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