Merge branch 'feature_2023070514_liqian_add_ad_abtest'

ad_threshold_update_task.sh

@@ -4,12 +4,14 @@ if [[ $ROV_OFFLINE_ENV == 'test' ]]; then
     cd /data2/rov-offline &&
     /root/anaconda3/bin/python /data2/rov-offline/ad_users_data_update.py &&
     /root/anaconda3/bin/python /data2/rov-offline/ad_users_data_update_new.py &&
+    /root/anaconda3/bin/python /data2/rov-offline/ad_user_data_with_out_update.py &&
     /root/anaconda3/bin/python /data2/rov-offline/ad_video_data_update.py &&
     /root/anaconda3/bin/python /data2/rov-offline/ad_user_video_predict.py
 elif [[ $ROV_OFFLINE_ENV == 'pro' ]]; then
     cd /data/rov-offline &&
     /root/anaconda3/bin/python /data/rov-offline/ad_users_data_update.py &&
     /root/anaconda3/bin/python /data/rov-offline/ad_users_data_update_new.py &&
+    /root/anaconda3/bin/python /data/rov-offline/ad_user_data_with_out_update.py &&
     /root/anaconda3/bin/python /data/rov-offline/ad_video_data_update.py &&
     /root/anaconda3/bin/python /data/rov-offline/ad_user_video_predict.py
 fi

ad_user_data_with_out_update.py

@@ -0,0 +1,115 @@
+import datetime
+import traceback
+from threading import Timer
+from utils import RedisHelper, data_check, get_feature_data, send_msg_to_feishu
+from config import set_config
+from log import Log
+config_, _ = set_config()
+log_ = Log()
+redis_helper = RedisHelper()
+
+features = [
+    'apptype',
+    'group',
+    'adrate',
+    'outrate',
+    'adrate_out'
+]
+
+
+def predict_user_group_out_rate(user_group_initial_df, dt, data_params, rule_params, param):
+    """预估用户组对应的有广告时不直接跳出的概率"""
+    # 获取对应的参数
+    data_key = param.get('data')
+    data_param = data_params.get(data_key)
+    rule_key = param.get('rule')
+    rule_param = rule_params.get(rule_key)
+
+    # 获取对应的用户组特征
+    user_group_df = user_group_initial_df.copy()
+    user_group_df['apptype'] = user_group_df['apptype'].astype(int)
+    user_group_df = user_group_df[user_group_df['apptype'] == data_param]
+    user_group_df['adrate'].fillna(0, inplace=True)
+    user_group_df['outrate'].fillna(0, inplace=True)
+    user_group_df['adrate_out'].fillna(0, inplace=True)
+    user_group_df['adrate'] = user_group_df['adrate'].astype(float)
+    user_group_df['outrate'] = user_group_df['outrate'].astype(float)
+    user_group_df['adrate_out'] = user_group_df['adrate_out'].astype(float)
+
+    # 获取对应的用户分组数据
+    user_group_list = rule_param.get('group_list')
+    user_group_df = user_group_df[user_group_df['group'].isin(user_group_list)]
+
+    # 去除对应无广告用户组
+    if rule_param.get('remove_no_ad_group') is True:
+        user_group_df = user_group_df[~user_group_df['group'].isin(rule_param.get('no_ad_mid_group_list'))]
+
+    # 计算用户组有广告时直接跳出的概率
+    user_group_df = user_group_df[user_group_df['adrate'] != 0]
+    user_group_df['group_ad_out_rate'] = \
+        user_group_df['adrate_out'] * user_group_df['outrate'] / user_group_df['adrate']
+    user_group_df['group_ad_out_rate'].fillna(0, inplace=True)
+    # 计算用户组有广告时不直接跳出的概率
+    user_group_df['group_ad_no_out_rate'] = 1 - user_group_df['group_ad_out_rate']
+    # log_.info(f"user_group_df:\n{user_group_df}")
+    # 结果写入redis
+    key_name = f"{config_.KEY_NAME_PREFIX_AD_GROUP}{data_key}:{rule_key}:{dt}"
+    redis_data = {}
+    for index, item in user_group_df.iterrows():
+        redis_data[item['group']] = item['group_ad_no_out_rate']
+    group_ad_out_rate_mean = user_group_df['group_ad_no_out_rate'].mean()
+    redis_data['mean_group'] = group_ad_out_rate_mean
+    if len(redis_data) > 0:
+        redis_helper = RedisHelper()
+        redis_helper.add_data_with_zset(key_name=key_name, data=redis_data, expire_time=2 * 24 * 3600)
+    return user_group_df
+
+
+def update_users_data(project, table, dt, update_params):
+    """预估用户组有广告时直接跳出的概率"""
+    # 获取用户组特征
+    user_group_initial_df = get_feature_data(project=project, table=table, features=features, dt=dt)
+    data_params = update_params.get('data_params')
+    rule_params = update_params.get('rule_params')
+    for param in update_params.get('params_list'):
+        log_.info(f"param = {param} update start...")
+        predict_user_group_out_rate(user_group_initial_df=user_group_initial_df,
+                                    dt=dt,
+                                    data_params=data_params,
+                                    rule_params=rule_params,
+                                    param=param)
+        log_.info(f"param = {param} update end!")
+
+
+def timer_check():
+    try:
+        update_params = config_.AD_USER_WITH_OUT_PARAMS
+        project = config_.ad_model_data['users_data_with_out'].get('project')
+        table = config_.ad_model_data['users_data_with_out'].get('table')
+        now_date = datetime.datetime.today()
+        dt = datetime.datetime.strftime(now_date, '%Y%m%d')
+        log_.info(f"now_date: {dt}")
+        # 查看当前更新的数据是否已准备好
+        data_count = data_check(project=project, table=table, dt=dt)
+        if data_count > 0:
+            log_.info(f"ad user group data count = {data_count}")
+            # 数据准备好，进行更新
+            update_users_data(project=project, table=table, dt=dt, update_params=update_params)
+            log_.info(f"ad user group data update end!")
+        else:
+            # 数据没准备好，1分钟后重新检查
+            Timer(60, timer_check).start()
+
+    except Exception as e:
+        log_.error(f"用户组直接跳出的概率预测数据更新失败, exception: {e}, traceback: {traceback.format_exc()}")
+        send_msg_to_feishu(
+            webhook=config_.FEISHU_ROBOT['server_robot'].get('webhook'),
+            key_word=config_.FEISHU_ROBOT['server_robot'].get('key_word'),
+            msg_text=f"rov-offline{config_.ENV_TEXT} - 用户组直接跳出的概率预测数据更新失败\n"
+                     f"exception: {e}\n"
+                     f"traceback: {traceback.format_exc()}"
+        )
+
+
+if __name__ == '__main__':
+    timer_check()

