import datetime import sys import numpy as np import pandas as pd from odps import ODPS from utils import data_check, get_feature_data, send_msg_to_feishu, RedisHelper from config import set_config from log import Log config_, _ = set_config() log_ = Log() redis_helper = RedisHelper() def predict_user_group_share_rate(dt, app_type): """预估用户组对应的有广告时分享率""" # 获取用户组特征 project = config_.ad_model_data['users_share_rate'].get('project') table = config_.ad_model_data['users_share_rate'].get('table') features = [ 'apptype', 'group', 'sharerate_all', 'sharerate_ad' ] user_group_df = get_feature_data(project=project, table=table, features=features, dt=dt) user_group_df['apptype'] = user_group_df['apptype'].astype(int) user_group_df = user_group_df[user_group_df['apptype'] == app_type] user_group_df['sharerate_all'] = user_group_df['sharerate_all'].astype(float) user_group_df['sharerate_ad'] = user_group_df['sharerate_ad'].astype(float) # 获取有广告时所有用户组近30天的分享率 ad_all_group_share_rate = user_group_df[user_group_df['group'] == 'allmids']['sharerate_ad'].values[0] user_group_df = user_group_df[user_group_df['group'] != 'allmids'] # 计算用户组有广告时分享率 user_group_df['group_ad_share_rate'] = \ user_group_df['sharerate_ad'] * float(ad_all_group_share_rate) / user_group_df['sharerate_all'] return user_group_df def predict_video_share_rate(dt, app_type): """预估视频有广告时分享率""" # 获取视频特征 project = config_.ad_model_data['videos_share_rate'].get('project') table = config_.ad_model_data['videos_share_rate'].get('table') features = [ 'apptype', 'videoid', 'sharerate_all', 'sharerate_ad' ] video_df = get_feature_data(project=project, table=table, features=features, dt=dt) video_df['apptype'] = video_df['apptype'].astype(int) video_df = video_df[video_df['apptype'] == app_type] video_df['sharerate_all'] = video_df['sharerate_all'].astype(float) video_df['sharerate_ad'] = video_df['sharerate_ad'].astype(float) # 获取有广告时所有视频近30天的分享率 ad_all_videos_share_rate = video_df[video_df['videoid'] == 'allvideos']['sharerate_ad'].values[0] video_df = video_df[video_df['videoid'] != 'allvideos'] # 计算视频有广告时分享率 video_df['video_ad_share_rate'] = \ video_df['sharerate_ad'] * float(ad_all_videos_share_rate) / video_df['sharerate_all'] return video_df def predict_ad_group_video(dt, config_key, config_param): log_.info(f"config_key = {config_key} update start ...") # 获取用户组预测值 user_data_key = config_param['user'].get('data') user_rule_key = config_param['user'].get('rule') group_key_name = f"{config_.KEY_NAME_PREFIX_AD_GROUP}{user_data_key}:{user_rule_key}:{dt}" group_data = redis_helper.get_all_data_from_zset(key_name=group_key_name, with_scores=True) if group_data is None: log_.info(f"group data is None!") group_df = pd.DataFrame(data=group_data, columns=['group', 'group_ad_share_rate']) group_df = group_df[group_df['group'] != 'mean_group'] log_.info(f"group_df count = {len(group_df)}") # 获取视频预测值 video_data_key = config_param['video'].get('data') video_key_name = f"{config_.KEY_NAME_PREFIX_AD_VIDEO}{video_data_key}:{dt}" video_data = redis_helper.get_all_data_from_zset(key_name=video_key_name, with_scores=True) if video_data is None: log_.info(f"video data is None!") video_df = pd.DataFrame(data=video_data, columns=['videoid', 'video_ad_share_rate']) video_df = video_df[video_df['videoid'] != -1] log_.info(f"video_df count = {len(video_df)}") if len(group_df) == 0 or len(video_df) == 0: sys.exit(1) predict_df = video_df all_group_data = [] for index, item in group_df.iterrows(): predict_df[item['group']] = predict_df['video_ad_share_rate'] * item['group_ad_share_rate'] all_group_data.extend(predict_df[item['group']].tolist()) # 计算对应的阈值 ad_threshold_mappings = config_.AD_ABTEST_THRESHOLD_CONFIG.get(config_key.split('-')[0]) for abtest_group, ad_threshold_mapping in ad_threshold_mappings.items(): threshold_data = {} for _, item in group_df.iterrows(): # 获取分组对应的均值作为阈值 threshold_data[item['group']] = predict_df[item['group']].mean() * ad_threshold_mapping['group'] threshold_data['mean_group'] = np.mean(all_group_data) * ad_threshold_mapping['mean_group'] log_.info(f"config_key = {config_key}, abtest_group = {abtest_group}, threshold_data = {threshold_data}") # 将阈值写入redis abtest_config_list = config_key.split('-') abtest_id, abtest_config_tag = abtest_config_list[0], abtest_config_list[1] for key, val in threshold_data.items(): key_name = f"{config_.KEY_NAME_PREFIX_AD_THRESHOLD}{abtest_id}:{abtest_config_tag}:{abtest_group}:{key}" redis_helper.set_data_to_redis(key_name=key_name, value=val, expire_time=2 * 24 * 3600) # 计算关怀模式实验阈值 并 写入Redis care_model = config_param.get('care_model', None) threshold_rate = config_param.get('threshold_rate', None) if care_model is True: care_model_threshold_data = {} for key, val in threshold_data.items(): up_val = val * threshold_rate care_model_threshold_data[key] = up_val up_key_name = \ f"{config_.KEY_NAME_PREFIX_AD_THRESHOLD_CARE_MODEL}{abtest_id}:{abtest_config_tag}:{abtest_group}:{key}" redis_helper.set_data_to_redis(key_name=up_key_name, value=up_val, expire_time=2 * 24 * 3600) log_.info(f"config_key = {config_key}, abtest_group = {abtest_group}, " f"care_model_threshold_data = {care_model_threshold_data}") # predict_df.to_csv(f'./data/ad_user_video_predict_{config_key}.csv') log_.info(f"config_key = {config_key} update end!") def predict(): now_date = datetime.datetime.today() dt = datetime.datetime.strftime(now_date, '%Y%m%d') log_.info(f"dt = {dt}") params = config_.AD_ABTEST_CONFIG threshold_record = {} for config_key, config_param in params.items(): predict_ad_group_video(dt=dt, config_key=config_key, config_param=config_param) # 阈值参数记录 ad_threshold_mappings = config_.AD_ABTEST_THRESHOLD_CONFIG.get(config_key.split('-')[0]) for abtest_group, ad_threshold_mapping in ad_threshold_mappings.items(): threshold_record[f"{config_key}-{abtest_group}"] = ad_threshold_mapping['group'] redis_helper.set_data_to_redis(key_name=config_.KEY_NAME_PREFIX_AD_THRESHOLD_RECORD, value=str(threshold_record), expire_time=24*3600) if __name__ == '__main__': # predict_ad_group_video() predict()