model_script/model/segment_calibration_check.py

import logging

import pandas as pd

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

SEGMENT_BASE_PATH = "/Users/zhao/Desktop/tzld/XGB/segment_csv"
logger = logging.getLogger(__name__)


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:
    #                     continue
    #                 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 read_cpa() -> pd.DataFrame:
    # 添加CPA
    cpa_target_df = pd.read_csv("/Users/zhao/Desktop/tzld/XGB/creative_package_target.csv")
    cpa_target_df = cpa_target_df[['creative_id', 'cpa_target']]
    cpa_target_df.rename(columns={'creative_id': 'cid'}, inplace=True)
    cpa_target_df['cid'] = cpa_target_df['cid'].astype(str)

    return cpa_target_df


def read_calibration_file(filepath: str) -> list:
    segment = []
    with open(filepath, "r") as f:
        for line in f.readlines():
            s = line.split("\t")
            if len(s) != 2:
                continue

            score = s[0].split(",")
            min_score = float(score[0].replace("(", ""))
            max_score = float(score[1].replace("]", ""))
            alpha = float(s[1])
            segment.append((min_score, max_score, alpha))
    return segment


def score_calibration(score, segment: list):
    for e in segment:
        if (score >= e[0]) & (score <= e[1]):
            return score * (1 + e[2])
    return score


def read_and_calibration_predict(predict_path: str, is_hdfs=True) -> [pd.DataFrame]:
    """
    读取评估结果，并进行校准
    """
    if is_hdfs:
        # 文件路径处理
        predicts = read_predict_from_hdfs(predict_path)
    else:
        predicts = read_predict_from_local_txt(predict_path)
    df = pd.DataFrame(predicts)

    # 每条曝光添加对应创意的CPA，并计算CPM
    cpa_target_df = read_cpa()
    df = pd.merge(df, cpa_target_df, on='cid', how='left')
    df['p_cpm'] = df['score'] * df['cpa_target'] * 1000
    df['t_cpm'] = df['score'] * df['cpa_target'] * 1000 * df['label']

    segment = read_calibration_file("/Users/zhao/Desktop/tzld/XGB/model_xgb_351_1000_v2_calibration.txt")
    # 生成校准后的分数并计算对应的p_cpm和t_cpm
    df['score_2'] = df['score'].apply(lambda x: score_calibration(x, segment))
    df['p_cpm_2'] = df['score_2'] * df['cpa_target'] * 1000
    df['t_cpm_2'] = df['score_2'] * df['cpa_target'] * 1000 * df['label']

    # 按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


def _main(old_predict_path: str, new_predict_path: str, calibration_file: str, analyse_file: str):
    old_group_df = read_and_calibration_predict(old_predict_path, is_hdfs=False)
    new_group_df = read_and_calibration_predict(new_predict_path, is_hdfs=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']]

    # 拼接CPA
    cpa_target_df = read_cpa()
    merged = (pd.merge(old_group_df, new_group_df, on='cid', how='left')
              .merge(cpa_target_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)
    merged['new_cpm'] = merged['new_score_avg'] * merged['cpa_target'] * 1000
    merged['old_cpm'] = merged['old_score_avg'] * merged['cpa_target'] * 1000

    # 计算校准后的模型分与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['new2_cpm'] = merged['new_score_2_avg'] * merged['cpa_target'] * 1000
    merged['old2_cpm'] = merged['old_score_2_avg'] * merged['cpa_target'] * 1000

    # 计算真实的CPM
    merged['true_cpm'] = merged['true_ctcvr'] * merged['cpa_target'] * 1000
    # 按照曝光排序，写入本地文件
    merged = merged.sort_values(by=['view'], ascending=False)
    merged = merged[[
        'cid', 'view', "conv", "true_ctcvr", 'true_cpm',
        "old_score_avg", "new_score_avg", "(old-true)/true", "(new-true)/true", "old_cpm", "new_cpm",
        "old_score_2_avg", "new_score_2_avg", "(old2-true)/true", "(new2-true)/true", "old2_cpm", "new2_cpm",
    ]]

    # 根据文件名保存不同的格式
    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__':
    _main(
        f"/Users/zhao/Desktop/tzld/XGB/predict_file/20241103_351_1000_1028_1102.txt",
        f"/Users/zhao/Desktop/tzld/XGB/predict_file/20241104_351_1000_1028_1102.txt",
        "/Users/zhao/Desktop/tzld/XGB/model_xgb_351_1000_v2_calibration.txt",
        f"/Users/zhao/Desktop/tzld/XGB/20241104.csv"
    )