Skip to content

decay_popularity

logger = logging.getLogger(__name__) module-attribute

DecayPopularity

Bases: TopKAlgorithm, PopularityPaddingMixin, IncrementalTrainingMixin

Source code in src/recnexteval/algorithms/baseline/decay_popularity.py
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
class DecayPopularity(TopKAlgorithm, PopularityPaddingMixin, IncrementalTrainingMixin):
    IS_BASE: bool = False
    X_: csr_matrix | None = None
    decay_type = "exponential"
    decay_rate = 0.01
    internal_clock = 0

    def _compute_decay_weights(self, time_delta: int) -> float:
        """Apply decay based on time."""
        if self.decay_type == "exponential":
            return np.exp(-self.decay_rate * time_delta)
        elif self.decay_type == "linear":
            return np.clip(1 - self.decay_rate * time_delta, 0, 1)
        else:
            return 1.0 / (1.0 + self.decay_rate * (time_delta**2))

    def _fit(self, X: csr_matrix) -> Self:
        if self.X_ is not None:
            self._append_training_data(X)
        else:
            self.X_ = X.copy()

        if not isinstance(self.X_, csr_matrix):
            raise ValueError("Training data is not initialized properly.")

        if self.X_.shape[1] < self.K:
            logger.warning("K is larger than the number of items.", UserWarning)

        # Compute decay weights
        decay_weights = self._compute_decay_weights(self.internal_clock)
        self.X_.data = self.X_.data * decay_weights

        # Get top-K items
        self.sorted_scores_ = self.get_popularity_scores(self.X_)
        self.internal_clock += 1
        return self

    def _predict(self, X: PredictionMatrix) -> csr_matrix:
        intended_shape = (X.filter_for_predict().num_interactions, X.user_item_shape[1])

        # Vectorized: repeat the sorted scores for each prediction row
        data = np.tile(self.sorted_scores_, (intended_shape[0], 1))
        X_pred = csr_matrix(data)

        return X_pred

IS_BASE = False class-attribute instance-attribute

X_ = None class-attribute instance-attribute

decay_type = 'exponential' class-attribute instance-attribute

decay_rate = 0.01 class-attribute instance-attribute

internal_clock = 0 class-attribute instance-attribute

pad_with_popularity = pad_with_popularity instance-attribute

name property

Name of the object's class.

:return: Name of the object's class :rtype: str

params property

Parameters of the object.

:return: Parameters of the object :rtype: dict

identifier property

Identifier of the object.

Identifier is made by combining the class name with the parameters passed at construction time.

Constructed by recreating the initialisation call. Example: Algorithm(param_1=value)

:return: Identifier of the object :rtype: str

ITEM_USER_BASED instance-attribute

seed = 42 instance-attribute

rand_gen = np.random.default_rng(seed=(self.seed)) instance-attribute

description property

Description of the algorithm.

:return: Description of the algorithm :rtype: str

K = K instance-attribute

get_popularity_scores(X)

Compute a popularity-based scoring vector for items.

This method calculates normalized interaction counts for each item, selects the top-K most popular items, and returns a vector where only those top-K items have their normalized scores (others are 0). This is used to pad predictions for unseen users with popular items.

:param X: The interaction matrix (user-item) to compute popularity from. :type X: csr_matrix :return: A 1D array of shape (num_items,) with popularity scores for top-K items. :rtype: np.ndarray

Source code in src/recnexteval/algorithms/core/popularity_padding.py
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
def get_popularity_scores(self, X: csr_matrix) -> np.ndarray:
    """Compute a popularity-based scoring vector for items.

    This method calculates normalized interaction counts for each item,
    selects the top-K most popular items, and returns a vector where
    only those top-K items have their normalized scores (others are 0).
    This is used to pad predictions for unseen users with popular items.

    :param X: The interaction matrix (user-item) to compute popularity from.
    :type X: csr_matrix
    :return: A 1D array of shape (num_items,) with popularity scores for top-K items.
    :rtype: np.ndarray
    """
    interaction_counts = X.sum(axis=0).A[0]
    normalized_scores = interaction_counts / interaction_counts.max()

    num_items = X.shape[1]
    if hasattr(self, "K"):
        k_value = self.K
    else:
        k_value = 100
    if num_items < k_value:
        logger.warning("K is larger than the number of items.")

    effective_k = min(k_value, num_items)
    # Get indices of top-K items by popularity
    top_k_indices = np.argpartition(normalized_scores, -effective_k)[-effective_k:]
    popularity_vector = np.zeros(num_items)
    popularity_vector[top_k_indices] = normalized_scores[top_k_indices]

    return popularity_vector

