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', 'sharerate', 'adrate_share' ] def predict_user_group_share_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['sharerate'].fillna(0, inplace=True) user_group_df['adrate_share'].fillna(0, inplace=True) user_group_df['adrate'] = user_group_df['adrate'].astype(float) user_group_df['sharerate'] = user_group_df['sharerate'].astype(float) user_group_df['adrate_share'] = user_group_df['adrate_share'].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_share_rate'] = \ user_group_df['adrate_share'] * user_group_df['sharerate'] / user_group_df['adrate'] user_group_df['group_ad_share_rate'].fillna(0, inplace=True) # 结果写入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_share_rate'] group_ad_share_rate_mean = user_group_df['group_ad_share_rate'].mean() redis_data['mean_group'] = group_ad_share_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_share_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_PARAMS_NEW project = config_.ad_model_data['users_data'].get('project') table = config_.ad_model_data['users_data'].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()