merge: update daily ROI revenue calculation

AGENTS.md

@@ -41,7 +41,7 @@
 
 - ROI 是指导渠道、账户、广告和创意动作的核心北极星指标,必须作为独立领域模块维护,不能散落在飞书、腾讯 API 或调度代码中。
 - ROI 指标计算必须保持纯数据输入/输出,不能依赖飞书审批、腾讯写操作或具体动作执行器。动作策略只能消费一个明确的 `metric_version`。
-- 当前日级 ROI 指标版本为 `north_star_roi_t15_v4`,报表版本为 `roi_report_v5`,使用已发布参数 `20260712_A0-A15_v1`。T0 裂变收入是实际值,只预测 T1-T15 增量:`T0实际裂变收入 * (传播裂变系数-对T0裂变 - 1)`;预测总收入为 `首层实际效率收入 + T0实际裂变收入 + 预测T1-T15裂变收入`,禁止重复计算 T0。
+- 当前日级 ROI 指标版本为 `north_star_roi_t15_v5`,报表版本为 `roi_report_v9`,使用已发布参数 `20260712_A0-A15_v2`。首层效率收入取 `SUM(效率收入)`,T0 实际裂变收入取当天的 `SUM(裂变效率收入)`;只预测 T1-T15 增量:`T0实际裂变收入 * (传播裂变系数 - 1)`,预测总收入为 `首层效率收入 + T0实际裂变收入 + 预测T1-T15裂变收入`,禁止继续读取已废弃的 `多层t0裂变收入`。
 - 小程序传播裂变系数按 `人群包+转化目标精确 -> 人群包回退 -> 转化目标回退 -> 渠道回退` 匹配;公众号按 `合作方+公众号精确 -> 合作方回退 -> 渠道回退` 匹配。样本不足的精确实体不能使用自身系数。
 - 小程序日级 ODPS 数据必须保留 `广告优化目标`;缺失目标只能进入人群包回退,不得默认成关键页面访问。企微 ROI 口径未复核,优先使用已发布的合作方 SQL 计算结果,合作方未匹配、样本不足或旧版本无企微参数时默认参考系数为 2.5 并明确标注;企微不进入阈值样本池且不得生成调控动作。
 - ROI 审批表金额和 UV 默认展示三日窗口的单日均值,UV 显示为整数;预测 ROI、实际 ROI、裂变率和阈值必须继续使用三日汇总后的加权口径,不能对每日比例做算术平均。原始三日汇总和每日明细字段保留为隐藏审计列。

examples/auto_put_ad_mini/docs/unified_services_deployment.md

@@ -80,7 +80,7 @@ ROI_APPROVAL_TTL_MINUTES=120
 ROI_SCALE_RATIO=1.10
 ROI_SCALE_COOLDOWN_DAYS=3
 ROI_MAX_BASE_RATIO=2.00
-ROI_FISSION_PARAMETER_VERSION=20260712_A0-A15_v1
+ROI_FISSION_PARAMETER_VERSION=20260712_A0-A15_v2
 ROI_FEISHU_CHAT_ID=oc_xxx
 ```
 
@@ -165,10 +165,10 @@ docker compose --env-file /dev/null run --rm \
 
 日级 ROI 单独分两阶段启用：
 
-当前 `north_star_roi_t15_v4` 使用版本化传播裂变参数，报表版本为 `roi_report_v5`。小程序按人群包和转化目标、
-公众号按合作方和公众号匹配传播裂变系数；企微优先按合作方精确匹配，合作方未匹配或样本不足时使用默认参考系数 2.5，不做渠道聚合回退。企微仅保留展示，不进入阈值和调控。T0 收入是实际值，
-只预测 T1-T15 增量。日级任务不会现场重算 cohort 参数。部署前应保持
-`ROI_FISSION_PARAMETER_VERSION=20260712_A0-A15_v1`。参数必须先发布到 MySQL，
+当前 `north_star_roi_t15_v5` 使用版本化传播裂变参数，报表版本为 `roi_report_v9`。小程序按人群包和转化目标、
+公众号按合作方和公众号匹配传播裂变系数；企微优先按合作方精确匹配，合作方未匹配或样本不足时使用默认参考系数 2.5，不做渠道聚合回退。企微仅保留展示，不进入阈值和调控。首层效率收入读取 `效率收入`，T0 实际裂变收入读取当天的 `裂变效率收入`，
+并以 T0 实际裂变收入乘传播裂变系数预测完整裂变收入。日级任务不会现场重算 cohort 参数。部署前应保持
+`ROI_FISSION_PARAMETER_VERSION=20260712_A0-A15_v2`。参数必须先发布到 MySQL，
 启动日志、数据库发布版本和 Excel 摘要中的参数版本、cohort 日期必须一致。
 默认企微参考值使用独立版本 `qiwei_reference_2p5_v1`，并进入 ROI 批次幂等键，
 避免与其他临时系数结果复用同一个运行批次。
@@ -202,6 +202,18 @@ docker compose run --rm ad_daily_service \
 确认数据库快照和本地报告后，再增加 `--send-feishu` 验证在线表、黄色审批列和链接编辑权限。
 此时必须保持 `ROI_APPLY_ENABLED=0`、`ROI_SHEET_APPROVAL_ENABLED=0`，不会轮询或执行表格选择。
 
+如果同一日期窗口的上游历史数据被修订，必须提供稳定的数据修订标识生成新批次，不能覆盖或删除已发布快照：
+
+```bash
+docker compose run --rm ad_daily_service \
+  python /app/examples/auto_put_ad_mini/run_daily_roi.py \
+  --end-date 20260729 \
+  --source-revision datafix_20260730_1 \
+  --send-feishu
+```
+
+相同 `--source-revision` 重复执行会复用同一批次；新的修订必须使用新的标识。日常调度不传此参数。
+
 离线复算新参数必须与日级任务分开执行。以下命令默认只生成 SQL，不访问 ODPS：
 
 ```bash
