neighbors.models.NNMF_mult

The non-negative matrix factorization algorithm tries to decompose a users x items matrix into two additional matrices: users x factors and factors x items.

Training is performed via multiplicative updating and continues until convergence or the maximum number of training iterations has been reached. Unlike the NNMF_sgd, this implementation takes no hyper-parameters and thus is simpler and faster to use, but less flexible, i.e. no regularization.

The number of factors, convergence, and maximum iterations can be controlled with the n_factors, tol, and max_iterations arguments to the .fit method. By default the number of factors = the number items.

The implementation here follows closely that of Lee & Seung, 2001 (eq 4): https://papers.nips.cc/paper/2000/file/f9d1152547c0bde01830b7e8bd60024c-Paper.pdf

Note: random_state does not control the sgd fit, only the initialization of the factor matrices

Source code in neighbors/models.py

class NNMF_mult(BaseNMF):
    """
    The non-negative matrix factorization algorithm tries to decompose a users x items matrix into two additional matrices: users x factors and factors x items.

    Training is performed via multiplicative updating and continues until convergence or the maximum number of training iterations has been reached. Unlike the `NNMF_sgd`, this implementation takes no hyper-parameters and thus is simpler and faster to use, but less flexible, i.e. no regularization.

    The number of factors, convergence, and maximum iterations can be controlled with the `n_factors`, `tol`, and `max_iterations` arguments to the `.fit` method. By default the number of factors = the number items.

    The implementation here follows closely that of Lee & Seung, 2001 (eq 4): https://papers.nips.cc/paper/2000/file/f9d1152547c0bde01830b7e8bd60024c-Paper.pdf

    *Note*: `random_state` does not control the sgd fit, only the initialization of the factor matrices
    """

    def __init__(
        self, data, mask=None, n_mask_items=None, verbose=True, random_state=None
    ):
        """
        Args:
            data (pd.DataFrame): users x items dataframe
            mask (pd.DataFrame, optional): A boolean dataframe used to split the data into 'observed' and 'missing' datasets. Defaults to None.
            n_mask_items (int/float, optional): number of items to mask out, while the rest are treated as observed; Defaults to None.
            data_range (int/float, optional): max - min of the data; Default computed from the input data. This is useful to set manually in case the input data do not span the full range of possible values
            random_state (None, int, RandomState): a seed or random state used for all internal random operations (e.g. randomly mask half the data given n_mask_item = .05). Passing None will generate a new random seed. Default None.
            verbose (bool; optional): print any initialization warnings; Default True

        """
        super().__init__(
            data, mask, n_mask_items, random_state=random_state, verbose=verbose
        )
        self.H = None  # factors x items
        self.W = None  # user x factors
        self.n_factors = None

    def __repr__(self):
        return f"{super().__repr__()[:-1]}, n_factors={self.n_factors})"

    def fit(
        self,
        n_factors=None,
        n_iterations=1000,
        tol=1e-6,
        eps=1e-6,
        verbose=False,
        dilate_by_nsamples=None,
        clip_predictions=True,
        **kwargs,
    ):
        """Fit NNMF collaborative filtering model to train data using multiplicative updating.

        Given non-negative matrix `V` find non-negative factors `W` and `H` by minimizing `||V - WH||^2`.

        Args:
            n_factors (int, optional): number of factors to learn. Defaults to None which includes all factors.
            n_iterations (int, optional): total number of training iterations if convergence is not achieved. Defaults to 5000.
            tol (float, optional): Convergence criteria. Model is considered converged if the change in error during training < tol. Defaults to 0.001.
            eps (float; optiona): small value added to denominator of update rules to avoid divide-by-zero errors; Default 1e-6.
            verbose (bool, optional): print information about training. Defaults to False.
            dilate_by_nsamples (int, optional): How many items to dilate by prior to training. Defaults to None.
            clip_predictions (bool, optional): clip predictions to the observed rating range, since factorization alone can produce predictions outside it. This is the same approach the [Surprise](https://surpriselib.com/) package takes when making predictions. Defaults to True.
        """

        # Call parent fit which acts as a guard for non-masked data
        super().fit()

        n_users, n_items = self.data.shape

        if (
            isinstance(n_factors, int) and (n_factors > n_items and n_factors > n_users)
        ) or isinstance(n_factors, np.floating):
            raise TypeError("n_factors must be an integer < number of items and users")

        if n_factors is None:
            n_factors = min([n_users, n_items])

        self.n_factors = n_factors
        self.clip_predictions = clip_predictions

        # Initialize W and H as non-negative scaled random values
        # We use random initialization scaled by the number of factors not unlike sklearn: https://github.com/scikit-learn/scikit-learn/blob/95119c13af77c76e150b753485c662b7c52a41a2/sklearn/decomposition/_nmf.py#L334
        self.W = np.abs(
            self.random_state.normal(scale=1.0 / n_factors, size=(n_users, n_factors))
        )
        self.H = np.abs(
            self.random_state.normal(scale=1.0 / n_factors, size=(n_factors, n_items))
        )

        # Whereas in SGD we explity pass in indices of training data for fitting, here we set testing indices to 0 so they have no impact on the multiplicative update. See Zhu, 2016 for more details: https://arxiv.org/pdf/1612.06037.pdf
        self.dilate_mask(n_samples=dilate_by_nsamples)

        # fillna(0) is equivalent to hadamard (element-wise) product with a binary mask
        X = self.masked_data.fillna(0).to_numpy()

        # Run multiplicative updating
        # Silence numba warning until this issue gets fixed: https://github.com/numba/numba/issues/4585
        with warnings.catch_warnings():
            warnings.filterwarnings("ignore", category=NumbaPerformanceWarning)
            error_history, converged, n_iter, delta, norm_rmse, W, H = mult(
                X,
                self.W,
                self.H,
                self.data_range,
                eps,
                tol,
                n_iterations,
                verbose,
            )

        # Save outputs to model
        self.W, self.H = W, H
        self.error_history = error_history
        self._n_iter = n_iter
        self._delta = delta
        self._norm_rmse = norm_rmse
        self.converged = converged

        if verbose:
            if self.converged:
                print("\n\tCONVERGED!")
                print(f"\n\tFinal Iteration: {self._n_iter}")
                print(f"\tFinal Delta: {np.round(self._delta)}")
            else:
                print("\tFAILED TO CONVERGE (n_iter reached)")
                print(f"\n\tFinal Iteration: {self._n_iter}")
                print(f"\tFinal delta exceeds tol: {tol} <= {self._delta}")

            print(f"\tFinal Norm Error: {np.round(100 * norm_rmse, 2)}%")
        self._predict()
        self.is_fit = True

    def _predict(self):
        """Predict subjects' missing items using NNMF with multiplicative updating"""

        predictions = self.W @ self.H
        if self.clip_predictions:
            predictions = np.clip(
                predictions,
                self.masked_data.min().min(),
                self.masked_data.max().max(),
            )
        self.predictions = pd.DataFrame(
            predictions, index=self.data.index, columns=self.data.columns
        )

