BayesCombine.py 14 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275
  1. import datetime
  2. import json
  3. import logging
  4. import traceback
  5. import uuid
  6. from itertools import product
  7. import pandas as pd
  8. import requests
  9. import yaml
  10. from concurrent_log import ConcurrentTimedRotatingFileHandler
  11. from config.url_and_db import estimate_people_number_url, headers, jeecg_db, jeecg_product_db, application_product_db
  12. from utils.BaseClass import NpEncoder
  13. from utils.commonFunc import get_db_engine, age_code_transform, gender_code_transform
  14. log_formatter = logging.Formatter('%(asctime)s %(levelname)s %(message)s', '%Y/%m/%d %I:%M:%S %p')
  15. log_handler = ConcurrentTimedRotatingFileHandler("/data/pythonProject/ai_target/logs/BayesCombine.log", when="midnight", backupCount=100)
  16. log_handler.setFormatter(log_formatter)
  17. logger = logging.getLogger('bayes_combine_logger')
  18. logger.addHandler(log_handler)
  19. logger.setLevel(logging.DEBUG)
  20. print('id of bayes_combine_logger %s' % id(logger))
  21. class GetSingleDimFeatures(object):
  22. """
  23. 计算素材单维度滑窗组合转化占比情况
  24. """
  25. def __init__(self, signature, dim, dim_config):
  26. self.signature = signature
  27. self.dim = dim
  28. self.dim_config = dim_config
  29. self.table = self.dim_config['table']
  30. self.filed_name = self.dim_config['fieldName']
  31. self.dim_value_lst = self.dim_config['Lst']
  32. self.ratio_threshold = self.dim_config['ratio_threshold']
  33. self.ratio_diff_threshold = self.dim_config['ratio_diff_threshold']
  34. self.window_size = self.dim_config['windowSize']
  35. self.window_combine = []
  36. self.features = {}
  37. def get_window_combine(self):
  38. """
  39. 根据size得到指定维度的滑窗组合。
  40. 如 age: ['18-23岁', '24-30岁','31-40岁',....] 和 size = 2
  41. 得到 ['18-23岁', '24-30岁'],['24-30岁','31-40岁'],......
  42. """
  43. start = 0
  44. while (start < len(self.dim_value_lst)) and (start + self.window_size <= len(self.dim_value_lst)):
  45. self.window_combine.append(self.dim_value_lst[start: start + self.window_size])
  46. start += 1
  47. # 添加一个全部的组合,等同于'不限'
  48. self.window_combine.append(['不限'])
  49. def get_features(self):
  50. # 1、获取素材在指定维度下的人群数据
  51. get_audience_sql = f"select signature, {self.filed_name}, sum(activation) activation " \
  52. f"from {self.table} where signature = '{self.signature}' group by signature, {self.filed_name}"
  53. audience_df = pd.read_sql(get_audience_sql, jeecg_product_db)
  54. for combine in self.window_combine:
  55. if combine != ['不限']:
  56. self.features['|'.join(combine)] = round(audience_df[audience_df[self.filed_name].isin(combine)].activation.sum() / \
  57. audience_df.activation.sum(), 4)
  58. else:
  59. if self.dim == 'gender':
  60. # 如果男/女 激活占比差异小于0.1, 说明素材的受众没有明显的性别倾向,可以对性别通投
  61. male_ratio = audience_df[audience_df['gender'] == '男'].activation.sum() / audience_df.activation.sum()
  62. female_ratio = audience_df[audience_df['gender'] == '女'].activation.sum() / audience_df.activation.sum()
  63. if abs(male_ratio - female_ratio) <= self.ratio_diff_threshold:
  64. self.features['不限'] = 1
  65. if self.dim == 'age':
  66. self.features['不限'] = 1
  67. # 过滤得到占比达到 ratio_threshold 的组合值
  68. self.features = {f"{self.dim}_{combine}": ratio for combine, ratio in self.features.items() if ratio > self.ratio_threshold}
  69. class GetCompositeDimFeatures(object):
  70. """
  71. """
  72. def __init__(self, signature, feature_lst, project_id):
  73. self.signature = signature
  74. self.feature_lst = feature_lst
  75. self.project_id = project_id
  76. def get_composite_features(self):
  77. feature_combine_lst = [list(item.keys()) for item in self.feature_lst]
  78. fea_combine_df = pd.DataFrame()
  79. for ele in product(*feature_combine_lst):
  80. fea_combine = dict(zip([e.split('_')[0] for e in ele], [e.split('_')[-1] for e in ele]))
  81. fea_combine['unlimited_cnt'] = list(fea_combine.values()).count('不限')
  82. fea_combine['features'] = str([(e.split('_')[0], e.split('_')[1], item[e]) for e in ele
  83. for item in self.feature_lst if e in item.keys()])
  84. # TODO 调用人群覆盖接口
  85. try:
  86. get_open_account_sql = f"select account_id from ctop_user_allocation where project_id = {self.project_id} and " \
  87. f"account_status = 0 limit 1"
  88. account_df = pd.read_sql(get_open_account_sql, jeecg_product_db)
  89. account_id = account_df.account_id.values[0]
  90. # TODO 测试用, 上线前需要注释掉
  91. account_id = 9774238
  92. request_data = {'ages_range': age_code_transform(fea_combine.get('age')),
  93. 'gender': gender_code_transform(fea_combine.get('gender')),
  94. 'advertiser_id': account_id}
  95. request = requests.post(url=estimate_people_number_url,
  96. headers=headers,
  97. data=json.dumps(request_data, cls=NpEncoder))
  98. response_data = json.loads(request.text)
  99. if response_data['code'] == 0:
  100. fea_combine['crowd_coverage_cnt'] = response_data['data'].get('audience_prediction_num')
  101. except:
  102. logger.error(f"人群预估覆盖接口调用报错,请求数据: {request_data}, 返回数据: {request.text}, 异常信息: {traceback.format_exc()}")
  103. fea_combine_df = fea_combine_df.append(pd.DataFrame([fea_combine]), ignore_index=True)
  104. # 计算结果写入数据库中
  105. fea_combine_df['signature'] = self.signature
  106. fea_combine_df['stat_date'] = str(datetime.datetime.now().date())
  107. fea_combine_df['project_id'] = self.project_id
  108. fea_combine_df.to_sql(name="ctop_ai_kuaishou_signature_recommended_target_combine_v2",
  109. con=jeecg_db,
  110. if_exists='append',
  111. index=False)
  112. # class BayesCombine(object):
  113. # """
  114. # 依据特征值,计算多维度的组合预估值
  115. # """
  116. #
  117. # def __init__(self, signature, project_id, target_type, dim_features):
  118. # self.signature = signature
  119. # self.project_id = project_id
  120. # self.target_type = target_type
  121. # self.dim_features = dim_features
  122. # self.bayes_combine_df = pd.DataFrame()
  123. # self.actual_prob = None
  124. # self.sample_size = None
  125. # self.sig_pos_pct = None
  126. # self.sig_neg_pct = None
  127. #
  128. # def get_bayes_estimate(self):
  129. # dim_combine_lst = [[key for key in fea.keys()] for fea in self.dim_features]
  130. # self.dim_features = reduce(update_dict, self.dim_features)
  131. #
  132. # # 通过 年龄对应的表: ctop_kuaishou_audience_daily_report_by_signature_age 来获取样本量,和正负比例值(其他表的数据可能会丢失导致不全)
  133. # get_sample_and_pct_sql = """
  134. # select sum(aclick) aclick, sum(bclick) bclick, sum(activation) activation from %s where
  135. # signature = '%s'
  136. # """ % (config['bayesDim']['age']['table'], self.signature)
  137. # sample_pct_df = pd.read_sql(get_sample_and_pct_sql, product_engine)
  138. # if self.target_type == 'action_ratio':
  139. # self.sig_pos_pct = sample_pct_df['bclick'].sum() / sample_pct_df['aclick'].sum()
  140. # self.sig_neg_pct = 1 - self.sig_pos_pct
  141. # self.actual_prob = self.sig_pos_pct
  142. # self.sample_size = int(sample_pct_df.aclick.sum())
  143. # elif self.target_type == 'convert_ratio':
  144. # self.sig_pos_pct = sample_pct_df['activation'].sum() / sample_pct_df['bclick'].sum()
  145. # self.sig_neg_pct = 1 - self.sig_pos_pct
  146. # self.actual_prob = self.sig_pos_pct
  147. # self.sample_size = int(sample_pct_df.bclick.sum())
  148. #
  149. # for ele in product(*dim_combine_lst):
  150. # pos_pct_lst = [self.dim_features[key]['pos_pct'] for key in ele]
  151. # prob_pos = reduce(lambda x, y: x * y, pos_pct_lst)
  152. #
  153. # neg_pct_lst = [self.dim_features[key]['neg_pct'] for key in ele]
  154. # prob_neg = reduce(lambda x, y: x * y, neg_pct_lst)
  155. #
  156. # prob = (prob_pos * self.sig_pos_pct) / (prob_neg * self.sig_neg_pct)
  157. # out_dict = dict(zip([e.split('_')[0] for e in ele], [e.split('_')[-1] for e in ele]))
  158. #
  159. # # 计算该组合的概率值 与 实际投放的概率值 是否存在显著性差异:显著好、 显著差
  160. # count = np.array([self.sample_size * prob, self.sample_size * self.actual_prob])
  161. # nobs = np.array([self.sample_size, self.sample_size])
  162. # z_score, p_value = proportions_ztest(count=count, nobs=nobs, value=None, alternative='two-sided', prop_var=False)
  163. # out_dict['z_score'] = z_score
  164. # out_dict['p_value'] = p_value
  165. # out_dict['sample_size'] = self.sample_size
  166. #
  167. # try:
  168. # # 获取该项目下在投的账号
  169. # get_acc_sql = """select account_id from ctop_user_allocation where project_id = %s and account_status=0 limit 1""" % \
  170. # self.project_id
  171. # account_df = pd.read_sql(get_acc_sql, product_engine)
  172. # account_id = account_df['account_id'].values[0]
  173. # account_id = 9774238
  174. # request_data = {'region': city_code_transform(out_dict.get('city')),
  175. # 'ages_range': age_code_transform(out_dict.get('age')),
  176. # 'gender': gender_code_transform(out_dict.get('gender')),
  177. # 'advertiser_id': account_id}
  178. #
  179. # request = requests.post(url=estimate_people_number_url,
  180. # headers=headers,
  181. # data=json.dumps(request_data, cls=NpEncoder))
  182. # response_data = json.loads(request.text)
  183. # if response_data['code'] == 0:
  184. # out_dict['crowd_coverage_cnt'] = response_data['data'].get('audience_prediction_num')
  185. # else:
  186. # logger.error("人群预估覆盖接口调用报错,请求数据为%s,返回数据为%s" % (str(request_data), str(response_data)))
  187. # except:
  188. # logger.error("人群预估覆盖接口调用报错,请求数据为%s,请求地址为%s, 返回数据为%s, 异常信息为%s" %
  189. # (str(request_data), estimate_people_number_url, str(response_data), traceback.format_exc()))
  190. #
  191. # out_dict['combine_estimate_prob'] = prob
  192. # out_dict['actual_prob'] = self.actual_prob
  193. # out_dict['signature'] = self.signature
  194. # out_dict['project_id'] = self.project_id
  195. # out_dict['target_type'] = self.target_type
  196. # out_dict['stat_date'] = str(datetime.now().date())
  197. #
  198. # combine = pd.DataFrame([out_dict])
  199. # self.bayes_combine_df = self.bayes_combine_df.append(combine, ignore_index=True)
  200. #
  201. # def write_to_db(self):
  202. # self.bayes_combine_df.to_sql(name="ctop_ai_kuaishou_signature_recommended_target_combine",
  203. # con=engine,
  204. # if_exists='append',
  205. # index=False)
  206. if __name__ == '__main__':
  207. # 1、读取配置文件
  208. with open('/data/pythonProject/ai_target/config/config.yaml', mode='r', encoding='utf-8') as f:
  209. config = yaml.load(f.read(), Loader=yaml.FullLoader)
  210. # 2、 参与定向组合的维度
  211. target_dim = [key for key in config['bayesDim'].keys() if config['bayesDim'][key]['isOn']]
  212. # 4、计算贝叶斯组合入库
  213. for project_id in config['projectId']:
  214. # 4.1 获取指定项目下当前活跃的素材信息
  215. # 如近60天内累计激活个数达到100个(开发环境或者生产环境,这部分都读取生产数据库)
  216. start_date = datetime.datetime.now().date() + datetime.timedelta(days=-config['activeMaterialFilterRule']['days'])
  217. sql = f"select signature from kuaishou_material_video_report_daily_dw where project_id = {project_id} " \
  218. f"and stat_date >= {start_date.year * 10000 + start_date.month * 100 + start_date.day} group by signature " \
  219. f"having sum(activation) >= {config['activeMaterialFilterRule']['activation']}"
  220. df = pd.read_sql(sql, application_product_db)
  221. active_signatures = list(df[~df.signature.isnull()].signature.values)
  222. active_signatures = active_signatures * 2 if len(active_signatures) == 1 else active_signatures
  223. # 4.2 在素材人群报表筛选里面累计100个激活
  224. sql = f"select signature from ctop_kuaishou_audience_report_daily_age_material where signature in {tuple(active_signatures)} " \
  225. f"group by signature having sum(activation) >= {config['getBaysCombineMaterialFilterRule']['activation']}"
  226. df = pd.read_sql(sql, jeecg_product_db)
  227. signature_lst = df['signature'].values
  228. # 这行代码用于测试
  229. # signature_lst = ['f7adee9229e58176d132cf0bd7590a95']
  230. # 4.2 计算指定素材定向
  231. for sig in signature_lst:
  232. try:
  233. feature_lst = []
  234. for dimension in target_dim:
  235. get_single_fea_ins = GetSingleDimFeatures(sig, dimension, config['bayesDim'][dimension])
  236. # 计算滑窗组合
  237. get_single_fea_ins.get_window_combine()
  238. get_single_fea_ins.get_features()
  239. feature_lst.append(get_single_fea_ins.features)
  240. # 依据特征值,计算多维度的组合预估值
  241. get_composite_fea_ins = GetCompositeDimFeatures(sig, feature_lst, project_id)
  242. get_composite_fea_ins.get_composite_features()
  243. logger.info('project_id=%s, signature= %s 完成贝叶斯组合计算!' % (project_id, sig))
  244. except:
  245. logger.error('project_id=%s, signature=%s 贝叶斯组合计算出错,异常信息为%s!' % (project_id, sig, traceback.format_exc()))
  246. logger.info('project_id is %s 完成贝叶斯组合计算!' % project_id)