ad_video_data_update.py

@@ -24,6 +24,14 @@ features_new = [
     'adrate_share'
 ]
 
+features_with_out = [
+    'apptype',
+    'videoid',
+    'adrate',  # 出广告的概率
+    'outrate',  # 被直接跳出的概率
+    'adrate_out'  # 被直接跳出时出广告的概率
+]
+
 
 def get_top10_abnormal_videos_return(dt, filter_param):
     """获取昨日各端top10中的异常视频（裂变视频）"""
@@ -185,6 +193,74 @@ def update_videos_data_new(project, table, dt, update_params, top10_abnormal_vid
         log_.info(f"data_key = {data_key} update end!")
 
 
+def predict_video_out_rate(video_initial_df, dt, data_key, data_param, top10_abnormal_videos):
+    """预估视频有广告时不被直接跳出的概率"""
+    # 获取对应的视频特征
+    video_df = video_initial_df.copy()
+    video_df['apptype'] = video_df['apptype'].astype(int)
+    video_df = video_df[video_df['apptype'] == int(data_param)]
+    log_.info(f"video_df length: {len(video_df)}")
+    video_df['adrate'].fillna(0, inplace=True)
+    video_df['outrate'].fillna(0, inplace=True)
+    video_df['adrate_out'].fillna(0, inplace=True)
+    video_df['adrate'] = video_df['adrate'].astype(float)
+    video_df['outrate'] = video_df['outrate'].astype(float)
+    video_df['adrate_out'] = video_df['adrate_out'].astype(float)
+
+    # 剔除异常视频数据
+    video_df['videoid'] = video_df['videoid'].astype(int)
+    top10_abnormal_video_ids = top10_abnormal_videos.get(int(data_param), None)
+    # print(int(data_param), len(video_df), top10_abnormal_video_ids)
+    if top10_abnormal_video_ids is not None:
+        video_df = video_df[~video_df['videoid'].isin(top10_abnormal_video_ids)]
+        # print(len(video_df))
+    # 计算视频有广告时被直接跳出的概率
+    video_df = video_df[video_df['adrate'] != 0]
+    video_df = video_df[video_df['adrate_out'] != 0]
+    video_df['video_ad_out_rate'] = \
+        video_df['adrate_out'] * video_df['outrate'] / video_df['adrate']
+    video_df['video_ad_out_rate'].fillna(0, inplace=True)
+    # 计算视频有广告时不被直接跳出的概率
+    video_df['video_ad_no_out_rate'] = 1 - video_df['video_ad_out_rate']
+    # print(len(video_df))
+    # video_df = video_df[video_df['video_ad_no_out_rate'] != 0]
+    # log_.info(f"video_df: {video_df}")
+    log_.info(f"video_df filtered 0 length: {len(video_df)}")
+    # video_df = video_df[video_df['video_ad_no_out_rate'] != 1]
+    # log_.info(f"video_df: {video_df}")
+    # log_.info(f"video_df filtered 0 length: {len(video_df)}")
+    # 结果写入redis
+    key_name = f"{config_.KEY_NAME_PREFIX_AD_VIDEO}{data_key}:{dt}"
+    redis_data = {}
+    for index, item in video_df.iterrows():
+        redis_data[int(item['videoid'])] = item['video_ad_no_out_rate']
+    group_ad_out_rate_mean = video_df['video_ad_no_out_rate'].mean()
+    redis_data[-1] = group_ad_out_rate_mean
+    # 异常视频给定值：mean/3
+    if top10_abnormal_video_ids is not None:
+        abnormal_video_param = config_.AD_ABNORMAL_VIDEOS_PARAM.get(data_key, 1)
+        print(data_key, data_param, abnormal_video_param)
+        for abnormal_video_id in top10_abnormal_video_ids:
+            print(abnormal_video_id, group_ad_out_rate_mean, group_ad_out_rate_mean * abnormal_video_param)
+            redis_data[int(abnormal_video_id)] = group_ad_out_rate_mean * abnormal_video_param
+    log_.info(f"redis_data count: {len(redis_data)}")
+    if len(redis_data) > 0:
+        redis_helper = RedisHelper()
+        redis_helper.add_data_with_zset(key_name=key_name, data=redis_data, expire_time=2 * 24 * 3600)
+    return video_df
+
+
+def update_videos_data_with_out(project, table, dt, update_params, top10_abnormal_videos):
+    """预估视频有广告时被直接跳出的概率"""
+    # 获取视频特征
+    video_initial_df = get_feature_data(project=project, table=table, features=features_with_out, dt=dt)
+    for data_key, data_param in update_params.items():
+        log_.info(f"data_key = {data_key} update start...")
+        predict_video_out_rate(video_initial_df=video_initial_df, dt=dt, data_key=data_key,
+                               data_param=data_param, top10_abnormal_videos=top10_abnormal_videos)
+        log_.info(f"data_key = {data_key} update end!")
+
+
 def timer_check(dt, video_key, video_params, top10_abnormal_videos):
     log_.info(f"video_key = {video_key}")
     project = config_.ad_model_data[video_key].get('project')
@@ -197,6 +273,9 @@ def timer_check(dt, video_key, video_params, top10_abnormal_videos):
         if video_key == 'videos_data_alladtype':
             update_videos_data_new(project=project, table=table, dt=dt, update_params=video_params,
                                    top10_abnormal_videos=top10_abnormal_videos)
+        elif video_key == 'videos_data_with_out_alladtype':
+            update_videos_data_with_out(project=project, table=table, dt=dt, update_params=video_params,
+                                        top10_abnormal_videos=top10_abnormal_videos)
         else:
             update_videos_data(project=project, table=table, dt=dt, update_params=video_params,
                                top10_abnormal_videos=top10_abnormal_videos)

config.py

@@ -771,6 +771,14 @@ class BaseConfig(object):
             'project': 'loghubods',
             'table': 'video_data_admodel_alladtype'
         },  # 新的视频侧数据：所有广告类型数据，按照videoId统计（各视频出广告的概率，各视频被分享的概率，各视频被分享的情况下出广告的概率）
