itemknn_rolling

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

@abstractmethod
def get_params(self) -> dict[str, Any]:
    """Get the parameters of the object.

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

@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

def set_params(self, **params) -> Self:
    """Set the parameters of the estimator.

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

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

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