import yaml import pandas as pd from sqlalchemy import create_engine from urllib import parse from commonFunc import update_dict from functools import reduce from itertools import product from datetime import datetime class BayesFeatures(object): """ 计算素材单维度的特征值 """ def __init__(self, signature, target_type, dim, dim_config): self.signature = signature self.target_type = target_type self.dim = dim self.dim_config = dim_config self.table = self.dim_config['table'] self.file_name = self.dim_config['fieldName'] self.dim_lst = self.dim_config['Lst'] self.window_size = self.dim_config['windowSize'] self.window_combine = [] self.bayes_feature = {} def get_window_combine(self): """ 根据size得到指定维度的滑窗组合。 如 age: ['18-23岁', '24-30岁','31-40岁',....] 和 size = 2 得到 ['18-23岁', '24-30岁'],['24-30岁','31-40岁'],...... """ start = 0 while (start < len(self.dim_lst)) and (start + self.window_size <= len(self.dim_lst)): self.window_combine.append(self.dim_lst[start: start + self.window_size]) start += 1 # 添加一个全部的组合,等同于'不限' self.window_combine.append(['不限']) def get_bayes_feature(self): # 1、获取指定素材在指定维度下的人群数据(近一个月内的数据表现) sql = ''' select signature, %s, sum(aclick) aclick, sum(bclick) bclick, sum(activation) activation from %s where signature = '%s' -- and datediff(now(),stat_date)<=30 group by signature, %s ''' % (self.file_name, self.table, self.signature, self.file_name) df = pd.read_sql(sql, test_engine) if self.dim != 'city': if self.target_type == 'action_ratio': # 计算行为率(bclick/aclick)的组合特征值 df['unbclick'] = df['aclick'] - df['bclick'] sig_pos_pct = df['bclick'].sum() / df['aclick'].sum() # 素材曝光-点击的概率 sig_neg_pct = 1 - sig_pos_pct # 素材曝光-未点击的概率 self.bayes_feature['sig_pos_pct'] = sig_pos_pct self.bayes_feature['sig_neg_pct'] = sig_neg_pct for sub_combine in self.window_combine: if sub_combine != ['不限']: key = self.dim + '_' + '|'.join(sub_combine) pos_pct = df[df[self.file_name].isin(sub_combine)].bclick.sum() / df.bclick.sum() neg_pct = df[df[self.file_name].isin(sub_combine)].unbclick.sum() / df.unbclick.sum() self.bayes_feature[key] = {'pos_pct': pos_pct, 'neg_pct': neg_pct} else: self.bayes_feature[self.dim + '_' + '不限'] = {'pos_pct': 1, 'neg_pct': 1} if self.target_type == 'convert_ratio': # 计算转化率(activation/bclick)的组合特征值 df['unactivation'] = df['bclick'] - df['activation'] sig_pos_pct = df['activation'].sum() / df['bclick'].sum() # 素材点击-转化的概率 sig_neg_pct = 1 - sig_pos_pct # 素材点击-未转化的概率 self.bayes_feature['sig_pos_pct'] = sig_pos_pct self.bayes_feature['sig_neg_pct'] = sig_neg_pct for sub_combine in self.window_combine: if sub_combine != '不限': key = self.dim + '_' + '|'.join(sub_combine) pos_pct = df[df[self.file_name].isin(sub_combine)].activation.sum() / df.activation.sum() neg_pct = df[df[self.file_name].isin(sub_combine)].unactivation.sum() / df.unactivation.sum() self.bayes_feature[key] = {'pos_pct': pos_pct, 'neg_pct': neg_pct} else: self.bayes_feature[self.dim + '_' + '不限'] = {'pos_pct': 1, 'neg_pct': 1} elif self.dim == 'city': pass else: pass class BayesCombine(object): """ 依据特征值,计算多维度的组合预估值 """ def __init__(self, signature, project_id, target_type, dim_features): self.signature = signature self.project_id = project_id self.target_type = target_type self.dim_features = dim_features self.bayes_combine_df = pd.DataFrame() self.actual_prob = self.dim_features[0]['sig_pos_pct'] def get_bayes_estimate(self): dim_combine_lst = [[key for key in fea.keys() if key not in ['sig_pos_pct', 'sig_neg_pct']] for fea in self.dim_features] self.dim_features = reduce(update_dict, self.dim_features) for ele in product(*dim_combine_lst): prob_pos = reduce(lambda x, y: self.dim_features[x]['pos_pct'] * self.dim_features[y]['pos_pct'], ele) prob_neg = reduce(lambda x, y: self.dim_features[x]['neg_pct'] * self.dim_features[y]['neg_pct'], ele) prob = (prob_pos * self.dim_features['sig_pos_pct']) / (prob_neg * self.dim_features['sig_neg_pct']) out_dict = dict(zip([e.split('_')[0] for e in ele], [e.split('_')[-1] for e in ele])) # TODO 调用人群预估覆盖接口,得到该组合的人群覆盖数 out_dict['population_cnt'] = 1000 out_dict['combine_estimate_prob'] = prob out_dict['actual_prob'] = self.actual_prob out_dict['signature'] = self.signature out_dict['project_id'] = self.project_id out_dict['target_type'] = self.target_type out_dict['stat_date'] = str(datetime.now().date()) combine = pd.DataFrame([out_dict]) self.bayes_combine_df = self.bayes_combine_df.append(combine, ignore_index=True) def write_to_db(self): self.bayes_combine_df.to_sql(name="ctop_ai_kuaishou_signature_recommended_target_combine", con=test_engine, if_exists='append', index=False) if __name__ == '__main__': # 1、读取配置文件 with open('config/config.yaml', mode='r', encoding='utf-8') as f: config = yaml.load(f.read(), Loader=yaml.FullLoader) project_ids = config['projectId'] target_types = config['targetType'] online_db = config['onlineDB'] test_db = config['testDB'] material_rule = config['materialRule'] bayes_dim = config['bayesDim'] # 2、数据库连接引擎 db_con_str = 'mysql+pymysql://%s:%s@%s:%d/%s' % \ (online_db['username'], parse.quote_plus(online_db['password']), online_db['host'], online_db['port'], online_db['database']) engine = create_engine(db_con_str, connect_args={'charset': 'utf8'}) db_con_str = 'mysql+pymysql://%s:%s@%s:%d/%s' % \ (test_db['username'], parse.quote_plus(test_db['password']), test_db['host'], test_db['port'], test_db['database']) test_engine = create_engine(db_con_str, connect_args={'charset': 'utf8'}) # 3、 参与定向组合的维度 target_dim = [key for key in bayes_dim.keys() if bayes_dim[key]['isOn']] # 4、计算贝叶斯组合入库 for project_id in project_ids: # 4.1 获取指定项目下当前活跃的素材信息, 如近3天内累计激活个数达到50个 sql = ''' select signature from ctop_kuaishou_report_daily_material where account_id in (select account_id from ctop_user_allocation where project_id = %s) and datediff(now(),stat_date) <= %s group by signature having sum(activation) >= %s ''' % (project_id, material_rule['days'], material_rule['activation']) df = pd.read_sql(sql, engine) signature_lst = df['signature'].values signature_lst = ['0070efb7557b2a04cf3d4a6f243c3cd8', '03b93728f0d82ea7865c3c7cf632bc1b'] # 4.2 计算指定素材的贝叶斯特征 for sig in signature_lst: for t_type in target_types: bayes_feature_lst = [] for dimension in target_dim: cls = BayesFeatures(sig, t_type, dimension, bayes_dim[dimension]) # 计算滑窗组合 cls.get_window_combine() cls.get_bayes_feature() bayes_feature_lst.append(cls.bayes_feature) # 依据特征值,计算多维度的组合预估值 cls = BayesCombine(sig, project_id, t_type, bayes_feature_lst) cls.get_bayes_estimate() cls.write_to_db()