扩展整体与裂变报表执行模式

run_bucket_source_offline_5day_report.py

@@ -12,48 +12,154 @@ from run_bucket_source_pv_uv_return_realtime import (
     column_mapping,
     format_fact_columns_as_integers,
     format_rate_columns,
-    group_name,
     reorder_columns,
 )
 
 
 BASE = Path(__file__).resolve().parent
-SQL_FILE = BASE / "bucket_source_pv_uv_return_offline_20260716_20260720_apptype4_all_versions.sql"
+CONFIGS = {
+    "offline": {
+        "sql": BASE / "bucket_source_pv_uv_return_offline_20260716_20260720_apptype4_all_versions.sql",
+        "output": BASE / "bucket_source_full_report_offline_20260716_20260720_apptype4_all_versions.csv",
+        "experiment_buckets": set("01"),
+        "experiment_label": "实验组（0、1）",
+        "control_label": "对照组（其余14桶）",
+        "experiment_tail_label": "0、1",
+        "control_tail_label": "其余14桶",
+    },
+    "offline_20260713_20260716": {
+        "sql": BASE / "bucket_source_pv_uv_return_offline_20260713_20260716_apptype4_all_versions.sql",
+        "output": BASE / "bucket_source_full_report_offline_20260713_20260716_apptype4_all_versions.csv",
+        "experiment_buckets": set("01"),
+        "experiment_label": "实验组（0、1）",
+        "control_label": "对照组（其余14桶）",
+        "experiment_tail_label": "0、1",
+        "control_tail_label": "其余14桶",
+    },
+    "offline_20260612_1508": {
+        "sql": BASE / "bucket_source_pv_uv_return_offline_20260612_apptype0_version1508.sql",
+        "output": BASE / "bucket_source_full_report_offline_20260612_apptype0_version1508.csv",
+        "experiment_buckets": set("0123456789ab"),
+        "experiment_label": "实验组（0-b）",
+        "control_label": "对照组（c-f）",
+        "experiment_tail_label": "0-b",
+        "control_tail_label": "c-f",
+    },
+    "offline_20260617_all": {
+        "sql": BASE / "bucket_source_pv_uv_return_offline_20260617_apptype0_all_versions.sql",
+        "output": BASE / "bucket_source_full_report_offline_20260617_apptype0_all_versions.csv",
+        "experiment_buckets": set("01"),
+        "experiment_label": "实验组（0、1）",
+        "control_label": "对照组（其余14桶）",
+        "experiment_tail_label": "0、1",
+        "control_tail_label": "其余14桶",
+    },
+    "offline_20260617_1512": {
+        "sql": BASE / "bucket_source_pv_uv_return_offline_20260617_apptype0_version1512.sql",
+        "output": BASE / "bucket_source_full_report_offline_20260617_apptype0_version1512.csv",
+        "experiment_buckets": set("01"),
+        "experiment_label": "实验组（0、1）",
+        "control_label": "对照组（其余14桶）",
+        "experiment_tail_label": "0、1",
+        "control_tail_label": "其余14桶",
+    },
+    "offline_20260721_1578": {
+        "sql": BASE / "bucket_source_pv_uv_return_offline_20260721_apptype4_version1578.sql",
+        "output": BASE / "bucket_source_full_report_offline_20260721_apptype4_version1578.csv",
+        "experiment_buckets": set("01"),
+        "experiment_label": "实验组（0、1）",
+        "control_label": "对照组（其余14桶）",
+        "experiment_tail_label": "0、1",
+        "control_tail_label": "其余14桶",
+    },
+    "offline_20260722_all": {
+        "sql": BASE / "bucket_source_pv_uv_return_offline_20260722_apptype4_all_versions.sql",
+        "output": BASE / "bucket_source_full_report_offline_20260722_apptype4_all_versions.csv",
+        "experiment_buckets": set("01"),
+        "experiment_label": "实验组（0、1）",
+        "control_label": "对照组（其余14桶）",
+        "experiment_tail_label": "0、1",
+        "control_tail_label": "其余14桶",
+    },
+    "offline_20260717_20260721_exclude_qywx": {
+        "sql": BASE / "bucket_source_pv_uv_return_offline_20260717_20260721_apptype4_all_versions_exclude_qywx.sql",
+        "output": BASE / "bucket_source_full_report_offline_20260717_20260721_apptype4_all_versions_exclude_qywx.csv",
+        "experiment_buckets": set("01"),
+        "experiment_label": "实验组（0、1）",
+        "control_label": "对照组（其余14桶）",
+        "experiment_tail_label": "0、1",
+        "control_tail_label": "其余14桶",
+    },
+    "offline_20260721_1578_exclude_qywx": {
+        "sql": BASE / "bucket_source_pv_uv_return_offline_20260721_apptype4_version1578_exclude_qywx.sql",
+        "output": BASE / "bucket_source_full_report_offline_20260721_apptype4_version1578_exclude_qywx.csv",
+        "experiment_buckets": set("01"),
+        "experiment_label": "实验组（0、1）",
+        "control_label": "对照组（其余14桶）",
+        "experiment_tail_label": "0、1",
+        "control_tail_label": "其余14桶",
+    },
+}
+MODE = next((arg for arg in sys.argv[1:] if not arg.startswith("--")), "offline")
+if MODE not in CONFIGS:
+    raise SystemExit("usage: python3 run_bucket_source_offline_5day_report.py [offline|offline_20260713_20260716|offline_20260612_1508|offline_20260617_all|offline_20260617_1512|offline_20260721_1578|offline_20260722_all|offline_20260717_20260721_exclude_qywx|offline_20260721_1578_exclude_qywx] [--rebuild-with-dau]")
+CONFIG = CONFIGS[MODE]
+SQL_FILE, OUTPUT_FILE = CONFIG["sql"], CONFIG["output"]
 DAU_SQL_FILE = BASE / "offline_dau_by_bucket_20260716_20260720_apptype4.sql"