+        'users_data_with_out': {
+            'project': 'loghubods',
+            'table': 'usergroup_data_with_out_admodel'
+        },  # 以是否直接跳出为目标的用户侧数据：按照用户分组统计（各用户组出广告的概率，各用户组的直接跳出率，各用户组直接跳出的情况下出广告的概率）
+        'videos_data_with_out_alladtype': {
+            'project': 'loghubods',
+            'table': 'video_data_with_out_admodel_alladtype'
+        },  # 以是否直接跳出为目标的视频侧数据：所有广告类型数据，按照videoId统计（各视频出广告的概率，各视频被直接跳出的概率，各视频被直接跳出的情况下出广告的概率）
     }
 
     # 自动调整广告模型阈值数据
@@ -832,6 +840,10 @@ class BaseConfig(object):
             'videos4new': APP_TYPE['LOVE_LIVE'],  # 票圈视频
             'videos0new': APP_TYPE['VLOG'],  # vlog
         },
+        # 以是否直接跳出为目标的视频侧数据：所有广告类型视频数据
+        'videos_data_with_out_alladtype': {
+            'videos5out': APP_TYPE['LONG_VIDEO'],  # 内容精选
+        },
     }
 
     # 广告模型异常视频数据处理参数
@@ -858,6 +870,7 @@ class BaseConfig(object):
         # 内容精选
         'videos5': 1 / 3,
         'videos5new': 1 / 3,
