feat(roi): add three-day P20/P80 offline control report

examples/auto_put_ad_mini/.env.example

@@ -37,11 +37,14 @@ ROI_APPLY_ENABLED=0
 ROI_SHEET_APPROVAL_ENABLED=0
 ROI_SHEET_APPROVAL_POLL_SECONDS=60
 ROI_APPROVAL_TTL_MINUTES=120
+# 合格小程序创意三日ROI前20%且广告age>=3时生成扩量建议。
 ROI_SCALE_RATIO=1.10
 ROI_SCALE_COOLDOWN_DAYS=3
 ROI_MAX_BASE_RATIO=2.00
+# 连续三天合格的小程序创意+公众号实体等权整体P20。
 ROI_STOP_QUANTILE=0.20
 ROI_UP_QUANTILE=0.80
+ROI_OBSERVE_MIN_LATEST_UV=200
 ROI_FISSION_PARAMETER_VERSION=20260712_A0-A15_v1
 # ROI报表发送到此群；同时抄送 FEISHU_OPERATOR_CHAT_ID，重复ID自动去重。
 # 审批表链接为获得链接者可编辑，黄色审批列选择“批准”即自动执行。

examples/auto_put_ad_mini/DEPLOYMENT.md

@@ -168,6 +168,31 @@ curl -X POST http://localhost:8080/trigger | jq .
 
 ## 配置说明
 
+### 日级 ROI 三日口径
+
+- 日级 ROI 读取 T-1 至 T-3 三个连续完整数据日，数据深度过滤使用最新业务 SQL 的 `usersharedepth<=1`。
+- 正式阈值样本为连续三天每天首层 UV>200、成本>0 且 ROI 有效的小程序创意级和公众号实体；两类实体合并后按实体等权计算整体 P20 关停线。
+- 小程序广告级从 ODPS 原始数据按广告直接 `COUNT(DISTINCT mid)`，复用同一 P20，但不重复进入样本池；低于关停线且广告 age 达标时生成广告级关停建议。
+- 金额、UV 和人数在三日汇总表展示为三日总量/3；ROI 与裂变率使用三日总分子/总分母的加权口径。
+- 未进入正式样本、但最新日首层 UV>200 的实体放在各汇总 Sheet 末尾，仅观察、不参与 P20、不自动执行。
+- 报表包含小程序创意级、小程序广告级、公众号的三日汇总与每日明细共 6 个主 Sheet；广告级两个 Sheet 默认隐藏，企微暂不进入本版计算和报表。
+- 每日明细 Sheet 的 `dt` 位于第一列，整表按日期倒序排列。
+- 可见 ROI 列简化为“当日效率ROI”和“预测总效率ROI”；其后依次展示“关停线（P20）”和“扩量线（P80）”，公众号不参与扩量故扩量线留空；整体三日排名百分位保留为隐藏审计字段。
+- “动作”可见列改为“建议动作”；动作原因、阈值样本状态、执行状态和执行结果保留为隐藏审计字段。
+- 当前三日策略以创意级+公众号合格实体统一等权 P20 生成低 ROI 关停建议；以合格小程序创意实体等权 P80 识别头部20%，广告 age≥3 时生成扩量建议。创意级关停批准后只暂停对应动态创意，广告级关停批准后暂停整个广告。
+- 汇总表将三日总 T0 裂变人数 / 三日总首层 UV 的加权比例展示为“日均T0裂变率”。汇总顺序为关停、扩量、其他中间 ROI、观察；第一条扩量行和扩量后第一条无动作行顶部均使用粗线分隔。“当日效率ROI”和“预测总效率ROI”均使用深红—黄—绿色阶，绿色代表表现好。普通数值显示两位小数，UV、人数、数量和广告年龄显示整数；两个“裂变系数-总裂变UV/…”比率列固定显示两位小数。“传播裂变系数匹配”保留审计数据但默认隐藏。
+- “审批选择”默认隐藏，需取消隐藏后审批；创意级“当前创意状态”只对关停建议只读腾讯状态，显示正常、已停止或读取失败，其他行留空。
+
+离线复算不发送飞书：
+
+```bash
+.venv/bin/python examples/auto_put_ad_mini/run_daily_roi.py \
+  --end-date YYYYMMDD \
+  --source-revision <稳定修订标识>
+```
+
+只有显式增加 `--send-feishu` 才会发布飞书表格；该命令本身不会直接执行腾讯写操作。
+
 ### 环境变量
 
 | 变量名 | 说明 | 默认值 | 必需 |

examples/auto_put_ad_mini/docs/unified_services_deployment.md

@@ -165,27 +165,26 @@ docker compose --env-file /dev/null run --rm \
 
 日级 ROI 单独分两阶段启用：
 
-当前 `north_star_roi_t15_v7` 使用版本化传播裂变参数，策略版本为 `roi_policy_v6`，报表版本为 `roi_report_v15`。T0 裂变人数取 `SUM(t0_fission_uv_root)`，不再读取旧字段 `t0裂变人数`。小程序按人群包和转化目标、
-公众号按合作方和公众号匹配传播裂变系数；企微优先按合作方精确匹配，合作方未匹配或样本不足时使用默认参考系数 2.5，不做渠道聚合回退。企微仅保留展示，不进入阈值和调控。首层效率收入读取 `效率收入`，T0 实际裂变收入读取当天的 `裂变效率收入`，
+当前 `north_star_roi_t15_v8` 使用版本化传播裂变参数，策略版本为 `roi_policy_v11`，报表版本为 `roi_report_v29`。T0 裂变人数取 `SUM(t0_fission_uv_root)`，不再读取旧字段 `t0裂变人数`。小程序按人群包和转化目标、
+公众号按合作方和公众号匹配传播裂变系数；企微暂不进入本版报表、阈值和调控。首层效率收入读取 `效率收入`，T0 实际裂变收入读取当天的 `裂变效率收入`，
 并以 T0 实际裂变收入乘传播裂变系数预测完整裂变收入。日级任务不会现场重算 cohort 参数。部署前应保持
 `ROI_FISSION_PARAMETER_VERSION=20260712_A0-A15_v2`。参数必须先发布到 MySQL，
 启动日志、数据库发布版本和 Excel 摘要中的参数版本、cohort 日期必须一致。
-默认企微参考值使用独立版本 `qiwei_reference_2p5_v1`，并进入 ROI 批次幂等键，
-避免与其他临时系数结果复用同一个运行批次。
 ROI 在线表只上传一次，并发送到 `ROI_FEISHU_CHAT_ID` 和
 `FEISHU_OPERATOR_CHAT_ID`；两个变量相同时只发送一次。
-ROI 表格使用获得链接者可编辑权限。黄色【审批选择】列只接受“批准”或“拒绝”，
+ROI 表格使用获得链接者可编辑权限。默认隐藏的黄色【审批选择】列取消隐藏后只接受“批准”或“拒绝”，
 数据库隐藏幂等键决定真实执行目标，表格中的账户、广告、创意、成本和 ROI 不作为写入参数。
 
-审批表使用长表结构，同一实体每天一行并用 `dt` 区分，每行展示当日首层 UV、T0 裂变人数、
-T0 裂变率、首层/T0/预测总效率收入、成本、效率 ROI、两个裂变系数、消耗加权分位和
-P25 关停线；动作、审批和执行状态仅出现在最新日期行，
-历史日期行不得保留动作幂等键。两日汇总 ROI 仅保留为隐藏审计列。小程序与公众号分别按日期计算
-阈值，关停线使用成本权重按 P95 封顶后的消耗
-加权 P25，扩量线保持实体等权 P80；连续两天单日首层 UV 都大于 200 且
-连续处于同一极端方向时才生成正式建议。最新日首层 UV 大于 100 但未满足连续两日
-条件的实体作为“观察”放在正式样本之后，不进入阈值和腾讯执行。
-企微 Sheet 展示全部参考实体；各 Sheet 内正式样本按预测 ROI 升序排列。
+审批表包含小程序创意级、小程序广告级和公众号的三日汇总与每日明细共 6 个主 Sheet，广告级两个 Sheet 默认隐藏。
+金额、UV 和人数展示三日总量/3，ROI 和裂变率使用三日汇总后的加权口径。连续三天每天
+首层 UV>200、成本>0 且 ROI 有效的小程序创意和公众号实体合并后按实体等权计算整体 P20；
+广告级直接从 ODPS 原始数据按广告去重，复用 P20 但不进入样本池；低于关停线且广告 age 达标时生成广告级关停建议，批准后暂停整个广告。
+最新日首层 UV>200 但未满足三日条件的实体作为“观察”放在正式样本之后，不进入阈值和腾讯执行。
+可见尾部列按“当日效率ROI → 预测总效率ROI → 关停线（P20） → 扩量线（P80） → 建议动作”排列；公众号不参与扩量，扩量线留空。
+整体三日排名百分位、是否位于后20%、最新日UV、覆盖天数、动作原因、阈值样本状态和执行结果只保留为隐藏审计字段。
+每日明细 Sheet 的 `dt` 位于第一列并按日期倒序。主 Sheet 冻结首行和前 7 列，
+关停线列为“关停线（P20）”。汇总表将三日总 T0 裂变人数 / 三日总首层 UV 的加权比例展示为“日均T0裂变率”。合格小程序创意实体等权 P80 为扩量线，处于头部20%且广告 age≥3 时生成扩量建议。汇总顺序为关停、扩量、其他中间 ROI、观察；第一条扩量行和扩量后第一条无动作行顶部均使用粗线分隔。“当日效率ROI”和“预测总效率ROI”均使用深红—黄—绿色阶，绿色代表表现好。普通数值显示两位小数，UV、人数、数量和广告年龄显示整数；两个“裂变系数-总裂变UV/…”比率列固定显示两位小数。“传播裂变系数匹配”保留审计数据但默认隐藏。
+创意级“当前创意状态”只对关停建议只读腾讯状态并显示正常、已停止或读取失败，其他行留空；“审批选择”默认隐藏。
 
 1. 首次部署先执行下方参数发布命令并完成回读校验。
 2. 设置 `DAILY_ROI_ENABLED=1`、`ROI_APPLY_ENABLED=0`、`ROI_SHEET_APPROVAL_ENABLED=0`，观察 11:00 的 ODPS 计算、数据库快照和可编辑飞书表。

examples/auto_put_ad_mini/roi_control/config.py

@@ -30,8 +30,8 @@ class RoiConfig:
     self_up_min_age: int = 3
     self_min_daily_uv: float = 200
     partner_min_daily_uv: float = 200
-    observe_min_latest_uv: float = 100
-    stop_quantile: float = 0.25
+    observe_min_latest_uv: float = 200
+    stop_quantile: float = 0.20
     up_quantile: float = 0.80
     stop_weight_cap_quantile: float = 0.95
 
@@ -67,9 +67,9 @@ class RoiConfig:
                 )
             ),
             observe_min_latest_uv=float(
-                os.getenv("ROI_OBSERVE_MIN_LATEST_UV", "100")
+                os.getenv("ROI_OBSERVE_MIN_LATEST_UV", "200")
             ),
-            stop_quantile=float(os.getenv("ROI_STOP_QUANTILE", "0.25")),
+            stop_quantile=float(os.getenv("ROI_STOP_QUANTILE", "0.20")),
             up_quantile=float(os.getenv("ROI_UP_QUANTILE", "0.80")),
             stop_weight_cap_quantile=float(
                 os.getenv("ROI_STOP_WEIGHT_CAP_QUANTILE", "0.95")
@@ -105,8 +105,12 @@ class RoiConfig:
             self.observe_min_latest_uv,
         ) <= 0:
             raise ValueError("ROI UV thresholds must be positive")
