data_query_miniprogram/run_t0_fission_rate_realtime.py

#!/usr/bin/env python3
from pathlib import Path
import sys

import pandas as pd

from odps_module import ODPSClient


BASE = Path(__file__).resolve().parent
CONFIGS = {
    "realtime_20260721": (
        BASE / "t0_fission_rate_realtime_20260721_apptype4_version1578.sql",
        BASE / "t0_fission_rate_realtime_20260721_apptype4_version1578.csv",
    ),
    "realtime_20260722_special_layer": (
        BASE / "t0_fission_rate_realtime_20260722_apptype4_version1578_no_qywx_filter_special_layer.sql",
        BASE / "t0_fission_rate_realtime_20260722_apptype4_version1578_no_qywx_filter_special_layer.csv",
    ),
}
MODE = next((arg for arg in sys.argv[1:] if not arg.startswith("--")), "realtime_20260721") if __name__ == "__main__" else "realtime_20260721"
if __name__ == "__main__" and MODE not in CONFIGS:
    raise SystemExit("usage: python3 run_t0_fission_rate_realtime.py [realtime_20260721|realtime_20260722_special_layer]")
SQL_FILE, OUTPUT_FILE = CONFIGS[MODE]


def group_name(bucket):
    return "实验组（0、1）" if bucket in {"0", "1"} else "对照组（其余14桶）"


def percent(value):
    return "" if pd.isna(value) else f"{value * 100:.4f}%"


def main():
    data = ODPSClient().execute_sql(SQL_FILE.read_text(encoding="utf-8")).rename(columns={
        "stat_date": "时间",
        "app_type": "产品类型",
        "version_code": "版本号",
        "bucket": "尾号",
        "first_layer_uv": "首层UV",
        "fission_layer_uv": "裂变层UV",
    })
    data["时间"] = data["时间"].astype(str)
    data["产品类型"] = data["产品类型"].astype(str)
    data["版本号"] = data["版本号"].astype(str)
    data["尾号"] = data["尾号"].astype(str)
    data["行类型"] = "尾号明细"
    data["分组"] = data["尾号"].map(group_name)
    data["T0裂变率"] = data["裂变层UV"] / data["首层UV"].replace(0, pd.NA)
    data["T0裂变率相对对照组变化率"] = ""

    stat_dates = data["时间"].unique()
    app_types = data["产品类型"].unique()
    versions = data["版本号"].unique()
    if len(stat_dates) != 1 or len(app_types) != 1 or len(versions) != 1:
        raise ValueError(f"日期、产品或版本不唯一：dates={stat_dates}, app_types={app_types}, versions={versions}")
    stat_date, app_type, version = stat_dates[0], app_types[0], versions[0]

    mean_rows = []
    groups = [
        ("实验组（0、1）", {"0", "1"}, "2桶均值"),
        ("对照组（其余14桶）", set("23456789abcdef"), "14桶均值"),
    ]
    for group, buckets, bucket_label in groups:
        selected = data[data["尾号"].isin(buckets)]
        first_mean = selected["首层UV"].mean()
        fission_mean = selected["裂变层UV"].mean()
        mean_rows.append({
            "时间": stat_date,
            "产品类型": app_type,
            "版本号": version,
            "行类型": "每桶均值",
            "分组": group,
            "尾号": bucket_label,
            "首层UV": round(first_mean, 2),
            "裂变层UV": round(fission_mean, 2),
            "T0裂变率": fission_mean / first_mean if first_mean else pd.NA,
            "T0裂变率相对对照组变化率": "",
        })

    result = pd.concat([data, pd.DataFrame(mean_rows)], ignore_index=True)
    exp_idx = result.index[(result["行类型"] == "每桶均值") & (result["分组"] == "实验组（0、1）")][0]
    ctrl_idx = result.index[(result["行类型"] == "每桶均值") & (result["分组"] == "对照组（其余14桶）")][0]
    control_rate = result.loc[ctrl_idx, "T0裂变率"]
    if pd.notna(control_rate) and control_rate != 0:
        result.loc[exp_idx, "T0裂变率相对对照组变化率"] = percent(
            result.loc[exp_idx, "T0裂变率"] / control_rate - 1
        )

    result["T0裂变率"] = result["T0裂变率"].map(percent)
    detail_mask = result["行类型"] == "尾号明细"
    for column in ["首层UV", "裂变层UV"]:
        result[column] = result[column].astype(object)
        result.loc[detail_mask, column] = result.loc[detail_mask, column].map(lambda value: f"{value:.0f}")
        result.loc[~detail_mask, column] = result.loc[~detail_mask, column].map(lambda value: f"{value:.2f}")
    result = result[[
        "时间", "产品类型", "版本号", "行类型", "分组", "尾号",
        "首层UV", "裂变层UV", "T0裂变率", "T0裂变率相对对照组变化率",
    ]]
    result.to_csv(OUTPUT_FILE, index=False, encoding="utf-8-sig")
    print(f"[CSV] {OUTPUT_FILE}", flush=True)
    print(result.to_string(index=False), flush=True)


if __name__ == "__main__":
    main()