+        'videos5out': 1 / 3,
         # 票圈短视频
         'videos6': 1 / 3,
         # 老好看视频
@@ -975,6 +988,32 @@ class BaseConfig(object):
         ]
     }
 
+    # 新的 - 广告模型用户数据（直接跳出）
+    AD_USER_WITH_OUT_PARAMS = {
+        'data_params': {
+            'user5out': APP_TYPE['LONG_VIDEO'],  # 内容精选
+        },
+        'rule_params': {
+            'rule1': {
+                'group_list': AD_MID_GROUP['class1'],
+                'no_ad_mid_group_list': NO_AD_MID_GROUP_LIST['class1'],
+            },
+            'rule2': {
+                'group_list': AD_MID_GROUP['class1'],
+                'no_ad_mid_group_list': NO_AD_MID_GROUP_LIST['class1'],
+                'remove_no_ad_group': True,  # mean_group 预测&计算阈值时，去除不出广告的用户组
+            },  # 优化阈值计算方式
+            'rule3': {
+                'group_list': AD_MID_GROUP['class2'],
+                'no_ad_mid_group_list': NO_AD_MID_GROUP_LIST['class2'],
+                'remove_no_ad_group': True,  # mean_group 预测&计算阈值时，去除不出广告的用户组
+            },  # 优化阈值计算方式 + 优化2
+        },
+        'params_list': [
+            {'data': 'user5out', 'rule': 'rule2'},  # 内容精选 + 优化阈值计算方式
+        ]
+    }
+
     # 广告模型abtest配置
     AD_ABTEST_CONFIG = {
         # 票圈vlog
@@ -1078,6 +1117,10 @@ class BaseConfig(object):
                   'user': {'data': 'user5new', 'rule': 'rule2'},
                   'care_model': True, 'threshold_rate': 0.7,
                   },  # 所有广告类型本端数据 + 优化阈值计算方式 + else非关怀模式人群多出广告 + 调整计算公式
+        '195-h': {'video': {'data': 'videos5out'},
+                  'user': {'data': 'user5out', 'rule': 'rule2'},
+                  'care_model': True, 'threshold_rate': 0.7,
+                  },  # 所有广告类型本端数据 + 优化阈值计算方式 + else非关怀模式人群多出广告 + 使用以是否直接跳出为目标的数据
 
         # 票圈短视频
         '196-a': {'video': {'data': 'data1'},