__init__(self, data, mask=None, n_mask_items=None, verbose=True, random_state=None) special

Parameters:

Name	Type	Description	Default
data	pd.DataFrame	users x items dataframe	required
mask	pd.DataFrame	A boolean dataframe used to split the data into 'observed' and 'missing' datasets. Defaults to None.	None
n_mask_items	int/float	number of items to mask out, while the rest are treated as observed; Defaults to None.	None
data_range	int/float	max - min of the data; Default computed from the input data. This is useful to set manually in case the input data do not span the full range of possible values	required
random_state	None, int, RandomState	a seed or random state used for all internal random operations (e.g. randomly mask half the data given n_mask_item = .05). Passing None will generate a new random seed. Default None.	None
verbose	bool; optional	print any initialization warnings; Default True	True

Source code in neighbors/models.py

def __init__(
    self, data, mask=None, n_mask_items=None, verbose=True, random_state=None
):
    """
    Args:
        data (pd.DataFrame): users x items dataframe
        mask (pd.DataFrame, optional): A boolean dataframe used to split the data into 'observed' and 'missing' datasets. Defaults to None.
        n_mask_items (int/float, optional): number of items to mask out, while the rest are treated as observed; Defaults to None.
        data_range (int/float, optional): max - min of the data; Default computed from the input data. This is useful to set manually in case the input data do not span the full range of possible values
        random_state (None, int, RandomState): a seed or random state used for all internal random operations (e.g. randomly mask half the data given n_mask_item = .05). Passing None will generate a new random seed. Default None.
        verbose (bool; optional): print any initialization warnings; Default True

    """
    super().__init__(
        data, mask, n_mask_items, random_state=random_state, verbose=verbose
    )
    self.H = None  # factors x items
    self.W = None  # user x factors
    self.n_factors = None

fit(self, n_factors=None, n_iterations=1000, tol=1e-06, eps=1e-06, verbose=False, dilate_by_nsamples=None, clip_predictions=True, **kwargs)

Fit NNMF collaborative filtering model to train data using multiplicative updating.

Given non-negative matrix V find non-negative factors W and H by minimizing ||V - WH||^2.

Parameters:

