tave-st's picture
initial commit
e618873
raw
history blame
3.63 kB
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
# This variable will be set during training phase
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":
# Balance the positive and negative (nan) entries
# http://stanford.edu/~rezab/nips2014workshop/submits/logmat.pdf
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