-        if not 0 < self.stop_quantile < self.up_quantile < 1:
-            raise ValueError("ROI quantiles must satisfy 0 < stop < up < 1")
+        if not 0 < self.stop_quantile < 1:
+            raise ValueError("ROI_STOP_QUANTILE must satisfy 0 < value < 1")
+        if not self.stop_quantile < self.up_quantile < 1:
+            raise ValueError(
+                "ROI_UP_QUANTILE must satisfy ROI_STOP_QUANTILE < value < 1"
+            )
         if not 0 < self.stop_weight_cap_quantile <= 1:
             raise ValueError(
                 "ROI_STOP_WEIGHT_CAP_QUANTILE must satisfy 0 < value <= 1"

examples/auto_put_ad_mini/roi_control/data_source.py

@@ -8,7 +8,7 @@ from zoneinfo import ZoneInfo
 
 import pandas as pd
 
-from .metrics import GZH_CHANNEL, QIWEI_CHANNEL, SELF_CHANNEL
+from .metrics import GZH_CHANNEL, SELF_CHANNEL
 from .odps_client import ODPSClient
 
 
@@ -25,7 +25,7 @@ def parse_yyyymmdd(value: str) -> datetime:
 
 def date_window(end_date: str) -> Tuple[str, str]:
     end = parse_yyyymmdd(end_date)
-    return (end - timedelta(days=1)).strftime("%Y%m%d"), end.strftime("%Y%m%d")
+    return (end - timedelta(days=2)).strftime("%Y%m%d"), end.strftime("%Y%m%d")
 
 
 def resolve_end_date(client: ODPSClient, requested: str | None = None) -> str:
@@ -50,7 +50,7 @@ def build_daily_sql(start_date: str, end_date: str) -> str:
     parse_yyyymmdd(end_date)
     common_filter = f"""
       dt BETWEEN '{start_date}' AND '{end_date}'
-      AND usersharedepth = '0'
+      AND usersharedepth <= 1
       AND videoid IS NOT NULL
       AND NVL(hotsencetype, '') <> '1167'
     """
@@ -91,18 +91,18 @@ UNION ALL
 
 SELECT
   dt,
-  'gzh' AS entity_type,
+  'self_ad' AS entity_type,
   channel,
-  '' AS 代理名称,
-  '' AS 账号id,
-  '' AS 账号名称,
-  '' AS 广告id,
-  '' AS 广告名称,
-  '' AS 包名,
-  '' AS 广告优化目标,
+  MAX(NVL(代理名称, '')) AS 代理名称,
+  NVL(账号id, '') AS 账号id,
+  MAX(NVL(账号名称, '')) AS 账号名称,
+  NVL(广告id, '') AS 广告id,
+  MAX(NVL(广告名称, '')) AS 广告名称,
+  NVL(包名, '') AS 包名,
+  {optimize_goal} AS 广告优化目标,
   '' AS 创意id,
-  NVL(合作方名, '') AS 合作方名,
-  NVL(公众号名, '') AS 公众号名,
+  '' AS 合作方名,
+  '' AS 公众号名,
   COUNT(DISTINCT mid) AS 首层UV,
   SUM(NVL(t0_fission_uv_root, 0)) AS T0裂变数,
   SUM(NVL(成本, 0)) AS 成本,
@@ -110,15 +110,16 @@ SELECT
   SUM(NVL(裂变效率收入, 0)) AS 裂变效率收入
 FROM {TABLE_NAME}
 WHERE {common_filter}
-  AND channel = '{GZH_CHANNEL}'
-  AND 公众号名 IS NOT NULL
-GROUP BY dt, channel, 合作方名, 公众号名
+  AND channel = '{SELF_CHANNEL}'
+  AND 广告id IS NOT NULL
+GROUP BY
+  dt, channel, 账号id, 广告id, 包名, {optimize_goal}
 
 UNION ALL
 
 SELECT
   dt,
-  'qiwei' AS entity_type,
+  'gzh' AS entity_type,
   channel,
   '' AS 代理名称,
   '' AS 账号id,
@@ -129,7 +130,7 @@ SELECT
   '' AS 广告优化目标,
   '' AS 创意id,
   NVL(合作方名, '') AS 合作方名,
-  MAX(NVL(公众号名, '')) AS 公众号名,
+  NVL(公众号名, '') AS 公众号名,
   COUNT(DISTINCT mid) AS 首层UV,
   SUM(NVL(t0_fission_uv_root, 0)) AS T0裂变数,
   SUM(NVL(成本, 0)) AS 成本,
@@ -137,9 +138,9 @@ SELECT
   SUM(NVL(裂变效率收入, 0)) AS 裂变效率收入
 FROM {TABLE_NAME}
 WHERE {common_filter}
-  AND channel = '{QIWEI_CHANNEL}'
-  AND 合作方名 IS NOT NULL
-GROUP BY dt, channel, 合作方名
+  AND channel = '{GZH_CHANNEL}'
+  AND 公众号名 IS NOT NULL
+GROUP BY dt, channel, 合作方名, 公众号名
 """.strip()
 
 

examples/auto_put_ad_mini/roi_control/execution.py

@@ -26,7 +26,7 @@ from tencent_client import (
 )
 
 from .config import RoiConfig
-from .policy import ACTION_PAUSE_CREATIVE, ACTION_SCALE_BID
+from .policy import ACTION_PAUSE_AD, ACTION_PAUSE_CREATIVE, ACTION_SCALE_BID
 from .repository import (
     FINAL_STATUSES,
     claim_run_for_execution,
@@ -190,6 +190,72 @@ def _execute_pause(
         )
 
 
+def _execute_pause_ad(
+    item: dict[str, Any],
+    *,
+    client: TencentClient,
+    now: datetime,
+) -> None:
+    """Pause the entire ad after an ad-level ROI row is approved."""
+
+    account_id = int(item["account_id"])
+    adgroup_id = int(item["adgroup_id"])
+    ad = client.get_ad(account_id, adgroup_id)
+    before_status = str(ad.get("configured_status") or "")
+    if item["execution_status"] == "PREPARED" and before_status == SUSPEND_STATUS:
+        update_action_item(
+            int(item["id"]),
+            before_status=item.get("before_status"),
+            target_status=SUSPEND_STATUS,
+            readback_status=before_status,
+            execution_status="SUCCESS",
+            readback_json=ad,
+            executed_at=now,
+        )
+        return
+    if before_status != ACTIVE_STATUS:
+        update_action_item(
+            int(item["id"]),
+            before_status=before_status,
+            target_status=SUSPEND_STATUS,
+            readback_status=before_status,
+            execution_status="SKIPPED_STATE_MISMATCH",
+            skip_reason="ad is no longer active",
+            pre_state_json=ad,
+            executed_at=now,
+        )
+        return
+
+    update_action_item(
+        int(item["id"]),
+        before_status=before_status,
+        target_status=SUSPEND_STATUS,
+        execution_status="PREPARED",
+        pre_state_json=ad,
+    )
+    try:
+        readback = client.update_ad(
+            account_id,
+            adgroup_id,
+            target_status=SUSPEND_STATUS,
+        )
+        update_action_item(
+            int(item["id"]),
+            readback_status=str(readback.get("configured_status") or ""),
+            execution_status="SUCCESS",
+            readback_json=readback,
+            executed_at=now,
+        )
+    except Exception as exc:
+        update_action_item(
+            int(item["id"]),
+            execution_status=_failure_status(exc),
+            error_message=str(exc),
+            readback_json=(exc.actual if isinstance(exc, PostWriteVerificationError) else None),
+            executed_at=now,
+        )
+
+
 def _resume_prepared_scale(
     item: dict[str, Any],
     ad: dict[str, Any],
@@ -427,6 +493,8 @@ def execute_roi_batch(
             try:
                 if item["action_type"] == ACTION_PAUSE_CREATIVE:
                     _execute_pause(item, client=client, now=now)
+                elif item["action_type"] == ACTION_PAUSE_AD:
+                    _execute_pause_ad(item, client=client, now=now)
                 elif item["action_type"] == ACTION_SCALE_BID:
                     _execute_scale(
                         item,
@@ -494,6 +562,8 @@ def execute_approved_roi_actions(
             try:
                 if item["action_type"] == ACTION_PAUSE_CREATIVE:
                     _execute_pause(item, client=client, now=now)
+                elif item["action_type"] == ACTION_PAUSE_AD:
+                    _execute_pause_ad(item, client=client, now=now)
                 elif item["action_type"] == ACTION_SCALE_BID:
                     _execute_scale(
                         item,

examples/auto_put_ad_mini/roi_control/feishu.py

@@ -147,8 +147,8 @@ class RoiFeishuPublisher:
             instructions = "\n\n本批次没有可执行动作，无需审批。"
         else:
             instructions = (
-                "\n\n请在有效期内打开表格，在小程序投流表黄色"
-                "【审批选择】列逐行选择批准或拒绝。批准即为最终确认，系统自动执行。"
+                "\n\n请在有效期内打开表格，取消隐藏小程序三日汇总表的黄色"
+                "【审批选择】列后逐行选择批准或拒绝。批准即为最终确认，系统自动执行。"
             )
         card = {
             "config": {"wide_screen_mode": True},

examples/auto_put_ad_mini/roi_control/fission_multiplier.py

@@ -864,8 +864,9 @@ def apply_fission_multiplier(
     levels: list[str] = []
     sources: list[str] = []
     for row in result.to_dict("records"):
+        entity_type = str(row["entity_type"])
         match = parameters.match(
-            str(row["entity_type"]),
+            "self" if entity_type == "self_ad" else entity_type,
             package=row["包名"],
             optimize_goal=row["广告优化目标"],
             partner=row["合作方名"],
@@ -889,6 +890,9 @@ def apply_fission_multiplier(
     qiwei_formal = qiwei & ~qiwei_reference
     result["传播裂变参数状态"] = "正式参数"
     result["调控参与状态"] = "参与阈值和调控"
+    result.loc[result["entity_type"].eq("self_ad"), "调控参与状态"] = (
+        "广告级复用统一阈值_审批后可暂停广告"
+    )
     result.loc[qiwei, "调控参与状态"] = "仅展示_不进入阈值和调控"
     result["ROI计算口径"] = (
         "首层实际效率收入+T0实际裂变收入+预测T1-T15裂变收入"

examples/auto_put_ad_mini/roi_control/metrics.py

@@ -19,14 +19,15 @@ from .fission_multiplier import (
 )
 
 
-METRIC_VERSION = "north_star_roi_t15_v7"
-METRIC_RUN_SUFFIX = "m7"
+METRIC_VERSION = "north_star_roi_t15_v8"
+METRIC_RUN_SUFFIX = "m8"
 
 SELF_CHANNEL = "小程序投流-稳定"
 GZH_CHANNEL = "公众号合作-即转-稳定"
 QIWEI_CHANNEL = "群/企微合作-稳定"
 
 ENTITY_SELF = "self"
+ENTITY_SELF_AD = "self_ad"
 ENTITY_GZH = "gzh"
 ENTITY_QIWEI = "qiwei"
 
@@ -61,6 +62,13 @@ ENTITY_KEYS: Mapping[str, Sequence[str]] = {
         "广告优化目标",
         "创意id",
     ),
+    ENTITY_SELF_AD: (
+        "channel",
+        "账号id",
+        "广告id",
+        "包名",
+        "广告优化目标",
+    ),
     ENTITY_GZH: ("channel", "合作方名", "公众号名"),
     ENTITY_QIWEI: ("channel", "合作方名"),
 }
@@ -137,13 +145,13 @@ def _age_map(ad_age: pd.DataFrame | None) -> Dict[str, int]:
     return dict(zip(ages["广告id"], ages["广告age"]))
 
 
-def _expected_two_days(expected_dates: Iterable[str]) -> List[str]:
+def _expected_three_days(expected_dates: Iterable[str]) -> List[str]:
     dates = sorted({str(value) for value in expected_dates})
-    if len(dates) != 2:
-        raise ValueError(f"必须提供连续两个日期，实际为: {dates}")
+    if len(dates) != 3:
+        raise ValueError(f"必须提供连续三个日期，实际为: {dates}")
     parsed = pd.to_datetime(dates, format="%Y%m%d")
     gaps = parsed.to_series().diff().dropna().dt.days.tolist()
-    if gaps != [1]:
+    if gaps != [1, 1]:
         raise ValueError(f"日期必须连续，实际为: {dates}")
     return dates
 
@@ -183,7 +191,7 @@ def _summarize_entity(
         ]
         if inconsistent:
             raise ValueError(
-                f"同一实体两日内传播裂变参数不一致: {key_values}; fields={inconsistent}"
+                f"同一实体三日内传播裂变参数不一致: {key_values}; fields={inconsistent}"
             )
         observed_dates = set(group["dt"].astype(str))
         by_date = group.groupby("dt", as_index=False)[
@@ -277,8 +285,21 @@ def _summarize_entity(
             if record["成本"] > 0
             else np.nan
         )
-        record["前一日首层UV"] = float(by_date.iloc[0]["首层UV"])
-        record["前一日效率ROI"] = (
+        window_days = float(len(expected_dates))
+        record["日均T0裂变数"] = record["窗口T0裂变数"] / window_days
+        record["日均成本"] = record["成本"] / window_days
+        record["日均效率收入"] = record["效率收入"] / window_days
+        record["日均T0裂变效率收入"] = (
+            record["T0实际裂变收入"] / window_days
+        )
+        record["日均总预估效率收入"] = (
+            record["预测全链路效率收入"] / window_days
+        )
+        record["三日加权平均T0裂变率"] = record["T0裂变率"]
+        record["三日加权平均实际ROI"] = record["实际ROI"]
+        record["三日加权平均效率ROI"] = record["ROI"]
+        record["首日首层UV"] = float(by_date.iloc[0]["首层UV"])
+        record["首日效率ROI"] = (
             float(daily_predicted_roi[0])
             if np.isfinite(daily_predicted_roi[0])
             else np.nan
@@ -306,10 +327,13 @@ def _summarize_entity(
             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
 
-        if entity_type == ENTITY_SELF:
+        if entity_type in (ENTITY_SELF, ENTITY_SELF_AD):
             record["广告age"] = int(ages.get(str(record.get("广告id", "")), 0))
         records.append(record)
 
@@ -323,14 +347,14 @@ def compute_roi_summary(
     *,
     fission_parameters: FissionMultiplierParameters,
 ) -> tuple[pd.DataFrame, list[str]]:
-    """Return two-day entity ROI snapshots and normalized dates."""
+    """Return three-day entity ROI snapshots and normalized dates."""
 
-    dates = _expected_two_days(expected_dates)
+    dates = _expected_three_days(expected_dates)
     daily = prepare_daily_metrics(raw_daily, fission_parameters)
     daily = daily[daily["dt"].isin(dates)].copy()
     ages = _age_map(ad_age)
     summaries = []
-    for entity_type in (ENTITY_SELF, ENTITY_GZH, ENTITY_QIWEI):
+    for entity_type in (ENTITY_SELF, ENTITY_SELF_AD, ENTITY_GZH):
         entity_daily = daily[daily["entity_type"].eq(entity_type)]
         if entity_daily.empty:
             continue
@@ -344,7 +368,7 @@ def compute_roi_summary(
             summaries.append(entity_summary)
 
     if not summaries:
-        raise ValueError("三个目标渠道均没有可用的最近两日数据")
+        raise ValueError("三个目标实体均没有可用的最近三日数据")
 
     summary = pd.concat(summaries, ignore_index=True, sort=False)
     return summary, dates

examples/auto_put_ad_mini/roi_control/policy.py

@@ -12,10 +12,11 @@ from .fission_multiplier import (
     DISPLAY_MULTIPLIER_COLUMN,
     DISPLAY_TOTAL_TO_FIRST_COLUMN,
 )
-from .metrics import ENTITY_KEYS, ENTITY_SELF
+from .metrics import ENTITY_KEYS, ENTITY_SELF, ENTITY_SELF_AD
 
 
 ACTION_PAUSE_CREATIVE = "PAUSE_CREATIVE"
+ACTION_PAUSE_AD = "PAUSE_AD"
 ACTION_SCALE_BID = "SCALE_BID"
 MODE_ACTIONABLE = "ACTIONABLE"
 MODE_NOTIFY_ONLY = "NOTIFY_ONLY"
@@ -80,17 +81,23 @@ def annotate_execution(
         if recommendation:
             if recommendation == "观察":
                 execution_reason = "观察行仅展示，不执行腾讯写操作"
-            elif entity_type != ENTITY_SELF:
+            elif entity_type not in (ENTITY_SELF, ENTITY_SELF_AD):
                 execution_reason = "合作渠道当前仅通知，不执行腾讯写操作"
             elif account_id not in managed_account_ids:
                 execution_reason = "账户不在自动化管理范围，仅通知"
             elif not adgroup_id:
                 execution_reason = "缺少有效广告ID，仅通知"
-            elif recommendation == "关停" and not creative_id:
+            elif (
+                recommendation == "关停"
+                and entity_type == ENTITY_SELF
+                and not creative_id
+            ):
                 execution_reason = "缺少有效动态创意ID，仅通知"
-            elif recommendation == "关停":
+            elif recommendation == "关停" and entity_type == ENTITY_SELF_AD:
+                action_type = ACTION_PAUSE_AD
+            elif recommendation == "关停" and entity_type == ENTITY_SELF:
                 action_type = ACTION_PAUSE_CREATIVE
-            elif recommendation == "扩量":
+            elif recommendation == "扩量" and entity_type == ENTITY_SELF:
                 action_type = ACTION_SCALE_BID
             else:
                 execution_reason = "该建议当前没有自动执行器，仅通知"
@@ -99,11 +106,12 @@ def annotate_execution(
             suffix = creative_id if action_type == ACTION_PAUSE_CREATIVE else adgroup_id
             idempotency_key = f"{run_id}:{action_type}:{account_id}:{suffix}"
             execution_mode = MODE_ACTIONABLE
-            execution_reason = (
-                "审批后暂停该动态创意"
-                if action_type == ACTION_PAUSE_CREATIVE
-                else "审批后按ROI策略提高广告永久基础出价"
-            )
+            if action_type == ACTION_PAUSE_CREATIVE:
+                execution_reason = "审批后暂停该动态创意"
+            elif action_type == ACTION_PAUSE_AD:
+                execution_reason = "审批后暂停整个广告"
+            else:
+                execution_reason = "审批后按ROI策略提高广告永久基础出价"
             if idempotency_key not in seen_actions:
                 actions.append(
                     {

examples/auto_put_ad_mini/roi_control/reporting.py

@@ -1,46 +1,61 @@
-"""调控结果 Excel 报告生成。"""
+"""Three-day ROI Excel report generation with hidden audit columns."""
 
 from __future__ import annotations
 
+import json
+import re
 from pathlib import Path
-from typing import Dict, Mapping, Sequence
+from typing import Mapping, Sequence
 
 import numpy as np
 import pandas as pd
 from openpyxl import Workbook
-from openpyxl.worksheet.datavalidation import DataValidation
 from openpyxl.formatting.rule import ColorScaleRule
-from openpyxl.styles import Alignment, Font, PatternFill
+from openpyxl.styles import Alignment, Border, Font, PatternFill, Side
 from openpyxl.utils import get_column_letter
+from openpyxl.worksheet.datavalidation import DataValidation
+
 from .fission_multiplier import (
     DISPLAY_MULTIPLIER_COLUMN,
     DISPLAY_TOTAL_TO_FIRST_COLUMN,
 )
-from .metrics import ENTITY_GZH, ENTITY_QIWEI, ENTITY_SELF
+from .metrics import ENTITY_GZH, ENTITY_SELF, ENTITY_SELF_AD
 
 
 HEADER_FILL = PatternFill("solid", fgColor="1F4E78")
 HEADER_FONT = Font(color="FFFFFF", bold=True)
-STOP_FILL = PatternFill("solid", fgColor="F4CCCC")
-UP_FILL = PatternFill("solid", fgColor="D9EAD3")
-ADJUST_FILL = PatternFill("solid", fgColor="FFF2CC")
 OBSERVE_FILL = PatternFill("solid", fgColor="FFF2CC")
 APPROVAL_FILL = PatternFill("solid", fgColor="FFD966")
 APPROVAL_HEADER_FILL = PatternFill("solid", fgColor="BF9000")
-REPORT_VERSION = "roi_report_v15"
-REPORT_RUN_SUFFIX = "r15"
+
+REPORT_VERSION = "roi_report_v29"
+REPORT_RUN_SUFFIX = "r29"
 
 T0_FISSION_MULTIPLIER_COLUMN = "裂变系数-总裂变UV/T0裂变UV"
 TOTAL_FISSION_TO_FIRST_UV_COLUMN = DISPLAY_TOTAL_TO_FIRST_COLUMN
-FINAL_ROI_COLUMN = "最终效率ROI"
+FINAL_ROI_COLUMN = "三日加权平均效率ROI"
 
+SUMMARY_SHEETS = {
+    ENTITY_SELF: "小程序创意级三日汇总",
+    ENTITY_SELF_AD: "小程序广告级三日汇总",
+    ENTITY_GZH: "公众号三日汇总",
+}
+DAILY_SHEETS = {
+    ENTITY_SELF: "小程序创意级每日明细",
+    ENTITY_SELF_AD: "小程序广告级每日明细",
+    ENTITY_GZH: "公众号每日明细",
+}
+SHEET_TO_ENTITY = {
+    **{value: key for key, value in SUMMARY_SHEETS.items()},
+    **{value: key for key, value in DAILY_SHEETS.items()},
+}
 
-BASE_COLUMNS: Dict[str, Sequence[str]] = {
-    "小程序投流": (
-        "dt",
+ENTITY_DIMENSIONS = {
+    ENTITY_SELF: (
         "渠道",
         "代理名称",
         "账号id",
+        "账号名称",
         "广告id",
         "广告名称",
         "包名",
@@ -48,20 +63,38 @@ BASE_COLUMNS: Dict[str, Sequence[str]] = {
         "创意id",
         "广告age",
     ),
-    "公众号即转": (
-        "dt",
+    ENTITY_SELF_AD: (
         "渠道",
-        "合作方名",
-        "公众号名",
-    ),
-    "企微群合作": (
-        "dt",
-        "渠道",
-        "合作方名",
+        "代理名称",
+        "账号id",
+        "账号名称",
+        "广告id",
+        "广告名称",
+        "包名",
+        "广告优化目标",
+        "广告age",
     ),
+    ENTITY_GZH: ("渠道", "合作方名", "公众号名"),
 }
 
-DAILY_COLUMNS = (
+SUMMARY_METRICS = (
+    "日均首层UV",
+    "日均T0裂变人数",
+    "日均T0裂变率",
+    "日均首层效率收入",
+    "日均T0裂变效率收入",
+    "日均总预估效率收入",
+    "日均成本",
+    T0_FISSION_MULTIPLIER_COLUMN,
+    TOTAL_FISSION_TO_FIRST_UV_COLUMN,
+    "当日效率ROI",
+    "预测总效率ROI",
+    "关停线（P20）",
+    "扩量线（P80）",
+    "建议动作",
+)
+
+DAILY_METRICS = (
     "首层UV",
     "T0裂变人数",
     "T0裂变率",
@@ -69,109 +102,66 @@ DAILY_COLUMNS = (
     "T0裂变效率收入",
     "预测总效率收入",
     "成本",
-    "效率ROI",
+    "当日效率ROI",
+    "预测总效率ROI",
     T0_FISSION_MULTIPLIER_COLUMN,
     TOTAL_FISSION_TO_FIRST_UV_COLUMN,
-    "消耗加权分位",
-)
-
-DECISION_COLUMNS: Dict[str, Sequence[str]] = {
-    "小程序投流": (
-        "动作",
-        "动作原因",
-        "阈值样本状态",
-        "审批选择",
-        "执行状态",
-        "执行结果",
-    ),
-    "公众号即转": (
-        "动作",
-        "动作原因",
-        "阈值样本状态",
-    ),
-    "企微群合作": (),
-}
-
-QIWEI_STATUS_COLUMNS = (
-    "传播裂变参数状态",
-    "调控参与状态",
 )
 
-ENTITY_TO_SHEET = {
-    ENTITY_SELF: "小程序投流",
-    ENTITY_GZH: "公众号即转",
-    ENTITY_QIWEI: "企微群合作",
-}
-
-FREEZE_PANES = {
-    "小程序投流": "G2",
-    "公众号即转": "E2",
-    "企微群合作": "D2",
-}
-
 
 def _visible_columns(
     sheet_name: str,
-    expected_dates: Sequence[str],
-    stop_quantile: float,
+    expected_dates: Sequence[str] | None = None,
+    stop_quantile: float = 0.20,
 ) -> list[str]:
-    stop_label = f"P{int(stop_quantile * 100)}"
-    threshold_columns = (
-        []
-        if sheet_name == "企微群合作"
-        else [f"消耗加权{stop_label}线"]
-    )
-    status_columns = (
-        list(QIWEI_STATUS_COLUMNS)
-        if sheet_name == "企微群合作"
-        else list(DECISION_COLUMNS[sheet_name])
-    )
-    return (
-        list(BASE_COLUMNS[sheet_name])
-        + list(DAILY_COLUMNS)
-        + threshold_columns
-        + status_columns
-    )
+    entity_type = SHEET_TO_ENTITY[sheet_name]
+    if sheet_name in SUMMARY_SHEETS.values():
+        columns = list(ENTITY_DIMENSIONS[entity_type]) + list(SUMMARY_METRICS)
+        if entity_type == ENTITY_SELF:
+            columns += ["当前创意状态"]
+        return columns
+    return [
+        "dt",
+        "渠道",
+        *[c for c in ENTITY_DIMENSIONS[entity_type] if c != "渠道"],
+        *DAILY_METRICS,
+    ]
 
 
-def _sheet_frame(
-    candidates: pd.DataFrame,
-    sheet_name: str,
-    expected_dates: Sequence[str] | None = None,
-    stop_quantile: float = 0.25,
-) -> pd.DataFrame:
-    entity_type = next(
-        key for key, value in ENTITY_TO_SHEET.items() if value == sheet_name
-    )
-    subset = candidates[candidates["entity_type"].eq(entity_type)].copy()
+def _ensure_columns(frame: pd.DataFrame, columns: Sequence[str]) -> pd.DataFrame:
+    result = frame.copy()
+    for column in columns:
+        if column not in result:
+            result[column] = ""
+    return result
+
+
+def _summary_frame(rows: pd.DataFrame, entity_type: str) -> pd.DataFrame:
+    subset = rows[rows["entity_type"].eq(entity_type)].copy()
     subset = subset.rename(columns={"channel": "渠道"})
-    coverage_source = (
-        subset["覆盖天数"]
-        if "覆盖天数" in subset
-        else pd.Series(2, index=subset.index)
-    )
-    coverage_days = pd.to_numeric(
-        coverage_source, errors="coerce"
-    ).replace(0, np.nan)
     subset["日均首层UV"] = pd.to_numeric(
         subset.get("日均首层UV"), errors="coerce"
     ).round().astype("Int64")
-    subset["首层效率收入"] = (
-        pd.to_numeric(subset.get("效率收入"), errors="coerce")
-        / coverage_days
+    subset["日均T0裂变人数"] = pd.to_numeric(
+        subset.get("日均T0裂变数"), errors="coerce"
     )
-    subset["T0裂变效率收入"] = (
-        pd.to_numeric(subset.get("T0实际裂变收入"), errors="coerce")
-        / coverage_days
+    subset["日均首层效率收入"] = pd.to_numeric(
+        subset.get("日均效率收入"), errors="coerce"
     )
-    subset["总预估效率收入"] = (
-        pd.to_numeric(
-            subset.get("预测全链路效率收入"), errors="coerce"
-        )
-        / coverage_days
+    subset["日均T0裂变效率收入"] = pd.to_numeric(
+        subset.get("日均T0裂变效率收入"), errors="coerce"
+    )
+    subset["日均总预估效率收入"] = pd.to_numeric(
+        subset.get("日均总预估效率收入"), errors="coerce"
+    )
+    subset["日均T0裂变率"] = pd.to_numeric(
+        subset.get("三日加权平均T0裂变率"), errors="coerce"
+    )
+    subset["当日效率ROI"] = pd.to_numeric(
+        subset.get("三日加权平均实际ROI"), errors="coerce"
     )
-    subset["日均成本"] = (
-        pd.to_numeric(subset.get("成本"), errors="coerce") / coverage_days
+    subset["预测总效率ROI"] = pd.to_numeric(
+        subset.get("三日加权平均效率ROI"), errors="coerce"
     )
     subset[T0_FISSION_MULTIPLIER_COLUMN] = pd.to_numeric(
         subset.get(DISPLAY_MULTIPLIER_COLUMN), errors="coerce"
@@ -179,94 +169,106 @@ def _sheet_frame(
     subset[TOTAL_FISSION_TO_FIRST_UV_COLUMN] = pd.to_numeric(
         subset.get(DISPLAY_TOTAL_TO_FIRST_COLUMN), errors="coerce"
     )
-    subset[FINAL_ROI_COLUMN] = subset["ROI"]
-    subset["关停线"] = subset["t_stop"]
-    subset["扩量线"] = subset["t_up"]
-    dates = list(expected_dates or sorted(
-        column.removeprefix("ROI_")
-        for column in subset.columns
-        if column.startswith("ROI_") and len(column) == 12
-    ))
-    stop_label = f"P{int(stop_quantile * 100)}"
-    daily_frames = []
-    latest_date = dates[-1] if dates else ""
-    for dt in dates:
-        daily = subset.copy()
+    subset["关停线（P20）"] = pd.to_numeric(
+        subset.get("t_stop"), errors="coerce"
+    )
+    subset["扩量线（P80）"] = (
+        pd.to_numeric(subset.get("t_up"), errors="coerce")
+        if entity_type == ENTITY_SELF
+        else np.nan
+    )
+    subset["建议动作"] = subset.get("动作", "")
+    visible = _visible_columns(SUMMARY_SHEETS[entity_type])
+    required = list(visible)
+    if entity_type in (ENTITY_SELF, ENTITY_SELF_AD):
+        required.append("审批选择")
+    subset = _ensure_columns(subset, required)
+
+    if not subset.empty:
+        subset["_观察排序"] = subset["阈值样本状态"].eq(
+            "补充观察_昨日UV>200"
+        ).astype(int)
+        subset["_动作排序"] = (
+            subset["建议动作"]
+            .fillna("")
+            .map({"关停": 0, "扩量": 1, "": 2, "观察": 3})
+            .fillna(4)
+            .astype(int)
+        )
+        roi_sort = pd.to_numeric(
+            subset["三日加权平均效率ROI"], errors="coerce"
+        ).fillna(np.inf)
+        subset["_ROI排序"] = np.where(
+            subset["建议动作"].eq("扩量"), -roi_sort, roi_sort
+        )
+        subset["_成本排序"] = pd.to_numeric(
+            subset["日均成本"], errors="coerce"
+        ).fillna(0)
+        subset["_UV排序"] = pd.to_numeric(
+            subset["最新日首层UV"], errors="coerce"
+        ).fillna(0)
+        observation = subset["_观察排序"].eq(1)
+        subset.loc[observation, "_ROI排序"] = np.inf
+        subset.loc[observation, "_成本排序"] = 0
+        subset = subset.sort_values(
+            ["_观察排序", "_动作排序", "_ROI排序", "_成本排序", "_UV排序"],
+            ascending=[True, True, True, False, False],
+            kind="stable",
+        ).drop(columns=["_观察排序", "_动作排序", "_ROI排序", "_成本排序", "_UV排序"])
+    hidden = [column for column in subset.columns if column not in visible]
+    return subset[visible + hidden]
+
+
+def _daily_frame(
+    rows: pd.DataFrame,
+    entity_type: str,
+    expected_dates: Sequence[str],
+) -> pd.DataFrame:
+    summary = _summary_frame(rows, entity_type)
+    daily_frames: list[pd.DataFrame] = []
+    for dt in expected_dates:
+        daily = summary.copy()
         daily["dt"] = dt
         daily["首层UV"] = daily.get(f"首层UV_{dt}")
         daily["T0裂变人数"] = daily.get(f"T0裂变数_{dt}")
-        daily["T0裂变率"] = np.where(
-            pd.to_numeric(daily["首层UV"], errors="coerce").gt(0),
-            pd.to_numeric(daily["T0裂变人数"], errors="coerce")
-            / pd.to_numeric(daily["首层UV"], errors="coerce"),
-            np.nan,
-        )
+        uv = pd.to_numeric(daily["首层UV"], errors="coerce")
+        fission = pd.to_numeric(daily["T0裂变人数"], errors="coerce")
+        daily["T0裂变率"] = np.where(uv.gt(0), fission / uv, np.nan)
         daily["首层效率收入"] = daily.get(f"效率收入_{dt}")
         daily["T0裂变效率收入"] = daily.get(f"裂变效率收入_{dt}")
+        daily["预测总效率收入"] = daily.get(f"预测全链路效率收入_{dt}")
         daily["成本"] = daily.get(f"成本_{dt}")
-        daily["效率ROI"] = daily.get(f"ROI_{dt}")
-        daily["预测总效率收入"] = (
-            pd.to_numeric(daily["效率ROI"], errors="coerce")
-            * pd.to_numeric(daily["成本"], errors="coerce")
-        )
-        daily["消耗加权分位"] = daily.get(f"消耗加权同日分位_{dt}")
-        if sheet_name != "企微群合作":
-            daily[f"消耗加权{stop_label}线"] = daily.get(f"t_stop_{dt}")
-        if dt != latest_date:
-            for column in (
-                "动作",
-                "动作原因",
-                "阈值样本状态",
-                "执行状态",
-                "执行结果",
-                "动作幂等键",
-                "执行模式",
-                "执行说明",
-            ):
-                daily[column] = ""
-            daily["审批选择"] = "历史日"
+        daily["当日效率ROI"] = daily.get(f"实际ROI_{dt}")
+        daily["预测总效率ROI"] = daily.get(f"ROI_{dt}")
         daily_frames.append(daily)
-    if daily_frames:
-        subset = pd.concat(daily_frames, ignore_index=True)
-    else:
-        subset["dt"] = ""
-    visible_columns = _visible_columns(sheet_name, dates, stop_quantile)
-    for column in visible_columns:
-        if column not in subset:
-            subset[column] = ""
-    hidden_columns = [
-        column for column in subset.columns if column not in visible_columns
-    ]
-    ordered = visible_columns + hidden_columns
-    subset = subset[ordered]
-    if not subset.empty:
-        subset["_日期排序"] = subset["dt"].map(
-            {dt: position for position, dt in enumerate(reversed(dates))}
-        ).fillna(len(dates))
-        subset["_补充观察排序"] = (
-            subset["阈值样本状态"].eq("观察_最新日UV达标").astype(int)
-        )
-        action_order = {"关停": 0, "扩量": 1, "观察": 2, "": 3}
-        subset["_动作排序"] = (
-            subset["动作"].fillna("").map(action_order).fillna(4)
+    result = pd.concat(daily_frames, ignore_index=True) if daily_frames else summary
+    visible = _visible_columns(DAILY_SHEETS[entity_type])
+    result = _ensure_columns(result, visible)
+    if not result.empty:
+        entity_dimensions = [c for c in ENTITY_DIMENSIONS[entity_type] if c in result]
+        result["_实体序"] = result.groupby(entity_dimensions, dropna=False, sort=False).ngroup()
+        result["_日期序"] = result["dt"].map(
+            {dt: i for i, dt in enumerate(reversed(expected_dates))}
         )
-        subset["_当日成本排序"] = pd.to_numeric(
-            subset.get("成本"),
-            errors="coerce",
-        ).fillna(0)
-        subset = subset.sort_values(
-            ["_日期排序", "_补充观察排序", "_动作排序", "_当日成本排序"],
-            ascending=[True, True, True, False],
-            kind="stable",
-        ).drop(
-            columns=[
-                "_日期排序",
-                "_补充观察排序",
-                "_动作排序",
-                "_当日成本排序",
-            ]
+        result = result.sort_values(["_日期序", "_实体序"], kind="stable").drop(
+            columns=["_实体序", "_日期序"]
         )
-    return subset
+    hidden = [column for column in result.columns if column not in visible]
+    return result[visible + hidden]
+
+
+def _sheet_frame(
+    rows: pd.DataFrame,
+    sheet_name: str,
+    expected_dates: Sequence[str] | None = None,
+    stop_quantile: float = 0.20,
+) -> pd.DataFrame:
+    entity_type = SHEET_TO_ENTITY[sheet_name]
+    if sheet_name in SUMMARY_SHEETS.values():
+        return _summary_frame(rows, entity_type)
+    if expected_dates is None:
+        raise ValueError("每日明细必须提供 expected_dates")
+    return _daily_frame(rows, entity_type, expected_dates)
 
 
 def _write_dataframe(ws, frame: pd.DataFrame) -> None:
@@ -274,308 +276,234 @@ def _write_dataframe(ws, frame: pd.DataFrame) -> None:
     for values in frame.itertuples(index=False, name=None):
         ws.append(
             [
-                None if isinstance(value, float) and np.isnan(value) else value
+                None
+                if isinstance(value, (float, np.floating)) and np.isnan(value)
+                else value
                 for value in values
             ]
         )
 
 
+def _number_format_for_header(header: object) -> str:
+    name = re.sub(r"_\d{8}$", "", str(header or ""))
+    if name in {
+        T0_FISSION_MULTIPLIER_COLUMN,
+        TOTAL_FISSION_TO_FIRST_UV_COLUMN,
+    }:
+        return "0.00"
+    lower = name.lower()
+    is_count = (
+        name == "dt"
+        or lower.endswith("id")
+        or "uv" in lower
+        or "人数" in name
+        or "数量" in name
+        or name.endswith("age")
+        or name.endswith("天数")
+        or (name.endswith("数") and not name.endswith("系数"))
+    )
+    return "0" if is_count else "0.00"
+
+
+def _apply_number_formats(ws, headers: Mapping[object, int]) -> None:
+    for header, column_index in headers.items():
+        number_format = _number_format_for_header(header)
+        for row_number in range(2, ws.max_row + 1):
+            cell = ws.cell(row_number, column_index)
+            if isinstance(
+                cell.value, (int, float, np.integer, np.floating)
+            ) and not isinstance(cell.value, bool):
+                cell.number_format = number_format
+
+
 def _format_sheet(
     ws,
-    sheet_name: str,
     visible_columns: Sequence[str],
+    *,
+    approval: bool = False,
+    freeze_panes: str = "A2",
 ) -> None:
     max_column = max(ws.max_column, 1)
     max_row = max(ws.max_row, 1)
-    ws.freeze_panes = FREEZE_PANES[sheet_name]
+    ws.freeze_panes = freeze_panes
     ws.auto_filter.ref = f"A1:{get_column_letter(max_column)}{max_row}"
     ws.row_dimensions[1].height = 28
-
-    headers = {}
+    headers = {cell.value: cell.column for cell in ws[1]}
     for cell in ws[1]:
         cell.fill = HEADER_FILL
         cell.font = HEADER_FONT
         cell.alignment = Alignment(horizontal="center", vertical="center")
-        headers[cell.value] = cell.column
+    for column_index in range(1, max_column + 1):
+        header = ws.cell(1, column_index).value
+        ws.column_dimensions[get_column_letter(column_index)].hidden = (
+            header not in visible_columns
+        )
+        if header in visible_columns:
+            ws.column_dimensions[get_column_letter(column_index)].width = min(
+                max(12, len(str(header)) * 2 + 2), 34
+            )
+    _apply_number_formats(ws, headers)
+
+    action_column = headers.get("建议动作") or headers.get("动作")
+    status_column = headers.get("阈值样本状态")
+    for row_number in range(2, max_row + 1):
+        action = ws.cell(row_number, action_column).value if action_column else ""
+        status = ws.cell(row_number, status_column).value if status_column else ""
+        fill = (
+            OBSERVE_FILL
+            if status == "补充观察_昨日UV>200" or action == "观察"
+            else None
+        )
+        if fill:
+            for column_index in range(1, len(visible_columns) + 1):
+                ws.cell(row_number, column_index).fill = fill
+
+    if action_column:
+        def add_top_separator(row_number: int) -> None:
+            separator = Side(style="medium", color="1F1F1F")
+            for column_index in range(1, max_column + 1):
+                if ws.cell(1, column_index).value in visible_columns:
+                    ws.cell(row_number, column_index).border = Border(
+                        top=separator
+                    )
+
+        first_scale_row = next(
+            (
+                row_number
+                for row_number in range(2, max_row + 1)
+                if ws.cell(row_number, action_column).value == "扩量"
+            ),
+            None,
+        )
+        if first_scale_row:
+            add_top_separator(first_scale_row)
+            first_no_action_row = next(
+                (
+                    row_number
+                    for row_number in range(first_scale_row + 1, max_row + 1)
+                    if not (ws.cell(row_number, action_column).value or "")
+                ),
+                None,
+            )
+            if first_no_action_row:
+                add_top_separator(first_no_action_row)
+
+    for roi_header in ("当日效率ROI", "预测总效率ROI"):
+        roi_column = headers.get(roi_header)
+        if not roi_column or max_row < 2:
+            continue
+        roi_letter = get_column_letter(roi_column)
+        ws.conditional_formatting.add(
+            f"{roi_letter}2:{roi_letter}{max_row}",
+            ColorScaleRule(
+                start_type="min",
+                start_color="C00000",
+                mid_type="percentile",
+                mid_value=50,
+                mid_color="FFEB84",
+                end_type="max",
+                end_color="00B050",
+            ),
+        )
 
-    approval_column = headers.get("审批选择")
-    if approval_column:
-        approval_header = ws.cell(1, approval_column)
-        approval_header.fill = APPROVAL_HEADER_FILL
-        approval_header.font = HEADER_FONT
+    if approval and "审批选择" in headers and max_row >= 2:
+        column_index = headers["审批选择"]
+        letter = get_column_letter(column_index)
+        ws.cell(1, column_index).fill = APPROVAL_HEADER_FILL
         validation = DataValidation(
-            type="list",
-            formula1='"批准,拒绝"',
-            allow_blank=True,
+            type="list", formula1='"批准,拒绝"', allow_blank=True
         )
-        validation.error = "请选择批准或拒绝"
-        validation.errorTitle = "审批值无效"
         ws.add_data_validation(validation)
-        for row in range(2, max_row + 1):
-            cell = ws.cell(row, approval_column)
+        for row_number in range(2, max_row + 1):
+            cell = ws.cell(row_number, column_index)
             if cell.value in (None, ""):
-                cell.fill = APPROVAL_FILL
                 validation.add(cell)
+                cell.fill = APPROVAL_FILL
 
-    for column_index in range(1, max_column + 1):
-        header = ws.cell(1, column_index).value or ""
-        values = [str(ws.cell(row, column_index).value or "") for row in range(1, max_row + 1)]
-        width = min(max(max(map(len, values)), len(str(header))) + 2, 42)
-        ws.column_dimensions[get_column_letter(column_index)].width = max(width, 11)
-
-        if (
-            "ROI" in str(header)
-            or "裂变率" in str(header)
-            or "百分位" in str(header)
-            or "分位" in str(header)
-            or "关停线" in str(header)
-            or str(header).startswith("裂变系数-")
-        ):
-            for row in range(2, max_row + 1):
-                ws.cell(row, column_index).number_format = "0.000"
-        elif "成本" in str(header) or header in {
-            "首层效率收入",
-            "T0裂变效率收入",
-            "总预估效率收入",
-            "效率收入",
-            "实际全链路效率收入",
-            "裂变效率收入",
-            "T0实际裂变收入",
-            "预测T1-T15裂变收入",
-            "预测T0-T15裂变收入",
-            "预测全链路效率收入",
-        }:
-            for row in range(2, max_row + 1):
-                ws.cell(row, column_index).number_format = "#,##0.00"
-        elif "UV" in str(header) or "裂变数" in str(header):
-            for row in range(2, max_row + 1):
-                ws.cell(row, column_index).number_format = "#,##0"
-
-        if "ROI" in str(header) and max_row >= 2:
-            letter = get_column_letter(column_index)
-            ws.conditional_formatting.add(
-                f"{letter}2:{letter}{max_row}",
-                ColorScaleRule(
-                    start_type="min",
-                    start_color="F8696B",
-                    mid_type="percentile",
-                    mid_value=50,
-                    mid_color="FFEB84",
-                    end_type="max",
-                    end_color="63BE7B",
-                ),
-            )
-
-        if header not in set(visible_columns):
-            ws.column_dimensions[get_column_letter(column_index)].hidden = True
 
-    action_column = headers.get("动作")
-    if action_column:
-        for row in range(2, max_row + 1):
-            action = ws.cell(row, action_column).value
-            fill = {
-                "关停": STOP_FILL,
-                "扩量": UP_FILL,
-                "调整封面&落地页视频": ADJUST_FILL,
-                "观察": OBSERVE_FILL,
-            }.get(action)
-            if fill:
-                ws.cell(row, action_column).fill = fill
-
-def _write_summary(
+def _write_summary_sheet(
     workbook: Workbook,
     thresholds: pd.DataFrame,
-    candidates: pd.DataFrame,
     expected_dates: Sequence[str],
-    rule_config: Mapping[str, object] | None,
+    config: Mapping[str, object],
 ) -> None:
     ws = workbook.create_sheet("运行摘要")
-    self_min_uv = (rule_config or {}).get("self_min_daily_uv", 200)
-    partner_min_uv = (rule_config or {}).get("partner_min_daily_uv", 200)
-    observe_min_uv = (rule_config or {}).get("observe_min_latest_uv", 100)
-    stop_quantile = float((rule_config or {}).get("stop_quantile", 0.25))
-    up_quantile = float((rule_config or {}).get("up_quantile", 0.80))
-    weight_cap_quantile = float(
-        (rule_config or {}).get("stop_weight_cap_quantile", 0.95)
-    )
+    threshold = thresholds.iloc[0].to_dict() if not thresholds.empty else {}
     rows = [
+        ("报表版本", REPORT_VERSION),
         ("统计窗口", f"{expected_dates[0]} 至 {expected_dates[-1]}"),
-        ("首层口径", "usersharedepth='0'"),
-        ("UV口径", "COUNT(DISTINCT mid)"),
-        (
-            "ROI口径",
-            "源表裂变效率收入为T0当天实际值；预测总收入=首层效率收入+T0实际裂变收入+T0实际裂变收入×(传播裂变系数-1)",
-        ),
-        (
-            "企微口径",
-            "企微按合作方匹配已发布参数；未匹配或样本不足时默认取2.5。"
-            "企微仅展示，不进入阈值样本池和调控。",
-        ),
-        (
-            "阈值口径",
-            "小程序与公众号分渠道、分日期计算阈值；"
-            f"关停线为成本权重按P{int(weight_cap_quantile * 100)}封顶后的"
-            f"消耗加权P{int(stop_quantile * 100)}，"
-            f"扩量线保持实体等权P{int(up_quantile * 100)}；企微不参与。",
-        ),
-        (
-            "阈值样本条件",
-            f"连续2天分别满足：小程序单日首层UV>{self_min_uv:g}，"
-            f"公众号单日首层UV>{partner_min_uv:g}，且当日成本>0、当日ROI有效。",
-        ),
-        (
-            "动作口径",
-            f"连续2天均处于同渠道当日消耗加权后"
-            f"{int(stop_quantile * 100)}%才建议关停，"
-            f"连续2天均处于实体等权前{int((1 - up_quantile) * 100)}%"
-            "才建议扩量；"
-            "方向不一致或广告age不足时观察。",
-        ),
-        (
-            "观察补充",
-            f"未满足连续2天正式样本条件，但最新日首层UV>{observe_min_uv:g}的实体也进入报表末尾；"
-            "只展示最新日ROI，不进入阈值计算和腾讯执行。",
-        ),
-        ("审批方式", "在小程序投流表黄色【审批选择】列逐行选择批准或拒绝；批准即为最终确认并自动执行"),
+        ("数据深度口径", "usersharedepth<=1，与最新业务SQL一致"),
+        ("关停线（P20）", threshold.get("t_stop")),
+        ("阈值样本数", threshold.get("阈值样本数")),
+        ("创意扩量线（P80）", threshold.get("t_up")),
+        ("扩量样本数", threshold.get("扩量样本数")),
+        ("阈值样本", "连续三天每天首层UV>200、成本>0且ROI有效的小程序创意和公众号实体"),
+        ("广告级", "直接按广告去重计算，复用统一P20但不进入样本池；低于关停线且广告age达标时，审批后暂停整个广告"),
+        ("日均字段", "三日总量/3，缺失日按0"),
+        ("ROI与裂变率", "三日汇总分子/三日汇总分母的加权口径"),
+        ("补充观察", "非正式样本中最新日首层UV>200，置于汇总表末尾且不执行"),
+        ("配置快照", json.dumps(dict(config), ensure_ascii=False, default=str)),
     ]
-    for label, value in rows:
-        ws.append([label, value])
-
-    if rule_config:
-        ws.append([])
-        ws.append(["规则配置", "值"])
-        for key in (
-            "self_stop_min_age",
-            "self_up_min_age",
-            "self_min_daily_uv",
-            "partner_min_daily_uv",
-            "observe_min_latest_uv",
-            "stop_quantile",
-            "up_quantile",
-            "stop_weight_cap_quantile",
-            "scale_ratio",
-            "scale_cooldown_days",
-            "max_base_ratio",
+    for row in rows:
+        ws.append(row)
+    ws.column_dimensions["A"].width = 28
+    ws.column_dimensions["B"].width = 110
+    for cell in ws[1]:
+        cell.font = Font(bold=True)
+    for row_number in range(1, ws.max_row + 1):
+        label = str(ws.cell(row_number, 1).value or "")
+        value_cell = ws.cell(row_number, 2)
+        if isinstance(value_cell.value, (int, float)) and not isinstance(
+            value_cell.value, bool
         ):
-            ws.append([key, rule_config.get(key)])
-        fission_multiplier = rule_config.get("fission_multiplier")
-        if isinstance(fission_multiplier, Mapping):
-            ws.append([])
-            ws.append(["传播裂变参数", "值"])
-            for key in (
-                "version",
-                "cohort_date",
-                "observation_end_date",
-                "horizon_days",
-                "miniapp_exact_available_rows",
-                "gzh_exact_available_rows",
-                "miniapp_channel_multiplier",
-                "gzh_channel_multiplier",
-                "qiwei_exact_available_rows",
-            ):
-                if key in fission_multiplier:
-                    ws.append([key, fission_multiplier[key]])
-
-    ws.append([])
-    display_thresholds = thresholds.rename(
-        columns={"t_stop": "关停线", "t_up": "扩量线"}
-    )
-    threshold_columns = [
-        "统计日期",
-        "渠道",
-        "关停线",
-        "扩量线",
-        "阈值样本数",
-        "关停线口径",
-        "权重封顶成本",
-        "权重封顶分位",
-        "扩量线口径",
-    ]
-    ws.append(threshold_columns)
-    for values in display_thresholds[threshold_columns].itertuples(
-        index=False,
-        name=None,
-    ):
-        ws.append(list(values))
-
-    ws.append([])
-    ws.append(["渠道", "建议动作数"])
-    for entity_type, sheet_name in ENTITY_TO_SHEET.items():
-        count = int(
-            (
-                candidates["entity_type"].eq(entity_type)
-                & candidates["动作"].ne("")
-            ).sum()
-        )
-        ws.append([sheet_name, count])
-
-    ws.column_dimensions["A"].width = 22
-    ws.column_dimensions["B"].width = 92
-    for row in ws.iter_rows():
-        if row[0].value in {"统计窗口", "渠道"}:
-            for cell in row:
-                cell.font = Font(bold=True)
-    ws.freeze_panes = "A2"
+            value_cell.number_format = "0" if label.endswith("数") else "0.00"
 
 
 def write_workbook(
-    candidates: pd.DataFrame,
+    rows: pd.DataFrame,
     thresholds: pd.DataFrame,
     expected_dates: Sequence[str],
     output_path: Path,
-    rule_config: Mapping[str, object] | None = None,
+    config: Mapping[str, object],
     fission_match_summary: pd.DataFrame | None = None,
-) -> Path:
-    output_path.parent.mkdir(parents=True, exist_ok=True)
+) -> None:
     workbook = Workbook()
     workbook.remove(workbook.active)
 
-    for sheet_name in BASE_COLUMNS:
+    for entity_type in (ENTITY_SELF, ENTITY_SELF_AD, ENTITY_GZH):
+        sheet_name = SUMMARY_SHEETS[entity_type]
+        frame = _summary_frame(rows, entity_type)
         ws = workbook.create_sheet(sheet_name)
-        stop_quantile = float((rule_config or {}).get("stop_quantile", 0.25))
-        visible_columns = _visible_columns(
-            sheet_name,
-            expected_dates,
-            stop_quantile,
-        )
-        frame = _sheet_frame(
-            candidates,
-            sheet_name,
-            expected_dates,
-            stop_quantile,
+        _write_dataframe(ws, frame)
+        visible = _visible_columns(sheet_name)
+        _format_sheet(
+            ws,
+            visible,
+            approval=entity_type in (ENTITY_SELF, ENTITY_SELF_AD),
+            freeze_panes="H2",
         )
+        if entity_type == ENTITY_SELF_AD:
+            ws.sheet_state = "hidden"
+
+    for entity_type in (ENTITY_SELF, ENTITY_SELF_AD, ENTITY_GZH):
+        sheet_name = DAILY_SHEETS[entity_type]
+        frame = _daily_frame(rows, entity_type, expected_dates)
+        ws = workbook.create_sheet(sheet_name)
         _write_dataframe(ws, frame)
-        _format_sheet(ws, sheet_name, visible_columns)
+        _format_sheet(
+            ws,
+            _visible_columns(sheet_name),
+            freeze_panes="H2",
+        )
+        if entity_type == ENTITY_SELF_AD:
+            ws.sheet_state = "hidden"
 
     if fission_match_summary is not None:
         ws = workbook.create_sheet("传播裂变系数匹配")
         _write_dataframe(ws, fission_match_summary)
-        ws.freeze_panes = "A2"
-        ws.auto_filter.ref = ws.dimensions
-        for cell in ws[1]:
-            cell.fill = HEADER_FILL
-            cell.font = HEADER_FONT
-        for column_index in range(1, ws.max_column + 1):
-            header = str(ws.cell(1, column_index).value or "")
-            values = [
-                str(ws.cell(row, column_index).value or "")
-                for row in range(1, ws.max_row + 1)
-            ]
-            width = min(max(max(map(len, values)), len(header)) + 2, 42)
-            ws.column_dimensions[get_column_letter(column_index)].width = max(
-                width, 11
-            )
-            if header == "匹配率":
-                for row in range(2, ws.max_row + 1):
-                    ws.cell(row, column_index).number_format = "0.00%"
-
-    _write_summary(
-        workbook,
-        thresholds,
-        candidates,
-        expected_dates,
-        rule_config,
-    )
+        _format_sheet(ws, list(fission_match_summary.columns))
+        ws.sheet_state = "hidden"
 
+    _write_summary_sheet(workbook, thresholds, expected_dates, config)
+    output_path.parent.mkdir(parents=True, exist_ok=True)
     workbook.save(output_path)
-    return output_path

examples/auto_put_ad_mini/roi_control/rules.py

@@ -1,4 +1,4 @@
-"""Versioned daily ROI thresholds and two-day action policy."""
+"""Versioned three-day P20 stop and creative-level P80 scale policy."""
 
 from __future__ import annotations
 
@@ -11,14 +11,14 @@ import pandas as pd
 from .fission_multiplier import FissionMultiplierParameters
 from .metrics import (
     ENTITY_GZH,
-    ENTITY_QIWEI,
     ENTITY_SELF,
+    ENTITY_SELF_AD,
     compute_roi_summary,
 )
 
 
-POLICY_VERSION = "roi_policy_v6"
-POLICY_RUN_SUFFIX = "p6"
+POLICY_VERSION = "roi_policy_v11"
+POLICY_RUN_SUFFIX = "p11"
 
 
 @dataclass(frozen=True)
@@ -27,54 +27,9 @@ class RuleConfig:
     self_up_min_age: int = 3
     self_min_daily_uv: float = 200
     partner_min_daily_uv: float = 200
-    observe_min_latest_uv: float = 100
-    stop_quantile: float = 0.25
+    observe_min_latest_uv: float = 200
+    stop_quantile: float = 0.20
     up_quantile: float = 0.80
-    stop_weight_cap_quantile: float = 0.95
-
-
-def _weighted_quantile(
-    values: pd.Series,
-    weights: pd.Series,
-    quantile: float,
-) -> float:
-    frame = pd.DataFrame(
-        {
-            "value": pd.to_numeric(values, errors="coerce"),
-            "weight": pd.to_numeric(weights, errors="coerce"),
-        }
-    ).dropna()
-    frame = frame[frame["weight"].gt(0)].sort_values("value", kind="stable")
-    if frame.empty:
-        raise ValueError("消耗加权分位没有有效ROI或成本")
-    cutoff = float(frame["weight"].sum()) * quantile
-    position = int(frame["weight"].cumsum().searchsorted(cutoff, side="left"))
-    return float(frame.iloc[min(position, len(frame) - 1)]["value"])
-
-
-def _weighted_percentiles(
-    values: pd.Series,
-    weights: pd.Series,
-) -> pd.Series:
-    frame = pd.DataFrame(
-        {
-            "value": pd.to_numeric(values, errors="coerce"),
-            "weight": pd.to_numeric(weights, errors="coerce"),
-        }
-    )
-    valid = frame["value"].notna() & frame["weight"].gt(0)
-    result = pd.Series(np.nan, index=frame.index)
-    if not valid.any():
-        return result
-    grouped = (
-        frame.loc[valid]
-        .groupby("value", sort=True)["weight"]
-        .sum()
-        .cumsum()
-    )
-    percentile_by_value = grouped / grouped.iloc[-1]
-    result.loc[valid] = frame.loc[valid, "value"].map(percentile_by_value)
-    return result
 
 
 def _daily_sample_mask(
@@ -84,9 +39,9 @@ def _daily_sample_mask(
     config: RuleConfig,
 ) -> pd.Series:
     min_uv = (
-        config.self_min_daily_uv
-        if entity_type == ENTITY_SELF
-        else config.partner_min_daily_uv
+        config.partner_min_daily_uv
+        if entity_type == ENTITY_GZH
+        else config.self_min_daily_uv
     )
     return (
         summary["entity_type"].eq(entity_type)
@@ -96,95 +51,102 @@ def _daily_sample_mask(
     )
 
 
-def threshold_eligibility_mask(
+def entity_eligibility_mask(
     summary: pd.DataFrame,
+    entity_type: str,
     config: RuleConfig,
     expected_dates: list[str],
 ) -> pd.Series:
-    eligible = pd.Series(False, index=summary.index)
-    for entity_type in (ENTITY_SELF, ENTITY_GZH):
-        entity_eligible = summary["entity_type"].eq(entity_type)
-        for dt in expected_dates:
-            entity_eligible &= _daily_sample_mask(
-                summary,
-                entity_type,
-                dt,
-                config,
-            )
-        eligible |= entity_eligible
+    eligible = summary["entity_type"].eq(entity_type)
+    for dt in expected_dates:
+        eligible &= _daily_sample_mask(summary, entity_type, dt, config)
     return eligible
 
 
+def threshold_eligibility_mask(
+    summary: pd.DataFrame,
+    config: RuleConfig,
+    expected_dates: list[str],
+) -> pd.Series:
+    """Only creative-level miniapp and official accounts enter global P20."""
+
+    return entity_eligibility_mask(
+        summary, ENTITY_SELF, config, expected_dates
+    ) | entity_eligibility_mask(summary, ENTITY_GZH, config, expected_dates)
+
+
+def report_formal_mask(
+    summary: pd.DataFrame,
+    config: RuleConfig,
+    expected_dates: list[str],
+) -> pd.Series:
+    return threshold_eligibility_mask(
+        summary, config, expected_dates
+    ) | entity_eligibility_mask(summary, ENTITY_SELF_AD, config, expected_dates)
+
+
 def observation_mask(
     summary: pd.DataFrame,
     config: RuleConfig,
     expected_dates: list[str],
 ) -> pd.Series:
     latest = expected_dates[-1]
-    eligible = threshold_eligibility_mask(summary, config, expected_dates)
+    formal = report_formal_mask(summary, config, expected_dates)
     return (
-        summary["entity_type"].isin([ENTITY_SELF, ENTITY_GZH])
-        & ~eligible
+        summary["entity_type"].isin([ENTITY_SELF, ENTITY_SELF_AD, ENTITY_GZH])
+        & ~formal
         & pd.to_numeric(summary[f"首层UV_{latest}"], errors="coerce").gt(
             config.observe_min_latest_uv
         )
-        & pd.to_numeric(summary[f"成本_{latest}"], errors="coerce").gt(0)
-        & np.isfinite(pd.to_numeric(summary[f"ROI_{latest}"], errors="coerce"))
     )
 
 
-def compute_daily_thresholds(
+def compute_global_threshold(
     summary: pd.DataFrame,
     expected_dates: list[str],
     config: RuleConfig,
 ) -> pd.DataFrame:
-    records: list[dict[str, object]] = []
-    labels = {ENTITY_SELF: "小程序投流", ENTITY_GZH: "公众号即转"}
-    formal_eligible = threshold_eligibility_mask(
-        summary,
-        config,
-        expected_dates,
+    eligible = threshold_eligibility_mask(summary, config, expected_dates)
+    sample = pd.to_numeric(summary.loc[eligible, "ROI"], errors="coerce")
+    sample = sample[np.isfinite(sample)]
+    if sample.empty:
+        raise ValueError("最近三日没有满足每日UV和成本门槛的统一阈值样本")
+    creative_eligible = entity_eligibility_mask(
+        summary, ENTITY_SELF, config, expected_dates
     )
-    for dt in expected_dates:
-        for entity_type in (ENTITY_SELF, ENTITY_GZH):
-            mask = formal_eligible & summary["entity_type"].eq(entity_type)
-            valid = summary.loc[mask, f"ROI_{dt}"]
-            if valid.empty:
-                continue
-            raw_cost = pd.to_numeric(
-                summary.loc[mask, f"成本_{dt}"],
-                errors="coerce",
-            )
-            weight_cap = float(
-                raw_cost.quantile(config.stop_weight_cap_quantile)
-            )
-            capped_cost = raw_cost.clip(upper=weight_cap)
-            records.append(
-                {
-                    "统计日期": dt,
-                    "entity_type": entity_type,
-                    "渠道": labels[entity_type],
-                    "t_stop": _weighted_quantile(
-                        valid,
-                        capped_cost,
-                        config.stop_quantile,
-                    ),
-                    "t_up": float(valid.quantile(config.up_quantile)),
-                    "阈值样本数": int(len(valid)),
-                    "关停线口径": (
-                        f"消耗加权P{int(config.stop_quantile * 100)}"
-                    ),
-                    "权重封顶成本": weight_cap,
-                    "权重封顶分位": config.stop_weight_cap_quantile,
-                    "扩量线口径": (
-                        f"实体等权P{int(config.up_quantile * 100)}"
-                    ),
-                }
-            )
-    thresholds = pd.DataFrame(records)
-    if thresholds.empty:
-        raise ValueError("最近两日没有满足单日UV门槛的阈值样本")
-    return thresholds
+    creative_sample = pd.to_numeric(
+        summary.loc[creative_eligible, "ROI"], errors="coerce"
+    )
+    creative_sample = creative_sample[np.isfinite(creative_sample)]
+    t_up = (
+        float(creative_sample.quantile(config.up_quantile))
+        if not creative_sample.empty
+        else np.nan
+    )
+    return pd.DataFrame(
+        [
+            {
+                "统计窗口": f"{expected_dates[0]} 至 {expected_dates[-1]}",
+                "entity_type": "global",
+                "渠道": "小程序创意级+公众号",
+                "t_stop": float(sample.quantile(config.stop_quantile)),
+                "t_up": t_up,
+                "阈值样本数": int(len(sample)),
+                "扩量样本数": int(len(creative_sample)),
+                "关停线口径": f"合格实体等权P{int(config.stop_quantile * 100)}",
+                "扩量线口径": f"合格小程序创意实体等权P{int(config.up_quantile * 100)}",
+                "广告级是否入池": "否_仅复用统一关停线",
+            }
+        ]
+    )
+
+
+def _sample_percentile(values: pd.Series, target: float) -> float:
+    clean = pd.to_numeric(values, errors="coerce")
+    clean = clean[np.isfinite(clean)]
+    if clean.empty or not np.isfinite(target):
+        return np.nan
+    return float((clean <= target).mean())
 
 
 def apply_actions(
@@ -196,139 +158,104 @@ def apply_actions(
     result = summary.copy()
     result["动作"] = ""
     result["动作原因"] = ""
-    eligible = threshold_eligibility_mask(result, config, expected_dates)
+    result["t_stop"] = float(thresholds.iloc[0]["t_stop"])
+    result["t_up"] = float(thresholds.iloc[0]["t_up"])
+
+    threshold_eligible = threshold_eligibility_mask(result, config, expected_dates)
+    ad_eligible = entity_eligibility_mask(
+        result, ENTITY_SELF_AD, config, expected_dates
+    )
     observe_only = observation_mask(result, config, expected_dates)
-    threshold_map = {
-        (str(row["统计日期"]), str(row["entity_type"])): (
-            float(row["t_stop"]),
-            float(row["t_up"]),
+    sample_roi = result.loc[threshold_eligible, "ROI"]
+    creative_eligible = entity_eligibility_mask(
+        result, ENTITY_SELF, config, expected_dates
+    )
+    creative_sample_roi = result.loc[creative_eligible, "ROI"]
+    result["整体三日ROI排名百分位"] = result["ROI"].apply(
+        lambda value: _sample_percentile(sample_roi, float(value))
+        if pd.notna(value)
+        else np.nan
+    )
+    result["是否位于三日ROI后20%"] = False
+    result["创意三日ROI排名百分位"] = result["ROI"].apply(
+        lambda value: _sample_percentile(creative_sample_roi, float(value))
+        if pd.notna(value)
+        else np.nan
+    )
+    result["是否位于创意三日ROI前20%"] = False
+    comparable = threshold_eligible | ad_eligible
+    result.loc[comparable, "是否位于三日ROI后20%"] = (
+        pd.to_numeric(result.loc[comparable, "ROI"], errors="coerce")
+        <= result.loc[comparable, "t_stop"]
+    )
+    if np.isfinite(float(result["t_up"].iloc[0])):
+        result.loc[creative_eligible, "是否位于创意三日ROI前20%"] = (
+            pd.to_numeric(result.loc[creative_eligible, "ROI"], errors="coerce")
+            >= result.loc[creative_eligible, "t_up"]
         )
-        for row in thresholds.to_dict("records")
-    }
-
-    for dt in expected_dates:
-        result[f"同日排名百分位_{dt}"] = np.nan
-        result[f"消耗加权同日分位_{dt}"] = np.nan
-        result[f"t_stop_{dt}"] = np.nan
-        result[f"t_up_{dt}"] = np.nan
-        for entity_type in (ENTITY_SELF, ENTITY_GZH):
-            key = (dt, entity_type)
-            if key not in threshold_map:
-                continue
-            mask = eligible & result["entity_type"].eq(entity_type)
-            stop, up = threshold_map[key]
-            result.loc[mask, f"同日排名百分位_{dt}"] = (
-                result.loc[mask, f"ROI_{dt}"].rank(
-                    method="average",
-                    ascending=True,
-                    pct=True,
-                )
-            )
-            raw_cost = pd.to_numeric(
-                result.loc[mask, f"成本_{dt}"],
-                errors="coerce",
-            )
-            weight_cap = float(
-                raw_cost.quantile(config.stop_weight_cap_quantile)
-            )
-            result.loc[mask, f"消耗加权同日分位_{dt}"] = (
-                _weighted_percentiles(
-                    result.loc[mask, f"ROI_{dt}"],
-                    raw_cost.clip(upper=weight_cap),
-                )
-            )
-            entity_mask = result["entity_type"].eq(entity_type)
-            result.loc[entity_mask, f"t_stop_{dt}"] = stop
-            result.loc[entity_mask, f"t_up_{dt}"] = up
-
-    first, latest = expected_dates
-    result["前一日同日排名百分位"] = result[f"同日排名百分位_{first}"]
-    result["最新日同日排名百分位"] = result[f"同日排名百分位_{latest}"]
-    result["前一日消耗加权分位"] = result[f"消耗加权同日分位_{first}"]
-    result["最新日消耗加权分位"] = result[f"消耗加权同日分位_{latest}"]
-    result["t_stop"] = result[f"t_stop_{latest}"]
-    result["t_up"] = result[f"t_up_{latest}"]
 
     for index, row in result.iterrows():
         entity_type = str(row["entity_type"])
-        if entity_type == ENTITY_QIWEI:
-            continue
         if bool(observe_only.loc[index]):
             result.at[index, "动作"] = "观察"
             result.at[index, "动作原因"] = (
-                f"最新日首层UV>{config.observe_min_latest_uv:g}，"
-                "但不满足连续两天首层UV均达到正式样本门槛；"
-                "已展示最新日ROI，不进入阈值和自动执行"
+                f"最新日首层UV>{config.observe_min_latest_uv:g}，但未满足连续三天"
+                "每天首层UV>200、成本>0且ROI有效；仅置底展示，不进入P20和自动执行"
             )
             continue
-        if not bool(eligible.loc[index]):
+        if bool(ad_eligible.loc[index]):
+            if float(row["ROI"]) <= float(row["t_stop"]):
+                raw_age = row.get("广告age")
+                age = int(raw_age) if pd.notna(raw_age) else 0
+                if age >= config.self_stop_min_age:
+                    result.at[index, "动作"] = "关停"
+                    result.at[index, "动作原因"] = (
+                        f"广告级三日加权平均效率ROI≤统一实体等权P20，"
+                        f"广告age≥{config.self_stop_min_age}天；审批后暂停整个广告"
+                    )
+                else:
+                    result.at[index, "动作"] = "观察"
+                    result.at[index, "动作原因"] = (
+                        f"广告级三日加权平均效率ROI≤统一实体等权P20，但广告age<"
+                        f"{config.self_stop_min_age}天"
+                    )
+            continue
+        if not bool(threshold_eligible.loc[index]):
             continue
-
-        states: list[str] = []
-        for dt in expected_dates:
-            roi = float(row[f"ROI_{dt}"])
-            stop, up = threshold_map[(dt, entity_type)]
-            if roi <= stop:
-                states.append("低")
-            elif roi > 0 and roi >= up:
-                states.append("高")
-            else:
-                states.append("中")
-
-        both_low = states == ["低", "低"]
-        both_high = states == ["高", "高"]
-        inconsistent = len(set(states)) > 1 and any(
-            state in {"低", "高"} for state in states
-        )
-
         if entity_type == ENTITY_SELF:
             raw_age = row.get("广告age")
             age = int(raw_age) if pd.notna(raw_age) else 0
-            if both_low and age >= config.self_stop_min_age:
-                result.at[index, "动作"] = "关停"
-                result.at[index, "动作原因"] = (
-                    f"连续两天分别处于同渠道当日消耗加权后"
-                    f"{config.stop_quantile:.0%}，"
-                    f"广告age≥{config.self_stop_min_age}天"
-                )
-            elif both_high and age >= config.self_up_min_age:
-                result.at[index, "动作"] = "扩量"
-                result.at[index, "动作原因"] = (
-                    f"连续两天分别处于同渠道当日前{1-config.up_quantile:.0%}，"
-                    f"广告age≥{config.self_up_min_age}天"
-                )
-            elif both_low or both_high:
-                result.at[index, "动作"] = "观察"
-                required_age = (
-                    config.self_stop_min_age if both_low else config.self_up_min_age
-                )
-                result.at[index, "动作原因"] = (
-                    f"连续两天表现方向一致，但广告age<{required_age}天"
-                )
-            elif inconsistent:
-                result.at[index, "动作"] = "观察"
-                result.at[index, "动作原因"] = (
-                    f"连续两天表现不一致（{first}:{states[0]}，"
-                    f"{latest}:{states[1]}）"
-                )
-        elif entity_type == ENTITY_GZH:
-            if both_low:
-                result.at[index, "动作"] = "关停"
-                result.at[index, "动作原因"] = (
-                    f"连续两天分别处于公众号当日消耗加权后"
-                    f"{config.stop_quantile:.0%}"
-                )
-            elif both_high:
-                result.at[index, "动作"] = "扩量"
-                result.at[index, "动作原因"] = (
-                    f"连续两天分别处于公众号当日前{1-config.up_quantile:.0%}"
-                )
-            elif inconsistent:
-                result.at[index, "动作"] = "观察"
-                result.at[index, "动作原因"] = (
-                    f"连续两天表现不一致（{first}:{states[0]}，"
-                    f"{latest}:{states[1]}）"
-                )
+            if float(row["ROI"]) <= float(row["t_stop"]):
+                if age >= config.self_stop_min_age:
+                    result.at[index, "动作"] = "关停"
+                    result.at[index, "动作原因"] = (
+                        f"三日加权平均效率ROI≤统一实体等权P20，"
+                        f"广告age≥{config.self_stop_min_age}天；审批后仅暂停动态创意"
+                    )
+                else:
+                    result.at[index, "动作"] = "观察"
+                    result.at[index, "动作原因"] = (
+                        f"三日加权平均效率ROI≤统一实体等权P20，但广告age<"
+                        f"{config.self_stop_min_age}天"
+                    )
+            elif bool(row["是否位于创意三日ROI前20%"]):
+                if age >= config.self_up_min_age:
+                    result.at[index, "动作"] = "扩量"
+                    result.at[index, "动作原因"] = (
+                        f"三日加权平均效率ROI≥合格创意实体等权P80，"
+                        f"广告age≥{config.self_up_min_age}天；审批后提高广告永久基础出价"
+                    )
+                else:
+                    result.at[index, "动作"] = "观察"
+                    result.at[index, "动作原因"] = (
+                        f"三日加权平均效率ROI≥合格创意实体等权P80，但广告age<"
+                        f"{config.self_up_min_age}天"
+                    )
+        elif entity_type == ENTITY_GZH and float(row["ROI"]) <= float(row["t_stop"]):
+            result.at[index, "动作"] = "关停"
+            result.at[index, "动作原因"] = (
+                "三日加权平均效率ROI≤统一实体等权P20；公众号当前仅通知参考"
+            )
     return result
 
 
@@ -346,16 +273,14 @@ def evaluate_rules(
         ad_age,
         fission_parameters=fission_parameters,
     )
-    thresholds = compute_daily_thresholds(summary, dates, config)
+    thresholds = compute_global_threshold(summary, dates, config)
     evaluated = apply_actions(summary, thresholds, config, dates)
-    eligible = threshold_eligibility_mask(evaluated, config, dates)
+    formal_threshold = threshold_eligibility_mask(evaluated, config, dates)
+    formal_ad = entity_eligibility_mask(evaluated, ENTITY_SELF_AD, config, dates)
     observe_only = observation_mask(evaluated, config, dates)
-    evaluated["阈值样本状态"] = "未达到连续两日样本门槛"
-    evaluated.loc[eligible, "阈值样本状态"] = "进入连续两日阈值样本池"
-    evaluated.loc[observe_only, "阈值样本状态"] = "观察_最新日UV达标"
-    evaluated.loc[
-        evaluated["entity_type"].eq(ENTITY_QIWEI),
-        "阈值样本状态",
-    ] = "企微仅展示_不进入阈值样本池"
+    evaluated["阈值样本状态"] = "未达到三日正式样本门槛"
+    evaluated.loc[formal_threshold, "阈值样本状态"] = "进入三日统一阈值样本池"
+    evaluated.loc[formal_ad, "阈值样本状态"] = "广告级三日合格_不进入阈值样本池"
+    evaluated.loc[observe_only, "阈值样本状态"] = "补充观察_昨日UV>200"
     candidates = evaluated[evaluated["动作"].ne("")].copy()
     return candidates, thresholds, evaluated

examples/auto_put_ad_mini/roi_control/service.py

@@ -12,6 +12,7 @@ from typing import Any
 from zoneinfo import ZoneInfo
 
 from storage import initialize_schema, load_managed_accounts
+from tencent_client import ACTIVE_STATUS, TencentClient
 
 from .config import RoiConfig
 from .data_source import (
@@ -22,13 +23,10 @@ from .data_source import (
     resolve_end_date,
 )
 from .feishu import RoiFeishuPublisher
-from .metrics import ENTITY_QIWEI, METRIC_RUN_SUFFIX, METRIC_VERSION
+from .metrics import ENTITY_SELF, METRIC_RUN_SUFFIX, METRIC_VERSION
 from .fission_multiplier import (
     DEFAULT_FISSION_PARAMETER_VERSION,
     FissionMultiplierParameters,
-    QIWEI_REFERENCE_MULTIPLIER,
-    QIWEI_REFERENCE_RUN_SUFFIX,
-    QIWEI_REFERENCE_VERSION,
     parameters_from_database,
 )
 from .policy import annotate_execution
@@ -53,6 +51,53 @@ SHANGHAI = ZoneInfo("Asia/Shanghai")
 logger = logging.getLogger("auto_put_ad_mini.roi_control")
 
 
+def _annotate_current_creative_status(
+    rows,
+    tencent: TencentClient | None = None,
+):
+    """Read current Tencent status only for creative-level stop decisions."""
+
+    result = rows.copy()
+    result["当前创意状态"] = ""
+    mask = result["entity_type"].eq(ENTITY_SELF) & result["动作"].eq("关停")
+    if not mask.any():
+        return result
+
+    client = tencent or TencentClient()
+    cache: dict[tuple[int, int], str] = {}
+    try:
+        for index, row in result.loc[mask].iterrows():
+            try:
+                account_id = int(row["账号id"])
+                creative_id = int(row["创意id"])
+            except (TypeError, ValueError):
+                result.at[index, "当前创意状态"] = "读取失败"
+                continue
+            key = (account_id, creative_id)
+            if key not in cache:
+                try:
+                    creative = client.get_dynamic_creative(account_id, creative_id)
+                    status = str(creative.get("configured_status") or "")
+                    cache[key] = (
+                        "正常"
+                        if status == ACTIVE_STATUS
+                        else "已停止" if status else "读取失败"
+                    )
+                except Exception as exc:
+                    logger.warning(
+                        "Failed to read creative status account=%s creative=%s: %s",
+                        account_id,
+                        creative_id,
+                        exc,
+                    )
+                    cache[key] = "读取失败"
+            result.at[index, "当前创意状态"] = cache[key]
+    finally:
+        if tencent is None:
+            client.session.close()
+    return result
+
+
 def _rule_config(config: RoiConfig) -> RuleConfig:
     return RuleConfig(
         self_stop_min_age=config.self_stop_min_age,
@@ -62,7 +107,6 @@ def _rule_config(config: RoiConfig) -> RuleConfig:
         observe_min_latest_uv=config.observe_min_latest_uv,
         stop_quantile=config.stop_quantile,
         up_quantile=config.up_quantile,
-        stop_weight_cap_quantile=config.stop_weight_cap_quantile,
     )
 
 
@@ -70,7 +114,7 @@ def _dates(start_date: str) -> list[str]:
     start = datetime.strptime(start_date, "%Y%m%d")
     return [
         (start + timedelta(days=offset)).strftime("%Y%m%d")
-        for offset in range(2)
+        for offset in range(3)
     ]
 
 
@@ -86,8 +130,8 @@ def _run_identity(
             "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]
+    run_suffixes = [release.run_suffix]
+    run_key_versions = [release.version]
     run_id = (
         f"roi_{end_date}_{METRIC_RUN_SUFFIX}_{POLICY_RUN_SUFFIX}_"
         f"{'_'.join(run_suffixes)}_{REPORT_RUN_SUFFIX}"
@@ -128,8 +172,6 @@ def run_daily_roi(
     )
     run_config = config.snapshot()
     run_config["fission_multiplier"] = fission_parameters.snapshot()
-    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
@@ -195,6 +237,7 @@ def run_daily_roi(
             managed_ids,
             run_id,
         )
+        annotated = _annotate_current_creative_status(annotated)
         fission_match_summary = (
             summary.groupby(
                 ["entity_type", "传播裂变系数匹配层级"],
@@ -222,11 +265,11 @@ def run_daily_roi(
         report_rows = annotated[
             annotated["阈值样本状态"].isin(
                 [
-                    "进入连续两日阈值样本池",
-                    "观察_最新日UV达标",
+                    "进入三日统一阈值样本池",
+                    "广告级三日合格_不进入阈值样本池",
+                    "补充观察_昨日UV>200",
                 ]
             )
-            | annotated["entity_type"].eq(ENTITY_QIWEI)
         ].copy()
         for row in snapshots:
             row["run_id"] = run_id
@@ -234,7 +277,7 @@ def run_daily_roi(
             row["run_id"] = run_id
         threshold_record = {
             "统计窗口": f"{start_date} 至 {end_date}",
-            "每日渠道阈值": thresholds.to_dict("records"),
+            "整体三日关停线": thresholds.to_dict("records"),
         }
         replace_run_results(
             run_id,
@@ -281,7 +324,6 @@ def run_daily_roi(
         summary_text = (
             f"统计窗口：{start_date} - {end_date}\n"
             f"关停建议：{counts.get('关停', 0)}\n"
-            f"扩量建议：{counts.get('扩量', 0)}\n"
             f"观察：{counts.get('观察', 0)}\n"
             f"可执行动作：{len(actions)}\n"
             f"审批有效期：发送后 {config.approval_ttl_minutes} 分钟"

examples/auto_put_ad_mini/roi_control/sheet_approval.py

@@ -27,7 +27,11 @@ from .repository import (
 
 
 SHANGHAI = ZoneInfo("Asia/Shanghai")
-APPROVAL_SHEET_NAME = "小程序投流"
+APPROVAL_SHEET_NAMES = (
+    "小程序创意级三日汇总",
+    "小程序广告级三日汇总",
+)
+LEGACY_APPROVAL_SHEET_NAME = "小程序投流"
 APPROVED_VALUES = {"批准", "approve", "approved"}
 REJECTED_VALUES = {"拒绝", "reject", "rejected"}
 logger = logging.getLogger("auto_put_ad_mini.roi_sheet_approval")
@@ -101,7 +105,7 @@ class RoiSheetClient:
     def _headers(token: str) -> dict[str, str]:
         return {"Authorization": f"Bearer {token}"}
 
-    def _sheet_id(self, token: str, sheet_token: str) -> str:
+    def _sheet_ids(self, token: str, sheet_token: str) -> list[tuple[str, str]]:
         response = self.client.get(
             f"{BASE_URL}/sheets/v3/spreadsheets/{sheet_token}/sheets/query",
             headers=self._headers(token),
@@ -111,13 +115,23 @@ class RoiSheetClient:
             .get("data", {})
             .get("sheets", [])
         )
-        target = next(
-            (sheet for sheet in sheets if sheet.get("title") == APPROVAL_SHEET_NAME),
-            None,
-        )
-        if not target:
-            raise RuntimeError(f"ROI审批表缺少工作表: {APPROVAL_SHEET_NAME}")
-        return str(target["sheet_id"])
+        by_title = {
+            str(sheet.get("title") or ""): str(sheet["sheet_id"])
+            for sheet in sheets
+        }
+        targets = [
+            (title, by_title[title])
+            for title in APPROVAL_SHEET_NAMES
+            if title in by_title
+        ]
+        if not targets and LEGACY_APPROVAL_SHEET_NAME in by_title:
+            targets.append(
+                (LEGACY_APPROVAL_SHEET_NAME, by_title[LEGACY_APPROVAL_SHEET_NAME])
+            )
+        if not targets:
+            expected = "、".join((*APPROVAL_SHEET_NAMES, LEGACY_APPROVAL_SHEET_NAME))
+            raise RuntimeError(f"ROI审批表缺少工作表: {expected}")
+        return targets
 
     def _read_values(
         self,
@@ -137,16 +151,35 @@ class RoiSheetClient:
             .get("values", [])
         ) or []
 
-    def read_approvals(self, sheet_token: str) -> tuple[str, list[dict[str, Any]]]:
+    def read_approvals(self, sheet_token: str) -> list[dict[str, Any]]:
         token = self._token()
-        sheet_id = self._sheet_id(token, sheet_token)
+        decisions: list[dict[str, Any]] = []
+        for sheet_name, sheet_id in self._sheet_ids(token, sheet_token):
+            decisions.extend(
+                self._read_sheet_approvals(
+                    token,
+                    sheet_token,
+                    sheet_name=sheet_name,
+                    sheet_id=sheet_id,
+                )
+            )
+        return decisions
+
+    def _read_sheet_approvals(
+        self,
+        token: str,
+        sheet_token: str,
+        *,
+        sheet_name: str,
+        sheet_id: str,
+    ) -> list[dict[str, Any]]:
         header_rows = self._read_values(
             token,
             sheet_token,
             f"{sheet_id}!A1:ZZ1",
         )
         if not header_rows:
-            return sheet_id, []
+            return []
         headers = {
             str(value or "").strip(): index
             for index, value in enumerate(header_rows[0])
@@ -188,14 +221,15 @@ class RoiSheetClient:
                     "idempotency_key": key,
                     "decision": decision,
                     "headers": headers,
+                    "sheet_id": sheet_id,
+                    "sheet_name": sheet_name,
                 }
             )
-        return sheet_id, decisions
+        return decisions
 
     def write_results(
         self,
         sheet_token: str,
-        sheet_id: str,
         decisions: list[dict[str, Any]],
         actions: list[dict[str, Any]],
     ) -> None:
@@ -223,7 +257,10 @@ class RoiSheetClient:
             end = get_column_letter(result_column + 1)
             value_ranges.append(
                 {
-                    "range": f"{sheet_id}!{start}{decision['row_number']}:{end}{decision['row_number']}",
+                    "range": (
+                        f"{decision['sheet_id']}!{start}{decision['row_number']}:"
+                        f"{end}{decision['row_number']}"
+                    ),
                     "values": [[status, result]],
                 }
             )
@@ -274,7 +311,7 @@ class RoiSheetApprovalService:
         }
 
     def _process_run(self, run: dict[str, Any], now: datetime) -> int:
-        sheet_id, sheet_decisions = self.client.read_approvals(run["sheet_token"])
+        sheet_decisions = self.client.read_approvals(run["sheet_token"])
         approved_ids: list[int] = []
         processed_decisions: list[dict[str, Any]] = []
         changed_count = 0
@@ -307,7 +344,6 @@ class RoiSheetApprovalService:
             try:
                 self.client.write_results(
                     run["sheet_token"],
-                    sheet_id,
                     processed_decisions,
                     actions,
                 )

examples/auto_put_ad_mini/test_roi_control_metrics.py

@@ -1,36 +1,37 @@
+import tempfile
 import unittest
-from dataclasses import replace
+from pathlib import Path
 
 import pandas as pd
+from openpyxl import load_workbook
 
+from roi_control.data_source import build_daily_sql, date_window
 from roi_control.fission_multiplier import (
-    DISPLAY_TOTAL_TO_FIRST_COLUMN,
-    MATCH_QIWEI_PARTNER,
-    QIWEI_REFERENCE_RUN_SUFFIX,
-    QIWEI_REFERENCE_VERSION,
+    DISPLAY_MULTIPLIER_COLUMN,
     load_fission_multiplier_parameters,
 )
 from roi_control.metrics import (
     ENTITY_GZH,
-    ENTITY_QIWEI,
     ENTITY_SELF,
+    ENTITY_SELF_AD,
     GZH_CHANNEL,
-    QIWEI_CHANNEL,
     SELF_CHANNEL,
+    prepare_daily_metrics,
 )
 from roi_control.reporting import (
-    FINAL_ROI_COLUMN,
-    T0_FISSION_MULTIPLIER_COLUMN,
-    TOTAL_FISSION_TO_FIRST_UV_COLUMN,
-    _sheet_frame,
+    DAILY_SHEETS,
+    SUMMARY_SHEETS,
+    _daily_frame,
+    _summary_frame,
     _visible_columns,
+    write_workbook,
 )
-from roi_control.data_source import build_daily_sql
 from roi_control.rules import evaluate_rules as _evaluate_rules
-from roi_control.service import _run_identity
+from roi_control.service import _annotate_current_creative_status, _run_identity
+from tencent_client import ACTIVE_STATUS, SUSPEND_STATUS
 
 
-DATES = ["20260721", "20260722"]
+DATES = ["20260720", "20260721", "20260722"]
 FISSION_PARAMETERS = load_fission_multiplier_parameters()
 
 
@@ -39,11 +40,11 @@ def evaluate_rules(*args, **kwargs):
     return _evaluate_rules(*args, **kwargs)
 
 
-def row(entity_type, channel, entity_id, dt, roi, uv=600, cost=200.0):
+def row(entity_type, entity_id, dt, roi, *, uv=600, cost=200.0):
     common = {
         "dt": dt,
         "entity_type": entity_type,
-        "channel": channel,
+        "channel": SELF_CHANNEL if entity_type != ENTITY_GZH else GZH_CHANNEL,
         "代理名称": "",
         "账号id": "",
         "账号名称": "",
@@ -60,7 +61,7 @@ def row(entity_type, channel, entity_id, dt, roi, uv=600, cost=200.0):
         "效率收入": roi * cost,
         "裂变效率收入": 0,
     }
-    if entity_type == ENTITY_SELF:
+    if entity_type in (ENTITY_SELF, ENTITY_SELF_AD):
         common.update(
             {
                 "代理名称": "代理",
@@ -68,246 +69,363 @@ def row(entity_type, channel, entity_id, dt, roi, uv=600, cost=200.0):
                 "账号名称": "账户",
                 "广告id": entity_id,
                 "广告名称": f"广告{entity_id}",
-                "包名": "人群包",
+                "包名": "泛人群",
                 "广告优化目标": "关键页面访问次数",
-                "创意id": f"creative-{entity_id}",
+                "创意id": f"creative-{entity_id}" if entity_type == ENTITY_SELF else "",
             }
         )
-    elif entity_type == ENTITY_GZH:
-        common.update({"合作方名": "合作方", "公众号名": entity_id})
     else:
-        common.update({"合作方名": entity_id})
+        common.update({"合作方名": "合作方", "公众号名": entity_id})
     return common
 
 
-class RoiRulesTest(unittest.TestCase):
+class RoiThreeDayRulesTest(unittest.TestCase):
     def build_daily(self):
         rows = []
         for dt in DATES:
-            for index, roi in enumerate(
-                [0.1, 0.5, 0.8, 1.0, 1.2, 1.5, 2.0, 3.0, 4.0, 5.0]
-            ):
-                rows.append(
-                    row(ENTITY_SELF, SELF_CHANNEL, f"ad-{index}", dt, roi)
-                )
-            for index, roi in enumerate([0.1, 0.5, 1.0, 2.0, 4.0]):
-                rows.append(
-                    row(
-                        ENTITY_GZH,
-                        GZH_CHANNEL,
-                        f"公众号-{index}",
-                        dt,
-                        roi,
-                        uv=300,
-                    )
-                )
-            rows.append(
-                row(ENTITY_QIWEI, QIWEI_CHANNEL, "企微合作方", dt, 2.0, uv=300)
-            )
+            for index, roi in enumerate([0.1, 1.0, 3.0, 4.0]):
+                rows.append(row(ENTITY_SELF, f"creative-ad-{index}", dt, roi))
+            for index, roi in enumerate([0.5, 2.0, 5.0, 6.0]):
+                rows.append(row(ENTITY_GZH, f"公众号-{index}", dt, roi, uv=300))
+            for index, roi in enumerate([0.2, 3.5]):
+                rows.append(row(ENTITY_SELF_AD, f"ad-{index}", dt, roi))
         return pd.DataFrame(rows)
 
-    def test_run_identity_tracks_metric_policy_and_qiwei_versions(self):
-        run_id, run_key = _run_identity("20260727", FISSION_PARAMETERS)
-        self.assertIn("m7", run_id)
-        self.assertIn("p6", run_id)
-        self.assertIn("r15", run_id)
-        self.assertIn(QIWEI_REFERENCE_RUN_SUFFIX, run_id)
-        self.assertIn(QIWEI_REFERENCE_VERSION, run_key)
-
-    def test_daily_sql_uses_root_deduplicated_t0_fission_uv(self):
-        sql = build_daily_sql(DATES[0], DATES[1])
-        self.assertEqual(sql.count("SUM(NVL(t0_fission_uv_root, 0))"), 3)
-        self.assertNotIn("t0裂变人数", sql)
-
-        formal = replace(
-            FISSION_PARAMETERS,
-            qiwei_by_partner={"合作方": 2.0},
-            total_to_first={
-                **FISSION_PARAMETERS.total_to_first,
-                ("qiwei", MATCH_QIWEI_PARTNER, "合作方", ""): 0.75,
-            },
-            qiwei_exact_rows=1,
-            qiwei_exact_available_rows=1,
-        )
-        formal_id, _ = _run_identity("20260727", formal)
-        self.assertIn(QIWEI_REFERENCE_RUN_SUFFIX, formal_id)
-
-    def test_two_daily_thresholds_drive_consecutive_actions(self):
-        ages = pd.DataFrame(
+    def ages(self):
+        return pd.DataFrame(
             {
-                "广告id": [f"ad-{index}" for index in range(10)],
-                "广告age": [10] * 10,
+                "广告id": [f"creative-ad-{i}" for i in range(4)]
+                + [f"ad-{i}" for i in range(2)],
+                "广告age": [10] * 6,
             }
         )
-        candidates, thresholds, summary = evaluate_rules(
-            self.build_daily(), DATES, ages
-        )
 
-        self.assertEqual(len(thresholds), 4)
-        self.assertEqual(
-            set(zip(thresholds["统计日期"], thresholds["entity_type"])),
-            {
-                (DATES[0], ENTITY_SELF),
-                (DATES[1], ENTITY_SELF),
-                (DATES[0], ENTITY_GZH),
-                (DATES[1], ENTITY_GZH),
-            },
-        )
-        self.assertIn("关停", set(candidates["动作"]))
-        self.assertIn("扩量", set(candidates["动作"]))
-        self.assertTrue((summary["覆盖天数"] == 2).all())
-        self.assertTrue(
-            summary[
-                summary["entity_type"].isin([ENTITY_SELF, ENTITY_GZH])
-            ]["阈值样本状态"].eq("进入连续两日阈值样本池").all()
+    def test_versions_and_three_day_window(self):
+        self.assertEqual(date_window("20260722"), ("20260720", "20260722"))
+        run_id, _ = _run_identity("20260722", FISSION_PARAMETERS)
+        self.assertIn("m8", run_id)
+        self.assertIn("p11", run_id)
+        self.assertIn("r29", run_id)
+
+    def test_current_creative_status_only_reads_stop_decisions(self):
+        rows = pd.DataFrame(
+            [
+                {"entity_type": ENTITY_SELF, "动作": "关停", "账号id": "1", "创意id": "11"},
+                {"entity_type": ENTITY_SELF, "动作": "扩量", "账号id": "1", "创意id": "12"},
+                {"entity_type": ENTITY_SELF, "动作": "关停", "账号id": "1", "创意id": "13"},
+                {"entity_type": ENTITY_GZH, "动作": "关停", "账号id": "", "创意id": ""},
+            ]
         )
 
-    def test_inconsistent_daily_direction_is_observe(self):
-        daily = self.build_daily()
-        mask = daily["广告id"].eq("ad-0")
-        daily.loc[mask & daily["dt"].eq(DATES[1]), "效率收入"] = 1000
-        ages = pd.DataFrame({"广告id": ["ad-0"], "广告age": [10]})
+        class FakeTencent:
+            def __init__(self):
+                self.calls = []
 
-        candidates, _, _ = evaluate_rules(daily, DATES, ages)
-        target = candidates[candidates["广告id"].eq("ad-0")].iloc[0]
-        self.assertEqual(target["动作"], "观察")
-        self.assertIn("表现不一致", target["动作原因"])
+            def get_dynamic_creative(self, account_id, creative_id):
+                self.calls.append((account_id, creative_id))
+                return {
+                    "configured_status": (
+                        ACTIVE_STATUS if creative_id == 11 else SUSPEND_STATUS
+                    )
+                }
 
-    def test_latest_day_uv_over_100_is_reported_as_observe(self):
-        daily = self.build_daily()
-        extra = row(
-            ENTITY_SELF,
-            SELF_CHANNEL,
-            "latest-only",
-            DATES[1],
-            1.5,
-            uv=150,
+        client = FakeTencent()
+        result = _annotate_current_creative_status(rows, client)
+        self.assertEqual(client.calls, [(1, 11), (1, 13)])
+        self.assertEqual(
+            result["当前创意状态"].tolist(),
+            ["正常", "", "已停止", ""],
         )
-        daily = pd.concat([daily, pd.DataFrame([extra])], ignore_index=True)
 
-        candidates, _, summary = evaluate_rules(daily, DATES)
-        target = summary[summary["广告id"].eq("latest-only")].iloc[0]
-        self.assertEqual(target["覆盖天数"], 1)
-        self.assertEqual(target["动作"], "观察")
-        self.assertEqual(target["阈值样本状态"], "观察_最新日UV达标")
-        self.assertAlmostEqual(target["最新日效率ROI"], 1.5)
-        self.assertTrue(candidates["广告id"].eq("latest-only").any())
+    def test_daily_sql_has_direct_ad_grain_and_excludes_qiwei(self):
+        sql = build_daily_sql(DATES[0], DATES[-1])
+        self.assertEqual(sql.count("SUM(NVL(t0_fission_uv_root, 0))"), 3)
+        self.assertIn("'self_ad' AS entity_type", sql)
+        self.assertNotIn("'qiwei' AS entity_type", sql)
+        self.assertIn("COUNT(DISTINCT mid) AS 首层UV", sql)
+        self.assertIn("usersharedepth <= 1", sql)
+        self.assertNotIn("usersharedepth = '0'", sql)
 
-    def test_both_days_must_exceed_daily_uv_gate(self):
-        daily = self.build_daily()
-        mask = daily["广告id"].eq("ad-9")
-        daily.loc[mask & daily["dt"].eq(DATES[0]), "首层UV"] = 180
-        ages = pd.DataFrame({"广告id": ["ad-9"], "广告age": [10]})
+    def test_global_p20_and_creative_only_p80_scale_actions(self):
+        candidates, thresholds, summary = evaluate_rules(
+            self.build_daily(), DATES, self.ages()
+        )
+        self.assertEqual(len(thresholds), 1)
+        self.assertAlmostEqual(float(thresholds.iloc[0]["t_stop"]), 0.7)
+        self.assertAlmostEqual(float(thresholds.iloc[0]["t_up"]), 3.4)
+        self.assertEqual(int(thresholds.iloc[0]["阈值样本数"]), 8)
+        self.assertEqual(int(thresholds.iloc[0]["扩量样本数"]), 4)
+        self.assertEqual(thresholds.iloc[0]["关停线口径"], "合格实体等权P20")
+        self.assertEqual(thresholds.iloc[0]["扩量线口径"], "合格小程序创意实体等权P80")
+        scale_rows = candidates[candidates["动作"].eq("扩量")]
+        self.assertEqual(len(scale_rows), 1)
+        self.assertEqual(scale_rows.iloc[0]["entity_type"], ENTITY_SELF)
+        self.assertEqual(scale_rows.iloc[0]["广告id"], "creative-ad-3")
+        self.assertTrue(scale_rows.iloc[0]["是否位于创意三日ROI前20%"])
+        self.assertTrue((summary["覆盖天数"] == 3).all())
+        pool = summary[summary["entity_type"].isin([ENTITY_SELF, ENTITY_GZH])]
+        self.assertTrue(pool["阈值样本状态"].eq("进入三日统一阈值样本池").all())
 
-        _, _, summary = evaluate_rules(daily, DATES, ages)
-        target = summary[summary["广告id"].eq("ad-9")].iloc[0]
+    def test_creative_p80_scale_requires_ad_age_at_least_three_days(self):
+        ages = self.ages()
+        ages.loc[ages["广告id"].eq("creative-ad-3"), "广告age"] = 2
+        candidates, _, _ = evaluate_rules(self.build_daily(), DATES, ages)
+        target = candidates[candidates["广告id"].eq("creative-ad-3")].iloc[0]
         self.assertEqual(target["动作"], "观察")
-        self.assertEqual(target["阈值样本状态"], "观察_最新日UV达标")
+        self.assertIn("广告age<3天", target["动作原因"])
 
-    def test_stop_threshold_uses_cost_weighted_p30(self):
-        rows = []
-        specifications = [
-            ("weighted-0", 0.1, 10),
-            ("weighted-1", 0.2, 10),
-            ("weighted-2", 1.0, 1000),
-            ("weighted-3", 2.0, 1000),
-        ]
-        for dt in DATES:
-            for entity_id, roi, cost in specifications:
-                rows.append(
-                    row(
-                        ENTITY_SELF,
-                        SELF_CHANNEL,
-                        entity_id,
-                        dt,
-                        roi,
-                        cost=cost,
-                    )
-                )
-        _, thresholds, _ = evaluate_rules(pd.DataFrame(rows), DATES)
-        self_thresholds = thresholds[
-            thresholds["entity_type"].eq(ENTITY_SELF)
-        ]
-        self.assertTrue(self_thresholds["t_stop"].eq(1.0).all())
+    def test_ad_level_reuses_threshold_without_entering_sample_and_can_stop(self):
+        _, thresholds, summary = evaluate_rules(self.build_daily(), DATES, self.ages())
+        ad_rows = summary[summary["entity_type"].eq(ENTITY_SELF_AD)]
         self.assertTrue(
-            self_thresholds["关停线口径"].eq("消耗加权P25").all()
+            ad_rows["阈值样本状态"].eq("广告级三日合格_不进入阈值样本池").all()
         )
+        actions = dict(zip(ad_rows["广告id"], ad_rows["动作"]))
+        self.assertEqual(actions["ad-0"], "关停")
+        self.assertEqual(actions["ad-1"], "")
+        self.assertTrue(ad_rows["t_stop"].eq(float(thresholds.iloc[0]["t_stop"])).all())
+        self.assertTrue(ad_rows["调控参与状态"].str.contains("审批后可暂停广告").all())
 
-    def test_qiwei_is_display_only(self):
-        candidates, _, summary = evaluate_rules(self.build_daily(), DATES)
-        self.assertFalse(candidates["entity_type"].eq(ENTITY_QIWEI).any())
-        qiwei = summary[summary["entity_type"].eq(ENTITY_QIWEI)]
-        self.assertTrue(
-            qiwei["调控参与状态"].eq("仅展示_不进入阈值和调控").all()
-        )
-
-    def test_report_uses_daily_rows_and_only_latest_day_is_actionable(self):
+    def test_latest_day_uv_over_200_is_appended_with_fixed_three_day_average(self):
         daily = self.build_daily()
-        extra = row(
-            ENTITY_SELF,
-            SELF_CHANNEL,
-            "latest-only",
-            DATES[1],
-            0.01,
-            uv=150,
-        )
-        daily = pd.concat([daily, pd.DataFrame([extra])], ignore_index=True)
-        candidates, _, _ = evaluate_rules(daily, DATES)
-        frame = _sheet_frame(candidates, "小程序投流", DATES, 0.25)
-        visible = _visible_columns("小程序投流", DATES, 0.25)
+        latest_only = row(ENTITY_SELF, "latest-only", DATES[-1], 1.5, uv=300)
+        daily = pd.concat([daily, pd.DataFrame([latest_only])], ignore_index=True)
+        candidates, _, summary = evaluate_rules(daily, DATES, self.ages())
+        target = summary[summary["广告id"].eq("latest-only")].iloc[0]
+        self.assertEqual(target["覆盖天数"], 1)
+        self.assertEqual(target["阈值样本状态"], "补充观察_昨日UV>200")
+        self.assertEqual(target["动作"], "观察")
+        self.assertEqual(target["日均首层UV"], 100)
+        self.assertTrue(candidates["广告id"].eq("latest-only").any())
 
-        for column in (
-            "dt",
-            "首层UV",
-            "T0裂变人数",
-            "T0裂变率",
-            "首层效率收入",
-            "T0裂变效率收入",
-            "预测总效率收入",
-            "成本",
-            "效率ROI",
-            T0_FISSION_MULTIPLIER_COLUMN,
-            TOTAL_FISSION_TO_FIRST_UV_COLUMN,
-            "消耗加权分位",
-            "消耗加权P25线",
-            "动作原因",
-            "阈值样本状态",
-        ):
-            self.assertIn(column, visible)
-        self.assertNotIn(FINAL_ROI_COLUMN, visible)
-        self.assertIn(FINAL_ROI_COLUMN, frame.columns)
-        latest_row = frame[
-            frame["广告id"].eq("ad-0") & frame["dt"].eq(DATES[1])
-        ].iloc[0]
-        self.assertEqual(latest_row["T0裂变人数"], 120)
-        self.assertAlmostEqual(latest_row["T0裂变率"], 0.2)
-        self.assertEqual(latest_row["首层效率收入"], 20)
-        self.assertEqual(latest_row["预测总效率收入"], 20)
-        self.assertEqual(set(frame["dt"]), set(DATES))
-        self.assertTrue((frame.groupby("广告id").size() == 2).all())
+    def test_daily_prediction_uses_same_day_t0_fission_revenue_once(self):
+        raw = pd.DataFrame([row(ENTITY_SELF, "formula", DATES[-1], 0)])
+        raw.loc[0, ["成本", "效率收入", "裂变效率收入"]] = [100, 50, 30]
+        result = prepare_daily_metrics(raw, FISSION_PARAMETERS).iloc[0]
+        multiplier = result[DISPLAY_MULTIPLIER_COLUMN]
+        self.assertEqual(result["T0实际裂变收入"], 30)
+        self.assertAlmostEqual(result["预测全链路效率收入"], 50 + 30 * multiplier)
+        self.assertAlmostEqual(result["ROI"], (50 + 30 * multiplier) / 100)
 
-        history = frame[frame["dt"].eq(DATES[0])]
-        latest = frame[frame["dt"].eq(DATES[1])]
-        self.assertTrue(history["动作"].fillna("").eq("").all())
-        self.assertTrue(history["审批选择"].eq("历史日").all())
-        self.assertTrue(history["动作幂等键"].fillna("").eq("").all())
-        self.assertTrue(latest["动作"].ne("").any())
-        self.assertTrue(
-            latest["阈值样本状态"].eq("观察_最新日UV达标").any()
+    def test_summary_and_daily_report_frames(self):
+        daily = self.build_daily()
+        daily = pd.concat(
+            [
+                daily,
+                pd.DataFrame(
+                    [
+                        row(ENTITY_SELF, "latest-only-low", DATES[-1], 0.1, uv=220),
+                        row(ENTITY_SELF, "latest-only-high", DATES[-1], 9.0, uv=280),
+                    ]
+                ),
+            ],
+            ignore_index=True,
         )
+        _, thresholds, summary = evaluate_rules(daily, DATES, self.ages())
+        frame = _summary_frame(summary, ENTITY_SELF)
+        detail = _daily_frame(summary, ENTITY_SELF, DATES)
+        self.assertEqual(frame.iloc[-1]["阈值样本状态"], "补充观察_昨日UV>200")
+        observations = frame[frame["阈值样本状态"].eq("补充观察_昨日UV>200")]
+        self.assertEqual(observations.iloc[0]["广告id"], "latest-only-high")
+        formal_actions = frame[
+            ~frame["阈值样本状态"].eq("补充观察_昨日UV>200")
+        ]["建议动作"].tolist()
+        action_rank = {"关停": 0, "扩量": 1, "": 2, "观察": 3}
         self.assertEqual(
-            frame.iloc[0]["dt"],
-            DATES[1],
+            [action_rank[action] for action in formal_actions],
+            sorted(action_rank[action] for action in formal_actions),
         )
+        self.assertIn("日均总预估效率收入", frame.columns)
+        self.assertIn("日均T0裂变率", frame.columns)
+        self.assertNotIn("三日加权平均T0裂变率", _visible_columns(SUMMARY_SHEETS[ENTITY_SELF]))
+        self.assertIn("当日效率ROI", frame.columns)
+        self.assertIn("预测总效率ROI", frame.columns)
+        visible = _visible_columns(SUMMARY_SHEETS[ENTITY_SELF])
+        self.assertNotIn("整体三日ROI排名百分位", visible)
+        self.assertNotIn("是否位于三日ROI后20%", visible)
+        self.assertNotIn("最新日首层UV", visible)
+        self.assertNotIn("覆盖天数", visible)
+        self.assertNotIn("审批选择", visible)
+        self.assertEqual(visible[-1], "当前创意状态")
+        for removed in ("动作", "动作原因", "阈值样本状态", "执行状态", "执行结果"):
+            self.assertNotIn(removed, visible)
         self.assertEqual(
-            frame.iloc[-1]["dt"],
-            DATES[0],
-        )
-        pd.testing.assert_series_equal(
-            frame[TOTAL_FISSION_TO_FIRST_UV_COLUMN],
-            frame[DISPLAY_TOTAL_TO_FIRST_COLUMN],
-            check_names=False,
+            visible[visible.index("当日效率ROI") : visible.index("建议动作") + 1],
+            [
+                "当日效率ROI",
+                "预测总效率ROI",
+                "关停线（P20）",
+                "扩量线（P80）",
+                "建议动作",
+            ],
         )
+        self.assertIn("关停线（P20）", frame.columns)
+        self.assertTrue(frame["扩量线（P80）"].eq(float(thresholds.iloc[0]["t_up"])).all())
+        gzh_frame = _summary_frame(summary, ENTITY_GZH)
+        self.assertTrue(gzh_frame["扩量线（P80）"].isna().all())
+        self.assertIn("是否位于三日ROI后20%", frame.columns)
+        self.assertNotIn("关停线（P25）", frame.columns)
+        self.assertNotIn("整体实体等权P25关停线", frame.columns)
+        self.assertEqual(set(detail["dt"]), set(DATES))
+        self.assertTrue((detail.groupby("广告id").size() == 3).all())
+        self.assertEqual(detail.columns[0], "dt")
+        self.assertTrue(detail["dt"].astype(str).is_monotonic_decreasing)
+        self.assertIn("当日效率ROI", detail.columns)
+        self.assertIn("预测总效率ROI", detail.columns)
+
+        with tempfile.TemporaryDirectory() as directory:
+            output = Path(directory) / "roi.xlsx"
+            write_workbook(
+                summary,
+                thresholds,
+                DATES,
+                output,
+                {},
+                pd.DataFrame(
+                    [
+                        {
+                            "实体类型": "self",
+                            "匹配层级": "miniapp_package_goal_exact",
+                            "实体数": 1,
+                            "渠道实体数": 1,
+                            "匹配率": 1.0,
+                            "参数版本": "test",
+                        }
+                    ]
+                ),
+            )
+            workbook = load_workbook(output, read_only=False)
+            expected = set(SUMMARY_SHEETS.values()) | set(DAILY_SHEETS.values())
+            self.assertTrue(expected.issubset(workbook.sheetnames))
+            self.assertNotIn("企微群合作", workbook.sheetnames)
+            self.assertIn("传播裂变系数匹配", workbook.sheetnames)
+            self.assertEqual(
+                workbook["传播裂变系数匹配"].sheet_state,
+                "hidden",
+            )
+            for sheet_name in expected:
+                self.assertEqual(workbook[sheet_name].freeze_panes, "H2")
+                self.assertEqual(len(workbook[sheet_name].conditional_formatting), 2)
+            self.assertEqual(
+                workbook[SUMMARY_SHEETS[ENTITY_SELF_AD]].sheet_state, "hidden"
+            )
+            self.assertEqual(
+                workbook[DAILY_SHEETS[ENTITY_SELF_AD]].sheet_state, "hidden"
+            )
+            creative_sheet = workbook[SUMMARY_SHEETS[ENTITY_SELF]]
+            color_rules = [
+                rule
+                for rules in creative_sheet.conditional_formatting._cf_rules.values()
+                for rule in rules
+            ]
+            self.assertEqual(len(color_rules), 2)
+            for rule in color_rules:
+                self.assertEqual(rule.type, "colorScale")
+                colors = [color.rgb[-6:] for color in rule.colorScale.color]
+                self.assertEqual(colors, ["C00000", "FFEB84", "00B050"])
+            creative_headers = {
+                cell.value: cell.column for cell in creative_sheet[1]
+            }
+            first_scale_row = next(
+                row_number
+                for row_number in range(2, creative_sheet.max_row + 1)
+                if creative_sheet.cell(
+                    row_number, creative_headers["建议动作"]
+                ).value
+                == "扩量"
+            )
+            self.assertEqual(
+                creative_sheet.cell(first_scale_row, 1).border.top.style,
+                "medium",
+            )
+            self.assertEqual(
+                creative_sheet.cell(
+                    first_scale_row, creative_headers["建议动作"]
+                ).border.top.style,
+                "medium",
+            )
+            first_no_action_row = next(
+                row_number
+                for row_number in range(first_scale_row + 1, creative_sheet.max_row + 1)
+                if not creative_sheet.cell(
+                    row_number, creative_headers["建议动作"]
+                ).value
+            )
+            self.assertEqual(
+                creative_sheet.cell(first_no_action_row, 1).border.top.style,
+                "medium",
+            )
+            self.assertEqual(
+                creative_sheet.cell(
+                    first_no_action_row, creative_headers["建议动作"]
+                ).border.top.style,
+                "medium",
+            )
+            self.assertEqual(
+                creative_sheet.cell(2, creative_headers["日均首层UV"]).number_format,
+                "0",
+            )
+            self.assertEqual(
+                creative_sheet.cell(
+                    2, creative_headers["日均T0裂变人数"]
+                ).number_format,
+                "0",
+            )
+            self.assertEqual(
+                creative_sheet.cell(2, creative_headers["广告age"]).number_format,
+                "0",
+            )
+            self.assertEqual(
+                creative_sheet.cell(
+                    2, creative_headers["当日效率ROI"]
+                ).number_format,
+                "0.00",
+            )
+            self.assertEqual(
+                creative_sheet.cell(
+                    2, creative_headers["预测总效率ROI"]
+                ).number_format,
+                "0.00",
+            )
+            for header in (
+                "裂变系数-总裂变UV/T0裂变UV",
+                "裂变系数-总裂变UV/首层UV",
+            ):
+                self.assertEqual(
+                    creative_sheet.cell(
+                        2, creative_headers[header]
+                    ).number_format,
+                    "0.00",
+                )
+            self.assertEqual(
+                creative_sheet.cell(
+                    2, creative_headers["扩量线（P80）"]
+                ).number_format,
+                "0.00",
+            )
+            run_summary = workbook["运行摘要"]
+            summary_rows = {
+                run_summary.cell(row_number, 1).value: row_number
+                for row_number in range(1, run_summary.max_row + 1)
+            }
+            self.assertEqual(
+                run_summary.cell(summary_rows["创意扩量线（P80）"], 2).number_format,
+                "0.00",
+            )
+            self.assertEqual(
+                run_summary.cell(summary_rows["扩量样本数"], 2).number_format,
+                "0",
+            )
+            ad_sheet = workbook[SUMMARY_SHEETS[ENTITY_SELF_AD]]
+            ad_headers = [cell.value for cell in ad_sheet[1]]
+            self.assertIn("审批选择", ad_headers)
+            approval_column = ad_headers.index("审批选择") + 1
+            self.assertTrue(
+                ad_sheet.column_dimensions[
+                    ad_sheet.cell(1, approval_column).column_letter
+                ].hidden
+            )
+            self.assertGreater(len(ad_sheet.data_validations.dataValidation), 0)
 
 
 if __name__ == "__main__":

examples/auto_put_ad_mini/test_roi_control_policy.py

@@ -2,15 +2,16 @@ import os
 import unittest
 from datetime import date, datetime
 from decimal import Decimal
-from unittest.mock import patch
+from unittest.mock import Mock, patch
 
 import pandas as pd
 
 from operator_commands import ACTION_CONFIRM, ACTION_REJECT, parse_command
 from roi_control.config import RoiConfig
-from roi_control.execution import plan_scale_bid
-from roi_control.metrics import ENTITY_GZH, ENTITY_SELF
+from roi_control.execution import _execute_pause_ad, plan_scale_bid
+from roi_control.metrics import ENTITY_GZH, ENTITY_SELF, ENTITY_SELF_AD
 from roi_control.policy import (
+    ACTION_PAUSE_AD,
     ACTION_PAUSE_CREATIVE,
     ACTION_SCALE_BID,
     MODE_ACTIONABLE,
@@ -18,7 +19,8 @@ from roi_control.policy import (
     annotate_execution,
 )
 from roi_control.repository import _is_expired
-from roi_control.sheet_approval import parse_approval_rows
+from roi_control.sheet_approval import RoiSheetClient, parse_approval_rows
+from tencent_client import ACTIVE_STATUS, SUSPEND_STATUS
 
 
 class RoiControlPolicyTest(unittest.TestCase):
@@ -65,6 +67,15 @@ class RoiControlPolicyTest(unittest.TestCase):
                     "动作": "扩量",
                     "动作原因": "高ROI",
                 },
+                {
+                    "entity_type": ENTITY_SELF_AD,
+                    "channel": "小程序投流-稳定",
+                    "账号id": "84502354",
+                    "广告id": "5001",
+                    "创意id": "",
+                    "动作": "关停",
+                    "动作原因": "广告级低ROI",
+                },
             ]
         )
         annotated, snapshots, actions = annotate_execution(
@@ -73,10 +84,10 @@ class RoiControlPolicyTest(unittest.TestCase):
             "roi_20260725_north_star_roi_v1_roi_policy_v1",
         )
 
-        self.assertEqual(len(snapshots), 4)
+        self.assertEqual(len(snapshots), 5)
         self.assertEqual(
             {action["action_type"] for action in actions},
-            {ACTION_PAUSE_CREATIVE, ACTION_SCALE_BID},
+            {ACTION_PAUSE_CREATIVE, ACTION_PAUSE_AD, ACTION_SCALE_BID},
         )
         self.assertEqual(
             annotated.iloc[0]["执行模式"],
@@ -90,6 +101,13 @@ class RoiControlPolicyTest(unittest.TestCase):
             annotated.iloc[0]["动作幂等键"], actions[0]["idempotency_key"]
         )
         self.assertEqual(annotated.iloc[2]["审批选择"], "不可执行")
+        self.assertEqual(annotated.iloc[4]["执行模式"], MODE_ACTIONABLE)
+        self.assertEqual(annotated.iloc[4]["审批选择"], "")
+        ad_action = next(
+            action for action in actions if action["action_type"] == ACTION_PAUSE_AD
+        )
+        self.assertIsNone(ad_action["dynamic_creative_id"])
+        self.assertTrue(ad_action["idempotency_key"].endswith(":5001"))
 
     def test_sheet_approval_parser_uses_hidden_idempotency_key(self):
         rows = [
@@ -107,6 +125,70 @@ class RoiControlPolicyTest(unittest.TestCase):
             ],
         )
 
+    def test_sheet_client_reads_both_current_approval_sheets(self):
+        client = RoiSheetClient.__new__(RoiSheetClient)
+        client._token = Mock(return_value="token")
+        client._sheet_ids = Mock(
+            return_value=[
+                ("小程序创意级三日汇总", "creative-sheet"),
+                ("小程序广告级三日汇总", "ad-sheet"),
+            ]
+        )
+        client._read_sheet_approvals = Mock(
+            side_effect=[
+                [{"sheet_id": "creative-sheet", "idempotency_key": "creative"}],
+                [{"sheet_id": "ad-sheet", "idempotency_key": "ad"}],
+            ]
+        )
+        decisions = client.read_approvals("spreadsheet")
+        self.assertEqual(
+            [decision["sheet_id"] for decision in decisions],
+            ["creative-sheet", "ad-sheet"],
+        )
+
+    def test_pause_ad_marks_prepared_then_success_after_readback(self):
+        now = datetime.fromisoformat("2026-08-03T12:00:00+08:00")
+
+        class FakeTencent:
+            def get_ad(self, account_id, adgroup_id):
+                self.get_args = (account_id, adgroup_id)
+                return {
+                    "adgroup_id": adgroup_id,
+                    "configured_status": ACTIVE_STATUS,
+                }
+
+            def update_ad(self, account_id, adgroup_id, **kwargs):
+                self.update_args = (account_id, adgroup_id, kwargs)
+                return {
+                    "adgroup_id": adgroup_id,
+                    "configured_status": SUSPEND_STATUS,
+                }
+
+        client = FakeTencent()
+        item = {
+            "id": 99,
+            "account_id": 84502354,
+            "adgroup_id": 5001,
+            "execution_status": "PENDING",
+        }
+        with patch("roi_control.execution.update_action_item") as update_item:
+            _execute_pause_ad(item, client=client, now=now)
+        self.assertEqual(client.get_args, (84502354, 5001))
+        self.assertEqual(
+            client.update_args,
+            (84502354, 5001, {"target_status": SUSPEND_STATUS}),
+        )
+        self.assertEqual(update_item.call_count, 2)
+        self.assertEqual(
+            update_item.call_args_list[0].kwargs["execution_status"], "PREPARED"
+        )
+        self.assertEqual(
+            update_item.call_args_list[1].kwargs["execution_status"], "SUCCESS"
+        )
+        self.assertEqual(
+            update_item.call_args_list[1].kwargs["readback_status"], SUSPEND_STATUS
+        )
+
     def test_scale_actions_are_deduplicated_by_ad(self):
         rows = [
             {