Dawid-Skene 算法¶

class DawidSkene(BaseClassificationAggregator):
    # 省略部分代码
    def fit(
        self,
        data: pd.DataFrame,
        true_labels: Optional["pd.Series[Any]"] = None,
        initial_error: Optional[pd.DataFrame] = None,
    ) -> "DawidSkene":
        # 省略部分代码
        # Initialization
        probas = MajorityVote().fit_predict_proba(data)  # (1)
        # correct the probas by true_labels
        if true_labels is not None:
            probas = self._correct_probas_with_golden(probas, true_labels)
        priors = probas.mean()  # (2)
        errors = self._m_step(data, probas, initial_error, self.initial_error_strategy)  # (3)
        loss = -np.inf
        self.loss_history_ = []

        # Updating proba and errors n_iter times
        for _ in range(self.n_iter):
            probas = self._e_step(data, priors, errors)  # (4)
            # correct the probas by true_labels
            if true_labels is not None:
                probas = self._correct_probas_with_golden(probas, true_labels)
            priors = probas.mean()  # (5)
            errors = self._m_step(data, probas)  # (6)
            new_loss = self._evidence_lower_bound(data, probas, priors, errors) / len(
                data
            )
            self.loss_history_.append(new_loss)

            if new_loss - loss < self.tol:
                break
            loss = new_loss
        # 省略部分代码

class DawidSkene(BaseClassificationAggregator):
    # 省略部分代码
    def fit(
        self,
        data: pd.DataFrame,
        true_labels: Optional["pd.Series[Any]"] = None,
        initial_error: Optional[pd.DataFrame] = None,
    ) -> "DawidSkene":
        # 省略部分代码
        # Initialization
        probas = MajorityVote().fit_predict_proba(data)
        # correct the probas by true_labels  # (1)
        if true_labels is not None:
            probas = self._correct_probas_with_golden(probas, true_labels)
        priors = probas.mean()
        errors = self._m_step(data, probas, initial_error, self.initial_error_strategy)  # (3)
        loss = -np.inf
        self.loss_history_ = []

        # Updating proba and errors n_iter times
        for _ in range(self.n_iter):
            probas = self._e_step(data, priors, errors)
            # correct the probas by true_labels  # (2)
            if true_labels is not None:
                probas = self._correct_probas_with_golden(probas, true_labels)
            priors = probas.mean()
        # 省略部分代码

class DawidSkene(BaseClassificationAggregator):
    # 省略部分代码
    initial_error_strategy: Optional[Literal["assign", "addition"]] = attr.ib(default=None)  # (1)
    # 省略部分代码
    def fit(
        self,
        data: pd.DataFrame,
        true_labels: Optional["pd.Series[Any]"] = None,
        initial_error: Optional[pd.DataFrame] = None,  # (2)
    ) -> "DawidSkene":
        # 省略部分代码
        # Initialization
        probas = MajorityVote().fit_predict_proba(data)
        # correct the probas by true_labels  # (1)
        if true_labels is not None:
            probas = self._correct_probas_with_golden(probas, true_labels)
        priors = probas.mean()
        errors = self._m_step(data, probas, initial_error, self.initial_error_strategy)  # (3)
        loss = -np.inf
        self.loss_history_ = []
        # 省略部分代码

class DawidSkene(BaseClassificationAggregator):
    # 省略部分代码
    @staticmethod
    def _m_step(
        data: pd.DataFrame,
        probas: pd.DataFrame,
        initial_error: Optional[pd.DataFrame] = None,
        initial_error_strategy: Optional[Literal["assign", "addition"]] = None,
    ) -> pd.DataFrame:
        """Performs M-step of the Dawid-Skene algorithm.

        Estimates the workers' error probability matrix using the specified workers' responses and the true task label probabilities.
        """
        joined = data.join(probas, on="task")
        joined.drop(columns=["task"], inplace=True)
        errors = joined.groupby(["worker", "label"], sort=False).sum()
        # Apply the initial error matrix
        errors = initial_error_apply(errors, initial_error, initial_error_strategy)
        # Normalize the error matrix
        errors.clip(lower=_EPS, inplace=True)
        errors /= errors.groupby("worker", sort=False).sum()

        return errors

def initial_error_apply(
    errors: pd.DataFrame,
    initial_error: Optional[pd.DataFrame],
    initial_error_strategy: Optional[Literal["assign", "addition"]],
) -> pd.DataFrame:
    if initial_error_strategy is None or initial_error is None:
        return errors
    # check the index names of initial_error
    if initial_error.index.names != errors.index.names:  # (1)
        raise ValueError(
            f"The index of initial_error must be: {errors.index.names},"
            f"but got: {initial_error.index.names}"
        )
    if initial_error_strategy == "assign":
        # check the completeness of initial_error: all the workers in data should be in initial_error
        mask = errors.index.isin(initial_error.index)
        if not mask.all():  # (2)
            not_found_workers = errors.index[~mask].get_level_values("worker").unique()
            raise ValueError(
                f"All the workers in data should be in initial_error: "
                f"Can not find {len(not_found_workers)} workers' error matrix in initial_error"
            )
        # if the values in initial_error are probability, check the sum of each worker's error matrix
        if (initial_error <= 1.0).all().all() and not np.allclose(
            initial_error.groupby("worker", sort=False).sum(), 1.0
        ):  # (3)
            raise ValueError(
                "The sum of each worker's error matrix in initial_error should be 1.0"
            )
        errors = initial_error
    elif initial_error_strategy == "addition":  # (4)
        errors = errors.add(initial_error, axis="index", fill_value=0.0)
    else:
        raise ValueError(
            f"Invalid initial_error_strategy: {initial_error_strategy},"
            f"should be 'assign' or 'addition'"
        )
    return errors