recommend-emr-dataprocess/ad/model_predict_analyse.py

import argparse
import gzip
import os.path
from collections import OrderedDict

import pandas as pd
from hdfs import InsecureClient

client = InsecureClient("http://master-1-1.c-7f31a3eea195cb73.cn-hangzhou.emr.aliyuncs.com:9870", user="spark")

SEGMENT_BASE_PATH = "/dw/recommend/model/36_score_calibration_file"


def read_predict_from_local_txt(txt_file) -> list:
    result = []
    with open(txt_file, "r") as f:
        for line in f.readlines():
            sp = line.replace("\n", "").split("\t")
            if len(sp) == 4:
                label = int(sp[0])
                cid = sp[3].split("_")[0]
                score = float(sp[2].replace("[", "").replace("]", "").split(",")[1])
                result.append({
                    "label": label,
                    "cid": cid,
                    "score": score
                })
    return result


def read_predict_from_hdfs(hdfs_path: str) -> list:
    if not hdfs_path.endswith("/"):
        hdfs_path += "/"
    result = []
    for file in client.list(hdfs_path):
        with client.read(hdfs_path + file) as reader:
            with gzip.GzipFile(fileobj=reader, mode="rb") as gz_file:
                for line in gz_file.read().decode("utf-8").split("\n"):
                    split = line.split("\t")
                    if len(split) == 4:
                        cid = split[3].split("_")[0]
                        label = int(split[0])
                        score = float(split[2].replace("[", "").replace("]", "").split(",")[1])
                        result.append({
                            "cid": cid,
                            "label": label,
                            "score": score
                        })

    return result


def _segment_v1(scores, step):
    bins = []
    for i in range(0, len(scores), int((len(scores) / step))):
        if i == 0:
            bins.append(0)
        else:
            bins.append(scores[i])
    bins.append(1)
    return list(OrderedDict.fromkeys(bins))


def segment_calc_diff_rate_by_score(df: pd.DataFrame, segment_file_path: str, step=100) -> [pd.DataFrame, pd.DataFrame]:
    sored_df = df.sort_values(by=['score'])
    # 评估分数分段
    scores = sored_df['score'].values

    bins = _segment_v1(scores, step)

    # 等分分桶
    # split_indices = np.array_split(np.arange(len(scores)), step)
    # bins = [scores[index[0]] for index in split_indices] + [scores[split_indices[-1][-1]]]

    sored_df['score_segment'] = pd.cut(sored_df['score'], bins=bins)

    # 计算分段内分数的差异
    group_df = sored_df.groupby("score_segment", observed=True).agg(
        segment_label_sum=('label', 'sum'),
        segment_label_cnt=('label', 'count'),
        segment_score_avg=('score', 'mean'),
    ).reset_index()
    group_df['segment_true_score'] = group_df['segment_label_sum'] / group_df['segment_label_cnt']
    group_df['segment_diff_rate_origin'] = (group_df['segment_score_avg'] / group_df['segment_true_score'] - 1).mask(group_df['segment_true_score'] == 0, 0)

    # 使用滑动窗口计算当前值以及上下两行的平均值，作为新的diff_rate
    group_df['segment_diff_rate'] = group_df['segment_diff_rate_origin'].rolling(window=5, center=True, min_periods=1).mean()

    # 完整的分段文件保存
    csv_data = group_df.to_csv(sep="\t", index=False)
    with client.write(segment_file_path, encoding='utf-8', overwrite=True) as writer:
        writer.write(csv_data)

    filtered_df = group_df[(abs(group_df['segment_diff_rate']) >= 0.2) & (group_df['segment_label_cnt'] >= 1000)]
    filtered_df = filtered_df[['score_segment', 'segment_diff_rate']]
    # 每条曝光数据添加对应分数的diff
    merged_df = pd.merge(sored_df, filtered_df, on="score_segment", how="left")

    merged_df['segment_diff_rate'] = merged_df['segment_diff_rate'].fillna(0)
    return merged_df, filtered_df


