import datetime import json import logging import traceback import uuid from itertools import product import pandas as pd import requests import yaml from concurrent_log import ConcurrentTimedRotatingFileHandler from config.url_and_db import estimate_people_number_url, headers, jeecg_db, jeecg_product_db, application_product_db from utils.BaseClass import NpEncoder from utils.commonFunc import get_db_engine, age_code_transform, gender_code_transform log_formatter = logging.Formatter('%(asctime)s %(levelname)s %(message)s', '%Y/%m/%d %I:%M:%S %p') log_handler = ConcurrentTimedRotatingFileHandler("/data/pythonProject/ai_target/logs/BayesCombine.log", when="midnight", backupCount=100) log_handler.setFormatter(log_formatter) logger = logging.getLogger('bayes_combine_logger') logger.addHandler(log_handler) logger.setLevel(logging.DEBUG) print('id of bayes_combine_logger %s' % id(logger)) class GetSingleDimFeatures(object): """ 计算素材单维度滑窗组合转化占比情况 """ def __init__(self, signature, dim, dim_config): self.signature = signature self.dim = dim self.dim_config = dim_config self.table = self.dim_config['table'] self.filed_name = self.dim_config['fieldName'] self.dim_value_lst = self.dim_config['Lst'] self.ratio_threshold = self.dim_config['ratio_threshold'] self.ratio_diff_threshold = self.dim_config['ratio_diff_threshold'] self.window_size = self.dim_config['windowSize'] self.window_combine = [] self.features = {} 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_value_lst)) and (start + self.window_size <= len(self.dim_value_lst)): self.window_combine.append(self.dim_value_lst[start: start + self.window_size]) start += 1 # 添加一个全部的组合,等同于'不限' self.window_combine.append(['不限']) def get_features(self): # 1、获取素材在指定维度下的人群数据 get_audience_sql = f"select signature, {self.filed_name}, sum(activation) activation " \ f"from {self.table} where signature = '{self.signature}' group by signature, {self.filed_name}" audience_df = pd.read_sql(get_audience_sql, jeecg_product_db) for combine in self.window_combine: if combine != ['不限']: self.features['|'.join(combine)] = round(audience_df[audience_df[self.filed_name].isin(combine)].activation.sum() / \ audience_df.activation.sum(), 4) else: if self.dim == 'gender': # 如果男/女 激活占比差异小于0.1, 说明素材的受众没有明显的性别倾向,可以对性别通投 male_ratio = audience_df[audience_df['gender'] == '男'].activation.sum() / audience_df.activation.sum() female_ratio = audience_df[audience_df['gender'] == '女'].activation.sum() / audience_df.activation.sum() if abs(male_ratio - female_ratio) <= self.ratio_diff_threshold: self.features['不限'] = 1 if self.dim == 'age': self.features['不限'] = 1 # 过滤得到占比达到 ratio_threshold 的组合值 self.features = {f"{self.dim}_{combine}": ratio for combine, ratio in self.features.items() if ratio > self.ratio_threshold} class GetCompositeDimFeatures(object): """ """ def __init__(self, signature, feature_lst, project_id): self.signature = signature self.feature_lst = feature_lst self.project_id = project_id def get_composite_features(self): feature_combine_lst = [list(item.keys()) for item in self.feature_lst] fea_combine_df = pd.DataFrame() for ele in product(*feature_combine_lst): fea_combine = dict(zip([e.split('_')[0] for e in ele], [e.split('_')[-1] for e in ele])) fea_combine['unlimited_cnt'] = list(fea_combine.values()).count('不限') fea_combine['features'] = str([(e.split('_')[0], e.split('_')[1], item[e]) for e in ele for item in self.feature_lst if e in item.keys()]) # TODO 调用人群覆盖接口 try: get_open_account_sql = f"select account_id from ctop_user_allocation where project_id = {self.project_id} and " \ f"account_status = 0 limit 1" account_df = pd.read_sql(get_open_account_sql, jeecg_product_db) account_id = account_df.account_id.values[0] # TODO 测试用, 上线前需要注释掉 account_id = 9774238 request_data = {'ages_range': age_code_transform(fea_combine.get('age')), 'gender': gender_code_transform(fea_combine.get('gender')), 'advertiser_id': account_id} request = requests.post(url=estimate_people_number_url, headers=headers, data=json.dumps(request_data, cls=NpEncoder)) response_data = json.loads(request.text) if response_data['code'] == 0: fea_combine['crowd_coverage_cnt'] = response_data['data'].get('audience_prediction_num') except: logger.error(f"人群预估覆盖接口调用报错,请求数据: {request_data}, 返回数据: {request.text}, 异常信息: {traceback.format_exc()}") fea_combine_df = fea_combine_df.append(pd.DataFrame([fea_combine]), ignore_index=True) # 计算结果写入数据库中 fea_combine_df['signature'] = self.signature fea_combine_df['stat_date'] = str(datetime.datetime.now().date()) fea_combine_df['project_id'] = self.project_id fea_combine_df.to_sql(name="ctop_ai_kuaishou_signature_recommended_target_combine_v2", con=jeecg_db, if_exists='append', index=False) # 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 = None # self.sample_size = None # self.sig_pos_pct = None # self.sig_neg_pct = None # # def get_bayes_estimate(self): # dim_combine_lst = [[key for key in fea.keys()] for fea in self.dim_features] # self.dim_features = reduce(update_dict, self.dim_features) # # # 通过 年龄对应的表: ctop_kuaishou_audience_daily_report_by_signature_age 来获取样本量,和正负比例值(其他表的数据可能会丢失导致不全) # get_sample_and_pct_sql = """ # select sum(aclick) aclick, sum(bclick) bclick, sum(activation) activation from %s where # signature = '%s' # """ % (config['bayesDim']['age']['table'], self.signature) # sample_pct_df = pd.read_sql(get_sample_and_pct_sql, product_engine) # if self.target_type == 'action_ratio': # self.sig_pos_pct = sample_pct_df['bclick'].sum() / sample_pct_df['aclick'].sum() # self.sig_neg_pct = 1 - self.sig_pos_pct # self.actual_prob = self.sig_pos_pct # self.sample_size = int(sample_pct_df.aclick.sum()) # elif self.target_type == 'convert_ratio': # self.sig_pos_pct = sample_pct_df['activation'].sum() / sample_pct_df['bclick'].sum() # self.sig_neg_pct = 1 - self.sig_pos_pct # self.actual_prob = self.sig_pos_pct # self.sample_size = int(sample_pct_df.bclick.sum()) # # for ele in product(*dim_combine_lst): # pos_pct_lst = [self.dim_features[key]['pos_pct'] for key in ele] # prob_pos = reduce(lambda x, y: x * y, pos_pct_lst) # # neg_pct_lst = [self.dim_features[key]['neg_pct'] for key in ele] # prob_neg = reduce(lambda x, y: x * y, neg_pct_lst) # # prob = (prob_pos * self.sig_pos_pct) / (prob_neg * self.sig_neg_pct) # out_dict = dict(zip([e.split('_')[0] for e in ele], [e.split('_')[-1] for e in ele])) # # # 计算该组合的概率值 与 实际投放的概率值 是否存在显著性差异:显著好、 显著差 # count = np.array([self.sample_size * prob, self.sample_size * self.actual_prob]) # nobs = np.array([self.sample_size, self.sample_size]) # z_score, p_value = proportions_ztest(count=count, nobs=nobs, value=None, alternative='two-sided', prop_var=False) # out_dict['z_score'] = z_score # out_dict['p_value'] = p_value # out_dict['sample_size'] = self.sample_size # # try: # # 获取该项目下在投的账号 # get_acc_sql = """select account_id from ctop_user_allocation where project_id = %s and account_status=0 limit 1""" % \ # self.project_id # account_df = pd.read_sql(get_acc_sql, product_engine) # account_id = account_df['account_id'].values[0] # account_id = 9774238 # request_data = {'region': city_code_transform(out_dict.get('city')), # 'ages_range': age_code_transform(out_dict.get('age')), # 'gender': gender_code_transform(out_dict.get('gender')), # 'advertiser_id': account_id} # # request = requests.post(url=estimate_people_number_url, # headers=headers, # data=json.dumps(request_data, cls=NpEncoder)) # response_data = json.loads(request.text) # if response_data['code'] == 0: # out_dict['crowd_coverage_cnt'] = response_data['data'].get('audience_prediction_num') # else: # logger.error("人群预估覆盖接口调用报错,请求数据为%s,返回数据为%s" % (str(request_data), str(response_data))) # except: # logger.error("人群预估覆盖接口调用报错,请求数据为%s,请求地址为%s, 返回数据为%s, 异常信息为%s" % # (str(request_data), estimate_people_number_url, str(response_data), traceback.format_exc())) # # 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=engine, # if_exists='append', # index=False) if __name__ == '__main__': # 1、读取配置文件 with open('/data/pythonProject/ai_target/config/config.yaml', mode='r', encoding='utf-8') as f: config = yaml.load(f.read(), Loader=yaml.FullLoader) # 2、 参与定向组合的维度 target_dim = [key for key in config['bayesDim'].keys() if config['bayesDim'][key]['isOn']] # 4、计算贝叶斯组合入库 for project_id in config['projectId']: # 4.1 获取指定项目下当前活跃的素材信息 # 如近60天内累计激活个数达到100个(开发环境或者生产环境,这部分都读取生产数据库) start_date = datetime.datetime.now().date() + datetime.timedelta(days=-config['activeMaterialFilterRule']['days']) sql = f"select signature from kuaishou_material_video_report_daily_dw where project_id = {project_id} " \ f"and stat_date >= {start_date.year * 10000 + start_date.month * 100 + start_date.day} group by signature " \ f"having sum(activation) >= {config['activeMaterialFilterRule']['activation']}" df = pd.read_sql(sql, application_product_db) active_signatures = list(df[~df.signature.isnull()].signature.values) active_signatures = active_signatures * 2 if len(active_signatures) == 1 else active_signatures # 4.2 在素材人群报表筛选里面累计100个激活 sql = f"select signature from ctop_kuaishou_audience_report_daily_age_material where signature in {tuple(active_signatures)} " \ f"group by signature having sum(activation) >= {config['getBaysCombineMaterialFilterRule']['activation']}" df = pd.read_sql(sql, jeecg_product_db) signature_lst = df['signature'].values # 这行代码用于测试 # signature_lst = ['f7adee9229e58176d132cf0bd7590a95'] # 4.2 计算指定素材定向 for sig in signature_lst: try: feature_lst = [] for dimension in target_dim: get_single_fea_ins = GetSingleDimFeatures(sig, dimension, config['bayesDim'][dimension]) # 计算滑窗组合 get_single_fea_ins.get_window_combine() get_single_fea_ins.get_features() feature_lst.append(get_single_fea_ins.features) # 依据特征值,计算多维度的组合预估值 get_composite_fea_ins = GetCompositeDimFeatures(sig, feature_lst, project_id) get_composite_fea_ins.get_composite_features() logger.info('project_id=%s, signature= %s 完成贝叶斯组合计算!' % (project_id, sig)) except: logger.error('project_id=%s, signature=%s 贝叶斯组合计算出错,异常信息为%s!' % (project_id, sig, traceback.format_exc())) logger.info('project_id is %s 完成贝叶斯组合计算!' % project_id)