get_params() abstractmethod

Get the parameters of the object.

:return: Parameters of the object :rtype: dict

Source code in src/recnexteval/models/base.py
38
39
40
41
42
43
44
45
@abstractmethod
def get_params(self) -> dict[str, Any]:
    """Get the parameters of the object.

    :return: Parameters of the object
    :rtype: dict
    """
    ...

get_default_params() classmethod

Get default parameters without instantiation.

Uses inspect.signature to extract init parameters and their default values without instantiating the class.

Returns:

Type Description
dict

Dictionary of parameter names to default values.

dict

Parameters without defaults map to None.

Source code in src/recnexteval/algorithms/core/base.py
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
@classmethod
def get_default_params(cls) -> dict:
    """Get default parameters without instantiation.

    Uses inspect.signature to extract __init__ parameters and their
    default values without instantiating the class.

    Returns:
        Dictionary of parameter names to default values.
        Parameters without defaults map to None.
    """
    try:
        sig = signature(cls.__init__)
    except (ValueError, TypeError):
        # Fallback for built-in types or special cases
        return {}

    params = {}
    for param_name, param in sig.parameters.items():
        if param_name == "self":
            continue

        if param.kind in (Parameter.VAR_POSITIONAL, Parameter.VAR_KEYWORD):
            # Skip *args, **kwargs
            continue

        # Extract the default value
        if param.default is not Parameter.empty:
            params[param_name] = param.default
        else:
            params[param_name] = None

    return params

set_params(**params)

Set the parameters of the estimator.

:param params: Estimator parameters :type params: dict

Source code in src/recnexteval/algorithms/core/base.py
 94
 95
 96
 97
 98
 99
100
def set_params(self, **params) -> Self:
    """Set the parameters of the estimator.

    :param params: Estimator parameters
    :type params: dict
    """
    return super().set_params(**params)

fit(X)

Fit the model to the input interaction matrix.

The input data is transformed to the expected type using :meth:_transform_fit_input. The fitting is done using the :meth:_fit method. Finally the method checks that the fitting was successful using :meth:_check_fit_complete.

:param X: The interactions to fit the model on. :type X: InteractionMatrix :return: Fitted algorithm :rtype: Algorithm

Source code in src/recnexteval/algorithms/core/base.py
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
def fit(self, X: InteractionMatrix) -> Self:
    """Fit the model to the input interaction matrix.

    The input data is transformed to the expected type using
    :meth:`_transform_fit_input`. The fitting is done using the
    :meth:`_fit` method. Finally the method checks that the fitting
    was successful using :meth:`_check_fit_complete`.

    :param X: The interactions to fit the model on.
    :type X: InteractionMatrix
    :return: Fitted algorithm
    :rtype: Algorithm
    """
    start = time.time()
    X_transformed = to_csr_matrix(X, binary=True)
    self._fit(X_transformed)

    self._check_fit_complete()
    end = time.time()
    logger.debug(f"Fitting {self.name} complete - Took {end - start:.3}s")
    return self

predict(X)

Predicts scores, given the interactions in X

The input data is transformed to the expected type using :meth:_transform_predict_input. The predictions are made using the :meth:_predict method. Finally the predictions are then padded with random items for users that are not in the training data.

:param X: interactions to predict from. :type X: InteractionMatrix :return: The recommendation scores in a sparse matrix format. :rtype: csr_matrix

Source code in src/recnexteval/algorithms/core/base.py
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
def predict(self, X: PredictionMatrix) -> csr_matrix:
    """Predicts scores, given the interactions in X

    The input data is transformed to the expected type using
    :meth:`_transform_predict_input`. The predictions are made
    using the :meth:`_predict` method. Finally the predictions
    are then padded with random items for users that are not in the
    training data.

    :param X: interactions to predict from.
    :type X: InteractionMatrix
    :return: The recommendation scores in a sparse matrix format.
    :rtype: csr_matrix
    """
    self._check_fit_complete()
    X_pred = self._predict(X)
    return X_pred