Merge remote-tracking branch 'origin/agent_auto_ad_put_ad_adjust_0509_creative' into agent_auto_ad_put_ad_adjust_0509_creative

AGENTS.md

@@ -17,13 +17,14 @@
 ## 生产服务架构
 
 - 生产使用同一个完整 `ad-put-agent` 镜像启动两个容器:`ad-control-service` 和 `ad-daily-service`。
-- `ad-control-service` 负责唯一飞书 WebSocket、运营暂停/停止/恢复、ROI 整批审批执行和每 10 分钟实时 CPM 调控。
+- `ad-control-service` 负责唯一飞书 WebSocket、运营暂停/停止/恢复、ROI 表格逐行审批执行和每 10 分钟实时 CPM 调控。
 - 实时调控账户范围是历史 `ad_creation_account_config` 与启用白名单的交集;飞书关闭创建配置只停止新建/补创意,不能让既有广告退出实时管理。
 - `ad-daily-service` 每天 10:30 调用现有广告/创意创建流程,每 2 小时扫描腾讯创意正式审核结果,并可在每天 11:00 计算和发布日级 ROI 批次。
 - 旧 `server.py`、`execute_once.py` 和 `run.py` 的模型调控链路不再作为生产入口,但暂不删除历史代码。
 - 飞书生产控制命令使用确定性解析,不能让模型直接决定账户范围或执行腾讯写操作。
 - `暂停` 表示仅暂停到下一投放日 06:00;`停止` 表示持续停止;`恢复` 只解除运营暂停,不能解除仍生效的 CPM 暂停。
-- 所有飞书触发的腾讯写操作都必须二次确认。群聊命令必须来自配置群并 @机器人,发送人必须在允许列表。
+- 运营暂停、停止、恢复等群聊命令必须二次确认,且必须来自配置群并 @机器人,发送人必须在允许列表。
+- ROI 表格逐行审批是独立入口:黄色【审批选择】列填写“批准”即为最终确认,不再经过群聊二次确认。审批表使用获得链接者可编辑权限,但执行目标必须只按数据库中的隐藏幂等键回读,不能信任表格内可编辑的账户、广告、创意、成本或 ROI 字段。
 - 运营暂停状态和 CPM 暂停状态必须分开持久化。原本人工暂停的广告不能被系统认领或自动开启;腾讯后台人工重新开启时以人工操作为准。
 - 实时控制和飞书写操作必须共用 MySQL advisory lock,并执行腾讯写后回读校验。
 - `RTC_APPLY_ENABLED`、`DAILY_ROI_ENABLED`、`ROI_APPLY_ENABLED` 默认关闭;生产切换前必须先完成 dry-run 和只通知验证。
@@ -32,21 +33,21 @@
 
 - ROI 是指导渠道、账户、广告和创意动作的核心北极星指标,必须作为独立领域模块维护,不能散落在飞书、腾讯 API 或调度代码中。
 - ROI 指标计算必须保持纯数据输入/输出,不能依赖飞书审批、腾讯写操作或具体动作执行器。动作策略只能消费一个明确的 `metric_version`。
-- 当前日级 ROI 指标版本为 `north_star_roi_t15_v3`,报表版本为 `roi_report_v4`,使用已发布参数 `20260712_A0-A15_v1`。T0 裂变收入是实际值,只预测 T1-T15 增量:`T0实际裂变收入 * (传播裂变系数-对T0裂变 - 1)`;预测总收入为 `首层实际效率收入 + T0实际裂变收入 + 预测T1-T15裂变收入`,禁止重复计算 T0。
+- 当前日级 ROI 指标版本为 `north_star_roi_t15_v4`,报表版本为 `roi_report_v5`,使用已发布参数 `20260712_A0-A15_v1`。T0 裂变收入是实际值,只预测 T1-T15 增量:`T0实际裂变收入 * (传播裂变系数-对T0裂变 - 1)`;预测总收入为 `首层实际效率收入 + T0实际裂变收入 + 预测T1-T15裂变收入`,禁止重复计算 T0。
 - 小程序传播裂变系数按 `人群包+转化目标精确 -> 人群包回退 -> 转化目标回退 -> 渠道回退` 匹配;公众号按 `合作方+公众号精确 -> 合作方回退 -> 渠道回退` 匹配。样本不足的精确实体不能使用自身系数。
 - 小程序日级 ODPS 数据必须保留 `广告优化目标`;缺失目标只能进入人群包回退,不得默认成关键页面访问。企微 ROI 口径未复核,当前按渠道/合作方保留快照和报表,临时参考系数为 1.0 并明确标注待补算;企微不进入阈值样本池且不得生成调控动作。
 - ROI 审批表金额和 UV 默认展示三日窗口的单日均值,UV 显示为整数;预测 ROI、实际 ROI、裂变率和阈值必须继续使用三日汇总后的加权口径,不能对每日比例做算术平均。原始三日汇总和每日明细字段保留为隐藏审计列。
-- ROI 审批表应展示所有进入阈值样本池的小程序和公众号实体,以及全部企微参考实体;不能只展示有动作的数据。每个渠道 Sheet 默认按预测 ROI 倒序排列。
+- ROI 审批表应展示所有进入阈值样本池的小程序和公众号实体,以及全部企微参考实体;不能只展示有动作的数据。每个渠道 Sheet 默认按预测 ROI 升序排列。
 - 成熟参数是独立版本化只读资产。审计后的参数必须通过更新脚本发布到 MySQL 的 `roi_fission_parameter_release` 和 `roi_fission_parameter_value`;日级 ROI 只从数据库消费 `ROI_FISSION_PARAMETER_VERSION` 指定且通过内容哈希校验的版本,不能在日常任务中现场重算或静默回退到镜像文件。离线复算只能生成草稿,经恒等式、行数、匹配率和新旧结果对账后才能发布。
 - ROI 报表只上传一次,同一在线表链接发送到 ROI 通知群和 `FEISHU_OPERATOR_CHAT_ID` 投放审批群;群 ID 相同时必须去重。
 - ROI 公式、收入/成本归属、裂变口径或实体粒度发生语义变化时,必须升级指标版本并保存新快照;不得覆盖或重算成旧版本历史结果。
 - 策略阈值和动作语义使用独立 `policy_version`;指标版本与策略版本必须同时写入每个运行批次。
 - 每次运行必须保存完整实体快照、阈值配置、动作建议、可执行性原因和最终执行审计,不能只保存候选或飞书表格。
-- 日级 ROI 当前读取 T-1 至 T-3 的连续三日数据。所有渠道都进入指标和通知,只有当前自动化腾讯账户允许在整批审批后执行。
+- 日级 ROI 当前读取 T-1 至 T-3 的连续三日数据。所有渠道都进入指标和通知,只有当前自动化腾讯账户的可执行行允许在表格逐行批准后执行。
 - 当前低 ROI 动作只暂停 `dynamic_creative_id`,不能暂停整个广告;高 ROI 动作只调整广告永久基础出价,默认上调 10%。
 - 同广告永久基础出价 3 天内最多上调一次,且不超过首次纳管基础价的 2 倍。调整后必须同步实时 CPM 模块的基础出价,避免恢复旧值。
-- ROI 整批审批有效期默认 120 分钟。腾讯写操作必须与实时调控共用数据库 advisory lock,执行前回读映射和状态,执行后再次回读校验。
-- `DAILY_ROI_ENABLED=1`、`ROI_APPLY_ENABLED=0` 只允许计算、快照、报表和审批预览;只有两个开关都为 1 时,审批通过后才允许执行腾讯写操作。
+- ROI 逐行审批有效期默认 120 分钟。腾讯写操作必须与实时调控共用数据库 advisory lock,执行前回读映射和状态,执行后再次回读校验;执行结果必须回写表格并发送飞书通知,通知失败只能重试通知,不能重复腾讯写操作。
+- `DAILY_ROI_ENABLED=1`、`ROI_APPLY_ENABLED=0` 只允许计算、快照、报表和审批预览;只有 `DAILY_ROI_ENABLED=1`、`ROI_APPLY_ENABLED=1`、`ROI_SHEET_APPROVAL_ENABLED=1` 时,表格批准后才允许自动执行腾讯写操作。
 - 禁止配置 `DAILY_ROI_ENABLED=0`、`ROI_APPLY_ENABLED=1`;服务启动时必须拒绝这种不完整配置。
 
 ## 模块 B 创意创建规则

examples/auto_put_ad_mini/.env.example

