<commit_before>"""
Expected error reduction framework for active learning.
"""

from typing import Tuple

import numpy as np

from scipy.stats import entropy

from sklearn.base import clone
from sklearn.exceptions import NotFittedError

from modAL.models import ActiveLearner
from modAL.utils.data import modALinput, data_vstack
from modAL.utils.selection import multi_argmax


def expected_error_reduction(learner: ActiveLearner, X: modALinput,
                             p_subsample: np.float = 1.0, n_instances: int = 1) -> Tuple[np.ndarray, modALinput]:
    """
    Expected error reduction query strategy.

    References:
        <NAME> McCallum, 2001 (http://groups.csail.mit.edu/rrg/papers/icml01.pdf)

    Args:
        learner: The ActiveLearner object for which the expected error is to be estimated.
        X: The samples.
        p_subsample: Probability of keeping a sample from the pool when calculating expected error.
            Significantly improves runtime for large sample pools.
        n_instances: The number of instances to be sampled.


    Returns:
        The indices of the instances from X chosen to be labelled; the instances from X chosen to be labelled.
    """

    assert 0.0 <= p_subsample <= 1.0, 'p_subsample subsampling keep ratio must be between 0.0 and 1.0'

    expected_error = np.zeros(shape=(len(X), ))
    possible_labels = np.unique(learner.y_training)

    try:
        X_proba = learner.predict_proba(X)
    except NotFittedError:
        # TODO: implement a proper cold-start
        return 0, X[0]

    for x_idx, x in enumerate(X):
        # subsample the data if needed
        if np.random.rand() <= p_subsample:
            # estimate the expected error
            for y_idx, y in enumerate(possible_labels):
                X_new = data_vstack((learner.X_training, x.reshape(1, -1)))
                y_new = data_vstack((learner.y_training, np.array(y).reshape(1, )))

                refitted_estimator = clone(learner.estimator).fit(X_new, y_new)
                uncertainty = 1 - np.max(refitted_estimator.predict_proba(X), axis=1)

                expected_error[x_idx] += np.sum(uncertainty)*X_proba[x_idx, y_idx]

        else:
            expected_error[x_idx] -np.nan

    query_idx = multi_argmax(expected_error, n_instances)

    return query_idx, X[query_idx]


def expected_log_loss_reduction(learner: ActiveLearner, X: modALinput,
                                p_subsample: np.float = 1.0, n_instances: int = 1) -> Tuple[np.ndarray, modALinput]:
    """
    Expected log loss reduction query strategy.

    References:
        Roy and McCallum, 2001 (http://groups.csail.mit.edu/rrg/papers/icml01.pdf)

    Args:
        learner: The ActiveLearner object for which the expected log loss is to be estimated.
        X: The samples.
        p_subsample: Probability of keeping a sample from the pool when calculating expected log loss.
            Significantly improves runtime for large sample pools.
        n_instances: The number of instances to be sampled.


    Returns:
        The indices of the instances from X chosen to be labelled; the instances from X chosen to be labelled.
    """

    assert 0.0 <= p_subsample <= 1.0, 'p_subsample subsampling keep ratio must be between 0.0 and 1.0'

    expected_log_loss = np.zeros(shape=(len(X), ))
    possible_labels = np.unique(learner.y_training)

    try:
        X_proba = learner.predict_proba(X)
    except NotFittedError:
        # TODO: implement a proper cold-start
        return 0, X[0]

    for x_idx, x in enumerate(X):
        # subsample the data if needed
        if np.random.rand() <= p_subsample:
            # estimate the expected error
            for y_idx, y in enumerate(possible_labels):
                X_new = data_vstack((learner.X_training, x.reshape(1, -1)))
                y_new = data_vstack((learner.y_training, np.array(y).reshape(1, )))

                refitted_estimator = clone(learner.estimator).fit(X_new, y_new)
                refitted_proba = refitted_estimator.predict_proba(X)
                entr = np.transpose(entropy(np.transpose(refitted_proba)))

                expected_log_loss[x_idx] += np.sum(entr)*X_proba[x_idx, y_idx]

        else:
            expected_log_loss[x_idx] -np.nan

    query_idx = multi_argmax(expected_log_loss, n_instances)
<commit_msg>fix: bugfix for subsampled eer<commit_after>"""
Expected error reduction framework for active learning.
"""

from typing import Tuple

import numpy as np

from scipy.stats import entropy

