data_query_miniprogram/run_t0_fission_rate_offline_5day.py

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

import pandas as pd

from odps_module import ODPSClient
from run_t0_fission_rate_realtime import group_name, percent


BASE = Path(__file__).resolve().parent
CONFIGS = {
    "offline": (
        BASE / "t0_fission_rate_offline_20260716_20260720_apptype4_all_versions.sql",
        BASE / "t0_fission_rate_offline_20260716_20260720_apptype4_all_versions.csv",
    ),
    "offline_20260717_20260720_special_layer": (
        BASE / "t0_fission_rate_offline_20260717_20260720_apptype4_all_versions_no_qywx_filter_special_layer.sql",
        BASE / "t0_fission_rate_offline_20260717_20260720_apptype4_all_versions_no_qywx_filter_special_layer.csv",
    ),
    "offline_20260721_1578_special_layer": (
        BASE / "t0_fission_rate_offline_20260721_apptype4_version1578_no_qywx_filter_special_layer.sql",
        BASE / "t0_fission_rate_offline_20260721_apptype4_version1578_no_qywx_filter_special_layer.csv",
    ),
}
MODE = next((arg for arg in sys.argv[1:] if not arg.startswith("--")), "offline")
if MODE not in CONFIGS:
    raise SystemExit("usage: python3 run_t0_fission_rate_offline_5day.py [offline|offline_20260717_20260720_special_layer|offline_20260721_1578_special_layer]")
SQL_FILE, OUTPUT_FILE = CONFIGS[MODE]


def build_daily(data, stat_date):
    daily = data[data["时间"] == stat_date].copy()
    app_types = daily["产品类型"].astype(str).unique()
    versions = daily["版本号"].astype(str).unique()
    if len(app_types) != 1 or len(versions) != 1:
        raise ValueError(f"日期{stat_date}的产品或版本不唯一：app_types={app_types}, versions={versions}")
    app_type, version = app_types[0], versions[0]
    daily["尾号"] = daily["尾号"].astype(str)
    daily["行类型"] = "尾号明细"
    daily["分组"] = daily["尾号"].map(group_name)
    daily["T0裂变率"] = daily["裂变层UV"] / daily["首层UV"].replace(0, pd.NA)
    daily["T0裂变率相对对照组变化率"] = ""

    mean_rows = []
    groups = [
        ("实验组（0、1）", {"0", "1"}, "2桶均值"),
        ("对照组（其余14桶）", set("23456789abcdef"), "14桶均值"),
    ]
    for group, buckets, bucket_label in groups:
        selected = daily[daily["尾号"].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([daily, 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}")
    return result[[
        "时间", "产品类型", "版本号", "行类型", "分组", "尾号",
        "首层UV", "裂变层UV", "T0裂变率", "T0裂变率相对对照组变化率",
    ]]


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)
    result = pd.concat([
        build_daily(data, stat_date)
        for stat_date in sorted(data["时间"].unique(), reverse=True)
    ], ignore_index=True)
    result.to_csv(OUTPUT_FILE, index=False, encoding="utf-8-sig")
    print(f"[CSV] {OUTPUT_FILE}", flush=True)
    print(result[result["行类型"] == "每桶均值"].to_string(index=False), flush=True)


if __name__ == "__main__":
    main()