neighbors.models.KNN
The K-Nearest Neighbors algorithm makes predictions using a weighted mean of a subset of similar users. Similarity can be controlled via the metric argument to the .fit method, and the number of other users can be controlled with the k argument to the .predict method. NOTE: If user similiarity cannot be computed or no observed ratings have been made by the top k simililar users, this algorithm will fallback to the global mean on observed data for prediction (i.e. like the Mean model).
Source code in neighbors/models.py
class KNN(Base):
"""
The K-Nearest Neighbors algorithm makes predictions using a weighted mean of a subset of similar users. Similarity can be controlled via the `metric` argument to the `.fit` method, and the number of other users can be controlled with the `k` argument to the `.predict` method. NOTE: If user similiarity cannot be computed or no observed ratings have been made by the top k simililar users, this algorithm will fallback to the global mean on observed data for prediction (i.e. like the `Mean` model).
"""
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.user_similarity = None
self.metric = None
def __repr__(self):
return f"{super().__repr__()[:-1]}, similarity_metric={self.metric})"
def fit(
self,
k=10,
metric="correlation",
axis=0,
dilate_by_nsamples=None,
skip_refit=False,
**kwargs,
):
"""Fit collaborative model to train data. Calculate similarity between subjects across items. Repeated called to fit with different k, but the same previous arguments will re-use the computed user x user similarity matrix.
Args:
k (int): maximum number of other users to use when making a prediction for a single user. If set to None will use all users. Default 10. Note: it's possible for predictions to come from fewer than k other users if a particular user has fewer similar neighbors with positive similarity scores.
metric (str; optional): type of similarity. One of 'correlation', 'spearman', 'kendall', 'cosine', or 'pearson'. 'pearson' is just an alias for 'correlation'. Default 'correlation'.
axis (int): dimension along which to compute mean, 0 = mean across users separately by item, 1 = mean across items separately by user; Default 0
skip_refit (bool; optional): skip re-estimation of user x user similarity matrix. Faster if only exploring different k and no other model parameters or masks are changing. Default False.
"""
metrics = ["pearson", "spearman", "kendall", "cosine", "correlation"]
if metric not in metrics:
raise ValueError(f"metric must be one of {metrics}")
self.metric = metric
if metric == "correlation":
metric = "pearson"
# Call parent fit which acts as a guard for non-masked data
super().fit()
# If fit is being called more than once in a row with different k, but no other arguments are changing, reuse the last computed similarity matrix to save time. Otherwise re-calculate it
if not skip_refit:
self.dilate_mask(n_samples=dilate_by_nsamples)
# Store the mean because we'll use it in cases we can't make a prediction
self.mean = self.masked_data.mean(skipna=True, axis=axis)
if metric in ["pearson", "kendall", "spearman"]:
# Fall back to pandas
sim = self.masked_data.T.corr(method=metric)
else:
# Convert distance metrics to similarity (currently only cosine)
sim = pd.DataFrame(
1 - nanpdist(self.masked_data.to_numpy(), metric=metric),
index=self.masked_data.index,
columns=self.masked_data.index,
)
self.user_similarity = sim
self._predict(k=k)
self.is_fit = True
def _predict(self, k):
"""Make predictions using computed user similarities.
Args:
k (int): number of closest neighbors to use
"""
predictions = self.masked_data.copy()
for row_idx, _ in self.masked_data.iterrows():
user_prediction_error = False
# Get the similarity of this user to all other users, ignoring self-similarity
top_user_sims = self.user_similarity.loc[row_idx].drop(row_idx)
if top_user_sims.isnull().all():
warnings.warn(
f"User {row_idx} has no variance in their ratings. Impossible to compute similarity with other users. Falling back to global mean for all predictions",
stacklevel=2,
)
user_prediction_error = True # can't predict
else:
# Remove nan users and sort
top_user_sims = top_user_sims[~top_user_sims.isnull()].sort_values(
ascending=False
)
if len(top_user_sims) == 0:
user_prediction_error = True # can't predict
else:
# Get top k if requested
if k is not None:
top_user_sims = top_user_sims[: k + 1]
# Rescale similarity scores to the range 0 - 1, which has the effect of zeroing out negative similarities for currently supported similarity metrics.
# NOTE: we should revisit this approach for non-normalized similarity metrics e.g. euclidean distance
top_user_sims = top_user_sims.clip(lower=0, upper=1)
# No top users with positive correlations
if len(np.nonzero(top_user_sims.to_numpy())[0]) == 0:
user_prediction_error = True
else:
# NOTE: this code block is just a vectorized version of looping over every item for the current user and seeing whether we have observed ratings for each of the k other users to make a prediction with. We do this because for each item the *actual* number of other users' data availble for prediction will vary between 0-k based the pattern of sparsity
# Get the observed ratings from top users
top_user_ratings = self.masked_data.loc[top_user_sims.index, :]
# Make predictions = user_similarity_scores (column vector) * user x item (matrix of observed ratings)
# Do this in pandas rather than numpy because numpy will return nans when summing items if any item is nan
# Yields user x item matrix of ratings scaled by similarities
preds = (top_user_sims * top_user_ratings.T).T
# Add up the ratings from other users ignoring NaNs; this serves as the numerator of the formula
rating_sums = preds.sum()
# Now some of the values in preds will be nan because we never observed a rating for that user + item combo. We need to know how many are nans and which exact ones, because we need to sum down users for preds and then divide by the sum of the similarity weights we did end up using.
# Get locations of where we were able to make a prediction.
preds_mask = ~preds.isnull()
# Broadcast the user similarity vector over the user x item matrix so each column now contains the user similarity score if observed a prediction from that user and a 0 if not (True is converted to 1 during this multiplication whereas False is converted to 0)
user_sims_mat = (preds_mask.T * top_user_sims).T
# Now we can just sum down the rows which will give us the sum of the similarity weights we actually used
sim_sums = user_sims_mat.sum()
# Finally get the predictions by dividing the sum of ratings by sum of similarities we ended up using. This is how Surprise does it too: https://github.com/NicolasHug/Surprise/blob/master/surprise/prediction_algorithms/knns.py#L124
preds = rating_sums / sim_sums
# For items we can't predict because we never observed any ratings from the top k users for that item, fill in with the global mean for that item
if preds.isnull().any():
preds[preds.isnull()] = self.mean[preds.isnull()]
predictions.loc[row_idx] = preds.to_numpy()
# Handle cases where we were unable to make any predictions for this user
if user_prediction_error:
warnings.warn(
f"Not enough similar users with data to make any predictions for user {row_idx}. Falling back to global mean for all predictions",
stacklevel=2,
)
predictions.loc[row_idx, :] = self.mean.to_numpy()
self.predictions = predictions
def plot_user_similarity(
self, figsize=(8, 8), label_fontsize=16, hide_title=False, heatmap_kwargs=None
):
"""
Plot a heatmap of user x user similarities learned on the observed data
Args:
figsize (tuple, optional): matplotlib figure size. Defaults to (8, 8).
label_fontsize (int; optional): fontsize for title text; Default 16
hide_title (bool; optional): hide title containing metric information; Default False
heatmap_kwargs (dict, optional): addition arguments to seaborn.heatmap.
Returns:
ax: matplotib axis handle
"""
if not self.is_fit:
raise ValueError("Model as not been fit")
if self.metric in ["correlation", "pearson", "spearman"]:
vmin, vmax = -1, 1
cmap = "RdBu_r"
else:
vmin, vmax = 0, 1
cmap = None
_, ax = plt.subplots(1, 1, figsize=figsize)
_ = ax.set(xlabel=None, ylabel=None)
ax = sns.heatmap(
self.user_similarity,
vmin=vmin,
vmax=vmax,
cmap=cmap,
square=True,
ax=ax,
**(heatmap_kwargs or {}),
)
if not hide_title:
_ = ax.set_title(f"Metric: {self.metric}", fontsize=label_fontsize)
return ax
__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.user_similarity = None
self.metric = None
fit(self, k=10, metric='correlation', axis=0, dilate_by_nsamples=None, skip_refit=False, **kwargs)
Fit collaborative model to train data. Calculate similarity between subjects across items. Repeated called to fit with different k, but the same previous arguments will re-use the computed user x user similarity matrix.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
k |
int |
maximum number of other users to use when making a prediction for a single user. If set to None will use all users. Default 10. Note: it's possible for predictions to come from fewer than k other users if a particular user has fewer similar neighbors with positive similarity scores. |
10 |
metric |
str; optional |
type of similarity. One of 'correlation', 'spearman', 'kendall', 'cosine', or 'pearson'. 'pearson' is just an alias for 'correlation'. Default 'correlation'. |
'correlation' |
axis |
int |
dimension along which to compute mean, 0 = mean across users separately by item, 1 = mean across items separately by user; Default 0 |
0 |
skip_refit |
bool; optional |
skip re-estimation of user x user similarity matrix. Faster if only exploring different k and no other model parameters or masks are changing. Default False. |
False |
Source code in neighbors/models.py
def fit(
self,
k=10,
metric="correlation",
axis=0,
dilate_by_nsamples=None,
skip_refit=False,
**kwargs,
):
"""Fit collaborative model to train data. Calculate similarity between subjects across items. Repeated called to fit with different k, but the same previous arguments will re-use the computed user x user similarity matrix.
Args:
k (int): maximum number of other users to use when making a prediction for a single user. If set to None will use all users. Default 10. Note: it's possible for predictions to come from fewer than k other users if a particular user has fewer similar neighbors with positive similarity scores.
metric (str; optional): type of similarity. One of 'correlation', 'spearman', 'kendall', 'cosine', or 'pearson'. 'pearson' is just an alias for 'correlation'. Default 'correlation'.
axis (int): dimension along which to compute mean, 0 = mean across users separately by item, 1 = mean across items separately by user; Default 0
skip_refit (bool; optional): skip re-estimation of user x user similarity matrix. Faster if only exploring different k and no other model parameters or masks are changing. Default False.
"""
metrics = ["pearson", "spearman", "kendall", "cosine", "correlation"]
if metric not in metrics:
raise ValueError(f"metric must be one of {metrics}")
self.metric = metric
if metric == "correlation":
metric = "pearson"
# Call parent fit which acts as a guard for non-masked data
super().fit()
# If fit is being called more than once in a row with different k, but no other arguments are changing, reuse the last computed similarity matrix to save time. Otherwise re-calculate it
if not skip_refit:
self.dilate_mask(n_samples=dilate_by_nsamples)
# Store the mean because we'll use it in cases we can't make a prediction
self.mean = self.masked_data.mean(skipna=True, axis=axis)
if metric in ["pearson", "kendall", "spearman"]:
# Fall back to pandas
sim = self.masked_data.T.corr(method=metric)
else:
# Convert distance metrics to similarity (currently only cosine)
sim = pd.DataFrame(
1 - nanpdist(self.masked_data.to_numpy(), metric=metric),
index=self.masked_data.index,
columns=self.masked_data.index,
)
self.user_similarity = sim
self._predict(k=k)
self.is_fit = True
plot_user_similarity(self, figsize=(8, 8), label_fontsize=16, hide_title=False, heatmap_kwargs=None)
Plot a heatmap of user x user similarities learned on the observed data
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
figsize |
tuple |
matplotlib figure size. Defaults to (8, 8). |
(8, 8) |
label_fontsize |
int; optional |
fontsize for title text; Default 16 |
16 |
hide_title |
bool; optional |
hide title containing metric information; Default False |
False |
heatmap_kwargs |
dict |
addition arguments to seaborn.heatmap. |
None |
Returns:
| Type | Description |
|---|---|
ax |
matplotib axis handle |
Source code in neighbors/models.py
def plot_user_similarity(
self, figsize=(8, 8), label_fontsize=16, hide_title=False, heatmap_kwargs=None
):
"""
Plot a heatmap of user x user similarities learned on the observed data
Args:
figsize (tuple, optional): matplotlib figure size. Defaults to (8, 8).
label_fontsize (int; optional): fontsize for title text; Default 16
hide_title (bool; optional): hide title containing metric information; Default False
heatmap_kwargs (dict, optional): addition arguments to seaborn.heatmap.
Returns:
ax: matplotib axis handle
"""
if not self.is_fit:
raise ValueError("Model as not been fit")
if self.metric in ["correlation", "pearson", "spearman"]:
vmin, vmax = -1, 1
cmap = "RdBu_r"
else:
vmin, vmax = 0, 1
cmap = None
_, ax = plt.subplots(1, 1, figsize=figsize)
_ = ax.set(xlabel=None, ylabel=None)
ax = sns.heatmap(
self.user_similarity,
vmin=vmin,
vmax=vmax,
cmap=cmap,
square=True,
ax=ax,
**(heatmap_kwargs or {}),
)
if not hide_title:
_ = ax.set_title(f"Metric: {self.metric}", fontsize=label_fontsize)
return ax