@@ -28,18 +28,23 @@ DAILY_RUN_ON_STARTUP=0
 DAILY_REVIEW_RUN_ON_STARTUP=0
 DAILY_SYNC_DELIVERY_TEMPLATE=0
 
-# 日级 ROI 北极星指标与整批审批。默认只关闭,分阶段启用。
+# 日级 ROI 北极星指标与表格逐行审批。默认关闭,分阶段启用。
 DAILY_ROI_ENABLED=0
 DAILY_ROI_HOUR=11
 DAILY_ROI_MINUTE=0
 DAILY_ROI_RUN_ON_STARTUP=0
 ROI_APPLY_ENABLED=0
+ROI_SHEET_APPROVAL_ENABLED=0
+ROI_SHEET_APPROVAL_POLL_SECONDS=60
 ROI_APPROVAL_TTL_MINUTES=120
 ROI_SCALE_RATIO=1.10
 ROI_SCALE_COOLDOWN_DAYS=3
 ROI_MAX_BASE_RATIO=2.00
+ROI_STOP_QUANTILE=0.20
+ROI_UP_QUANTILE=0.80
 ROI_FISSION_PARAMETER_VERSION=20260712_A0-A15_v1
-# ROI报表发送到此群；同时抄送 FEISHU_OPERATOR_CHAT_ID，重复ID自动去重
+# ROI报表发送到此群；同时抄送 FEISHU_OPERATOR_CHAT_ID，重复ID自动去重。
+# 审批表链接为获得链接者可编辑，黄色审批列选择“批准”即自动执行。
 # 为空时依次复用 FEISHU_AD_PROJECT_CHAT_ID / RTC_COMMAND_CHAT_ID
 ROI_FEISHU_CHAT_ID=
 

examples/auto_put_ad_mini/docs/unified_services_deployment.md

@@ -67,6 +67,8 @@ DAILY_ROI_HOUR=11
 DAILY_ROI_MINUTE=0
 DAILY_ROI_RUN_ON_STARTUP=0
 ROI_APPLY_ENABLED=0
+ROI_SHEET_APPROVAL_ENABLED=0
+ROI_SHEET_APPROVAL_POLL_SECONDS=60
 ROI_APPROVAL_TTL_MINUTES=120
 ROI_SCALE_RATIO=1.10
 ROI_SCALE_COOLDOWN_DAYS=3
@@ -155,28 +157,28 @@ docker compose --env-file /dev/null run --rm \
 
 日级 ROI 单独分两阶段启用：
 
-当前 `north_star_roi_t15_v4` 使用版本化传播裂变参数，报表版本为 `roi_report_v4`。小程序按人群包和转化目标、
+当前 `north_star_roi_t15_v4` 使用版本化传播裂变参数，报表版本为 `roi_report_v5`。小程序按人群包和转化目标、
 公众号按合作方和公众号匹配传播裂变系数；企微口径未确认，当前按渠道/合作方统一使用临时参考系数 2.5 保留展示，但不进入阈值和调控。T0 收入是实际值，
 只预测 T1-T15 增量。日级任务不会现场重算 cohort 参数。部署前应保持
 `ROI_FISSION_PARAMETER_VERSION=20260712_A0-A15_v1`。参数必须先发布到 MySQL，
 启动日志、数据库发布版本和 Excel 摘要中的参数版本、cohort 日期必须一致。
 ROI 在线表只上传一次，并发送到 `ROI_FEISHU_CHAT_ID` 和
 `FEISHU_OPERATOR_CHAT_ID`；两个变量相同时只发送一次。
-生产环境应让 `RTC_COMMAND_CHAT_ID` 与 `FEISHU_OPERATOR_CHAT_ID` 指向同一个
-投放审批群，确保审批群中的 `确认/拒绝` 命令由唯一 WebSocket 消费者处理。
+ROI 表格使用获得链接者可编辑权限。黄色【审批选择】列只接受“批准”或“拒绝”，
+数据库隐藏幂等键决定真实执行目标，表格中的账户、广告、创意、成本和 ROI 不作为写入参数。
 
 审批表默认展示三日窗口的日均首层 UV、日均首层效率收入、日均 LTV 预测效率收入和
 日均成本；预测 ROI 仍按三日预测收入总和除以三日成本总和计算。三日汇总和每日明细
 保留为隐藏审计列，不参与日均展示。
 小程序和公众号 Sheet 展示所有进入阈值样本池的实体，企微 Sheet 展示全部参考实体；
-各 Sheet 默认按预测 ROI 倒序排列，动作列不参与排序。
+各 Sheet 默认按预测 ROI 升序排列，动作列不参与排序。
 
 1. 首次部署先执行下方参数发布命令并完成回读校验。
-2. 设置 `DAILY_ROI_ENABLED=1`、`ROI_APPLY_ENABLED=0`，观察 11:00 的 ODPS 计算、数据库快照、只读飞书表和整批审批提示。
-3. 使用 `确认 roi_<run_id>` 验证控制服务明确提示写操作关闭。
-4. 核对阈值、账户可执行范围和动作数量后，再设置 `ROI_APPLY_ENABLED=1`。
-5. 生产审批有效期默认 120 分钟；过期批次不能执行。拒绝整批使用 `拒绝 roi_<run_id>`。
-6. 不允许 `DAILY_ROI_ENABLED=0`、`ROI_APPLY_ENABLED=1`，两个服务都会在启动阶段拒绝该配置。
+2. 设置 `DAILY_ROI_ENABLED=1`、`ROI_APPLY_ENABLED=0`、`ROI_SHEET_APPROVAL_ENABLED=0`，观察 11:00 的 ODPS 计算、数据库快照和可编辑飞书表。
+3. 核对阈值、账户可执行范围、黄色审批列和动作数量。
+4. 再设置 `ROI_APPLY_ENABLED=1`、`ROI_SHEET_APPROVAL_ENABLED=1`，重建 `ad-control-service`；批准一条低风险测试行，核对腾讯回读、表格结果和群通知。
+5. 生产审批有效期默认 120 分钟；过期行不能执行。表格逐行选择“拒绝”不会产生腾讯写操作。
+6. 不允许 `DAILY_ROI_ENABLED=0`、`ROI_APPLY_ENABLED=1`，也不允许仅开启表格审批而关闭 ROI 写操作；服务启动阶段会拒绝不完整配置。
 
 首次启用前可手工运行一个“不发飞书、不写腾讯”的计算批次。该命令会读取 ODPS、
 初始化/写入 ROI 数据库快照并在挂载目录生成本地 Excel：
