import copy import datetime import traceback import math import numpy as np from threading import Timer import pandas as pd from utils import RedisHelper, data_check, get_feature_data, send_msg_to_feishu, request_get from config import set_config from log import Log config_, _ = set_config() log_ = Log() redis_helper = RedisHelper() features = [ 'apptype', '分组', '广告uv' ] def get_threshold_record_new(ad_abtest_abcode_config, feature_df, threshold_record): """根据活跃人数变化计算新的阈值参数""" robot_msg_record = [] threshold_record_new = threshold_record.copy() for app_type, config_params in ad_abtest_abcode_config.items(): # 获取对应端的数据, 更新阈值参数 # log_.info(f"app_type = {app_type}") temp_df = feature_df[feature_df['apptype'] == app_type] ab_test_id = config_params.get('ab_test_id') ab_test_config = config_params.get('ab_test_config') up_threshold_update = config_params.get('up_threshold_update') down_threshold_update = config_params.get('down_threshold_update') for config_name, ab_code_list in ab_test_config.items(): ad_abtest_tag = f"{ab_test_id}-{config_name}" # log_.info(f"ad_abtest_tag = {ad_abtest_tag}") if len(ab_code_list) > 0: b_mean = temp_df[temp_df['adcode'].isin(ab_code_list)]['b'].mean() if b_mean < 0: # 阈值按梯度调高 gradient = up_threshold_update[config_name].get('gradient') update_range = up_threshold_update[config_name].get('update_range') b_i = (b_mean * -1) // gradient + 1 threshold_param_new = float(threshold_record.get(ad_abtest_tag)) + update_range * b_i elif b_mean > 0.1: # 阈值按梯度调低 gradient = down_threshold_update[config_name].get('gradient') update_range = down_threshold_update[config_name].get('update_range') b_i = (b_mean - 0.1) // gradient + 1 threshold_param_new = float(threshold_record.get(ad_abtest_tag)) - update_range * b_i else: continue if threshold_param_new > 0: threshold_record_new[ad_abtest_tag] = threshold_param_new robot_msg_record.append({'appType': app_type, 'abtestTag': ad_abtest_tag, 'gradient': round(gradient, 4), 'range': round(update_range, 4), 'i': int(b_i), 'paramOld': round(float(threshold_record.get(ad_abtest_tag)), 4), 'paramNew': round(threshold_param_new, 4)}) return threshold_record_new, robot_msg_record def get_threshold_record_new_by_uv(ad_abtest_abcode_config, feature_df, threshold_record, ad_target_uv): """根据广告uv计算新的阈值参数""" robot_msg_record = [] threshold_record_new = copy.deepcopy(threshold_record) # 根据目标uv进行调整 for app_type, target_uv_mapping in ad_target_uv.items(): # 获取app_type对应的目标uv temp_df = feature_df[feature_df['apptype'] == int(app_type)] # 获取app_type对应的阈值调整参数 update_threshold_params = ad_abtest_abcode_config.get(int(app_type)) ab_test_id = update_threshold_params.get('ab_test_id') not_update = update_threshold_params.get('not_update') gradient = update_threshold_params.get('gradient') max_update_step = update_threshold_params.get('max_update_step') threshold_update_mapping = update_threshold_params.get('threshold_update') threshold_record_old = threshold_record.get(ab_test_id) # print(ab_test_id, threshold_record, threshold_record_old) for ab_test_group, target_uv in target_uv_mapping.items(): if target_uv is None: continue gradient, max_update_step = get_ad_uv_update_config(app_type=app_type, ab_group=ab_test_group) update_range = threshold_update_mapping.get(ab_test_group) # 获取对应组的当前uv try: current_uv = temp_df[temp_df['分组'] == ab_test_group]['广告uv'].values[0] except Exception as e: continue # 计算uv差值 uv_differ = current_uv - target_uv if abs(uv_differ) <= not_update: continue # 获取当前阈值参数 threshold_param_old = threshold_record_old[ab_test_group].get('group') if uv_differ < 0: # 当前uv < 目标uv，阈值按梯度调低（第一个梯度区间：向上取整，之后：四舍五入） if abs(uv_differ) < gradient: step = math.ceil(abs(uv_differ) / gradient) else: step = round(abs(uv_differ) / gradient) step = max_update_step if step > max_update_step else step threshold_param_new = float(threshold_param_old) - update_range * step elif uv_differ > 0: # 当前uv > 目标uv，阈值按梯度调高（第一个梯度区间：向上取整，之后：四舍五入） if uv_differ < gradient: step = math.ceil(uv_differ / gradient) else: step = round(uv_differ / gradient) step = max_update_step if step > max_update_step else step threshold_param_new = float(threshold_param_old) + update_range * step else: continue if threshold_param_new <= 0: threshold_param_new = 0 log_.info( { 'appType': app_type, 'abtestid': ab_test_id, 'abTestGroup': ab_test_group, 'targetUv': target_uv, 'currentUv': round(current_uv, 4), 'uvDiffer': round(uv_differ, 4), 'gradient': round(gradient, 4), 'step': step, 'range': round(update_range, 4), 'paramOld': round(float(threshold_param_old), 4), 'paramNew': round(threshold_param_new, 4) } ) threshold_record_new[ab_test_id][ab_test_group]['group'] = threshold_param_new threshold_record_new[ab_test_id][ab_test_group]['mean_group'] = threshold_param_new robot_msg_record.append( { 'appType': app_type, 'abtestid': ab_test_id, 'abTestGroup': ab_test_group, 'targetUv': target_uv, 'currentUv': round(current_uv, 4), 'uvDiffer': round(uv_differ, 4), 'gradient': round(gradient, 4), 'step': step, 'range': round(update_range, 4), 'paramOld': round(float(threshold_param_old), 4), 'paramNew': round(threshold_param_new, 4) } ) return threshold_record_new, robot_msg_record def update_threshold(threshold_record_old, threshold_record_new): """更新阈值""" # 获取用户组列表 ad_mid_group_list = [group for class_key, group_list in config_.AD_MID_GROUP.items() for group in group_list] ad_mid_group_list.append("mean_group") ad_mid_group_list = list(set(ad_mid_group_list)) # 获取实验配置列表 ad_abtest_config_mapping = {} abtest_id_list = [] for key, val in config_.AD_ABTEST_CONFIG.items(): abtest_id, abtest_config_tag = key.split('-') if abtest_id in abtest_id_list: ad_abtest_config_mapping[abtest_id].append((abtest_config_tag, val)) else: abtest_id_list.append(abtest_id) ad_abtest_config_mapping[abtest_id] = [(abtest_config_tag, val)] log_.info(f"ad_abtest_config_mapping = {ad_abtest_config_mapping}") # 计算新的阈值并更新 for abtest_id, threshold_param_mapping in threshold_record_new.items(): for abtest_group, threshold_param_new in threshold_param_mapping.items(): threshold_param_old = threshold_record_old[abtest_id].get(abtest_group) if str(threshold_param_old) == str(threshold_param_new): # print(abtest_id, abtest_group, threshold_param_old, threshold_param_new) continue log_.info(f"abtest_id = {abtest_id}, abtest_group = {abtest_group}, " f"threshold_param_old = {threshold_param_old}, threshold_param_new = {threshold_param_new}") for abtest_config_tag, config_val in ad_abtest_config_mapping.get(abtest_id, []): for group_key in ad_mid_group_list: # 获取对应的阈值 key_name = \ f"{config_.KEY_NAME_PREFIX_AD_THRESHOLD}{abtest_id}:{abtest_config_tag}:{abtest_group}:{group_key}" threshold_old = redis_helper.get_data_from_redis(key_name=key_name) if threshold_old is None: continue # 原阈值为0时，加10**(-5)兜底处理 if float(threshold_old) == 0: threshold_old = float(threshold_old) + 10**(-5) # 计算新的阈值 if group_key == 'mean_group': if threshold_param_old['mean_group'] == 0: threshold_new = \ float(threshold_old) / 10**(-5) * threshold_param_new['mean_group'] else: threshold_new = \ float(threshold_old) / threshold_param_old['mean_group'] * threshold_param_new['mean_group'] else: if threshold_param_old['group'] == 0: threshold_new = \ float(threshold_old) / 10**(-5) * threshold_param_new['group'] else: threshold_new = \ float(threshold_old) / threshold_param_old['group'] * threshold_param_new['group'] # 更新redis redis_helper.set_data_to_redis(key_name=key_name, value=threshold_new, expire_time=2 * 24 * 3600) log_.info(f"abtest_id = {abtest_id}, abtest_config_tag = {abtest_config_tag}, " f"abtest_group = {abtest_group}, group_key = {group_key}, " f"threshold_old = {threshold_old}, threshold_new = {threshold_new}") # 关怀模式实验阈值更新 care_model = config_val.get('care_model', None) threshold_rate = config_val.get('threshold_rate', None) if care_model is True: care_model_key_name = \ f"{config_.KEY_NAME_PREFIX_AD_THRESHOLD_CARE_MODEL}{abtest_id}:{abtest_config_tag}:{abtest_group}:{group_key}" care_model_threshold_old = redis_helper.get_data_from_redis(key_name=care_model_key_name) care_model_threshold_new = threshold_new * threshold_rate redis_helper.set_data_to_redis(key_name=care_model_key_name, value=care_model_threshold_new, expire_time=2 * 24 * 3600) log_.info(f"abtest_id = {abtest_id}, abtest_config_tag = {abtest_config_tag}, " f"abtest_group = {abtest_group}, group_key = {group_key}, " f"care_model_threshold_old = {care_model_threshold_old}, " f"care_model_threshold_new = {care_model_threshold_new}") def update_ad_abtest_threshold(project, table, dt, ad_abtest_abcode_config, ad_target_uv): # 获取当前阈值参数值 threshold_record = redis_helper.get_data_from_redis(key_name=config_.KEY_NAME_PREFIX_AD_THRESHOLD_RECORD) threshold_record = eval(threshold_record) log_.info(f"threshold_record = {threshold_record}") # 获取uv数据 feature_df = get_feature_data(project=project, table=table, features=features, dt=dt) feature_df['apptype'] = feature_df['apptype'].astype(int) feature_df['广告uv'] = feature_df['广告uv'].astype(float) # 根据广告uv变化计算新的阈值参数 threshold_record_new, robot_msg_record = get_threshold_record_new_by_uv( ad_abtest_abcode_config=ad_abtest_abcode_config, feature_df=feature_df, threshold_record=threshold_record, ad_target_uv=ad_target_uv) log_.info(f"threshold_record_new = {threshold_record_new}") # 更新阈值 update_threshold(threshold_record_old=threshold_record, threshold_record_new=threshold_record_new) # 更新阈值参数 redis_helper.set_data_to_redis(key_name=config_.KEY_NAME_PREFIX_AD_THRESHOLD_RECORD, value=str(threshold_record_new), expire_time=2 * 24 * 3600) return robot_msg_record def get_ad_target_uv(now_h): """获取管理后台开启自动调整阈值开关的目标uv值""" ad_target_uv = {} result = request_get(request_url=config_.GET_AD_TARGET_UV_URL) if result is None: log_.info('获取管理后台广告目标uv值失败！') return ad_target_uv if result['code'] != 0: log_.info('获取管理后台广告目标uv值失败！') return ad_target_uv if not result['content']: return ad_target_uv for item in result['content']: app_type = item['productId'] target_uv_mapping = {} target_uv_param = config_.AD_ABTEST_ABCODE_CONFIG.get(int(app_type)).get('target_uv_param', {}) for uv_item in item['uvTargetDetails']: ab_group = uv_item['abParam'] target_uv = uv_item['uvTarget'] target_uv_param_group = target_uv_param.get(ab_group, None) if target_uv_param_group is not None: update_hours = target_uv_param_group.get('update_hours') update_param = target_uv_param_group.get('update_param') if now_h in update_hours: target_uv *= update_param target_uv_mapping[ab_group] = target_uv ad_target_uv[app_type] = target_uv_mapping return ad_target_uv def get_ad_uv_update_config(app_type, ab_group): """获取对应组自动调整阈值参数：梯度，最大步长""" now_h = datetime.datetime.now().hour update_threshold_params = config_.AD_ABTEST_ABCODE_CONFIG.get(int(app_type)) gradient = update_threshold_params.get('gradient') max_update_step = update_threshold_params.get('max_update_step') target_uv_param = update_threshold_params.get('target_uv_param', {}) target_uv_param_group = target_uv_param.get(ab_group, None) if target_uv_param_group is not None: special_update_config = target_uv_param_group.get('special_update_config', None) if special_update_config is not None: special_hours = special_update_config.get('special_hours', []) if now_h in special_hours: gradient = special_update_config.get('special_gradient') max_update_step = special_update_config.get('special_max_update_step') return gradient, max_update_step def timer_check(): try: # 获取自动调整阈值参数 ad_abtest_abcode_config = config_.AD_ABTEST_ABCODE_CONFIG # 自动调整阈值参数存储至redis redis_helper.set_data_to_redis(key_name=config_.KEY_NAME_PREFIX_AD_THRESHOLD_PARAM_RECORD, value=str(ad_abtest_abcode_config), expire_time=24 * 3600) project = config_.AD_THRESHOLD_AUTO_UPDATE_DATA.get('project') table = config_.AD_THRESHOLD_AUTO_UPDATE_DATA.get('table') now_date = datetime.datetime.today() now_h = datetime.datetime.now().hour now_min = datetime.datetime.now().minute log_.info(f"now_date: {datetime.datetime.strftime(now_date, '%Y%m%d%H')}") # 00:00 - 09:00 不做阈值参数调整 if 0 <= now_h < 9: log_.info(f"00:00 - 09:00 不做阈值参数调整") return # 管理后台获取开启自动调整阈值开关的目标uv值 ad_target_uv = get_ad_target_uv(now_h=now_h) log_.info(f"ad_target_uv: {ad_target_uv}") if len(ad_target_uv) == 0: return # 查看当前更新的数据是否已准备好 dt = datetime.datetime.strftime(now_date - datetime.timedelta(hours=1), '%Y%m%d%H') data_count = data_check(project=project, table=table, dt=dt) if data_count > 0: log_.info(f"data count = {data_count}") # 数据准备好，进行更新 robot_msg_record = update_ad_abtest_threshold( project=project, table=table, dt=dt, ad_abtest_abcode_config=ad_abtest_abcode_config, ad_target_uv=ad_target_uv) if len(robot_msg_record) > 0: robot_msg_record_text = "\n".join([str(item) for item in robot_msg_record]) msg = f"threshold_param_update: \n{robot_msg_record_text.replace(', ', ', ')}\n" else: msg = "无需更新！\n" send_msg_to_feishu( webhook=config_.FEISHU_ROBOT['ad_threshold_auto_update_robot'].get('webhook'), key_word=config_.FEISHU_ROBOT['ad_threshold_auto_update_robot'].get('key_word'), msg_text=f"rov-offline{config_.ENV_TEXT} - 阈值更新完成！\n{msg}" ) log_.info(f"threshold update end!") elif now_min > 30: log_.info('threshold update data is None!') send_msg_to_feishu( webhook=config_.FEISHU_ROBOT['ad_threshold_auto_update_robot'].get('webhook'), key_word=config_.FEISHU_ROBOT['ad_threshold_auto_update_robot'].get('key_word'), msg_text=f"rov-offline{config_.ENV_TEXT} - 阈值更新相关数据未准备好！\n" ) 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['ad_threshold_auto_update_robot'].get('webhook'), key_word=config_.FEISHU_ROBOT['ad_threshold_auto_update_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()