| from implicit.als import AlternatingLeastSquares
|
| from implicit.lmf import LogisticMatrixFactorization
|
| from implicit.bpr import BayesianPersonalizedRanking
|
| from implicit.nearest_neighbours import bm25_weight
|
| from scipy.sparse import csr_matrix
|
| from typing import Dict, Any
|
|
|
| MODEL = {
|
| "lmf": LogisticMatrixFactorization,
|
| "als": AlternatingLeastSquares,
|
| "bpr": BayesianPersonalizedRanking,
|
| }
|
|
|
|
|
| def _get_sparse_matrix(values, user_idx, product_idx):
|
| return csr_matrix(
|
| (values, (user_idx, product_idx)),
|
| shape=(len(user_idx.unique()), len(product_idx.unique())),
|
| )
|
|
|
|
|
| def _get_model(name: str, **params):
|
| model = MODEL.get(name)
|
| if model is None:
|
| raise ValueError("No model with name {}".format(name))
|
| return model(**params)
|
|
|
|
|
| class InternalStatusError(Exception):
|
| pass
|
|
|
|
|
| class Recommender:
|
| def __init__(
|
| self,
|
| values,
|
| user_idx,
|
| product_idx,
|
| ):
|
| self.user_product_matrix = _get_sparse_matrix(values, user_idx, product_idx)
|
| self.user_idx = user_idx
|
| self.product_idx = product_idx
|
|
|
|
|
| self.model = None
|
| self.fitted = False
|
|
|
| def create_and_fit(
|
| self,
|
| model_name: str,
|
| weight_strategy: str = "bm25",
|
| model_params: Dict[str, Any] = {},
|
| ):
|
| weight_strategy = weight_strategy.lower()
|
| if weight_strategy == "bm25":
|
| data = bm25_weight(
|
| self.user_product_matrix,
|
| K1=1.2,
|
| B=0.75,
|
| )
|
| elif weight_strategy == "balanced":
|
|
|
|
|
| total_size = (
|
| self.user_product_matrix.shape[0] * self.user_product_matrix.shape[1]
|
| )
|
| sum = self.user_product_matrix.sum()
|
| num_zeros = total_size - self.user_product_matrix.count_nonzero()
|
| data = self.user_product_matrix.multiply(num_zeros / sum)
|
| elif weight_strategy == "same":
|
| data = self.user_product_matrix
|
| else:
|
| raise ValueError("Weight strategy not supported")
|
|
|
| self.model = _get_model(model_name, **model_params)
|
| self.fitted = True
|
|
|
| self.model.fit(data)
|
|
|
| return self
|
|
|
| def recommend_products(
|
| self,
|
| user_id,
|
| items_to_recommend=5,
|
| ):
|
| """Finds the recommended items for the user.
|
|
|
| Returns:
|
| (items, scores) pair, where item is already the name of the suggested item.
|
| """
|
|
|
| if not self.fitted:
|
| raise InternalStatusError(
|
| "Cannot recommend products without previously fitting the model."
|
| " Please, consider fitting the model before recommening products."
|
| )
|
|
|
| return self.model.recommend(
|
| user_id,
|
| self.user_product_matrix[user_id],
|
| filter_already_liked_items=True,
|
| N=items_to_recommend,
|
| )
|
|
|
| def explain_recommendation(
|
| self,
|
| user_id,
|
| suggested_item_id,
|
| recommended_items,
|
| ):
|
| _, items_score_contrib, _ = self.model.explain(
|
| user_id,
|
| self.user_product_matrix,
|
| suggested_item_id,
|
| N=recommended_items,
|
| )
|
|
|
| return items_score_contrib
|
|
|
| def similar_users(self, user_id):
|
| return self.model.similar_users(user_id)
|
|
|
| @property
|
| def item_factors(self):
|
| return self.model.item_factors
|
|
|
|
|