@@ -220,7 +232,7 @@ docker compose run --rm ad_daily_service \
 ```bash
 docker compose run --rm ad_daily_service \
   python /app/examples/auto_put_ad_mini/refresh_roi_fission_multiplier.py \
-  --publish-version 20260712_A0-A15_v1 \
+  --publish-version 20260712_A0-A15_v2 \
   --published-by production-deploy
 ```
 

examples/auto_put_ad_mini/roi_control/data_source.py

@@ -78,7 +78,7 @@ SELECT
   SUM(NVL(t0裂变人数, 0)) AS T0裂变数,
   SUM(NVL(成本, 0)) AS 成本,
   SUM(NVL(效率收入, 0)) AS 效率收入,
-  SUM(NVL(多层t0裂变收入, 0)) AS 多层裂变收入
+  SUM(NVL(裂变效率收入, 0)) AS 裂变效率收入
 FROM {TABLE_NAME}
 WHERE {common_filter}
   AND channel = '{SELF_CHANNEL}'
@@ -107,7 +107,7 @@ SELECT
   SUM(NVL(t0裂变人数, 0)) AS T0裂变数,
   SUM(NVL(成本, 0)) AS 成本,
   SUM(NVL(效率收入, 0)) AS 效率收入,
-  SUM(NVL(多层t0裂变收入, 0)) AS 多层裂变收入
+  SUM(NVL(裂变效率收入, 0)) AS 裂变效率收入
 FROM {TABLE_NAME}
 WHERE {common_filter}
   AND channel = '{GZH_CHANNEL}'
@@ -134,7 +134,7 @@ SELECT
   SUM(NVL(t0裂变人数, 0)) AS T0裂变数,
   SUM(NVL(成本, 0)) AS 成本,
   SUM(NVL(效率收入, 0)) AS 效率收入,
-  SUM(NVL(多层t0裂变收入, 0)) AS 多层裂变收入
+  SUM(NVL(裂变效率收入, 0)) AS 裂变效率收入
 FROM {TABLE_NAME}
 WHERE {common_filter}
   AND channel = '{QIWEI_CHANNEL}'

examples/auto_put_ad_mini/roi_control/metrics.py

@@ -19,8 +19,8 @@ from .fission_multiplier import (
 )
 
 
-METRIC_VERSION = "north_star_roi_t15_v4"
-METRIC_RUN_SUFFIX = "m4"
+METRIC_VERSION = "north_star_roi_t15_v5"
+METRIC_RUN_SUFFIX = "m5"
 
 SELF_CHANNEL = "小程序投流-稳定"
 GZH_CHANNEL = "公众号合作-即转-稳定"