from sklearn.base import clone
from sklearn.exceptions import NotFittedError

from modAL.models import ActiveLearner
from modAL.utils.data import modALinput, data_vstack
from modAL.utils.selection import multi_argmax


def expected_error_reduction(learner: ActiveLearner, X: modALinput,
                             p_subsample: np.float = 1.0, n_instances: int = 1) -> Tuple[np.ndarray, modALinput]:
    """
    Expected error reduction query strategy.

    References:
        Roy and McCallum, 2001 (http://groups.csail.mit.edu/rrg/papers/icml01.pdf)

    Args:
        learner: The ActiveLearner object for which the expected error is to be estimated.
        X: The samples.
        p_subsample: Probability of keeping a sample from the pool when calculating expected error.
            Significantly improves runtime for large sample pools.
        n_instances: The number of instances to be sampled.


    Returns:
        The indices of the instances from X chosen to be labelled; the instances from X chosen to be labelled.
    """

    assert 0.0 <= p_subsample <= 1.0, 'p_subsample subsampling keep ratio must be between 0.0 and 1.0'

    expected_error = np.zeros(shape=(len(X), ))
    possible_labels = np.unique(learner.y_training)

    try:
        X_proba = learner.predict_proba(X)
    except NotFittedError:
        # TODO: implement a proper cold-start
        return 0, X[0]

    for x_idx, x in enumerate(X):
        # subsample the data if needed
        if np.random.rand() <= p_subsample:
            # estimate the expected error
            for y_idx, y in enumerate(possible_labels):
                X_new = data_vstack((learner.X_training, x.reshape(1, -1)))
                y_new = data_vstack((learner.y_training, np.array(y).reshape(1, )))

                refitted_estimator = clone(learner.estimator).fit(X_new, y_new)
                uncertainty = 1 - np.max(refitted_estimator.predict_proba(X), axis=1)

                expected_error[x_idx] += np.sum(uncertainty)*X_proba[x_idx, y_idx]

        else:
            expected_error[x_idx] = np.inf

    query_idx = multi_argmax(expected_error, n_instances)

    return query_idx, X[query_idx]


def expected_log_loss_reduction(learner: ActiveLearner, X: modALinput,
                                p_subsample: np.float = 1.0, n_instances: int = 1) -> Tuple[np.ndarray, modALinput]:
    """
    Expected log loss reduction query strategy.

    References:
        Roy and McCallum, 2001 (http://groups.csail.mit.edu/rrg/papers/icml01.pdf)

    Args:
        learner: The ActiveLearner object for which the expected log loss is to be estimated.
        X: The samples.
        p_subsample: Probability of keeping a sample from the pool when calculating expected log loss.
            Significantly improves runtime for large sample pools.
        n_instances: The number of instances to be sampled.


    Returns:
        The indices of the instances from X chosen to be labelled; the instances from X chosen to be labelled.
    """

    assert 0.0 <= p_subsample <= 1.0, 'p_subsample subsampling keep ratio must be between 0.0 and 1.0'

    expected_log_loss = np.zeros(shape=(len(X), ))
    possible_labels = np.unique(learner.y_training)

    try:
        X_proba = learner.predict_proba(X)
    except NotFittedError:
        # TODO: implement a proper cold-start
        return 0, X[0]

    for x_idx, x in enumerate(X):
        # subsample the data if needed
        if np.random.rand() <= p_subsample:
            # estimate the expected error
            for y_idx, y in enumerate(possible_labels):
                X_new = data_vstack((learner.X_training, x.reshape(1, -1)))
                y_new = data_vstack((learner.y_training, np.array(y).reshape(1, )))

                refitted_estimator = clone(learner.estimator).fit(X_new, y_new)
                refitted_proba = refitted_estimator.predict_proba(X)
                entr = np.transpose(entropy(np.transpose(refitted_proba)))

                expected_log_loss[x_idx] += np.sum(entr)*X_proba[x_idx, y_idx]

        else:
            expected_log_loss[x_idx] = np.inf

    query_idx = multi_argmax(expected_log_loss, n_instances)

{'fraction_non_alphanumeric': 0.0855731679301631, 'fraction_numerical': 0.010567424764530209, 'mean_word_length': 2.652265100671141, 'pattern_counts': {'":': 0, '<': 16, '<?xml version=': 0, '>': 8, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5684483', 'n_tokens_mistral': 2818, 'n_tokens_neox': 2631, 'n_words': 757}