@@ -187,8 +189,8 @@ docker compose run --rm ad_daily_service \
   --end-date 20260725
 ```
 
-确认数据库快照和本地报告后，再增加 `--send-feishu` 验证只读在线表和整批命令。
-此时必须保持 `ROI_APPLY_ENABLED=0`；回复确认只应得到“写操作关闭”的提示。
+确认数据库快照和本地报告后，再增加 `--send-feishu` 验证在线表、黄色审批列和链接编辑权限。
+此时必须保持 `ROI_APPLY_ENABLED=0`、`ROI_SHEET_APPROVAL_ENABLED=0`，不会轮询或执行表格选择。
 
 离线复算新参数必须与日级任务分开执行。以下命令默认只生成 SQL，不访问 ODPS：
 

examples/auto_put_ad_mini/roi_control/config.py

@@ -18,6 +18,8 @@ def env_flag(name: str, default: bool = False) -> bool:
 class RoiConfig:
     daily_enabled: bool = False
     apply_enabled: bool = False
+    sheet_approval_enabled: bool = False
+    sheet_approval_poll_seconds: int = 60
     report_hour: int = 11
     report_minute: int = 0
     approval_ttl_minutes: int = 120
@@ -29,7 +31,7 @@ class RoiConfig:
     self_min_avg_uv: float = 200
     partner_min_avg_uv: float = 200
     min_daily_cost: float = 100
-    stop_quantile: float = 0.10
+    stop_quantile: float = 0.20
     up_quantile: float = 0.80
     gzh_adjust_rank_min: float = 0.10
     gzh_adjust_rank_max: float = 0.30
@@ -39,6 +41,10 @@ class RoiConfig:
         config = cls(
             daily_enabled=env_flag("DAILY_ROI_ENABLED"),
             apply_enabled=env_flag("ROI_APPLY_ENABLED"),
+            sheet_approval_enabled=env_flag("ROI_SHEET_APPROVAL_ENABLED"),
+            sheet_approval_poll_seconds=int(
+                os.getenv("ROI_SHEET_APPROVAL_POLL_SECONDS", "60")
+            ),
             report_hour=int(os.getenv("DAILY_ROI_HOUR", "11")),
             report_minute=int(os.getenv("DAILY_ROI_MINUTE", "0")),
             approval_ttl_minutes=int(
@@ -54,7 +60,7 @@ class RoiConfig:
                 os.getenv("ROI_PARTNER_MIN_AVG_UV", "200")
             ),
             min_daily_cost=float(os.getenv("ROI_MIN_DAILY_COST", "100")),
-            stop_quantile=float(os.getenv("ROI_STOP_QUANTILE", "0.10")),
+            stop_quantile=float(os.getenv("ROI_STOP_QUANTILE", "0.20")),
             up_quantile=float(os.getenv("ROI_UP_QUANTILE", "0.80")),
             gzh_adjust_rank_min=float(
                 os.getenv("ROI_GZH_ADJUST_RANK_MIN", "0.10")
@@ -71,6 +77,12 @@ class RoiConfig:
             raise ValueError(
                 "ROI_APPLY_ENABLED=1 requires DAILY_ROI_ENABLED=1"
             )
+        if self.sheet_approval_enabled and not self.apply_enabled:
+            raise ValueError(
+                "ROI_SHEET_APPROVAL_ENABLED=1 requires ROI_APPLY_ENABLED=1"
+            )
+        if self.sheet_approval_poll_seconds < 30:
+            raise ValueError("ROI_SHEET_APPROVAL_POLL_SECONDS must be at least 30")
         if not 0 <= self.report_hour <= 23 or not 0 <= self.report_minute <= 59:
             raise ValueError("DAILY_ROI_HOUR/MINUTE is invalid")
         if self.approval_ttl_minutes < 1:

examples/auto_put_ad_mini/roi_control/execution.py

@@ -31,6 +31,8 @@ from .repository import (
     FINAL_STATUSES,
     claim_run_for_execution,
     finalize_run,
+    finalize_sheet_run_if_resolved,
+    load_actions_by_ids,
     load_action_items,
     load_run,
     reject_run,
@@ -414,7 +416,7 @@ def execute_roi_batch(
                 continue
             account_id = int(item["account_id"])
             account = accounts.get(account_id)
-            if not account:
+            if account is None:
                 update_action_item(
                     int(item["id"]),
                     execution_status="SKIPPED_NOT_MANAGED",
@@ -453,6 +455,79 @@ def execute_roi_batch(
         return {"run_id": run_id, **finalize_run(run_id, now=now)}
 
 
+def execute_approved_roi_actions(
+    item_ids: list[int],
+    *,
+    now: datetime,
+    lock_name: str,
+    tencent: TencentClient | None = None,
+) -> dict[str, Any]:
+    """Execute only explicitly approved sheet rows without claiming the batch."""
+
+    config = RoiConfig.from_env()
+    if not config.apply_enabled:
+        raise RuntimeError("ROI_APPLY_ENABLED=0, ROI Tencent writes are disabled")
+    with advisory_lock(lock_name) as acquired:
+        if not acquired:
+            raise RuntimeError("实时控制正在执行，ROI表格审批稍后自动重试")
+        accounts = _managed_accounts()
+        client = tencent or TencentClient()
+        run_ids: set[str] = set()
+        processed = 0
+        for item in load_actions_by_ids(item_ids):
+            run_ids.add(str(item["run_id"]))
+            if item.get("approval_status") != "APPROVED":
+                continue
+            if item["execution_status"] not in {"PENDING", "PREPARED"}:
+                continue
+            processed += 1
+            account_id = int(item["account_id"])
+            account = accounts.get(account_id)
+            if account is None:
+                update_action_item(
+                    int(item["id"]),
+                    execution_status="SKIPPED_NOT_MANAGED",
+                    skip_reason="account left automated management scope",
+                    executed_at=now,
+                )
+                continue
+            try:
+                if item["action_type"] == ACTION_PAUSE_CREATIVE:
+                    _execute_pause(item, client=client, now=now)
+                elif item["action_type"] == ACTION_SCALE_BID:
+                    _execute_scale(
+                        item,
+                        account,
+                        client=client,
+                        config=config,
+                        now=now,
+                    )
+                else:
+                    update_action_item(
+                        int(item["id"]),
+                        execution_status="SKIPPED_UNSUPPORTED",
+                        skip_reason=f"unsupported action={item['action_type']}",
+                        executed_at=now,
+                    )
+            except Exception as exc:
+                logger.exception(
+                    "Approved ROI action failed run=%s item=%s",
+                    item["run_id"],
+                    item["id"],
+                )
+                update_action_item(
+                    int(item["id"]),
+                    execution_status=_failure_status(exc),
+                    error_message=str(exc),
+                    executed_at=now,
+                )
+        final = {
+            run_id: finalize_sheet_run_if_resolved(run_id, now=now)
+            for run_id in run_ids
+        }
+        return {"processed": processed, "runs": final}
+
+
 def reject_roi_batch(
     run_id: str,
     *,

examples/auto_put_ad_mini/roi_control/feishu.py

@@ -1,4 +1,4 @@
-"""Publish read-only ROI workbooks and whole-batch approval instructions."""
+"""Publish editable ROI workbooks and row-level approval instructions."""
 
 from __future__ import annotations
 
@@ -31,7 +31,6 @@ class RoiFeishuPublisher:
                 if chat_id
             )
         )
-        self.approval_chat_id = operator_chat_id or primary_chat_id
         missing = [
             name
             for name, value in {
@@ -121,17 +120,17 @@ class RoiFeishuPublisher:
             time.sleep(2)
         raise RuntimeError("ROI sheet import timed out after 60 seconds")
 
-    def _set_read_only_link(self, token: str, sheet_token: str) -> None:
+    def _set_editable_link(self, token: str, sheet_token: str) -> None:
         response = self.client.patch(
             f"{BASE_URL}/drive/v1/permissions/{sheet_token}/public",
             headers={**self._headers(token), "Content-Type": "application/json"},
             params={"type": "sheet"},
             json={
                 "external_access_entity": "open",
-                "link_share_entity": "anyone_readable",
+                "link_share_entity": "anyone_editable",
             },
         )
-        self._json(response, "set ROI sheet read-only permission")
+        self._json(response, "set ROI sheet editable permission")
 
     def _send_card(
         self,
@@ -140,19 +139,17 @@ class RoiFeishuPublisher:
         run_id: str,
         batch_name: str,
         chat_id: str,
-        can_approve: bool,
         url: str,
         summary: str,
         requires_approval: bool,
     ) -> str:
         if not requires_approval:
             instructions = "\n\n本批次没有可执行动作，无需审批。"
-        elif can_approve:
+        else:
             instructions = (
-                f"\n\n请在有效期内回复：`确认 {run_id}` 或 `拒绝 {run_id}`"
+                "\n\n请在有效期内打开表格，在小程序投流表黄色"
+                "【审批选择】列逐行选择批准或拒绝。批准即为最终确认，系统自动执行。"
             )
-        else:
-            instructions = "\n\n本批次需要审批，请前往投放审批群处理。"
         card = {
             "config": {"wide_screen_mode": True},
             "header": {
@@ -196,6 +193,67 @@ class RoiFeishuPublisher:
         )
         return self._json(response, "send ROI batch card")["data"]["message_id"]
 
+    def send_execution_results(self, rows: list[dict[str, Any]]) -> None:
+        if not rows:
+            return
+        token = self._token()
+        for offset in range(0, len(rows), 15):
+            chunk = rows[offset : offset + 15]
+            details = []
+            for row in chunk:
+                cost = float(row.get("cost") or 0)
+                roi = float(row.get("roi") or 0)
+                result = str(row.get("execution_status") or "")
+                error = str(row.get("error_message") or row.get("skip_reason") or "")
+                details.append(
+                    f"- 账户 `{row['account_id']}` / 广告 `{row['adgroup_id']}` / "
+                    f"创意 `{row.get('dynamic_creative_id') or '-'}`\n"
+                    f"  {row.get('adgroup_name') or ''} | 三日成本 {cost:.2f} 元 | "
+                    f"预测ROI {roi:.3f} | **{result}**"
+                    + (f" | {error[:160]}" if error else "")
+                )
+            success_count = sum(
+                row.get("execution_status") == "SUCCESS" for row in chunk
+            )
+            content = (
+                f"本次处理 {len(chunk)} 条，成功 {success_count} 条，"
+                f"其他 {len(chunk) - success_count} 条。\n\n"
+                + "\n".join(details)
+            )
+            card = {
+                "config": {"wide_screen_mode": True},
+                "header": {
+                    "template": "blue" if success_count == len(chunk) else "orange",
+                    "title": {"tag": "plain_text", "content": "日级 ROI 审批执行结果"},
+                },
+                "elements": [
+                    {"tag": "div", "text": {"tag": "lark_md", "content": content}},
+                    {
+                        "tag": "action",
+                        "actions": [
+                            {
+                                "tag": "button",
+                                "type": "primary",
+                                "text": {"tag": "plain_text", "content": "查看审批表"},
+                                "url": str(chunk[0].get("sheet_url") or ""),
+                            }
+                        ],
+                    },
+                ],
+            }
+            for chat_id in self.chat_ids:
+                response = self.client.post(
+                    f"{BASE_URL}/im/v1/messages",
+                    headers={**self._headers(token), "Content-Type": "application/json"},
+                    params={"receive_id_type": "chat_id"},
+                    json={
+                        "receive_id": chat_id,
+                        "msg_type": "interactive",
+                        "content": json.dumps(card, ensure_ascii=False),
+                    },
+                )
+                self._json(response, "send ROI execution result")
+
     def publish(
         self,
         path: Path,
@@ -215,14 +273,13 @@ class RoiFeishuPublisher:
         sheet_token = str(result.get("token") or "")
         if not url or not sheet_token:
             raise RuntimeError("ROI sheet import returned no URL/token")
-        self._set_read_only_link(token, sheet_token)
+        self._set_editable_link(token, sheet_token)
         message_ids = [
             self._send_card(
                 token,
                 run_id=run_id,
                 batch_name=batch_name,
                 chat_id=chat_id,
-                can_approve=chat_id == self.approval_chat_id,
                 url=url,
                 summary=summary,
                 requires_approval=requires_approval,

examples/auto_put_ad_mini/roi_control/policy.py

@@ -56,6 +56,10 @@ def annotate_execution(
     annotated = summary.copy()
     annotated["执行模式"] = MODE_NOTIFY_ONLY
     annotated["执行说明"] = "无调控建议"
+    annotated["审批选择"] = "不可执行"
+    annotated["执行状态"] = ""
+    annotated["执行结果"] = ""
+    annotated["动作幂等键"] = ""
     actions: list[dict[str, Any]] = []
     snapshots: list[dict[str, Any]] = []
     seen_actions: set[str] = set()
@@ -110,6 +114,10 @@ def annotate_execution(
 
         annotated.at[index, "执行模式"] = execution_mode
         annotated.at[index, "执行说明"] = execution_reason
+        if execution_mode == MODE_ACTIONABLE:
+            annotated.at[index, "审批选择"] = ""
+            annotated.at[index, "执行状态"] = "待审批"
+            annotated.at[index, "动作幂等键"] = idempotency_key
 
         daily_metrics = {
             column: _finite(row.get(column))

examples/auto_put_ad_mini/roi_control/reporting.py

@@ -8,6 +8,7 @@ from typing import Dict, 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.utils import get_column_letter
@@ -20,8 +21,10 @@ 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")
-REPORT_VERSION = "roi_report_v4"
-REPORT_RUN_SUFFIX = "r4"
+APPROVAL_FILL = PatternFill("solid", fgColor="FFD966")
+APPROVAL_HEADER_FILL = PatternFill("solid", fgColor="BF9000")
+REPORT_VERSION = "roi_report_v5"
+REPORT_RUN_SUFFIX = "r5"
 
 
 BASE_COLUMNS: Dict[str, Sequence[str]] = {
@@ -44,6 +47,9 @@ BASE_COLUMNS: Dict[str, Sequence[str]] = {
         "预测ROI",
         "动作",
         "关停线",
+        "审批选择",
+        "执行状态",
+        "执行结果",
     ),
     "公众号即转": (
         "渠道",
@@ -165,6 +171,25 @@ def _format_sheet(ws, sheet_name: str) -> None:
         cell.alignment = Alignment(horizontal="center", vertical="center")
         headers[cell.value] = cell.column
 
+    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
+        validation = DataValidation(
+            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)
+            if cell.value != "不可执行":
+                cell.fill = APPROVAL_FILL
+                validation.add(cell)
+
     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)]
@@ -244,7 +269,7 @@ def _write_summary(
             "企微口径",
             "企微按渠道/合作方保留展示，临时参考系数为2.5，待补算正式参数；不进入阈值样本池和调控。",
         ),
-        ("阈值口径", "小程序和公众号合格实体进入同一全局池：三日聚合ROI的P10=关停线、P80=扩量线；企微不参与"),
+        ("阈值口径", "小程序和公众号合格实体进入同一全局池：三日聚合ROI的P20=关停线、P80=扩量线；企微不参与"),
         (
             "阈值样本条件",
             f"连续3天数据完整；小程序日均首层UV>{self_min_uv:g}；"
@@ -252,7 +277,7 @@ def _write_summary(
             f"{min_daily_cost:g}；三日成本>0且预测ROI有效；企微仅展示、不进入样本池",
         ),
         ("动作优先级", "关停 > 扩量 > 调整封面&落地页视频"),
-        ("审批方式", "飞书消息整批确认或拒绝；表格仅用于查看，不做逐行审批"),
+        ("审批方式", "在小程序投流表黄色【审批选择】列逐行选择批准或拒绝；批准即为最终确认并自动执行"),
     ]
     for label, value in rows:
         ws.append([label, value])

examples/auto_put_ad_mini/roi_control/repository.py

@@ -439,12 +439,268 @@ def load_action_items(run_id: str) -> list[dict[str, Any]]:
         connection.close()
 
 
+def load_pending_sheet_runs(now: datetime) -> list[dict[str, Any]]:
+    connection = connect()
+    try:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                """
+                SELECT * FROM roi_metric_run
+                WHERE status='PENDING_APPROVAL'
+                  AND sheet_token IS NOT NULL
+                  AND (expires_at IS NULL OR expires_at>=%s)
+                ORDER BY created_at
+                """,
+                (now,),
+            )
+            return list(cursor.fetchall())
+    finally:
+        connection.close()
+
+
+def expire_pending_sheet_runs(now: datetime) -> int:
+    connection = connect()
+    connection.autocommit(False)
+    try:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                """
+                SELECT run_id FROM roi_metric_run
+                WHERE status='PENDING_APPROVAL'
+                  AND expires_at IS NOT NULL
+                  AND expires_at<%s
+                FOR UPDATE
+                """,
+                (now,),
+            )
+            run_ids = [row["run_id"] for row in cursor.fetchall()]
+            if run_ids:
+                placeholders = ",".join(["%s"] * len(run_ids))
+                cursor.execute(
+                    f"""
+                    UPDATE roi_action_item
+                    SET approval_status='EXPIRED', execution_status='EXPIRED',
+                        executed_at=%s
+                    WHERE run_id IN ({placeholders})
+                      AND approval_status='PENDING'
+                    """,
+                    [now, *run_ids],
+                )
+                cursor.execute(
+                    f"""
+                    UPDATE roi_metric_run SET status='EXPIRED'
+                    WHERE run_id IN ({placeholders})
+                    """,
+                    run_ids,
+                )
+        connection.commit()
+        return len(run_ids)
+    except Exception:
+        connection.rollback()
+        raise
+    finally:
+        connection.close()
+
+
+def record_sheet_decision(
+    *,
+    run_id: str,
+    idempotency_key: str,
+    decision: str,
+    sheet_row_number: int,
+    now: datetime,
+) -> dict[str, Any] | None:
+    if decision not in {"APPROVED", "REJECTED"}:
+        raise ValueError(f"Unsupported ROI sheet decision: {decision}")
+    connection = connect()
+    connection.autocommit(False)
+    try:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                "SELECT status, expires_at FROM roi_metric_run "
+                "WHERE run_id=%s FOR UPDATE",
+                (run_id,),
+            )
+            run = cursor.fetchone()
+            if not run or run["status"] != "PENDING_APPROVAL":
+                connection.rollback()
+                return None
+            if _is_expired(run.get("expires_at"), now):
+                connection.rollback()
+                return None
+            cursor.execute(
+                """
+                SELECT * FROM roi_action_item
+                WHERE run_id=%s AND idempotency_key=%s
+                FOR UPDATE
+                """,
+                (run_id, idempotency_key),
+            )
+            item = cursor.fetchone()
+            if not item:
+                connection.rollback()
+                return None
+            if item["approval_status"] != "PENDING":
+                connection.commit()
+                return {**item, "decision_changed": False}
+            if decision == "APPROVED":
+                cursor.execute(
+                    """
+                    UPDATE roi_action_item
+                    SET approval_status='APPROVED', approval_source='FEISHU_SHEET',
+                        approved_at=%s, sheet_row_number=%s
+                    WHERE id=%s AND approval_status='PENDING'
+                    """,
+                    (now, sheet_row_number, item["id"]),
+                )
+            else:
+                cursor.execute(
+                    """
+                    UPDATE roi_action_item
+                    SET approval_status='REJECTED', approval_source='FEISHU_SHEET',
+                        rejected_at=%s, sheet_row_number=%s,
+                        execution_status='REJECTED', executed_at=%s
+                    WHERE id=%s AND approval_status='PENDING'
+                    """,
+                    (now, sheet_row_number, now, item["id"]),
+                )
+        connection.commit()
+        return {
+            **item,
+            "approval_status": decision,
+            "sheet_row_number": sheet_row_number,
+            "decision_changed": True,
+        }
+    except Exception:
+        connection.rollback()
+        raise
+    finally:
+        connection.close()
+
+
+def load_actions_by_ids(item_ids: Iterable[int]) -> list[dict[str, Any]]:
+    ids = [int(value) for value in item_ids]
+    if not ids:
+        return []
+    connection = connect()
+    try:
+        with connection.cursor() as cursor:
+            placeholders = ",".join(["%s"] * len(ids))
+            cursor.execute(
+                f"SELECT * FROM roi_action_item WHERE id IN ({placeholders}) ORDER BY id",
+                ids,
+            )
+            return list(cursor.fetchall())
+    finally:
+        connection.close()
+
+
+def load_unnotified_action_details(
+    run_ids: Iterable[str] | None = None,
+) -> list[dict[str, Any]]:
+    values = [str(value) for value in run_ids] if run_ids is not None else []
+    run_filter = ""
+    params: list[Any] = []
+    if run_ids is not None:
+        if not values:
+            return []
+        placeholders = ",".join(["%s"] * len(values))
+        run_filter = f"AND a.run_id IN ({placeholders})"
+        params.extend(values)
+    connection = connect()
+    try:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                f"""
+                SELECT a.*, r.sheet_url, s.adgroup_name, s.audience_name,
+                       s.cost, s.roi
+                FROM roi_action_item a
+                JOIN roi_metric_run r ON r.run_id=a.run_id
+                LEFT JOIN roi_entity_snapshot s ON s.id=(
+                    SELECT MIN(s2.id) FROM roi_entity_snapshot s2
+                    WHERE s2.run_id=a.run_id
+                      AND s2.account_id=a.account_id
+                      AND s2.adgroup_id=a.adgroup_id
+                      AND (
+                          a.dynamic_creative_id IS NULL
+                          OR s2.dynamic_creative_id=a.dynamic_creative_id
+                      )
+                )
+                WHERE a.approval_status IN ('APPROVED','REJECTED')
+                  AND a.execution_status NOT IN ('PENDING','PREPARED')
+                  AND a.result_notified_at IS NULL
+                  {run_filter}
+                ORDER BY a.run_id, a.id
+                LIMIT 100
+                """,
+                params,
+            )
+            return list(cursor.fetchall())
+    finally:
+        connection.close()
+
+
+def mark_action_notifications(
+    item_ids: Iterable[int],
+    *,
+    notified_at: datetime | None,
+    error: str | None,
+) -> None:
+    ids = [int(value) for value in item_ids]
+    if not ids:
+        return
+    connection = connect()
+    try:
+        with connection.cursor() as cursor:
+            placeholders = ",".join(["%s"] * len(ids))
+            cursor.execute(
+                f"""
+                UPDATE roi_action_item
+                SET result_notified_at=%s, notification_error=%s
+                WHERE id IN ({placeholders})
+                """,
+                [notified_at, error[:4000] if error else None, *ids],
+            )
+    finally:
+        connection.close()
+
+
+def finalize_sheet_run_if_resolved(run_id: str, *, now: datetime) -> dict[str, Any]:
+    connection = connect()
+    try:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                """
+                SELECT
+                    SUM(approval_status='PENDING') AS pending_approvals,
+                    SUM(execution_status='SUCCESS') AS successes,
+                    SUM(execution_status IN ('FAILED','OUTCOME_UNKNOWN','VERIFY_FAILED')) AS failures,
+                    SUM(execution_status LIKE 'SKIPPED%%') AS skipped,
+                    SUM(approval_status='REJECTED') AS rejected
+                FROM roi_action_item WHERE run_id=%s
+                """,
+                (run_id,),
+            )
+            counts = cursor.fetchone() or {}
+            normalized = {key: int(value or 0) for key, value in counts.items()}
+            if normalized["pending_approvals"]:
+                return {"status": "PENDING_APPROVAL", **normalized}
+            status = "PARTIAL" if normalized["failures"] else "COMPLETED"
+            cursor.execute(
+                "UPDATE roi_metric_run SET status=%s, executed_at=%s WHERE run_id=%s",
+                (status, now, run_id),
+            )
+            return {"status": status, **normalized}
+    finally:
+        connection.close()
+
+
 def update_action_item(item_id: int, **values: Any) -> None:
     allowed = {
         "bid_field", "initial_base_bid_fen", "base_bid_fen", "before_bid_fen",
         "target_bid_fen", "before_status", "target_status", "readback_status",
         "execution_status", "skip_reason", "error_message", "pre_state_json",
-        "readback_json", "executed_at",
+        "readback_json", "executed_at", "notification_error",
     }
     unknown = set(values) - allowed
     if unknown:

examples/auto_put_ad_mini/roi_control/rules.py

@@ -17,8 +17,8 @@ from .metrics import (
 from .fission_multiplier import FissionMultiplierParameters
 
 
-POLICY_VERSION = "roi_policy_v1"
-POLICY_RUN_SUFFIX = "p1"
+POLICY_VERSION = "roi_policy_v2"
+POLICY_RUN_SUFFIX = "p2"
 
 @dataclass(frozen=True)
 class RuleConfig:
@@ -27,7 +27,7 @@ class RuleConfig:
     self_min_avg_uv: float = 200
     partner_min_avg_uv: float = 200
     min_daily_cost: float = 100
-    stop_quantile: float = 0.10
+    stop_quantile: float = 0.20
     up_quantile: float = 0.80
     gzh_adjust_rank_min: float = 0.10
     gzh_adjust_rank_max: float = 0.30

examples/auto_put_ad_mini/roi_control/sheet_approval.py

@@ -0,0 +1,354 @@
+"""Poll editable ROI sheets and execute approved database action items."""
+
+from __future__ import annotations
+
+import logging
+import os
+import threading
+from datetime import datetime
+from typing import Any
+from zoneinfo import ZoneInfo
+
+import httpx
+from openpyxl.utils import get_column_letter
+
+from .config import RoiConfig
+from .execution import execute_approved_roi_actions
+from .feishu import BASE_URL, RoiFeishuPublisher
+from .repository import (
+    expire_pending_sheet_runs,
+    finalize_sheet_run_if_resolved,
+    load_actions_by_ids,
+    load_pending_sheet_runs,
+    load_unnotified_action_details,
+    mark_action_notifications,
+    record_sheet_decision,
+)
+
+
+SHANGHAI = ZoneInfo("Asia/Shanghai")
+APPROVAL_SHEET_NAME = "小程序投流"
+APPROVED_VALUES = {"批准", "approve", "approved"}
+REJECTED_VALUES = {"拒绝", "reject", "rejected"}
+logger = logging.getLogger("auto_put_ad_mini.roi_sheet_approval")
+
+
+def parse_approval_rows(values: list[list[Any]]) -> list[dict[str, Any]]:
+    if not values:
+        return []
+    headers = {
+        str(value or "").strip(): index
+        for index, value in enumerate(values[0])
+        if str(value or "").strip()
+    }
+    required = {"审批选择", "动作幂等键"}
+    missing = required - headers.keys()
+    if missing:
+        raise RuntimeError(f"ROI审批表缺少列: {', '.join(sorted(missing))}")
+    decisions: list[dict[str, Any]] = []
+    for row_number, row in enumerate(values[1:], start=2):
+        approval_index = headers["审批选择"]
+        key_index = headers["动作幂等键"]
+        approval = str(row[approval_index] if approval_index < len(row) else "").strip()
+        key = str(row[key_index] if key_index < len(row) else "").strip()
+        normalized = approval.lower()
+        if not key:
+            continue
+        if normalized in APPROVED_VALUES:
+            decision = "APPROVED"
+        elif normalized in REJECTED_VALUES:
+            decision = "REJECTED"
+        else:
+            continue
+        decisions.append(
+            {
+                "row_number": row_number,
+                "idempotency_key": key,
+                "decision": decision,
+                "headers": headers,
+            }
+        )
+    return decisions
+
+
+class RoiSheetClient:
+    def __init__(self, timeout: float = 30.0) -> None:
+        self.app_id = os.getenv("FEISHU_APP_ID", "").strip()
+        self.app_secret = os.getenv("FEISHU_APP_SECRET", "").strip()
+        if not self.app_id or not self.app_secret:
+            raise RuntimeError("FEISHU_APP_ID and FEISHU_APP_SECRET are required")
+        self.client = httpx.Client(timeout=timeout)
+
+    def close(self) -> None:
+        self.client.close()
+
+    @staticmethod
+    def _json(response: httpx.Response, action: str) -> dict[str, Any]:
+        response.raise_for_status()
+        payload = response.json()
+        if payload.get("code") != 0:
+            raise RuntimeError(f"{action} failed: {payload.get('msg', payload)}")
+        return payload
+
+    def _token(self) -> str:
+        response = self.client.post(
+            f"{BASE_URL}/auth/v3/tenant_access_token/internal",
+            json={"app_id": self.app_id, "app_secret": self.app_secret},
+        )
+        return self._json(response, "get tenant token")["tenant_access_token"]
+
+    @staticmethod
+    def _headers(token: str) -> dict[str, str]:
+        return {"Authorization": f"Bearer {token}"}
+
+    def _sheet_id(self, token: str, sheet_token: str) -> str:
+        response = self.client.get(
+            f"{BASE_URL}/sheets/v3/spreadsheets/{sheet_token}/sheets/query",
+            headers=self._headers(token),
+        )
+        sheets = (
+            self._json(response, "query ROI sheets")
+            .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"])
+
+    def _read_values(
+        self,
+        token: str,
+        sheet_token: str,
+        cell_range: str,
+    ) -> list[list[Any]]:
+        response = self.client.get(
+            f"{BASE_URL}/sheets/v2/spreadsheets/{sheet_token}/values/{cell_range}",
+            headers=self._headers(token),
+            params={"valueRenderOption": "ToString"},
+        )
+        return (
+            self._json(response, "read ROI approvals")
+            .get("data", {})
+            .get("valueRange", {})
+            .get("values", [])
+        ) or []
+
+    def read_approvals(self, sheet_token: str) -> tuple[str, list[dict[str, Any]]]:
+        token = self._token()
+        sheet_id = self._sheet_id(token, sheet_token)
+        header_rows = self._read_values(
+            token,
+            sheet_token,
+            f"{sheet_id}!A1:ZZ1",
+        )
+        if not header_rows:
+            return sheet_id, []
+        headers = {
+            str(value or "").strip(): index
+            for index, value in enumerate(header_rows[0])
+            if str(value or "").strip()
+        }
+        required = {"审批选择", "动作幂等键"}
+        missing = required - headers.keys()
+        if missing:
+            raise RuntimeError(f"ROI审批表缺少列: {', '.join(sorted(missing))}")
+        approval_letter = get_column_letter(headers["审批选择"] + 1)
+        key_letter = get_column_letter(headers["动作幂等键"] + 1)
+        approval_rows = self._read_values(
+            token,
+            sheet_token,
+            f"{sheet_id}!{approval_letter}2:{approval_letter}5000",
+        )
+        key_rows = self._read_values(
+            token,
+            sheet_token,
+            f"{sheet_id}!{key_letter}2:{key_letter}5000",
+        )
+        decisions: list[dict[str, Any]] = []
+        for offset in range(max(len(approval_rows), len(key_rows))):
+            approval_row = approval_rows[offset] if offset < len(approval_rows) else []
+            key_row = key_rows[offset] if offset < len(key_rows) else []
+            approval = str(approval_row[0] if approval_row else "").strip().lower()
+            key = str(key_row[0] if key_row else "").strip()
+            if not key:
+                continue
+            if approval in APPROVED_VALUES:
+                decision = "APPROVED"
+            elif approval in REJECTED_VALUES:
+                decision = "REJECTED"
+            else:
+                continue
+            decisions.append(
+                {
+                    "row_number": offset + 2,
+                    "idempotency_key": key,
+                    "decision": decision,
+                    "headers": headers,
+                }
+            )
+        return sheet_id, decisions
+
+    def write_results(
+        self,
+        sheet_token: str,
+        sheet_id: str,
+        decisions: list[dict[str, Any]],
+        actions: list[dict[str, Any]],
+    ) -> None:
+        by_id = {int(item["id"]): item for item in actions}
+        token = self._token()
+        value_ranges = []
+        for decision in decisions:
+            item = by_id.get(int(decision["item_id"]))
+            if not item:
+                continue
+            headers = decision["headers"]
+            status_column = headers.get("执行状态")
+            result_column = headers.get("执行结果")
+            if status_column is None or result_column is None:
+                continue
+            execution_status = str(item.get("execution_status") or "")
+            status = {
+                "SUCCESS": "执行成功",
+                "REJECTED": "已拒绝",
+                "PENDING": "待执行",
+                "PREPARED": "执行中",
+            }.get(execution_status, "执行失败" if execution_status else "")
+            result = str(item.get("error_message") or item.get("skip_reason") or "")
+            start = get_column_letter(status_column + 1)
+            end = get_column_letter(result_column + 1)
+            value_ranges.append(
+                {
+                    "range": f"{sheet_id}!{start}{decision['row_number']}:{end}{decision['row_number']}",
+                    "values": [[status, result]],
+                }
+            )
+        if not value_ranges:
+            return
+        response = self.client.post(
+            f"{BASE_URL}/sheets/v2/spreadsheets/"
+            f"{sheet_token}/values_batch_update",
+            headers={**self._headers(token), "Content-Type": "application/json"},
+            json={"valueRanges": value_ranges},
+        )
+        self._json(response, "write ROI execution results")
+
+
+class RoiSheetApprovalService:
+    def __init__(self) -> None:
+        self.config = RoiConfig.from_env()
+        self.lock_name = os.getenv("RTC_DB_LOCK_NAME", "tencent_realtime_control")
+        self.client = RoiSheetClient()
+        self.stop_event = threading.Event()
+
+    def close(self) -> None:
+        self.stop_event.set()
+        self.client.close()
+
+    def process_once(self, now: datetime | None = None) -> dict[str, int]:
+        current = now or datetime.now(SHANGHAI)
+        expired = expire_pending_sheet_runs(current)
+        run_count = 0
+        decision_count = 0
+        failed_runs = 0
+        for run in load_pending_sheet_runs(current):
+            run_count += 1
+            try:
+                decision_count += self._process_run(run, current)
+            except Exception:
+                failed_runs += 1
+                logger.exception(
+                    "Skip incompatible or unavailable ROI sheet run=%s",
+                    run["run_id"],
+                )
+        self._notify_results(current)
+        return {
+            "runs": run_count,
+            "decisions": decision_count,
+            "expired": expired,
+            "failed_runs": failed_runs,
+        }
+
+    def _process_run(self, run: dict[str, Any], now: datetime) -> int:
+        sheet_id, sheet_decisions = self.client.read_approvals(run["sheet_token"])
+        approved_ids: list[int] = []
+        processed_decisions: list[dict[str, Any]] = []
+        changed_count = 0
+        for decision in sheet_decisions:
+            item = record_sheet_decision(
+                run_id=run["run_id"],
+                idempotency_key=decision["idempotency_key"],
+                decision=decision["decision"],
+                sheet_row_number=decision["row_number"],
+                now=now,
+            )
+            if not item:
+                continue
+            changed_count += int(bool(item.get("decision_changed")))
+            decision["item_id"] = int(item["id"])
+            processed_decisions.append(decision)
+            if item.get("approval_status") == "APPROVED":
+                approved_ids.append(int(item["id"]))
+        if approved_ids:
+            execute_approved_roi_actions(
+                approved_ids,
+                now=now,
+                lock_name=self.lock_name,
+            )
+        finalize_sheet_run_if_resolved(run["run_id"], now=now)
+        if processed_decisions:
+            actions = load_actions_by_ids(
+                [decision["item_id"] for decision in processed_decisions]
+            )
+            try:
+                self.client.write_results(
+                    run["sheet_token"],
+                    sheet_id,
+                    processed_decisions,
+                    actions,
+                )
+            except Exception:
+                logger.exception("Failed to write ROI execution result back to sheet")
+        return changed_count
+
+    def _notify_results(self, now: datetime) -> None:
+        rows = load_unnotified_action_details()
+        if not rows:
+            return
+        item_ids = [int(row["id"]) for row in rows]
+        publisher = RoiFeishuPublisher()
+        try:
+            publisher.send_execution_results(rows)
+        except Exception as exc:
+            mark_action_notifications(item_ids, notified_at=None, error=str(exc))
+            raise
+        finally:
+            publisher.close()
+        mark_action_notifications(item_ids, notified_at=now, error=None)
+
+    def run_forever(self) -> None:
+        logger.info(
+            "ROI sheet approval poller started interval=%ss",
+            self.config.sheet_approval_poll_seconds,
+        )
+        while not self.stop_event.is_set():
+            try:
+                result = self.process_once()
+                if result["runs"] or result["expired"]:
+                    logger.info("ROI sheet approval cycle result=%s", result)
+            except Exception:
+                logger.exception("ROI sheet approval cycle failed")
+            self.stop_event.wait(self.config.sheet_approval_poll_seconds)
+
+    def start(self) -> threading.Thread:
+        thread = threading.Thread(
+            target=self.run_forever,
+            name="roi-sheet-approval",
+            daemon=True,
+        )
+        thread.start()
+        return thread

examples/auto_put_ad_mini/run_daily_service.py

@@ -154,7 +154,7 @@ def main() -> None:
                 timezone="Asia/Shanghai",
             ),
             id="daily_roi",
-            name="日级ROI计算与整批审批",
+            name="日级ROI计算与逐行审批",
             max_instances=1,
             coalesce=True,
             misfire_grace_time=int(

examples/auto_put_ad_mini/test_roi_control_metrics.py

@@ -67,7 +67,7 @@ def row(entity_type, channel, entity_id, dt, roi, uv=600, age=10, cost=200.0):
 class RoiRulesTest(unittest.TestCase):
     def build_daily(self):
         rows = []
-        # 每日形成稳定分布，低值实体连续低于后10%，高值实体连续高于前20%。
+        # 每日形成稳定分布，低值实体连续低于后20%，高值实体连续高于前20%。
         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))
@@ -197,7 +197,7 @@ class RoiRulesTest(unittest.TestCase):
         _, thresholds, summary = evaluate_rules(pd.DataFrame(rows), DATES)
         threshold = thresholds.iloc[0]
         expected = pd.Series([1.0, 2.0])
-        self.assertAlmostEqual(threshold["t_stop"], expected.quantile(0.10))
+        self.assertAlmostEqual(threshold["t_stop"], expected.quantile(0.20))
         self.assertAlmostEqual(threshold["t_up"], expected.quantile(0.80))
         self.assertEqual(threshold["阈值样本数"], 2)
         self.assertEqual(threshold["小程序样本数"], 1)

examples/auto_put_ad_mini/test_roi_control_policy.py

@@ -18,6 +18,7 @@ from roi_control.policy import (
     annotate_execution,
 )
 from roi_control.repository import _is_expired
+from roi_control.sheet_approval import parse_approval_rows
 
 
 class RoiControlPolicyTest(unittest.TestCase):
@@ -83,6 +84,28 @@ class RoiControlPolicyTest(unittest.TestCase):
         )
         self.assertEqual(annotated.iloc[2]["执行模式"], MODE_NOTIFY_ONLY)
         self.assertEqual(annotated.iloc[3]["执行模式"], MODE_NOTIFY_ONLY)
+        self.assertEqual(annotated.iloc[0]["审批选择"], "")
+        self.assertEqual(annotated.iloc[0]["执行状态"], "待审批")
+        self.assertEqual(
+            annotated.iloc[0]["动作幂等键"], actions[0]["idempotency_key"]
+        )
+        self.assertEqual(annotated.iloc[2]["审批选择"], "不可执行")
+
+    def test_sheet_approval_parser_uses_hidden_idempotency_key(self):
+        rows = [
+            ["账号id", "审批选择", "动作幂等键", "执行状态", "执行结果"],
+            ["99999999", "批准", "roi_x:PAUSE_CREATIVE:1:2", "待审批", ""],
+            ["1", "拒绝", "roi_x:PAUSE_CREATIVE:1:3", "待审批", ""],
+            ["1", "随便填", "roi_x:PAUSE_CREATIVE:1:4", "待审批", ""],
+        ]
+        decisions = parse_approval_rows(rows)
+        self.assertEqual(
+            [(row["decision"], row["idempotency_key"]) for row in decisions],
+            [
+                ("APPROVED", "roi_x:PAUSE_CREATIVE:1:2"),
+                ("REJECTED", "roi_x:PAUSE_CREATIVE:1:3"),
+            ],
+        )
 
     def test_scale_actions_are_deduplicated_by_ad(self):
         rows = [
@@ -114,6 +137,16 @@ class RoiControlPolicyTest(unittest.TestCase):
             with self.assertRaisesRegex(ValueError, "requires DAILY_ROI_ENABLED"):
                 RoiConfig.from_env()
 
+    def test_sheet_approval_requires_roi_writes(self):
+        environment = {
+            "DAILY_ROI_ENABLED": "1",
+            "ROI_APPLY_ENABLED": "0",
+            "ROI_SHEET_APPROVAL_ENABLED": "1",
+        }
+        with patch.dict(os.environ, environment, clear=False):
+            with self.assertRaisesRegex(ValueError, "requires ROI_APPLY_ENABLED"):
+                RoiConfig.from_env()
+
     def test_roi_batch_commands_are_deterministic(self):
         confirm = parse_command("确认 roi_20260725_north_star_roi_v1")
         reject = parse_command("拒绝 roi_20260725_north_star_roi_v1")

examples/tencent_realtime_control/.env.example

@@ -24,6 +24,8 @@ DAILY_ROI_HOUR=11
 DAILY_ROI_MINUTE=0
 DAILY_ROI_RUN_ON_STARTUP=0
 ROI_APPLY_ENABLED=0
+ROI_SHEET_APPROVAL_ENABLED=0
+ROI_SHEET_APPROVAL_POLL_SECONDS=60
 ROI_FISSION_PARAMETER_VERSION=20260712_A0-A15_v1
 ROI_APPROVAL_TTL_MINUTES=120
 ROI_SCALE_RATIO=1.10

examples/tencent_realtime_control/README.md

@@ -155,22 +155,23 @@ dry-run 切换步骤见
 `examples/auto_put_ad_mini/docs/unified_services_deployment.md`。
 生产 Compose 文件兼容当前 ECS 的 Docker Compose `2.27.0`。
 
-## 日级 ROI 整批审批
+## 日级 ROI 逐行审批
 
-`ad-daily-service` 可在每天 11:00 计算版本化的三日 ROI 指标并发布只读飞书表；
-`ad-control-service` 复用本服务的唯一飞书 WebSocket 处理整批命令：
+`ad-daily-service` 可在每天 11:00 计算版本化的三日 ROI 指标并发布可编辑飞书表；
+`ad-control-service` 每 60 秒读取黄色审批列并处理逐行动作：
 
 ```text