-OUTPUT_FILE = BASE / "bucket_source_full_report_offline_20260716_20260720_apptype4_all_versions.csv"
 
 
 def build_daily_report(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}的产品或版本不唯一：apptype={app_types}, versions={versions}")
+    app_type, version = app_types[0], versions[0]
     daily["尾号"] = daily["尾号"].astype(str)
     daily["行类型"] = "尾号明细"
-    daily["分组"] = daily["尾号"].map(group_name)
+    experiment_buckets = CONFIG["experiment_buckets"]
+    control_buckets = set("0123456789abcdef") - experiment_buckets
+    experiment_label = CONFIG["experiment_label"]
+    control_label = CONFIG["control_label"]
+    tail_labels = {
+        experiment_label: CONFIG["experiment_tail_label"],
+        control_label: CONFIG["control_tail_label"],
+    }
+    daily["分组"] = daily["尾号"].map(lambda bucket: experiment_label if bucket in experiment_buckets else control_label)
     detail = daily[["日期", "产品类型", "版本号", "行类型", "分组", "尾号", *FACT_COLUMNS]]
 
     aggregate_rows, mean_rows = [], []
-    for group, buckets in [("实验组（0、1）", {"0", "1"}), ("对照组（其余14桶）", set("23456789abcdef"))]:
+    for group, buckets in [(experiment_label, experiment_buckets), (control_label, control_buckets)]:
         selected = detail[detail["尾号"].isin(buckets)]
-        common = {"日期": stat_date, "产品类型": "4", "版本号": "全部", "分组": group}
+        common = {"日期": stat_date, "产品类型": app_type, "版本号": version, "分组": group}
         aggregate_rows.append({
             **common,
             "行类型": "分组聚合",
-            "尾号": "0、1" if len(buckets) == 2 else "其余14桶",
+            "尾号": tail_labels[group],
             **selected[FACT_COLUMNS].sum().to_dict(),
         })
         mean_rows.append({
             **common,
             "行类型": "每桶均值",
-            "尾号": "2桶均值" if len(buckets) == 2 else "14桶均值",
+            "尾号": f"{len(buckets)}桶均值",
             **selected[FACT_COLUMNS].mean().to_dict(),
         })
 
     result = pd.concat([detail, pd.DataFrame(aggregate_rows), pd.DataFrame(mean_rows)], ignore_index=True)