Name	Type	Description	Default
n_factors	int	number of factors to learn. Defaults to None which includes all factors.	None
n_iterations	int	total number of training iterations if convergence is not achieved. Defaults to 5000.	1000
tol	float	Convergence criteria. Model is considered converged if the change in error during training < tol. Defaults to 0.001.	1e-06
eps	float; optiona	small value added to denominator of update rules to avoid divide-by-zero errors; Default 1e-6.	1e-06
verbose	bool	print information about training. Defaults to False.	False
dilate_by_nsamples	int	How many items to dilate by prior to training. Defaults to None.	None
clip_predictions	bool	clip predictions to the observed rating range, since factorization alone can produce predictions outside it. This is the same approach the Surprise package takes when making predictions. Defaults to True.	True

Source code in neighbors/models.py

def fit(
    self,
    n_factors=None,
    n_iterations=1000,
    tol=1e-6,
    eps=1e-6,
    verbose=False,
    dilate_by_nsamples=None,
    clip_predictions=True,
    **kwargs,
):
    """Fit NNMF collaborative filtering model to train data using multiplicative updating.

    Given non-negative matrix `V` find non-negative factors `W` and `H` by minimizing `||V - WH||^2`.

    Args:
        n_factors (int, optional): number of factors to learn. Defaults to None which includes all factors.
        n_iterations (int, optional): total number of training iterations if convergence is not achieved. Defaults to 5000.
        tol (float, optional): Convergence criteria. Model is considered converged if the change in error during training < tol. Defaults to 0.001.
        eps (float; optiona): small value added to denominator of update rules to avoid divide-by-zero errors; Default 1e-6.
        verbose (bool, optional): print information about training. Defaults to False.
        dilate_by_nsamples (int, optional): How many items to dilate by prior to training. Defaults to None.
        clip_predictions (bool, optional): clip predictions to the observed rating range, since factorization alone can produce predictions outside it. This is the same approach the [Surprise](https://surpriselib.com/) package takes when making predictions. Defaults to True.
    """

    # Call parent fit which acts as a guard for non-masked data
    super().fit()

    n_users, n_items = self.data.shape

    if (
        isinstance(n_factors, int) and (n_factors > n_items and n_factors > n_users)
    ) or isinstance(n_factors, np.floating):
        raise TypeError("n_factors must be an integer < number of items and users")

    if n_factors is None:
        n_factors = min([n_users, n_items])

    self.n_factors = n_factors
    self.clip_predictions = clip_predictions

    # Initialize W and H as non-negative scaled random values
    # We use random initialization scaled by the number of factors not unlike sklearn: https://github.com/scikit-learn/scikit-learn/blob/95119c13af77c76e150b753485c662b7c52a41a2/sklearn/decomposition/_nmf.py#L334
    self.W = np.abs(
        self.random_state.normal(scale=1.0 / n_factors, size=(n_users, n_factors))
    )
    self.H = np.abs(
        self.random_state.normal(scale=1.0 / n_factors, size=(n_factors, n_items))
    )

    # Whereas in SGD we explity pass in indices of training data for fitting, here we set testing indices to 0 so they have no impact on the multiplicative update. See Zhu, 2016 for more details: https://arxiv.org/pdf/1612.06037.pdf
    self.dilate_mask(n_samples=dilate_by_nsamples)

    # fillna(0) is equivalent to hadamard (element-wise) product with a binary mask
    X = self.masked_data.fillna(0).to_numpy()

    # Run multiplicative updating
    # Silence numba warning until this issue gets fixed: https://github.com/numba/numba/issues/4585
    with warnings.catch_warnings():
        warnings.filterwarnings("ignore", category=NumbaPerformanceWarning)
        error_history, converged, n_iter, delta, norm_rmse, W, H = mult(
            X,
            self.W,
            self.H,
            self.data_range,
            eps,
            tol,
            n_iterations,
            verbose,
        )

    # Save outputs to model
    self.W, self.H = W, H
    self.error_history = error_history
    self._n_iter = n_iter
    self._delta = delta
    self._norm_rmse = norm_rmse
    self.converged = converged

    if verbose:
        if self.converged:
            print("\n\tCONVERGED!")
            print(f"\n\tFinal Iteration: {self._n_iter}")
            print(f"\tFinal Delta: {np.round(self._delta)}")
        else:
            print("\tFAILED TO CONVERGE (n_iter reached)")
            print(f"\n\tFinal Iteration: {self._n_iter}")
            print(f"\tFinal delta exceeds tol: {tol} <= {self._delta}")

        print(f"\tFinal Norm Error: {np.round(100 * norm_rmse, 2)}%")
    self._predict()
    self.is_fit = True