本文實例為大家分享了python3實現基于用戶協同過濾的具體代碼，供大家參考，具體內容如下

廢話不多說，直接看代碼。

#!/usr/bin/python3 # -*- coding: utf-8 -*- #20170916號協同過濾電影推薦基稿 #字典等格式數據處理及直接寫入文件   ##from numpy import * import time from math import sqrt ##from texttable import Texttable   class CF:   def __init__(self, movies, ratings, k=5, n=20):   self.movies = movies#[MovieID,Title,Genres]   (self.train_data,self.test_data) = (ratings[0], ratings[1])#[UserID::MovieID::Rating::Timestamp]   # 鄰居個數   self.k = k   # 推薦個數   self.n = n   # 用戶對電影的評分   # 數據格式{'UserID用戶ID':[(MovieID電影ID,Rating用戶對電影的評星)]}   self.userDict = {}   # 對某電影評分的用戶   # 數據格式：{'MovieID電影ID':[UserID,用戶ID]}   # {'1',[1,2,3..],...}   self.ItemUser = {}   # 鄰居的信息   self.neighbors = []   # 推薦列表   self.recommandList = []#包含dist和電影id   self.recommand = [] #訓練集合測試集的交集，且僅有電影id   #用戶評過電影信息   self.train_user = []   self.test_user = []   #給用戶的推薦列表，僅含movieid   self.train_rec =[]   self.test_rec = []   #test中的電影評分預測數據集合,   self.forecast = {}#前k個近鄰的評分集合   self.score = {}#最終加權平均后的評分集合{“電影id”：預測評分}   #召回率和準確率   self.pre = [0.0,0.0]   self.z = [0.0, 0.0]  '''''  userDict數據格式：  '3': [('3421', 0.8), ('1641', 0.4), ('648', 0.6), ('1394', 0.8), ('3534', 0.6), ('104', 0.8),  ('2735', 0.8), ('1210', 0.8), ('1431', 0.6), ('3868', 0.6), ('1079', 1.0), ('2997', 0.6),  ('1615', 1.0), ('1291', 0.8), ('1259', 1.0), ('653', 0.8), ('2167', 1.0), ('1580', 0.6),  ('3619', 0.4), ('260', 1.0), ('2858', 0.8), ('3114', 0.6), ('1049', 0.8), ('1261', 0.2),  ('552', 0.8), ('480', 0.8), ('1265', 0.4), ('1266', 1.0), ('733', 1.0), ('1196', 0.8),  ('590', 0.8), ('2355', 1.0), ('1197', 1.0), ('1198', 1.0), ('1378', 1.0), ('593', 0.6),  ('1379', 0.8), ('3552', 1.0), ('1304', 1.0), ('1270', 0.6), ('2470', 0.8), ('3168', 0.8),  ('2617', 0.4), ('1961', 0.8), ('3671', 1.0), ('2006', 0.8), ('2871', 0.8), ('2115', 0.8),  ('1968', 0.8), ('1136', 1.0), ('2081', 0.8)]}  ItemUser數據格式：  {'42': ['8'], '2746': ['10'], '2797': ['1'], '2987': ['5'], '1653': ['5', '8', '9'],  '194': ['5'], '3500': ['8', '10'], '3753': ['6', '7'], '1610': ['2', '5', '7'],  '1022': ['1', '10'], '1244': ['2'], '25': ['8', '9']  '''   # 將ratings轉換為userDict和ItemUser  def formatRate(self,train_or_test):   self.userDict = {}   self.ItemUser = {}   for i in train_or_test:#[UserID,MovieID,Rating,Timestamp]    # 評分最高為5 除以5 進行數據歸一化 ##   temp = (i[1], float(i[2]) / 5)    temp = (i[1], float(i[2])) ##   temp = (i[1], i[2])    # 計算userDict {'用戶id':[(電影id,評分),(2,5)...],'2':[...]...}一個觀眾對每一部電影的評分集合    if(i[0] in self.userDict):     self.userDict[i[0]].append(temp)    else:     self.userDict[i[0]] = [temp]    # 計算ItemUser {'電影id',[用戶id..],...}同一部電影的觀眾集合    if(i[1] in self.ItemUser):     self.ItemUser[i[1]].append(i[0])    else:     self.ItemUser[i[1]] = [i[0]]     # 格式化userDict數據  def formatuserDict(self, userId, p):#userID為待查詢目標，p為近鄰對象   user = {}   #user數據格式為：電影id：[userID的評分，近鄰用戶的評分]   for i in self.userDict[userId]:#i為userDict數據中的每個括號同81行    user[i[0]] = [i[1], 0]   for j in self.userDict[p]:    if(j[0] not in user):     user[j[0]] = [0, j[1]]#說明目標用戶和近鄰用戶沒有同時對一部電影評分    else:     user[j[0]][1] = j[1]#說明兩者對同一部電影都有評分   return user        # 計算余弦距離  def getCost(self, userId, p):   # 獲取用戶userId和p評分電影的并集   # {'電影ID'：[userId的評分，p的評分]} 沒有評分為0   user = self.formatuserDict(userId, p)   x = 0.0   y = 0.0   z = 0.0   for k, v in user.items():#k是鍵，v是值    x += float(v[0]) * float(v[0])    y += float(v[1]) * float(v[1])    z += float(v[0]) * float(v[1])   if(z == 0.0):    return 0   return z / sqrt(x * y)  #計算皮爾遜相似度 ##  def getCost(self, userId, p): ##   # 獲取用戶userId和l評分電影的并集 ##   # {'電影ID'：[userId的評分，l的評分]} 沒有評分為0 ##   user = self.formatuserDict(userId, p) ##   sumxsq = 0.0 ##   sumysq = 0.0 ##   sumxy = 0.0 ##   sumx = 0.0 ##   sumy = 0.0 ##   n = len(user) ##   for k, v in user.items(): ##    sumx +=float(v[0]) ##    sumy +=float(v[1]) ##    sumxsq += float(v[0]) * float(v[0]) ##    sumysq += float(v[1]) * float(v[1]) ##    sumxy += float(v[0]) * float(v[1]) ##   up = sumxy -sumx*sumy/n ##   down = sqrt((sumxsq - pow(sumxsq,2)/n)*(sumysq - pow(sumysq,2)/n)) ##   if(down == 0.0): ##    return 0 ##   return up/down  # 找到某用戶的相鄰用戶  def getNearestNeighbor(self, userId):   neighbors = []   self.neighbors = []   # 獲取userId評分的電影都有那些用戶也評過分   for i in self.userDict[userId]:#i為userDict數據中的每個括號同95行#user數據格式為：電影id：[userID的評分，近鄰用戶的評分]    for j in self.ItemUser[i[0]]:#i[0]為電影編號，j為看同一部電影的每位用戶     if(j != userId and j not in neighbors):      neighbors.append(j)   # 計算這些用戶與userId的相似度并排序   for i in neighbors:#i為用戶id    dist = self.getCost(userId, i)    self.neighbors.append([dist, i])   # 排序默認是升序，reverse=True表示降序   self.neighbors.sort(reverse=True)   self.neighbors = self.neighbors[:self.k]#切片操作，取前k個 ##  print('neighbors',len(neighbors))    # 獲取推薦列表  def getrecommandList(self, userId):   self.recommandList = []   # 建立推薦字典   recommandDict = {}   for neighbor in self.neighbors:#這里的neighbor數據格式為[[dist，用戶id],[],....]    movies = self.userDict[neighbor[1]]#movies數據格式為[(電影id，評分),()，。。。。]    for movie in movies:     if(movie[0] in recommandDict):      recommandDict[movie[0]] += neighbor[0]####？？？？     else:      recommandDict[movie[0]] = neighbor[0]    # 建立推薦列表   for key in recommandDict:#recommandDict數據格式{電影id：累計dist，。。。}    self.recommandList.append([recommandDict[key], key])#recommandList數據格式【【累計dist，電影id】，【】，。。。?！?  self.recommandList.sort(reverse=True) ##  print(len(self.recommandList))   self.recommandList = self.recommandList[:self.n] ##  print(len(self.recommandList))  # 推薦的準確率  def getPrecision(self, userId): ##  print("開始?。?！") #先運算test_data，這樣最終self.neighbors等保留的是后來計算train_data后的數據(不交換位置的話就得在gR函數中增加參數保留各自的neighbor)   (self.test_user,self.test_rec) = self.getRecommand(self.test_data,userId)#測試集的用戶userId所評價的電影和給該用戶推薦的電影列表   (self.train_user,self.train_rec) = self.getRecommand(self.train_data,userId)#訓練集的用戶userId所評價的所有電影集合(self.train_user)和給該用戶推薦的電影列表(self.train_rec) #西安電大的張海朋：基于協同過濾的電影推薦系統的構建（2015）中的準確率召回率計算   for i in self.test_rec:    if i in self.train_rec:     self.recommand.append(i)   self.pre[0] = len(self.recommand)/len(self.train_rec)   self.z[0] = len(self.recommand)/len(self.test_rec)   #北京交大黃宇：基于協同過濾的推薦系統設計與實現（2015）中的準、召計算   self.recommand = []#這里沒有歸零的話，下面計算初始recommand不為空   for i in self.train_rec:    if i in self.test_user:     self.recommand.append(i)   self.pre[1] = len(self.recommand)/len(self.train_rec)   self.z[1] = len(self.recommand)/len(self.test_user) ##  print(self.train_rec,self.test_rec,"20",len(self.train_rec),len(self.train_rec))   #對同一用戶分別通過訓練集和測試集處理  def getRecommand(self,train_or_test,userId):   self.formatRate(train_or_test)   self.getNearestNeighbor(userId)   self.getrecommandList(userId)   user = [i[0] for i in self.userDict[userId]]#用戶userId評分的所有電影集合   recommand = [i[1] for i in self.recommandList]#推薦列表僅有電影id的集合，區別于recommandList（還含有dist） ##  print("userid該用戶已通過訓練集測試集處理")   return (user,recommand)  #對test的電影進行評分預測  def foreCast(self):   self.forecast = {}#？？？？？前面變量統一定義初始化后，函數內部是否需要該初始化？？？？   same_movie_id = []   neighbors_id = [i[1] for i in self.neighbors] #近鄰用戶數據僅含用戶id的集合         for i in self.test_user:#i為電影id，即在test里的i有被推薦到    if i in self.train_rec:     same_movie_id.append(i)     for j in self.ItemUser[i]:#j為用戶id,即尋找近鄰用戶的評分和相似度      if j in neighbors_id:       user = [i[0] for i in self.userDict[j]]#self.userDict[userId]數據格式：數據格式為[(電影id，評分),()，。。。。]；這里的userid應為近鄰用戶p       a = self.neighbors[neighbors_id.index(j)]#找到該近鄰用戶的數據【dist，用戶id】       b = self.userDict[j][user.index(i)]#找到該近鄰用戶的數據【電影id，用戶id】       c = [a[0], b[1], a[1]]       if (i in self.forecast):        self.forecast[i].append(c)       else:        self.forecast[i] = [c]#數據格式：字典{“電影id”：【dist，評分，用戶id】【】}{'589': [[0.22655856915174025, 0.6, '419'], [0.36264561173211646, 1.0, '1349']。。。} ##  print(same_movie_id)   #每個近鄰用戶的評分加權平均計算得預測評分   self.score = {}   if same_movie_id :#在test里的電影是否有在推薦列表里，如果為空不做判斷，下面的處理會報錯    for movieid in same_movie_id:     total_d = 0     total_down = 0     for d in self.forecast[movieid]:#此時的d已經是最里層的列表了【】；self.forecast[movieid]的數據格式[[]]      total_d += d[0]*d[1]      total_down += d[0]     self.score[movieid] = [round(total_d/total_down,3)]#加權平均后取3位小數的精度    #在test里但是推薦沒有的電影id,這里先按零計算    for i in self.test_user:     if i not in movieid:      self.score[i] = [0]   else:    for i in self.test_user:     self.score[i] = [0] ##  return self.score  #計算平均絕對誤差MAE  def cal_Mae(self,userId):   self.formatRate(self.test_data) ##  print(self.userDict)   for item in self.userDict[userId]:    if item[0] in self.score:     self.score[item[0]].append(item[1])#self.score數據格式[[預測分，實際分]] ##  #過渡代碼 ##  for i in self.score: ##   pass   return self.score     # 基于用戶的推薦  # 根據對電影的評分計算用戶之間的相似度 ## def recommendByUser(self, userId): ##  print("親，請稍等片刻，系統正在快馬加鞭為你運作中")   #人機交互輔助解讀， ##  self.getPrecision(self,userId)   # 獲取數據 def readFile(filename):  files = open(filename, "r", encoding = "utf-8")  data = []  for line in files.readlines():   item = line.strip().split("::")   data.append(item)  return data  files.close() def load_dict_from_file(filepath):  _dict = {}  try:   with open(filepath, 'r',encoding = "utf -8") as dict_file:    for line in dict_file.readlines():     (key, value) = line.strip().split(':')     _dict[key] = value  except IOError as ioerr:   print ("文件 %s 不存在" % (filepath))  return _dict def save_dict_to_file(_dict, filepath):  try:   with open(filepath, 'w',encoding = "utf - 8") as dict_file:    for (key,value) in _dict.items():     dict_file.write('%s:%s/n' % (key, value))   except IOError as ioerr:   print ("文件 %s 無法創建" % (filepath)) def writeFile(data,filename):  with open(filename, 'w', encoding = "utf-8")as f:   f.write(data)   # -------------------------開始-------------------------------  def start3():  start1 = time.clock()  movies = readFile("D:/d/movies.dat")  ratings = [readFile("D:/d/201709train.txt"),readFile("D:/d/201709test.txt")]  demo = CF(movies, ratings, k=20)  userId = '1000'  demo.getPrecision(userId) ## print(demo.foreCast())  demo.foreCast()  print(demo.cal_Mae(userId)) ## demo.recommendByUser(ID)  #上一句只能實現固定用戶查詢，這句可以實現“想查哪個查哪個”，后期可以加個循環，挨個查，查到你不想查  print("處理的數據為%d條" % (len(ratings[0])+len(ratings[1]))) ## print("____---",len(ratings[0]),len(ratings[1])) ## print("準確率： %.2f %%" % (demo.pre * 100)) ## print("召回率： %.2f %%" % (demo.z * 100))  print(demo.pre)  print(demo.z)  end1 = time.clock()  print("耗費時間： %f s" % (end1 - start1)) def start1():  start1 = time.clock()  movies = readFile("D:/d/movies.dat")  ratings = [readFile("D:/d/201709train.txt"),readFile("D:/d/201709test.txt")]  demo = CF(movies, ratings, k = 20)  demo.formatRate(ratings[0])  writeFile(str(demo.userDict),"D:/d/dd/userDict.txt")  writeFile(str(demo.ItemUser), "D:/d/dd/ItemUser.txt") ## save_dict_to_file(demo.userDict,"D:/d/dd/userDict.txt") ## save_dict_to_file(demo.ItemUser,"D:/d/dd/ItemUser.txt")  print("處理結束") ## with open("D:/d/dd/userDict.txt",'r',encoding = 'utf-8') as f: ##  diction = f.read() ##  i = 0 ##  for j in eval(diction): ##   print(j) ##   i += 1 ##   if i == 4: ##    break def start2():  start1 = time.clock()  movies = readFile("D:/d/movies.dat")  ratings = [readFile("D:/d/201709train.txt"),readFile("D:/d/201709test.txt")]  demo = CF(movies, ratings, k = 20)  demo.formatRate_toMovie(ratings[0])  writeFile(str(demo.movieDict),"D:/d/dd/movieDict.txt") ## writeFile(str(demo.userDict),"D:/d/dd/userDict.txt") ## writeFile(str(demo.ItemUser), "D:/d/dd/ItemUser.txt") ## save_dict_to_file(demo.userDict,"D:/d/dd/userDict.txt") ## save_dict_to_file(demo.ItemUser,"D:/d/dd/ItemUser.txt")  print("處理結束")   if __name__ == '__main__':  start1()