def read_and_calibration_predict(predict_path: str, step=100) -> [pd.DataFrame, pd.DataFrame]:
    """
    读取评估结果，并进行校准
    """
    # 本地调试使用
    # predicts = read_predict_from_local_txt(predict_path)
    predicts = read_predict_from_hdfs(predict_path)
    df = pd.DataFrame(predicts)

    # 模型分分段计算与真实ctcvr的dff_rate
    predict_basename = os.path.basename(predict_path)
    if predict_basename.endswith("/"):
        predict_basename = predict_basename[:-1]
    df, segment_df = segment_calc_diff_rate_by_score(df, segment_file_path=f"{SEGMENT_BASE_PATH}/{predict_basename}.txt", step=100)

    # 生成校准后的分数
    df['score_2'] = df['score'] / (1 + df['segment_diff_rate'])

    # 按CID统计真实ctcvr和校准前后的平均模型分
    grouped_df = df.groupby("cid").agg(
        view=('cid', 'size'),
        conv=('label', 'sum'),
        score_avg=('score', lambda x: round(x.mean(), 6)),
        score_2_avg=('score_2', lambda x: round(x.mean(), 6)),
    ).reset_index()
    grouped_df['true_ctcvr'] = grouped_df['conv'] / grouped_df['view']

    return grouped_df, segment_df


def _main(old_predict_path: str, new_predict_path: str, calibration_file: str, analyse_file: str):
    old_group_df, old_segment_df = read_and_calibration_predict(old_predict_path)
    new_group_df, new_segment_df = read_and_calibration_predict(new_predict_path)

    # 分段文件保存, 此处保留的最后使用的分段文件，不是所有的分段
    new_segment_df.to_csv(calibration_file, sep='\t', index=False, header=False)

    # 字段重命名，和列过滤
    old_group_df.rename(columns={'score_avg': 'old_score_avg', 'score_2_avg': 'old_score_2_avg'}, inplace=True)
    new_group_df.rename(columns={'score_avg': 'new_score_avg', 'score_2_avg': 'new_score_2_avg'}, inplace=True)
    old_group_df = old_group_df[['cid', 'view', 'conv', 'true_ctcvr', 'old_score_avg', 'old_score_2_avg']]
    new_group_df = new_group_df[['cid', 'new_score_avg', 'new_score_2_avg']]

    merged = pd.merge(old_group_df, new_group_df, on='cid', how='left')

    # 计算与真实ctcvr的差异值
    merged["(new-true)/true"] = (merged['new_score_avg'] / merged['true_ctcvr'] - 1).mask(merged['true_ctcvr'] == 0, 0)
    merged["(old-true)/true"] = (merged['old_score_avg'] / merged['true_ctcvr'] - 1).mask(merged['true_ctcvr'] == 0, 0)

    # 计算校准后的模型分与ctcvr的差异值
    merged["(new2-true)/true"] = (merged['new_score_2_avg'] / merged['true_ctcvr'] - 1).mask(merged['true_ctcvr'] == 0, 0)
    merged["(old2-true)/true"] = (merged['old_score_2_avg'] / merged['true_ctcvr'] - 1).mask(merged['true_ctcvr'] == 0, 0)

    # 按照曝光排序，写入本地文件
    merged = merged.sort_values(by=['view'], ascending=False)
    merged = merged[[
        'cid', 'view', "conv", "true_ctcvr",
        "old_score_avg", "new_score_avg", "(old-true)/true", "(new-true)/true",
        "old_score_2_avg", "new_score_2_avg", "(old2-true)/true", "(new2-true)/true",
    ]]

    # 根据文件名保存不同的格式
    if analyse_file.endswith(".csv"):
        merged.to_csv(analyse_file, index=False)
    else:
        with open(analyse_file, "w") as writer:
            writer.write(merged.to_string(index=False))
    print("0")


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description="model_predict_analyse.py")
    parser.add_argument("-op", "--old_predict_path", required=True, help="老模型评估结果")
    parser.add_argument("-np", "--new_predict_path", required=True, help="新模型评估结果")
    parser.add_argument("-af", "--analyse_file", required=True, help="最后计算结果的保存路径")
    parser.add_argument("-cf", "--calibration_file", required=True, help="线上使用的segment文件保存路径")
    args = parser.parse_args()

    _main(
        old_predict_path=args.old_predict_path,
        new_predict_path=args.new_predict_path,
        calibration_file=args.calibration_file,
        analyse_file=args.analyse_file
    )