@@ -49,7 +49,7 @@ NUMERIC_COLUMNS = [
     "T0裂变数",
     "成本",
     "效率收入",
-    "多层裂变收入",
+    "裂变效率收入",
 ]
 
 ENTITY_KEYS: Mapping[str, Sequence[str]] = {
@@ -69,7 +69,7 @@ def prepare_daily_metrics(
     raw: pd.DataFrame,
     fission_parameters: FissionMultiplierParameters,
 ) -> pd.DataFrame:
-    """清洗日聚合结果并计算实际 T0 与 T15 预测 ROI。"""
+    """清洗日聚合结果并用当日裂变效率收入计算 T15 预测 ROI。"""
 
     raw = raw.rename(
         columns={
@@ -92,7 +92,7 @@ def prepare_daily_metrics(
         daily[column] = pd.to_numeric(daily[column], errors="coerce").fillna(0.0)
 
     daily = apply_fission_multiplier(daily, fission_parameters)
-    daily["T0实际裂变收入"] = daily["多层裂变收入"]
+    daily["T0实际裂变收入"] = daily["裂变效率收入"]
     daily["实际全链路效率收入"] = (
         daily["效率收入"] + daily["T0实际裂变收入"]
     )
@@ -191,7 +191,7 @@ def _summarize_entity(
                 "T0裂变数",
                 "成本",
                 "效率收入",
-                "多层裂变收入",
+                "裂变效率收入",
                 "T0实际裂变收入",
                 "实际全链路效率收入",
                 "预测T1-T15裂变收入",
@@ -236,7 +236,9 @@ def _summarize_entity(
         record["成本"] = float(by_date["成本"].sum())
         record["三日最小单日成本"] = float(by_date["成本"].min())
         record["效率收入"] = float(by_date["效率收入"].sum())
-        record["多层裂变收入"] = float(by_date["多层裂变收入"].sum())
+        record["裂变效率收入"] = float(
+            by_date["裂变效率收入"].sum()
+        )
         record["T0实际裂变收入"] = float(
             by_date["T0实际裂变收入"].sum()
         )
@@ -287,8 +289,8 @@ def _summarize_entity(
             record[f"T0裂变数_{dt}"] = float(by_date.iloc[index]["T0裂变数"])
             record[f"成本_{dt}"] = float(by_date.iloc[index]["成本"])
             record[f"效率收入_{dt}"] = float(by_date.iloc[index]["效率收入"])
-            record[f"多层裂变收入_{dt}"] = float(
-                by_date.iloc[index]["多层裂变收入"]
+            record[f"裂变效率收入_{dt}"] = float(
+                by_date.iloc[index]["裂变效率收入"]
             )
             record[f"实际ROI_{dt}"] = actual_roi
             record[f"ROI_{dt}"] = predicted_roi

examples/auto_put_ad_mini/roi_control/policy.py

@@ -131,7 +131,7 @@ def annotate_execution(
                     "T0裂变数_20",
                     "成本_20",
                     "效率收入_20",
-                    "多层裂变收入_20",
+                    "裂变效率收入_20",
                     "实际ROI_20",
                     "ROI_20",
                 )
@@ -195,7 +195,7 @@ def annotate_execution(
                 "t0_fission_rate": _finite(row.get("T0裂变率")),
                 "cost": _finite(row.get("成本")),
                 "efficiency_revenue": _finite(row.get("效率收入")),
-                "fission_revenue": _finite(row.get("多层裂变收入")),
+                "fission_revenue": _finite(row.get("T0实际裂变收入")),
                 "actual_total_revenue": _finite(
                     row.get("实际全链路效率收入")
                 ),

examples/auto_put_ad_mini/roi_control/reporting.py

@@ -26,8 +26,8 @@ UP_FILL = PatternFill("solid", fgColor="D9EAD3")
 ADJUST_FILL = PatternFill("solid", fgColor="FFF2CC")
 APPROVAL_FILL = PatternFill("solid", fgColor="FFD966")
 APPROVAL_HEADER_FILL = PatternFill("solid", fgColor="BF9000")
-REPORT_VERSION = "roi_report_v8"
-REPORT_RUN_SUFFIX = "r8"
+REPORT_VERSION = "roi_report_v9"
+REPORT_RUN_SUFFIX = "r9"
 
 T0_FISSION_MULTIPLIER_COLUMN = "裂变系数-总裂变UV/T0裂变UV"
 TOTAL_FISSION_TO_FIRST_UV_COLUMN = DISPLAY_TOTAL_TO_FIRST_COLUMN
@@ -48,7 +48,7 @@ BASE_COLUMNS: Dict[str, Sequence[str]] = {
         "日均首层UV",
         "首层效率收入",
         "T0裂变效率收入",
-        "LTV预测效率收入",
+        "总预估效率收入",
         "日均成本",
         T0_FISSION_MULTIPLIER_COLUMN,
         TOTAL_FISSION_TO_FIRST_UV_COLUMN,
@@ -66,7 +66,7 @@ BASE_COLUMNS: Dict[str, Sequence[str]] = {
         "日均首层UV",
         "首层效率收入",
         "T0裂变效率收入",
-        "LTV预测效率收入",
+        "总预估效率收入",
         "日均成本",
         T0_FISSION_MULTIPLIER_COLUMN,
         TOTAL_FISSION_TO_FIRST_UV_COLUMN,
@@ -80,7 +80,7 @@ BASE_COLUMNS: Dict[str, Sequence[str]] = {
         "日均首层UV",
         "首层效率收入",
         "T0裂变效率收入",
-        "LTV预测效率收入",
+        "总预估效率收入",
         "日均成本",
         T0_FISSION_MULTIPLIER_COLUMN,
         TOTAL_FISSION_TO_FIRST_UV_COLUMN,
@@ -130,7 +130,7 @@ def _sheet_frame(candidates: pd.DataFrame, sheet_name: str) -> pd.DataFrame:
         pd.to_numeric(subset.get("T0实际裂变收入"), errors="coerce")
         / coverage_days
     )
-    subset["LTV预测效率收入"] = (
+    subset["总预估效率收入"] = (
         pd.to_numeric(
             subset.get("预测全链路效率收入"), errors="coerce"
         )
@@ -223,10 +223,11 @@ def _format_sheet(ws, sheet_name: str) -> None:
         elif "成本" in str(header) or header in {
             "首层效率收入",
             "T0裂变效率收入",
-            "LTV预测效率收入",
+            "总预估效率收入",
             "效率收入",
-            "T0实际裂变收入",
             "实际全链路效率收入",
+            "裂变效率收入",
+            "T0实际裂变收入",
             "预测T1-T15裂变收入",
             "预测T0-T15裂变收入",
             "预测全链路效率收入",
@@ -284,7 +285,7 @@ def _write_summary(
         ("UV口径", "COUNT(DISTINCT mid)"),
         (
             "ROI口径",
-            "T0裂变收入为实际值；仅预测T1-T15增量。预测总收入=首层实际效率收入+T0实际裂变收入+T0实际裂变收入×(裂变系数-总裂变UV/T0裂变UV-1)",
+            "源表裂变效率收入为T0当天实际值；预测总收入=首层效率收入+T0实际裂变收入+T0实际裂变收入×(传播裂变系数-1)",
         ),
         (
             "企微口径",

examples/auto_put_ad_mini/roi_control/repository.py

@@ -174,7 +174,20 @@ def create_or_load_run(record: dict[str, Any]) -> dict[str, Any]:
                 "SELECT * FROM roi_metric_run WHERE run_key=%s",
                 (record["run_key"],),
             )
-            return cursor.fetchone() or {}
+            run = cursor.fetchone()
+            if run:
+                return run
+            cursor.execute(
+                "SELECT * FROM roi_metric_run WHERE run_id=%s",
+                (record["run_id"],),
+            )
+            collision = cursor.fetchone()
+            if collision:
+                raise RuntimeError(
+                    f"ROI run_id already exists with a different run_key: "
+                    f"{record['run_id']}"
+                )
+            return {}
     finally:
         connection.close()
 

examples/auto_put_ad_mini/roi_control/service.py

@@ -2,8 +2,10 @@
 
 from __future__ import annotations
 
+import hashlib
 import logging
 import os
+import re
 from datetime import datetime, timedelta
 from pathlib import Path
 from typing import Any
@@ -76,7 +78,14 @@ def _dates(start_date: str) -> list[str]:
 def _run_identity(
     end_date: str,
     fission_parameters: FissionMultiplierParameters,
+    source_revision: str | None = None,
 ) -> tuple[str, str]:
+    if source_revision and not re.fullmatch(
+        r"[a-z0-9][a-z0-9_-]{0,63}", source_revision
+    ):
+        raise ValueError(
+            "source_revision must use lowercase letters, numbers, '_' or '-'"
+        )
     release = fission_parameters.release
     run_suffixes = [release.run_suffix, QIWEI_REFERENCE_RUN_SUFFIX]
     run_key_versions = [release.version, QIWEI_REFERENCE_VERSION]
@@ -88,6 +97,14 @@ def _run_identity(
         f"{METRIC_VERSION}:{POLICY_VERSION}:{':'.join(run_key_versions)}:"
         f"{REPORT_VERSION}:{end_date}"
     )
+    if source_revision:
+        run_id = f"{run_id}_{source_revision}"
+        revision_hash = hashlib.sha256(source_revision.encode("utf-8")).hexdigest()
+        run_key = f"{run_key}:sr:{revision_hash[:16]}"
+    if len(run_id) > 64:
+        raise ValueError("ROI run_id exceeds database limit")
+    if len(run_key) > 128:
+        raise ValueError("ROI run_key exceeds database limit")
     return run_id, run_key
 
 
@@ -97,6 +114,7 @@ def run_daily_roi(
     output_dir: Path,
     send_feishu: bool,
     now: datetime | None = None,
+    source_revision: str | None = None,
 ) -> dict[str, Any]:
     """Compute, snapshot, report, and optionally publish one ROI batch."""
 
@@ -114,11 +132,17 @@ def run_daily_roi(
     run_config["qiwei_reference_multiplier"] = QIWEI_REFERENCE_MULTIPLIER
     run_config["qiwei_reference_version"] = QIWEI_REFERENCE_VERSION
     run_config["report_version"] = REPORT_VERSION
+    if source_revision:
+        run_config["source_revision"] = source_revision
     client = ODPSClient(project=os.getenv("ODPS_PROJECT", "loghubods"))
     end_date = resolve_end_date(client, requested_end_date)
     start_date, end_date = date_window(end_date)
     expected_dates = _dates(start_date)
-    run_id, run_key = _run_identity(end_date, fission_parameters)
+    run_id, run_key = _run_identity(
+        end_date,
+        fission_parameters,
+        source_revision,
+    )
     run = create_or_load_run(
         {
             "run_id": run_id,
@@ -213,7 +237,9 @@ def run_daily_roi(
         )
 
         output_dir.mkdir(parents=True, exist_ok=True)
-        batch_name = f"ROI调控_{start_date}-{end_date}_LTV预测"
+        batch_name = f"ROI调控_{start_date}-{end_date}"
+        if source_revision:
+            batch_name = f"{batch_name}_{source_revision}"
         output_path = output_dir / f"{batch_name}.xlsx"
         write_workbook(
             report_rows,
@@ -231,6 +257,7 @@ def run_daily_roi(
             "reused": False,
             "start_date": start_date,
             "end_date": end_date,
+            "source_revision": source_revision,
             "entity_count": len(snapshots),
             "candidate_count": len(actionable_candidates),
             "actionable_count": len(actions),

examples/auto_put_ad_mini/run_daily_roi.py

@@ -34,6 +34,10 @@ def parse_args() -> argparse.Namespace:
         type=Path,
         default=HERE / "outputs" / "roi_control",
     )
+    parser.add_argument(
+        "--source-revision",
+        help="Stable revision label for corrected upstream data",
+    )
     parser.add_argument("--send-feishu", action="store_true")
     return parser.parse_args()
 
@@ -44,6 +48,7 @@ def main() -> None:
         requested_end_date=args.end_date,
         output_dir=args.output_dir,
         send_feishu=args.send_feishu,
+        source_revision=args.source_revision,
     )
     print(json.dumps(result, ensure_ascii=False, default=str))
 

examples/auto_put_ad_mini/test_roi_control_metrics.py

@@ -57,7 +57,7 @@ def row(entity_type, channel, entity_id, dt, roi, uv=600, age=10, cost=200.0):
         "T0裂变数": uv * 0.2,
         "成本": cost,
         "效率收入": roi * cost,
-        "多层裂变收入": 0,
+        "裂变效率收入": 0,
     }
     if entity_type == ENTITY_SELF:
         common.update(
@@ -99,6 +99,32 @@ class RoiRulesTest(unittest.TestCase):
         self.assertIn(QIWEI_REFERENCE_RUN_SUFFIX, formal_id)
         self.assertIn(QIWEI_REFERENCE_VERSION, formal_key)
 
+    def test_run_identity_isolates_corrected_source_revision(self):
+        default_id, default_key = _run_identity("20260729", FISSION_PARAMETERS)
+        revised_id, revised_key = _run_identity(
+            "20260729",
+            FISSION_PARAMETERS,
+            "datafix_20260730_1",
+        )
+
+        self.assertEqual(f"{default_id}_datafix_20260730_1", revised_id)
+        self.assertNotEqual(default_key, revised_key)
+        self.assertNotIn("datafix_20260730_1", revised_key)
+        self.assertLessEqual(len(revised_id), 64)
+        self.assertLessEqual(len(revised_key), 128)
+        self.assertEqual(
+            (revised_id, revised_key),
+            _run_identity(
+                "20260729",
+                FISSION_PARAMETERS,
+                "datafix_20260730_1",
+            ),
+        )
+        with self.assertRaisesRegex(ValueError, "source_revision"):
+            _run_identity("20260729", FISSION_PARAMETERS, "Invalid Revision")
+        with self.assertRaisesRegex(ValueError, "run_id"):
+            _run_identity("20260729", FISSION_PARAMETERS, "a" * 40)
+
     def build_daily(self):
         rows = []
         # 每日形成稳定分布，低值实体连续低于后20%，高值实体连续高于前20%。
@@ -150,14 +176,14 @@ class RoiRulesTest(unittest.TestCase):
         with self.assertRaisesRegex(ValueError, "连续三日数据"):
             evaluate_rules(incomplete, DATES)
 
-    def test_t0_income_is_scaled_once_without_duplicate_addition(self):
+    def test_daily_fission_income_is_scaled_once_without_duplicate_addition(self):
         daily = self.build_daily()
         mask = (
             daily["entity_type"].eq(ENTITY_GZH)
             & daily["公众号名"].eq("高ROI公众号")
         )
         daily.loc[mask, "效率收入"] = 100
-        daily.loc[mask, "多层裂变收入"] = 200
+        daily.loc[mask, "裂变效率收入"] = 200
         daily.loc[mask, "成本"] = 100
 
         _, _, summary = evaluate_rules(daily, DATES)
@@ -168,6 +194,8 @@ class RoiRulesTest(unittest.TestCase):
         expected = (
             100 + 200 * FISSION_PARAMETERS.gzh_channel
         ) / 100
+        self.assertEqual(target["裂变效率收入"], 600)
+        self.assertEqual(target["T0实际裂变收入"], 600)
         self.assertAlmostEqual(target["ROI"], expected)
         self.assertAlmostEqual(target["实际ROI"], 3.0)
 
@@ -294,6 +322,11 @@ class RoiRulesTest(unittest.TestCase):
         self.assertIn(FINAL_ROI_COLUMN, visible_columns)
         self.assertIn(T0_FISSION_MULTIPLIER_COLUMN, visible_columns)
         self.assertIn(TOTAL_FISSION_TO_FIRST_UV_COLUMN, visible_columns)
+        self.assertIn("T0裂变效率收入", visible_columns)
+        self.assertIn("总预估效率收入", visible_columns)
+        self.assertNotIn("LTV预测效率收入", visible_columns)
+        self.assertIn("裂变效率收入", frame.columns)
+        self.assertNotIn("裂变效率收入", visible_columns)
         self.assertTrue(
             frame[FINAL_ROI_COLUMN].equals(frame["ROI"])
         )

examples/auto_put_ad_mini/test_roi_fission_multiplier.py

@@ -157,7 +157,7 @@ class RoiFissionMultiplierTest(unittest.TestCase):
                     "T0裂变数": 20,
                     "成本": 100,
                     "效率收入": 50,
-                    "多层裂变收入": 30,
+                    "裂变效率收入": 30,
                 }
             ]
         )
@@ -371,7 +371,7 @@ class RoiFissionMultiplierTest(unittest.TestCase):
                     "T0裂变数": 20,
                     "成本": 100,
                     "效率收入": 50,
-                    "多层裂变收入": 30,
+                    "裂变效率收入": 30,
                 }
             ]
         )
@@ -398,6 +398,11 @@ class RoiFissionMultiplierTest(unittest.TestCase):
         self.assertIn("AS 广告优化目标", sql)
         self.assertIn("广告优化目标='null'", sql)
         self.assertIn("GROUP BY\n  dt, channel, 账号id, 广告id, 包名", sql)
+        self.assertEqual(
+            sql.count("SUM(NVL(裂变效率收入, 0)) AS 裂变效率收入"),
+            3,
+        )
+        self.assertNotIn("多层t0裂变收入", sql)
 
     def test_recompute_sql_matches_published_release_audit_sql(self):
         audit_dir = (