-    result = add_relative_changes(add_rates(result))
+    result = add_relative_changes(
+        add_rates(result),
+        experiment_label=experiment_label,
+        control_label=control_label,
+        experiment_bucket_count=len(experiment_buckets),
+        control_bucket_count=len(control_buckets),
+    )
     return reorder_columns(format_fact_columns_as_integers(format_rate_columns(result)))
 
 
 def main():
     if "--rebuild-with-dau" in sys.argv:
+        if MODE != "offline":
+            raise ValueError("--rebuild-with-dau 仅适用于五天全版本离线报表")
         existing = pd.read_csv(OUTPUT_FILE, dtype={"日期": str, "尾号": str})
         detail = existing[existing["行类型"] == "尾号明细"][[
             "日期", "产品类型", "版本号", "尾号", *FACT_COLUMNS

run_bucket_source_pv_uv_return_realtime.py

@@ -8,8 +8,40 @@ from odps_module import ODPSClient
 
 
 BASE = Path(__file__).resolve().parent
-SQL_FILE = BASE / "bucket_source_pv_uv_return_realtime_20260721_apptype4_version1578.sql"
-OUTPUT_FILE = BASE / "bucket_source_pv_uv_return_realtime_20260721_apptype4_version1578.csv"
+CONFIGS = {
+    "realtime_20260721": (
+        BASE / "bucket_source_pv_uv_return_realtime_20260721_apptype4_version1578.sql",
+        BASE / "bucket_source_pv_uv_return_realtime_20260721_apptype4_version1578.csv",
+    ),
+    "realtime_20260722": (
+        BASE / "bucket_source_pv_uv_return_realtime_20260722_apptype4_version1578.sql",
+        BASE / "bucket_source_pv_uv_return_realtime_20260722_apptype4_version1578.csv",
+    ),
+    "realtime_20260722_1580": (
+        BASE / "bucket_source_pv_uv_return_realtime_20260722_apptype4_version1580.sql",
+        BASE / "bucket_source_pv_uv_return_realtime_20260722_apptype4_version1580.csv",
+    ),
+    "realtime_20260723_1580": (
+        BASE / "bucket_source_pv_uv_return_realtime_20260723_apptype4_version1580.sql",
+        BASE / "bucket_source_pv_uv_return_realtime_20260723_apptype4_version1580.csv",
+    ),
+    "realtime_20260723_1578": (
+        BASE / "bucket_source_pv_uv_return_realtime_20260723_apptype4_version1578.sql",
+        BASE / "bucket_source_pv_uv_return_realtime_20260723_apptype4_version1578.csv",
+    ),
+    "realtime_20260723_all": (
+        BASE / "bucket_source_pv_uv_return_realtime_20260723_apptype4_all_versions.sql",
+        BASE / "bucket_source_pv_uv_return_realtime_20260723_apptype4_all_versions.csv",
+    ),
+    "realtime_20260722_exclude_qywx": (
+        BASE / "bucket_source_pv_uv_return_realtime_20260722_apptype4_version1578_exclude_qywx.sql",
+        BASE / "bucket_source_pv_uv_return_realtime_20260722_apptype4_version1578_exclude_qywx.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_bucket_source_pv_uv_return_realtime.py [realtime_20260721|realtime_20260722|realtime_20260722_1580|realtime_20260723_1578|realtime_20260723_1580|realtime_20260723_all|realtime_20260722_exclude_qywx] [--rebuild]")
+SQL_FILE, OUTPUT_FILE = CONFIGS[MODE]
 SOURCES = ["头部", "推荐", "全部"]
 SOURCE_KEYS = {"头部": "head", "推荐": "recommend", "全部": "all"}
 FACT_SUFFIXES = ["曝光PV", "曝光UV", "播放PV", "播放UV", "分享PV", "分享UV", "回流UV"]
@@ -54,25 +86,31 @@ def add_rates(data):
     return result
 
 
-def add_relative_changes(data):
+def add_relative_changes(
+    data,
+    experiment_label="实验组（0、1）",
+    control_label="对照组（其余14桶）",
+    experiment_bucket_count=2,
+    control_bucket_count=14,
+):
     result = data.copy()
     result["DAU相对对照组变化率"] = ""
-    exp_aggregate = result.index[(result["行类型"] == "分组聚合") & (result["分组"] == "实验组（0、1）")]
-    ctrl_aggregate = result.index[(result["行类型"] == "分组聚合") & (result["分组"] == "对照组（其余14桶）")]
+    exp_aggregate = result.index[(result["行类型"] == "分组聚合") & (result["分组"] == experiment_label)]
+    ctrl_aggregate = result.index[(result["行类型"] == "分组聚合") & (result["分组"] == control_label)]
     if len(exp_aggregate) == 1 and len(ctrl_aggregate) == 1:
-        exp_dau_per_bucket = result.loc[exp_aggregate[0], "DAU"] / 2
-        ctrl_dau_per_bucket = result.loc[ctrl_aggregate[0], "DAU"] / 14
+        exp_dau_per_bucket = result.loc[exp_aggregate[0], "DAU"] / experiment_bucket_count
+        ctrl_dau_per_bucket = result.loc[ctrl_aggregate[0], "DAU"] / control_bucket_count
         if ctrl_dau_per_bucket:
             dau_change = format_percent(exp_dau_per_bucket / ctrl_dau_per_bucket - 1)
-            result.loc[(result["分组"] == "实验组（0、1）") & result["行类型"].isin(["分组聚合", "每桶均值"]), "DAU相对对照组变化率"] = dau_change
+            result.loc[(result["分组"] == experiment_label) & result["行类型"].isin(["分组聚合", "每桶均值"]), "DAU相对对照组变化率"] = dau_change
     for source in SOURCES:
         for suffix in RATE_SUFFIXES:
             metric = f"{source}{suffix}"
             change = f"{metric}相对对照组变化率"
             result[change] = ""
             for row_type in ["分组聚合", "每桶均值"]:
-                exp_idx = result.index[(result["行类型"] == row_type) & (result["分组"] == "实验组（0、1）")]
-                ctrl_idx = result.index[(result["行类型"] == row_type) & (result["分组"] == "对照组（其余14桶）")]
+                exp_idx = result.index[(result["行类型"] == row_type) & (result["分组"] == experiment_label)]
+                ctrl_idx = result.index[(result["行类型"] == row_type) & (result["分组"] == control_label)]
                 if len(exp_idx) == 1 and len(ctrl_idx) == 1:
                     control = result.loc[ctrl_idx[0], metric]
                     if pd.notna(control) and control != 0:
@@ -131,15 +169,25 @@ def main():
         return
 
     data = ODPSClient().execute_sql(SQL_FILE.read_text(encoding="utf-8")).rename(columns=column_mapping())
+    data["日期"] = data["日期"].astype(str)
+    data["产品类型"] = data["产品类型"].astype(str)
+    data["版本号"] = data["版本号"].astype(str)
     data["尾号"] = data["尾号"].astype(str)
     data["行类型"] = "尾号明细"
     data["分组"] = data["尾号"].map(group_name)
     detail = data[["日期", "产品类型", "版本号", "行类型", "分组", "尾号", *FACT_COLUMNS]]
 
+    stat_dates = detail["日期"].unique()
+    app_types = detail["产品类型"].unique()
+    versions = detail["版本号"].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]
+
     aggregate_rows, mean_rows = [], []
     for group, buckets in [("实验组（0、1）", {"0", "1"}), ("对照组（其余14桶）", set("23456789abcdef"))]:
         selected = detail[detail["尾号"].isin(buckets)]
-        common = {"日期": "20260721", "产品类型": "4", "版本号": "1578", "分组": group}
+        common = {"日期": stat_date, "产品类型": app_type, "版本号": version, "分组": group}
         aggregate_rows.append({**common, "行类型": "分组聚合", "尾号": "0、1" if len(buckets) == 2 else "其余14桶", **selected[FACT_COLUMNS].sum().to_dict()})
         mean_rows.append({**common, "行类型": "每桶均值", "尾号": "2桶均值" if len(buckets) == 2 else "14桶均值", **selected[FACT_COLUMNS].mean().to_dict()})
 

run_first_layer_source_metrics.py

@@ -17,6 +17,30 @@ CONFIG = {
         BASE / "first_layer_source_metrics_offline_20260716_20260720_apptype4_all_versions.sql",
         BASE / "first_layer_source_metrics_offline_20260716_20260720_apptype4_all_versions.csv",
     ),
+    "offline_20260611_1507": (
+        BASE / "first_layer_source_metrics_offline_20260611_apptype0_version1507.sql",
+        BASE / "first_layer_source_metrics_offline_20260611_apptype0_version1507.csv",
+    ),
+    "offline_20260717_20260720_special_layer": (
+        BASE / "first_layer_source_metrics_offline_20260717_20260720_apptype4_all_versions_no_qywx_filter_special_layer.sql",
+        BASE / "first_layer_source_metrics_offline_20260717_20260720_apptype4_all_versions_no_qywx_filter_special_layer.csv",
+    ),
+    "offline_20260721_1578_special_layer": (
+        BASE / "first_layer_source_metrics_offline_20260721_apptype4_version1578_no_qywx_filter_special_layer.sql",
+        BASE / "first_layer_source_metrics_offline_20260721_apptype4_version1578_no_qywx_filter_special_layer.csv",
+    ),
+    "realtime_20260722_special_layer": (
+        BASE / "first_layer_source_metrics_realtime_20260722_apptype4_version1578_no_qywx_filter_special_layer.sql",
+        BASE / "first_layer_source_metrics_realtime_20260722_apptype4_version1578_no_qywx_filter_special_layer.csv",
+    ),
+    "offline_20260716_20260721_20260722_all_special_layer": (
+        BASE / "first_layer_source_metrics_offline_20260716_20260721_20260722_apptype4_all_versions_no_qywx_filter_special_layer.sql",
+        BASE / "first_layer_source_metrics_offline_20260716_20260721_20260722_apptype4_all_versions_no_qywx_filter_special_layer.csv",
+    ),
+    "realtime_20260723_all_special_layer": (
+        BASE / "first_layer_source_metrics_realtime_20260723_apptype4_all_versions_no_qywx_filter_special_layer.sql",
+        BASE / "first_layer_source_metrics_realtime_20260723_apptype4_all_versions_no_qywx_filter_special_layer.csv",
+    ),
 }
 SOURCES = ["头部", "推荐", "全部"]
 SOURCE_KEYS = {"头部": "head", "推荐": "recommend", "全部": "all"}
@@ -26,6 +50,20 @@ RATE_SUFFIXES = [
     "曝光PV/首层UV", "播放PV/首层UV", "分享PV/首层UV",
     "STR（分享PV/曝光PV）", "T0裂变率（裂变层UV/首层UV）", "裂变层UV/曝光PV",
 ]
+SPECIAL_LAYER_OUTPUTS = [
+    CONFIG["realtime_20260722_special_layer"][1],
+    CONFIG["offline_20260721_1578_special_layer"][1],
+    CONFIG["offline_20260717_20260720_special_layer"][1],
+]
+SPECIAL_LAYER_COMBINED = BASE / "first_layer_source_metrics_20260717_20260720_offline_all_20260721_offline_1578_20260722_realtime_1578_no_qywx_filter_special_layer.csv"
+SPECIAL_LAYER_DAILY = BASE / "first_layer_source_metrics_daily_compare_20260717_20260722_no_qywx_filter_special_layer.csv"
+ALL_VERSION_SPECIAL_LAYER_OUTPUTS = [
+    CONFIG["realtime_20260723_all_special_layer"][1],
+    CONFIG["offline_20260716_20260721_20260722_all_special_layer"][1],
+    CONFIG["offline_20260717_20260720_special_layer"][1],
+]
+ALL_VERSION_SPECIAL_LAYER_COMBINED = BASE / "first_layer_source_metrics_20260716_20260722_offline_20260723_realtime_apptype4_all_versions_no_qywx_filter_special_layer.csv"
+ALL_VERSION_SPECIAL_LAYER_DAILY = BASE / "first_layer_source_metrics_daily_compare_20260716_20260723_apptype4_all_versions_no_qywx_filter_special_layer.csv"
 
 
 def column_mapping():
@@ -146,8 +184,27 @@ def build_daily(data, stat_date):
 
 def main():
     mode = sys.argv[1] if len(sys.argv) > 1 else "realtime"
+    if mode == "combine_all_version_special_layer":
+        result = pd.concat([pd.read_csv(path, dtype=str) for path in ALL_VERSION_SPECIAL_LAYER_OUTPUTS], ignore_index=True)
+        result = result.sort_values("日期", ascending=False, kind="stable").reset_index(drop=True)
+        if len(result) != 160 or result["日期"].nunique() != 8:
+            raise ValueError(f"合并结果异常：rows={len(result)}, dates={result['日期'].nunique()}")
+        result.to_csv(ALL_VERSION_SPECIAL_LAYER_COMBINED, index=False, encoding="utf-8-sig")
+        daily = result[result["行类型"] == "分组聚合"].copy()
+        daily.to_csv(ALL_VERSION_SPECIAL_LAYER_DAILY, index=False, encoding="utf-8-sig")
+        print(f"[CSV] {ALL_VERSION_SPECIAL_LAYER_COMBINED} ({len(result)} rows)")
+        print(f"[CSV] {ALL_VERSION_SPECIAL_LAYER_DAILY} ({len(daily)} rows)")
+        return
+    if mode == "combine_special_layer":
+        result = pd.concat([pd.read_csv(path, dtype=str) for path in SPECIAL_LAYER_OUTPUTS], ignore_index=True)
+        result.to_csv(SPECIAL_LAYER_COMBINED, index=False, encoding="utf-8-sig")
+        daily = result[result["行类型"] == "分组聚合"].copy()
+        daily.to_csv(SPECIAL_LAYER_DAILY, index=False, encoding="utf-8-sig")
+        print(f"[CSV] {SPECIAL_LAYER_COMBINED} ({len(result)} rows)")
+        print(f"[CSV] {SPECIAL_LAYER_DAILY} ({len(daily)} rows)")
+        return
     if mode not in CONFIG:
-        raise SystemExit("usage: python3 run_first_layer_source_metrics.py [realtime|offline]")
+        raise SystemExit(f"usage: python3 run_first_layer_source_metrics.py [{'|'.join(CONFIG)}|combine_special_layer|combine_all_version_special_layer]")
     sql_file, output_file = CONFIG[mode]
     data = ODPSClient().execute_sql(sql_file.read_text(encoding="utf-8")).rename(columns=column_mapping())
     data["日期"] = data["日期"].astype(str)

run_t0_fission_rate_offline_5day.py

@@ -1,5 +1,6 @@
 #!/usr/bin/env python3
 from pathlib import Path
+import sys
 
 import pandas as pd
 
@@ -8,12 +9,33 @@ from run_t0_fission_rate_realtime import group_name, percent
 
 
 BASE = Path(__file__).resolve().parent
-SQL_FILE = BASE / "t0_fission_rate_offline_20260716_20260720_apptype4_all_versions.sql"
-OUTPUT_FILE = BASE / "t0_fission_rate_offline_20260716_20260720_apptype4_all_versions.csv"
+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)
@@ -31,8 +53,8 @@ def build_daily(data, stat_date):
         fission_mean = selected["裂变层UV"].mean()
         mean_rows.append({
             "时间": stat_date,
-            "产品类型": "4",
-            "版本号": "全部",
+            "产品类型": app_type,
+            "版本号": version,
             "行类型": "每桶均值",
             "分组": group,
             "尾号": bucket_label,

run_t0_fission_rate_realtime.py

@@ -1,5 +1,6 @@
 #!/usr/bin/env python3
 from pathlib import Path
+import sys
 
 import pandas as pd
 
@@ -7,8 +8,20 @@ from odps_module import ODPSClient
 
 
 BASE = Path(__file__).resolve().parent
-SQL_FILE = BASE / "t0_fission_rate_realtime_20260721_apptype4_version1578.sql"
-OUTPUT_FILE = BASE / "t0_fission_rate_realtime_20260721_apptype4_version1578.csv"
+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):
@@ -28,12 +41,22 @@ def main():
         "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桶均值"),
@@ -44,9 +67,9 @@ def main():
         first_mean = selected["首层UV"].mean()
         fission_mean = selected["裂变层UV"].mean()
         mean_rows.append({
-            "时间": "20260721",
-            "产品类型": "4",
-            "版本号": "1578",
+            "时间": stat_date,
+            "产品类型": app_type,
+            "版本号": version,
             "行类型": "每桶均值",
             "分组": group,
             "尾号": bucket_label,