import numpy as np
import numpy.ma as ma
from numpy import genfromtxt
from collections import defaultdict
from sklearn.preprocessing import StandardScaler, MinMaxScaler
from sklearn.model_selection import train_test_split
import pandas as pd
import tabulate
import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import TensorDataset, DataLoader
from recsysNN_utils import *
from public_tests import *
pd.set_option("display.precision", 1)
item_train, user_train, y_train, item_features, user_features, item_vecs, movie_dict, user_to_genre = load_data()
num_user_features = user_train.shape[1] - 3
num_item_features = item_train.shape[1] - 1
uvs = 3
ivs = 3
u_s = 3
i_s = 1
scaledata = True
print(f"训练向量数量: {len(item_train)}")
pprint_train(user_train, user_features, uvs, u_s, maxcount=5)
pprint_train(item_train, item_features, ivs, i_s, maxcount=5, user=False)
print(f"y_train[:5]: {y_train[:5]}")
if scaledata:
item_train_save = item_train
user_train_save = user_train
scalerItem = StandardScaler()
scalerItem.fit(item_train[:, i_s:])
item_train[:, i_s:] = scalerItem.transform(item_train[:, i_s:])
scalerUser = StandardScaler()
scalerUser.fit(user_train[:, u_s:])
user_train[:, u_s:] = scalerUser.transform(user_train[:, u_s:])
print(np.allclose(item_train_save[:, i_s:], scalerItem.inverse_transform(item_train[:, i_s:])))
print(np.allclose(user_train_save[:, u_s:], scalerUser.inverse_transform(user_train[:, u_s:])))
item_train, item_test = train_test_split(item_train, train_size=0.80, shuffle=True, random_state=1)
user_train, user_test = train_test_split(user_train, train_size=0.80, shuffle=True, random_state=1)
y_train, y_test = train_test_split(y_train, train_size=0.80, shuffle=True, random_state=1)
print(f"电影/物品训练数据形状: {item_train.shape}")
print(f"电影/物品测试数据形状: {item_test.shape}")
pprint_train(user_train, user_features, uvs, u_s, maxcount=5)
scaler = MinMaxScaler((-1, 1))
scaler.fit(y_train.reshape(-1, 1))
ynorm_train = scaler.transform(y_train.reshape(-1, 1))
ynorm_test = scaler.transform(y_test.reshape(-1, 1))
print(ynorm_train.shape, ynorm_test.shape)
num_outputs = 32
torch.manual_seed(1)
user_NN = [
nn.Linear(num_user_features, 256),
nn.Linear(256, 128),
nn.Linear(128, num_outputs)
]
item_NN = [
nn.Linear(num_item_features, 256),
nn.Linear(256, 128),
nn.Linear(128, num_outputs)
]
user_NN[0].activation = nn.ReLU()
user_NN[1].activation = nn.ReLU()
user_NN[2].activation = nn.Identity()
item_NN[0].activation = nn.ReLU()
item_NN[1].activation = nn.ReLU()
item_NN[2].activation = nn.Identity()
class RecommenderModel(nn.Module):
def __init__(self, user_layers, item_layers):
super(RecommenderModel, self).__init__()
self.user_layers = nn.ModuleList(user_layers)
self.item_layers = nn.ModuleList(item_layers)
def forward(self, user_input, item_input):
ux = user_input
for layer in self.user_layers:
ux = layer(ux)
ux = layer.activation(ux)
ix = item_input
for layer in self.item_layers:
ix = layer(ix)
ix = layer.activation(ix)
ux = nn.functional.normalize(ux, p=2, dim=1)
ix = nn.functional.normalize(ix, p=2, dim=1)
output = torch.sum(ux * ix, dim=1, keepdim=True)
return output
model = RecommenderModel(user_NN, item_NN)
print("用户网络结构:")
for i, layer in enumerate(user_NN):
print(f"第{i+1}层: {layer}，激活函数: {type(layer.activation).__name__}")
print("\n物品网络结构:")
for i, layer in enumerate(item_NN):
print(f"第{i+1}层: {layer}，激活函数: {type(layer.activation).__name__}")
test_tower(user_NN)
test_tower(item_NN)
def prepare_data(user_data, item_data, labels, batch_size=64):
user_tensor = torch.FloatTensor(user_data)
item_tensor = torch.FloatTensor(item_data)
labels_tensor = torch.FloatTensor(labels)
dataset = TensorDataset(user_tensor, item_tensor, labels_tensor)
dataloader = DataLoader(dataset, batch_size=batch_size, shuffle=True)
return dataloader
train_loader = prepare_data(
user_train[:, u_s:],
item_train[:, i_s:],
ynorm_train
)
test_loader = prepare_data(
user_test[:, u_s:],
item_test[:, i_s:],
ynorm_test
)
cost_fn = nn.MSELoss()
optimizer = optim.Adam(model.parameters(), lr=0.01)
torch.manual_seed(1)
epochs = 30
for epoch in range(epochs):
model.train()
total_loss = 0
for user_batch, item_batch, label_batch in train_loader:
outputs = model(user_batch, item_batch)
loss = cost_fn(outputs, label_batch)
optimizer.zero_grad()
loss.backward()
optimizer.step()
total_loss += loss.item()
avg_loss = total_loss / len(train_loader)
print(f"Epoch {epoch+1}/{epochs}, 平均损失: {avg_loss:.4f}")
model.eval()
test_loss = 0
with torch.no_grad():
for user_batch, item_batch, label_batch in test_loader:
outputs = model(user_batch, item_batch)
loss = cost_fn(outputs, label_batch)
test_loss += loss.item()
avg_test_loss = test_loss / len(test_loader)
print(f"测试集平均损失: {avg_test_loss:.4f}")
new_user_id = 5000
new_rating_ave = 1.0
new_action = 1.0
new_adventure = 1
new_animation = 1
new_childrens = 1
new_comedy = 5
new_crime = 1
new_documentary = 1
new_drama = 1
new_fantasy = 1
new_horror = 1
new_mystery = 1
new_romance = 5
new_scifi = 5
new_thriller = 1
new_rating_count = 3
user_vec = np.array([[new_user_id, new_rating_count, new_rating_ave,
new_action, new_adventure, new_animation, new_childrens,
new_comedy, new_crime, new_documentary,
new_drama, new_fantasy, new_horror, new_mystery,
new_romance, new_scifi, new_thriller]])
user_vecs = gen_user_vecs(user_vec, len(item_vecs))
sorted_index, sorted_ypu, sorted_items, sorted_user = predict_uservec(
user_vecs, item_vecs, model, u_s, i_s,
scaler, scalerUser, scalerItem, scaledata=scaledata
)
print_pred_movies(sorted_ypu, sorted_user, sorted_items, movie_dict, maxcount=10)
uid = 36
user_train_unscaled_part = scalerUser.inverse_transform(user_train[:, u_s:])
full_user_train_unscaled = np.concatenate((user_train[:, :u_s], user_train_unscaled_part), axis=1)
user_vecs, y_vecs = get_user_vecs(uid, full_user_train_unscaled, item_vecs, user_to_genre)
sorted_index, sorted_ypu, sorted_items, sorted_user = predict_uservec(
user_vecs, item_vecs, model, u_s, i_s, scaler,
scalerUser, scalerItem, scaledata=scaledata
)
sorted_y = y_vecs[sorted_index]
print_existing_user(
sorted_ypu, sorted_y.reshape(-1,1), sorted_user,
sorted_items, item_features, ivs, uvs, movie_dict, maxcount=10
)
def sq_dist(a, b):
"""
计算两个向量之间的平方距离
参数:
a (ndarray (n,)): 具有n个特征的向量
b (ndarray (n,)): 具有n个特征的向量
返回:
d (float) : 平方距离
"""
diff = a - b
d = np.dot(diff, diff)
return d
test_sq_dist(sq_dist)
a1 = np.array([1.0, 2.0, 3.0]); b1 = np.array([1.0, 2.0, 3.0])
a2 = np.array([1.1, 2.1, 3.1]); b2 = np.array([1.0, 2.0, 3.0])
a3 = np.array([0, 1, 0]); b3 = np.array([1, 0, 0])
print(f"a1和b1的平方距离: {sq_dist(a1, b1)}")
print(f"a2和b2的平方距离: {sq_dist(a2, b2)}")
print(f"a3和b3的平方距离: {sq_dist(a3, b3)}")
class ItemEmbeddingModel(nn.Module):
def __init__(self, item_layers):
super(ItemEmbeddingModel, self).__init__()
self.item_layers = nn.ModuleList(item_layers)
def forward(self, x):
for layer in self.item_layers:
x = layer(x)
x = layer.activation(x)
return nn.functional.normalize(x, p=2, dim=1)
model_m = ItemEmbeddingModel(item_NN)
scaled_item_vecs = scalerItem.transform(item_vecs[:, i_s:])
item_features_tensor = torch.FloatTensor(scaled_item_vecs)
model_m.eval()
with torch.no_grad():
vms = model_m(item_features_tensor).numpy()
print(f"所有预测的电影特征向量大小: {vms.shape}")
count = 50
dim = len(vms)
dist = np.zeros((dim, dim))
for i in range(dim):
for j in range(dim):
dist[i, j] = sq_dist(vms[i, :], vms[j, :])
m_dist = ma.masked_array(dist, mask=np.identity(dist.shape[0]))
disp = [["电影1", "类型", "电影2", "类型"]]
for i in range(count):
min_idx = np.argmin(m_dist[i])
movie1_id = int(item_vecs[i, 0])
movie2_id = int(item_vecs[min_idx, 0])
genre1, _ = get_item_genre(item_vecs[i, :], ivs, item_features)
genre2, _ = get_item_genre(item_vecs[min_idx, :], ivs, item_features)
disp.append([movie_dict[movie1_id]['title'], genre1,
movie_dict[movie2_id]['title'], genre2])
table = tabulate.tabulate(disp, tablefmt='html', headers="firstrow")
print(table)
with open("movie_similarity.html", "w", encoding="utf-8") as f:
f.write(table)
print("表格已保存为 movie_similarity.html，请用浏览器打开查看")