BayesCombine.py 9.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188
  1. import datetime
  2. import json
  3. import os
  4. import sys
  5. import traceback
  6. from itertools import product
  7. import pandas as pd
  8. import requests
  9. import yaml
  10. from loguru import logger
  11. curr_path = os.path.abspath(os.path.dirname(__file__))
  12. project_root_path = curr_path[:curr_path.find("ai_target") + len("ai_target")]
  13. sys.path.append(project_root_path)
  14. from config.url_and_db import estimate_people_number_url, headers, jeecg_db, jeecg_product_db, application_product_db, application_inner_db
  15. from utils.BaseClass import NpEncoder
  16. from utils.commonFunc import age_code_transform, gender_code_transform
  17. logger.remove() # 删去 import logger之后自动产生的handler,不删除的话会出现重复输出的现象
  18. logger.add("/data/pythonProject/ai_target/logs/bayes_combine.{time:YYYY-MM-DD}.log",
  19. rotation="00:00",
  20. format="{time:YYYY-MM-DD HH:mm:ss,SSS} [{process}] [{thread}] {level} {file} {line} - {message}",
  21. level="INFO")
  22. class GetSingleDimFeatures(object):
  23. """
  24. 计算素材单维度滑窗组合转化占比情况
  25. """
  26. def __init__(self, signature, dim, dim_config, project_id):
  27. self.project_id = project_id
  28. self.signature = signature
  29. self.dim = dim
  30. self.dim_config = dim_config
  31. self.table = self.dim_config['table']
  32. self.filed_name = self.dim_config['fieldName']
  33. self.dim_value_lst = self.dim_config['Lst']
  34. self.ratio_threshold = self.dim_config['ratio_threshold']
  35. self.ratio_diff_threshold = self.dim_config['ratio_diff_threshold']
  36. self.window_size = self.dim_config['windowSize']
  37. self.window_combine = []
  38. self.features = {}
  39. def get_window_combine(self):
  40. """
  41. 根据size得到指定维度的滑窗组合。
  42. 如 age: ['18-23岁', '24-30岁','31-40岁',....] 和 size = 2
  43. 得到 ['18-23岁', '24-30岁'],['24-30岁','31-40岁'],......
  44. """
  45. start = 0
  46. while (start < len(self.dim_value_lst)) and (start + self.window_size <= len(self.dim_value_lst)):
  47. self.window_combine.append(self.dim_value_lst[start: start + self.window_size])
  48. start += 1
  49. # 添加一个全部的组合,等同于'不限'
  50. self.window_combine.append(['不限'])
  51. def get_features(self):
  52. # 1、获取素材在指定维度下的人群数据
  53. get_audience_sql = f"select signature, {self.filed_name}, sum(activation) activation " \
  54. f"from {self.table} where signature = '{self.signature}' and project_id = {self.project_id} " \
  55. f"group by signature, {self.filed_name}"
  56. audience_df = pd.read_sql(get_audience_sql, application_inner_db)
  57. for combine in self.window_combine:
  58. if combine != ['不限']:
  59. self.features['|'.join(combine)] = round(audience_df[audience_df[self.filed_name].isin(combine)].activation.sum() / \
  60. audience_df.activation.sum(), 4)
  61. else:
  62. if self.dim == 'gender':
  63. # 如果男/女 激活占比差异小于0.1, 说明素材的受众没有明显的性别倾向,可以对性别通投
  64. male_ratio = audience_df[audience_df['gender'] == '男'].activation.sum() / audience_df.activation.sum()
  65. female_ratio = audience_df[audience_df['gender'] == '女'].activation.sum() / audience_df.activation.sum()
  66. if abs(male_ratio - female_ratio) <= self.ratio_diff_threshold:
  67. self.features['不限'] = 1
  68. if self.dim == 'age':
  69. self.features['不限'] = 1
  70. # 过滤得到占比达到 ratio_threshold 的组合值
  71. self.features = {f"{self.dim}_{combine}": ratio for combine, ratio in self.features.items() if ratio > self.ratio_threshold}
  72. class GetCompositeDimFeatures(object):
  73. """
  74. """
  75. def __init__(self, signature, feature_lst, project_id):
  76. self.signature = signature
  77. self.feature_lst = feature_lst
  78. self.project_id = project_id
  79. def get_composite_features(self):
  80. feature_combine_lst = [list(item.keys()) for item in self.feature_lst]
  81. fea_combine_df = pd.DataFrame()
  82. for ele in product(*feature_combine_lst):
  83. fea_combine = dict(zip([e.split('_')[0] for e in ele], [e.split('_')[-1] for e in ele]))
  84. fea_combine['unlimited_cnt'] = list(fea_combine.values()).count('不限')
  85. fea_combine['features'] = str([(e.split('_')[0], e.split('_')[1], item[e]) for e in ele
  86. for item in self.feature_lst if e in item.keys()])
  87. try:
  88. get_open_account_sql = f"select account_id from ctop_user_allocation where project_id = {self.project_id} and " \
  89. f"account_status = 0 limit 1"
  90. account_df = pd.read_sql(get_open_account_sql, jeecg_product_db)
  91. account_id = account_df.account_id.values[0]
  92. # account_id = 9774238
  93. request_data = {'ages_range': age_code_transform(fea_combine.get('age')),
  94. 'gender': gender_code_transform(fea_combine.get('gender')),
  95. 'advertiser_id': account_id}
  96. request = requests.post(url=estimate_people_number_url,
  97. headers=headers,
  98. data=json.dumps(request_data, cls=NpEncoder))
  99. response_data = json.loads(request.text)
  100. if response_data['code'] == 0:
  101. fea_combine['crowd_coverage_cnt'] = response_data['data'].get('audience_prediction_num')
  102. except:
  103. logger.error(f"人群预估覆盖接口调用报错,请求数据: {request_data}, 返回数据: {request.text}, 异常信息: {traceback.format_exc()}")
  104. fea_combine_df = fea_combine_df.append(pd.DataFrame([fea_combine]), ignore_index=True)
  105. # 计算结果写入数据库中
  106. fea_combine_df['signature'] = self.signature
  107. fea_combine_df['stat_date'] = str(datetime.datetime.now().date())
  108. fea_combine_df['project_id'] = self.project_id
  109. fea_combine_df.to_sql(name="ctop_ai_kuaishou_signature_recommended_target_combine_v2",
  110. con=jeecg_db,
  111. if_exists='append',
  112. index=False)
  113. if __name__ == '__main__':
  114. # 1、读取配置文件
  115. with open('/data/pythonProject/ai_target/config/config.yaml', mode='r', encoding='utf-8') as f:
  116. config = yaml.load(f.read(), Loader=yaml.FullLoader)
  117. # 2、 参与定向组合的维度
  118. target_dim = [key for key in config['bayesDim'].keys() if config['bayesDim'][key]['isOn']]
  119. # 4、计算贝叶斯组合入库
  120. for project_id in config['projectId']:
  121. logger.info(f"开始获取项目{project_id}的当前活跃素材")
  122. # 4.1 获取指定项目下当前活跃的素材信息
  123. # 如近60天内累计激活个数达到100个(开发环境或者生产环境,这部分都读取生产数据库)
  124. days = config['activeMaterialFilterRule']['days']
  125. activation_sum = config['activeMaterialFilterRule']['activation']
  126. pay_first_pct = config['activeMaterialFilterRule']['event_pay_first_day_pct']
  127. sql = f"select signature from kuaishou_material_video_report_daily_dw where project_id = {project_id} " \
  128. f"and stat_date >= date_format(date_sub(now(), interval {days} DAY),'%%Y%%m%%d') group by signature " \
  129. f"having sum(activation) >= {activation_sum} " \
  130. f"and sum(event_pay_first_day) / sum(activation) >= {pay_first_pct}"
  131. df = pd.read_sql(sql, application_inner_db)
  132. active_signatures = list(df[~df.signature.isnull()].signature.values)
  133. active_signatures = active_signatures * 2 if len(active_signatures) == 1 else active_signatures
  134. # 4.2 在素材人群报表筛选里面累计100个激活
  135. logger.info(f"开始获取项目{project_id}的活跃素材的人群分析报表")
  136. sql = f"select signature from kuaishou_unit_audience_age_report_daily_d where signature in {tuple(active_signatures)} " \
  137. f"and project_id = {project_id} " \
  138. f"group by signature having sum(activation) >= {config['getBaysCombineMaterialFilterRule']['activation']}"
  139. df = pd.read_sql(sql, application_inner_db)
  140. signature_lst = df['signature'].values
  141. # 这行代码用于测试
  142. # signature_lst = ['f7adee9229e58176d132cf0bd7590a95']
  143. # 4.2 计算指定素材定向
  144. for sig in signature_lst:
  145. try:
  146. feature_lst = []
  147. for dimension in target_dim:
  148. get_single_fea_ins = GetSingleDimFeatures(sig, dimension, config['bayesDim'][dimension], project_id)
  149. # 计算滑窗组合
  150. get_single_fea_ins.get_window_combine()
  151. get_single_fea_ins.get_features()
  152. feature_lst.append(get_single_fea_ins.features)
  153. # 依据特征值,计算多维度的组合预估值
  154. get_composite_fea_ins = GetCompositeDimFeatures(sig, feature_lst, project_id)
  155. get_composite_fea_ins.get_composite_features()
  156. logger.info('project_id=%s, signature= %s 完成贝叶斯组合计算!' % (project_id, sig))
  157. except:
  158. logger.error('project_id=%s, signature=%s 贝叶斯组合计算出错,异常信息为%s!' % (project_id, sig, traceback.format_exc()))
  159. logger.info('project_id is %s 完成贝叶斯组合计算!' % project_id)