-确认 roi_<run_id>
-拒绝 roi_<run_id>
+打开群内 ROI 在线表格，在“小程序投流”工作表黄色“审批选择”列逐行选择“批准”或“拒绝”。
+“批准”即为最终确认，`ad-control-service` 轮询后自动执行，不再需要群聊二次确认。
 ```
 
-`DAILY_ROI_ENABLED=1` 只开启计算和通知；`ROI_APPLY_ENABLED=1` 才允许审批后
+`DAILY_ROI_ENABLED=1` 只开启计算和通知；同时设置 `ROI_APPLY_ENABLED=1` 和
+`ROI_SHEET_APPROVAL_ENABLED=1` 才允许表格批准后
 暂停低 ROI 动态创意或提高高 ROI 广告的永久基础出价。ROI 腾讯写操作与实时
 CPM 调控共用 `RTC_DB_LOCK_NAME`，并在写入后回读验证。详细开关和分阶段上线
 步骤见统一部署文档。
 
-当前指标版本 `north_star_roi_t15_v3`、报表版本 `roi_report_v4` 消费 `ROI_FISSION_PARAMETER_VERSION` 指定的
+当前指标版本 `north_star_roi_t15_v4`、报表版本 `roi_report_v5` 消费 `ROI_FISSION_PARAMETER_VERSION` 指定的
 已发布传播裂变参数，默认 `20260712_A0-A15_v1`。T0 收入使用实际值，仅预测
 T1-T15 裂变增量。参数只影响日级 ROI 计算和建议，
 不参与实时 CPM 决策。正式参数由更新脚本发布到 MySQL，日级任务只读取数据库且

examples/tencent_realtime_control/feishu_command_service.py

@@ -32,8 +32,6 @@ from operator_control import (
     preview_write_command,
 )
 from realtime_config import RealtimeControlConfig
-from roi_control.config import RoiConfig
-from roi_control.execution import execute_roi_batch, reject_roi_batch
 
 
 SHANGHAI = ZoneInfo("Asia/Shanghai")
@@ -75,7 +73,6 @@ class FeishuCommandService:
             )
         self.apply = apply
         self.config = RealtimeControlConfig.from_env()
-        self.roi_config = RoiConfig.from_env()
         self.confirmation_ttl_minutes = int(
             os.getenv("RTC_COMMAND_CONFIRM_TTL_MINUTES", "10")
         )
@@ -137,7 +134,7 @@ class FeishuCommandService:
                 return
             if parsed.action == ACTION_CANCEL:
                 if (parsed.command_id or "").startswith("roi_"):
-                    raise ValueError("ROI批次请使用“拒绝 roi_<run_id>”")
+                    raise ValueError("ROI批次请在审批表黄色列逐行处理")
                 command = cancel_command(
                     parsed.command_id or "",
                     event.sender_open_id,
@@ -157,38 +154,12 @@ class FeishuCommandService:
                 )
                 return
             if parsed.action == ACTION_REJECT:
-                run = reject_roi_batch(
-                    parsed.command_id or "",
-                    sender_open_id=event.sender_open_id,
-                    now=now,
-                )
-                self._reply(
-                    event,
-                    f"ROI批次 {run['run_id']} 已处理\n- 状态：{run['status']}",
-                )
-                return
+                if (parsed.command_id or "").startswith("roi_"):
+                    self._reply(event, "ROI批次请在审批表黄色列逐行批准或拒绝。")
+                    return
             if parsed.action == ACTION_CONFIRM:
                 if (parsed.command_id or "").startswith("roi_"):
-                    if not self.roi_config.apply_enabled:
-                        self._reply(
-                            event,
-                            "ROI_APPLY_ENABLED=0，当前只生成报告，不执行腾讯写操作。",
-                        )
-                        return
-                    result = execute_roi_batch(
-                        parsed.command_id or "",
-                        sender_open_id=event.sender_open_id,
-                        now=now,
-                        lock_name=self.config.lock_name,
-                    )
-                    self._reply(
-                        event,
-                        f"ROI批次 {result['run_id']} 执行完成\n"
-                        f"- 状态：{result['status']}\n"
-                        f"- 成功：{result.get('successes', 0)} 条\n"
-                        f"- 跳过：{result.get('skipped', 0)} 条\n"
-                        f"- 失败：{result.get('failures', 0)} 条",
-                    )
+                    self._reply(event, "ROI批次不再整批确认，请在审批表黄色列逐行批准。")
                     return
                 if not self.apply:
                     self._reply(event, "当前服务为 dry-run，禁止执行腾讯写操作。")

examples/tencent_realtime_control/run_control_service.py

@@ -46,7 +46,7 @@ def main() -> None:
     load_environment()
     args = parse_args()
     apply = args.apply or _env_flag("RTC_APPLY_ENABLED")
-    RoiConfig.from_env()
+    roi_config = RoiConfig.from_env()
     initialize_schema()
     if os.getenv("RTC_COMMAND_ENABLED", "1").strip().lower() in {
         "1",
@@ -59,6 +59,13 @@ def main() -> None:
         logger.info("Feishu command WebSocket started")
     else:
         logger.warning("Feishu command WebSocket is disabled")
+    if roi_config.sheet_approval_enabled:
+        from roi_control.sheet_approval import RoiSheetApprovalService
+
+        RoiSheetApprovalService().start()
+        logger.info("ROI sheet approval polling started")
+    else:
+        logger.warning("ROI sheet approval polling is disabled")
     run_forever(apply=apply)
 
 

examples/tencent_realtime_control/schema.sql

@@ -69,7 +69,7 @@ CREATE TABLE IF NOT EXISTS roi_metric_run (
     UNIQUE KEY uk_roi_run_key (run_key),
     KEY idx_roi_run_status_expiry (status, expires_at),
     KEY idx_roi_run_end_date (end_date)
-) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='版本化ROI计算与整批审批运行';
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='版本化ROI计算与逐行审批运行';
 
 CREATE TABLE IF NOT EXISTS roi_fission_parameter_release (
     version VARCHAR(64) NOT NULL PRIMARY KEY,
@@ -162,16 +162,24 @@ CREATE TABLE IF NOT EXISTS roi_action_item (
     before_status VARCHAR(50) DEFAULT NULL,
     target_status VARCHAR(50) DEFAULT NULL,
     readback_status VARCHAR(50) DEFAULT NULL,
+    approval_status VARCHAR(32) NOT NULL DEFAULT 'PENDING',
+    approval_source VARCHAR(32) DEFAULT NULL,
+    approved_at DATETIME DEFAULT NULL,
+    rejected_at DATETIME DEFAULT NULL,
+    sheet_row_number INT DEFAULT NULL,
     execution_status VARCHAR(32) NOT NULL DEFAULT 'PENDING',
     skip_reason TEXT DEFAULT NULL,
     error_message TEXT DEFAULT NULL,
     pre_state_json LONGTEXT DEFAULT NULL,
     readback_json LONGTEXT DEFAULT NULL,
     executed_at DATETIME DEFAULT NULL,
+    result_notified_at DATETIME DEFAULT NULL,
+    notification_error TEXT DEFAULT NULL,
     created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
     updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
     UNIQUE KEY uk_roi_action_idempotency (idempotency_key),
     KEY idx_roi_action_run_status (run_id, execution_status),
+    KEY idx_roi_action_approval (run_id, approval_status),
     KEY idx_roi_action_ad (account_id, adgroup_id),
     KEY idx_roi_action_creative (account_id, dynamic_creative_id)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='ROI审批后腾讯动作与审计';

examples/tencent_realtime_control/storage.py

@@ -115,6 +115,15 @@ def initialize_schema() -> None:
                     "predicted_tail_revenue": "DECIMAL(20,4) DEFAULT NULL",
                     "predicted_fission_revenue": "DECIMAL(20,4) DEFAULT NULL",
                 },
+                "roi_action_item": {
+                    "approval_status": "VARCHAR(32) NOT NULL DEFAULT 'PENDING'",
+                    "approval_source": "VARCHAR(32) DEFAULT NULL",
+                    "approved_at": "DATETIME DEFAULT NULL",
+                    "rejected_at": "DATETIME DEFAULT NULL",
+                    "sheet_row_number": "INT DEFAULT NULL",
+                    "result_notified_at": "DATETIME DEFAULT NULL",
+                    "notification_error": "TEXT DEFAULT NULL",
+                },
             }
             for table_name, columns in roi_migrations.items():
                 cursor.execute(
@@ -134,6 +143,23 @@ def initialize_schema() -> None:
                             f"ALTER TABLE {table_name} "
                             f"ADD COLUMN {column} {definition}"
                         )
+            cursor.execute(
+                """
+                SELECT INDEX_NAME
+                FROM information_schema.STATISTICS
+                WHERE TABLE_SCHEMA=%s
+                  AND TABLE_NAME='roi_action_item'
+                  AND INDEX_NAME='idx_roi_action_approval'
+                """,
+                (os.environ["DB_NAME"],),
+            )
+            if not cursor.fetchone():
+                cursor.execute(
+                    """
+                    CREATE INDEX idx_roi_action_approval
+                    ON roi_action_item (run_id, approval_status)
+                    """
+                )
     finally:
         connection.close()