BayesCombine.py 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251
  1. import yaml
  2. import os
  3. import pandas as pd
  4. import numpy as np
  5. from utils.commonFunc import update_dict, get_db_engine, city_code_transform, age_code_transform, gender_code_transform
  6. from functools import reduce
  7. from itertools import product
  8. from datetime import datetime
  9. from statsmodels.stats.proportion import proportions_ztest
  10. class BayesFeatures(object):
  11. """
  12. 计算素材单维度的特征值
  13. """
  14. def __init__(self, signature, target_type, dim, dim_config):
  15. self.signature = signature
  16. self.target_type = target_type
  17. self.dim = dim
  18. self.dim_config = dim_config
  19. self.table = self.dim_config['table']
  20. self.file_name = self.dim_config['fieldName']
  21. self.dim_lst = self.dim_config['Lst']
  22. self.window_size = self.dim_config['windowSize']
  23. self.window_combine = []
  24. self.bayes_feature = {}
  25. def get_window_combine(self):
  26. """
  27. 根据size得到指定维度的滑窗组合。
  28. 如 age: ['18-23岁', '24-30岁','31-40岁',....] 和 size = 2
  29. 得到 ['18-23岁', '24-30岁'],['24-30岁','31-40岁'],......
  30. """
  31. start = 0
  32. while (start < len(self.dim_lst)) and (start + self.window_size <= len(self.dim_lst)):
  33. self.window_combine.append(self.dim_lst[start: start + self.window_size])
  34. start += 1
  35. # 添加一个全部的组合,等同于'不限'
  36. self.window_combine.append(['不限'])
  37. def get_bayes_feature(self):
  38. # 1、获取指定素材在指定维度下的人群数据(近一个月内的数据表现)
  39. sql = '''
  40. select signature,
  41. %s,
  42. sum(aclick) aclick,
  43. sum(bclick) bclick,
  44. sum(activation) activation
  45. from %s
  46. where signature = '%s'
  47. -- and datediff(now(),stat_date)<=30
  48. group by signature, %s
  49. ''' % (self.file_name, self.table, self.signature, self.file_name)
  50. df = pd.read_sql(sql, engine)
  51. if self.target_type == 'action_ratio':
  52. # 计算行为率(bclick/aclick)的组合特征值
  53. df['unbclick'] = df['aclick'] - df['bclick']
  54. for sub_combine in self.window_combine:
  55. if sub_combine != ['不限']:
  56. key = self.dim + '_' + '|'.join(sub_combine)
  57. if self.dim != 'city':
  58. pos_pct = df[df[self.file_name].isin(sub_combine)].bclick.sum() / df.bclick.sum()
  59. neg_pct = df[df[self.file_name].isin(sub_combine)].unbclick.sum() / df.unbclick.sum()
  60. if self.dim == 'city':
  61. # 获取 city_level 对应的 city_name
  62. city_lst = []
  63. for city_level in sub_combine:
  64. get_city_sql = """
  65. select city_name from ctop_kuaishou_city_level where city_level = '%s'
  66. """ % city_level
  67. city_df = pd.read_sql(get_city_sql, engine)
  68. city_lst.extend(city_df['city_name'].values)
  69. pos_pct = df[df[self.file_name].isin(city_lst)].bclick.sum() / df.bclick.sum()
  70. neg_pct = df[df[self.file_name].isin(city_lst)].unbclick.sum() / df.unbclick.sum()
  71. self.bayes_feature[key] = {'pos_pct': pos_pct, 'neg_pct': neg_pct}
  72. else:
  73. self.bayes_feature[self.dim + '_' + '不限'] = {'pos_pct': 1, 'neg_pct': 1}
  74. if self.target_type == 'convert_ratio':
  75. # 计算转化率(activation/bclick)的组合特征值
  76. df['unactivation'] = df['bclick'] - df['activation']
  77. for sub_combine in self.window_combine:
  78. if sub_combine != '不限':
  79. key = self.dim + '_' + '|'.join(sub_combine)
  80. if self.dim != 'city':
  81. pos_pct = df[df[self.file_name].isin(sub_combine)].activation.sum() / df.activation.sum()
  82. neg_pct = df[df[self.file_name].isin(sub_combine)].unactivation.sum() / df.unactivation.sum()
  83. if self.dim == 'city':
  84. # 获取 city_level 对应的 city_name
  85. city_lst = []
  86. for city_level in sub_combine:
  87. get_city_sql = """
  88. select city_name from ctop_kuaishou_city_level where city_level = '%s'
  89. """ % city_level
  90. city_df = pd.read_sql(get_city_sql, engine)
  91. city_lst.extend(city_df['city_name'].values)
  92. pos_pct = df[df[self.file_name].isin(city_lst)].activation.sum() / df.activation.sum()
  93. neg_pct = df[df[self.file_name].isin(city_lst)].unactivation.sum() / df.unactivation.sum()
  94. self.bayes_feature[key] = {'pos_pct': pos_pct, 'neg_pct': neg_pct}
  95. else:
  96. self.bayes_feature[self.dim + '_' + '不限'] = {'pos_pct': 1, 'neg_pct': 1}
  97. class BayesCombine(object):
  98. """
  99. 依据特征值,计算多维度的组合预估值
  100. """
  101. def __init__(self, signature, project_id, target_type, dim_features):
  102. self.signature = signature
  103. self.project_id = project_id
  104. self.target_type = target_type
  105. self.dim_features = dim_features
  106. self.bayes_combine_df = pd.DataFrame()
  107. self.actual_prob = None
  108. self.sample_size = None
  109. self.sig_pos_pct = None
  110. self.sig_neg_pct = None
  111. def get_bayes_estimate(self):
  112. dim_combine_lst = [[key for key in fea.keys()] for fea in self.dim_features]
  113. self.dim_features = reduce(update_dict, self.dim_features)
  114. # 通过 年龄对应的表: ctop_kuaishou_audience_daily_report_by_signature_age 来获取样本量,和正负比例值(其他表的数据可能会丢失导致不全)
  115. get_sample_and_pct_sql = """
  116. select sum(aclick) aclick, sum(bclick) bclick, sum(activation) activation from %s where
  117. signature = '%s'
  118. -- and datediff(now(),stat_date)<=30
  119. """ % (config['bayesDim']['age']['table'],self.signature)
  120. sample_pct_df = pd.read_sql(get_sample_and_pct_sql, engine)
  121. if self.target_type == 'action_ratio':
  122. self.sig_pos_pct = sample_pct_df['bclick'].sum() / sample_pct_df['aclick'].sum()
  123. self.sig_neg_pct = 1 - self.sig_pos_pct
  124. self.actual_prob = self.sig_pos_pct
  125. self.sample_size = int(sample_pct_df.aclick.sum())
  126. elif self.target_type == 'convert_ratio':
  127. self.sig_pos_pct = sample_pct_df['activation'].sum() / sample_pct_df['bclick'].sum()
  128. self.sig_neg_pct = 1 - self.sig_pos_pct
  129. self.actual_prob = self.sig_pos_pct
  130. self.sample_size = int(sample_pct_df.bclick.sum())
  131. for ele in product(*dim_combine_lst):
  132. pos_pct_lst = [self.dim_features[key]['pos_pct'] for key in ele]
  133. prob_pos = reduce(lambda x, y: x * y, pos_pct_lst)
  134. neg_pct_lst = [self.dim_features[key]['neg_pct'] for key in ele]
  135. prob_neg = reduce(lambda x, y: x * y, neg_pct_lst)
  136. prob = (prob_pos * self.sig_pos_pct) / (prob_neg * self.sig_neg_pct)
  137. out_dict = dict(zip([e.split('_')[0] for e in ele], [e.split('_')[-1] for e in ele]))
  138. # 计算该组合的概率值 与 实际投放的概率值 是否存在显著性差异:显著好、 显著差
  139. count = np.array([self.sample_size * prob, self.sample_size * self.actual_prob])
  140. nobs = np.array([self.sample_size, self.sample_size])
  141. z_score, p_value = proportions_ztest(count=count, nobs=nobs, value=None, alternative='two-sided', prop_var=False)
  142. out_dict['z_score'] = z_score
  143. out_dict['p_value'] = p_value
  144. out_dict['sample_size'] = self.sample_size
  145. # TODO 调用人群预估覆盖接口,得到该组合的人群覆盖数
  146. # 获取该项目下在投的账号
  147. get_acc_sql = """select account_id from ctop_user_allocation where project_id = %s and account_status=0 limit 1""" % \
  148. self.project_id
  149. account_df = pd.read_sql(get_acc_sql, engine)
  150. account_id = account_df['account_id'].values[0]
  151. request_data = {'region': city_code_transform(out_dict.get('city')),
  152. 'ages_range': age_code_transform(out_dict.get('age')),
  153. 'gender': gender_code_transform(out_dict.get('gender')),
  154. 'advertiser_id': account_id}
  155. out_dict['crowd_coverage_cnt'] = 1000
  156. out_dict['combine_estimate_prob'] = prob
  157. out_dict['actual_prob'] = self.actual_prob
  158. out_dict['signature'] = self.signature
  159. out_dict['project_id'] = self.project_id
  160. out_dict['target_type'] = self.target_type
  161. out_dict['stat_date'] = str(datetime.now().date())
  162. combine = pd.DataFrame([out_dict])
  163. self.bayes_combine_df = self.bayes_combine_df.append(combine, ignore_index=True)
  164. def write_to_db(self):
  165. self.bayes_combine_df.to_sql(name="ctop_ai_kuaishou_signature_recommended_target_combine",
  166. con=engine,
  167. if_exists='append',
  168. index=False)
  169. if __name__ == '__main__':
  170. # 1、读取配置文件
  171. with open('config/config.yaml', mode='r', encoding='utf-8') as f:
  172. config = yaml.load(f.read(), Loader=yaml.FullLoader)
  173. # 1-1 数据库连接引擎,依据开发环境/生产环境 进行切换
  174. if os.getenv('LYY_DEV', 'unknown') == 'dev':
  175. engine = get_db_engine(config['devDB'])
  176. else:
  177. engine = get_db_engine(config['productDB'])
  178. # 2、 参与定向组合的维度
  179. target_dim = [key for key in config['bayesDim'].keys() if config['bayesDim'][key]['isOn']]
  180. # 4、计算贝叶斯组合入库
  181. for project_id in config['projectId']:
  182. # 4.1 获取指定项目下当前活跃的素材信息, 如近3天内累计激活个数达到50个(开发环境或者生产环境,这部分都读取生产数据库)
  183. product_engine = get_db_engine(config['productDB'])
  184. sql = '''
  185. select signature from ctop_kuaishou_report_daily_material
  186. where account_id in (select account_id from ctop_user_allocation where project_id = %s)
  187. and datediff(now(),stat_date) <= %s
  188. group by signature
  189. having sum(activation) >= %s
  190. ''' % (project_id,
  191. config['activeMaterialFilterRule']['days'],
  192. config['activeMaterialFilterRule']['activation'])
  193. df = pd.read_sql(sql, product_engine)
  194. active_signatures = df[~df.signature.isnull()].signature.values
  195. # 4.2 在素材人群报表筛选里面近一个月累计100个激活
  196. # TODO 等生产的素材人群表存在之后,修改表名,添加 days
  197. sql = """
  198. select signature from ctop_kuaishou_audience_daily_report_by_signature_age
  199. where signature in %s
  200. group by signature
  201. having sum(activation) >= %s
  202. """ % (tuple(active_signatures),
  203. config['getBaysCombineMaterialFilterRule']['activation'])
  204. df = pd.read_sql(sql, engine)
  205. signature_lst = df['signature'].values
  206. # TODO 这行代码用于测试
  207. signature_lst = ['0070efb7557b2a04cf3d4a6f243c3cd8', '03b93728f0d82ea7865c3c7cf632bc1b']
  208. # 4.2 计算指定素材的贝叶斯特征
  209. for sig in signature_lst:
  210. for t_type in config['targetType']:
  211. bayes_feature_lst = []
  212. for dimension in target_dim:
  213. cls = BayesFeatures(sig, t_type, dimension, config['bayesDim'][dimension])
  214. # 计算滑窗组合
  215. cls.get_window_combine()
  216. cls.get_bayes_feature()
  217. bayes_feature_lst.append(cls.bayes_feature)
  218. # 依据特征值,计算多维度的组合预估值
  219. cls = BayesCombine(sig, project_id, t_type, bayes_feature_lst)
  220. cls.get_bayes_estimate()
  221. cls.write_to_db()