python3实现基于用户的协同过滤
本文实例为大家分享了python3实现基于用户协同过滤的具体代码，供大家参考，具体内容如下
废话不多说，直接看代码。
#!/usr/bin/python3
#-*-coding:utf-8-*-
#20170916号协同过滤电影推荐基稿
#字典等格式数据处理及直接写入文件
##fromnumpyimport*
importtime
frommathimportsqrt
##fromtexttableimportTexttable
classCF:
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
defformatRate(self,train_or_test):
self.userDict={}
self.ItemUser={}
foriintrain_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]inself.userDict):
self.userDict[i[0]].append(temp)
else:
self.userDict[i[0]]=[temp]
#计算ItemUser{'电影id',[用户id..],...}同一部电影的观众集合
if(i[1]inself.ItemUser):
self.ItemUser[i[1]].append(i[0])
else:
self.ItemUser[i[1]]=[i[0]]
#格式化userDict数据
defformatuserDict(self,userId,p):#userID为待查询目标，p为近邻对象
user={}
#user数据格式为：电影id：[userID的评分，近邻用户的评分]
foriinself.userDict[userId]:#i为userDict数据中的每个括号同81行
user[i[0]]=[i[1],0]
forjinself.userDict[p]:
if(j[0]notinuser):
user[j[0]]=[0,j[1]]#说明目标用户和近邻用户没有同时对一部电影评分
else:
user[j[0]][1]=j[1]#说明两者对同一部电影都有评分
returnuser
#计算余弦距离
defgetCost(self,userId,p):
#获取用户userId和p评分电影的并集
#{'电影ID'：[userId的评分，p的评分]}没有评分为0
user=self.formatuserDict(userId,p)
x=0.0
y=0.0
z=0.0
fork,vinuser.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):
return0
returnz/sqrt(x*y)
#计算皮尔逊相似度
##defgetCost(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)
##fork,vinuser.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):
##return0
##returnup/down
#找到某用户的相邻用户
defgetNearestNeighbor(self,userId):
neighbors=[]
self.neighbors=[]
#获取userId评分的电影都有那些用户也评过分
foriinself.userDict[userId]:#i为userDict数据中的每个括号同95行#user数据格式为：电影id：[userID的评分，近邻用户的评分]
forjinself.ItemUser[i[0]]:#i[0]为电影编号，j为看同一部电影的每位用户
if(j!=userIdandjnotinneighbors):
neighbors.append(j)
#计算这些用户与userId的相似度并排序
foriinneighbors:#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))
#获取推荐列表
defgetrecommandList(self,userId):
self.recommandList=[]
#建立推荐字典
recommandDict={}
forneighborinself.neighbors:#这里的neighbor数据格式为[[dist，用户id],[],....]
movies=self.userDict[neighbor[1]]#movies数据格式为[(电影id，评分),()，。。。。]
formovieinmovies:
if(movie[0]inrecommandDict):
recommandDict[movie[0]]+=neighbor[0]####？？？？
else:
recommandDict[movie[0]]=neighbor[0]
#建立推荐列表
forkeyinrecommandDict:#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))
#推荐的准确率
defgetPrecision(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）中的准确率召回率计算
foriinself.test_rec:
ifiinself.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不为空
foriinself.train_rec:
ifiinself.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))
#对同一用户分别通过训练集和测试集处理
defgetRecommand(self,train_or_test,userId):
self.formatRate(train_or_test)
self.getNearestNeighbor(userId)
self.getrecommandList(userId)
user=[i[0]foriinself.userDict[userId]]#用户userId评分的所有电影集合
recommand=[i[1]foriinself.recommandList]#推荐列表仅有电影id的集合，区别于recommandList（还含有dist）
##print("userid该用户已通过训练集测试集处理")
return(user,recommand)
#对test的电影进行评分预测
defforeCast(self):
self.forecast={}#？？？？？前面变量统一定义初始化后，函数内部是否需要该初始化？？？？
same_movie_id=[]
neighbors_id=[i[1]foriinself.neighbors]#近邻用户数据仅含用户id的集合
foriinself.test_user:#i为电影id，即在test里的i有被推荐到
ifiinself.train_rec:
same_movie_id.append(i)
forjinself.ItemUser[i]:#j为用户id,即寻找近邻用户的评分和相似度
ifjinneighbors_id:
user=[i[0]foriinself.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(iinself.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={}
ifsame_movie_id:#在test里的电影是否有在推荐列表里，如果为空不做判断，下面的处理会报错
formovieidinsame_movie_id:
total_d=0
total_down=0
fordinself.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,这里先按零计算
foriinself.test_user:
ifinotinmovieid:
self.score[i]=[0]
else:
foriinself.test_user:
self.score[i]=[0]
##returnself.score
#计算平均绝对误差MAE
defcal_Mae(self,userId):
self.formatRate(self.test_data)
##print(self.userDict)
foriteminself.userDict[userId]:
ifitem[0]inself.score:
self.score[item[0]].append(item[1])#self.score数据格式[[预测分，实际分]]
###过渡代码
##foriinself.score:
##pass
returnself.score
#基于用户的推荐
#根据对电影的评分计算用户之间的相似度
##defrecommendByUser(self,userId):
##print("亲，请稍等片刻，系统正在快马加鞭为你运作中")#人机交互辅助解读，
##self.getPrecision(self,userId)
#获取数据
defreadFile(filename):
files=open(filename,"r",encoding="utf-8")
data=[]
forlineinfiles.readlines():
item=line.strip().split("::")
data.append(item)
returndata
files.close()
defload_dict_from_file(filepath):
_dict={}
try:
withopen(filepath,'r',encoding="utf-8")asdict_file:
forlineindict_file.readlines():
(key,value)=line.strip().split(':')
_dict[key]=value
exceptIOErrorasioerr:
print("文件%s不存在"%(filepath))
return_dict
defsave_dict_to_file(_dict,filepath):
try:
withopen(filepath,'w',encoding="utf-8")asdict_file:
for(key,value)in_dict.items():
dict_file.write('%s:%s\n'%(key,value))
exceptIOErrorasioerr:
print("文件%s无法创建"%(filepath))
defwriteFile(data,filename):
withopen(filename,'w',encoding="utf-8")asf:
f.write(data)
#-------------------------开始-------------------------------
defstart3():
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("耗费时间：%fs"%(end1-start1))
defstart1():
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("处理结束")
##withopen("D:/d/dd/userDict.txt",'r',encoding='utf-8')asf:
##diction=f.read()
##i=0
##forjineval(diction):
##print(j)
##i+=1
##ifi==4:
##break
defstart2():
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()
以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持毛票票。