| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107 |
- #!/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()
|