feat: poll daily ROI until source data is ready

AGENTS.md

@@ -21,7 +21,7 @@
 - 实时调控账户范围是历史 `ad_creation_account_config` 与启用白名单的交集;飞书关闭创建配置只停止新建/补创意,不能让既有广告退出实时管理。
 - 自动化账户始终使用完整实时调控模式;同一账户同时出现在额外纳管范围时,不能被 `PAUSE_ONLY` 降级。
 - `realtime_control_account_scope` 可独立纳管非自动创建账户;其中 `PAUSE_ONLY` 账户只执行 CPM 关停,不能执行 BOOST 或 RESTORE 调价。
-- `ad-daily-service` 每天 10:30 调用现有广告/创意创建流程,每 2 小时扫描腾讯创意正式审核结果,并可在每天 11:00 计算和发布日级 ROI 批次。
+- `ad-daily-service` 每天 10:30 调用现有广告/创意创建流程,每 2 小时扫描腾讯创意正式审核结果;生产日级 ROI 从每天 09:00 开始检查 T-1 来源,未就绪时每 30 分钟轮询,18:00 做最后一次检查后截止。当天已有正式批次时不得重复计算或补发。
 - 旧 `server.py`、`execute_once.py` 和 `run.py` 的模型调控链路不再作为生产入口,但暂不删除历史代码。
 - 飞书生产控制命令优先使用确定性解析;模型只能分类为受限的动作和范围类型,最终账户集合必须由代码从原文和数据库确定,模型不能执行腾讯写操作。
 - 飞书自然语言控制只能接入 `ad-control-service` 的唯一 WebSocket,不能新增第二个监听服务。群聊消息只有来自配置群、发送人在允许列表且明确 @机器人时才进入解析;私聊和未 @ 消息必须在调用模型、数据库和腾讯接口前返回。

examples/auto_put_ad_mini/.env.example

@@ -30,8 +30,11 @@ DAILY_SYNC_DELIVERY_TEMPLATE=0
 
 # 日级 ROI 北极星指标与表格逐行审批。默认关闭,分阶段启用。
 DAILY_ROI_ENABLED=0
-DAILY_ROI_HOUR=11
+DAILY_ROI_HOUR=9
 DAILY_ROI_MINUTE=0
+DAILY_ROI_POLL_INTERVAL_MINUTES=30
+DAILY_ROI_POLL_CUTOFF_HOUR=18
+DAILY_ROI_POLL_CUTOFF_MINUTE=0
 DAILY_ROI_RUN_ON_STARTUP=0
 # 每天15:30运行独立内部测试，只将“自动化投放”分表发送到“内部”Webhook，不进入腾讯审批执行。
 ROI_INTERNAL_TEST_ENABLED=0

examples/auto_put_ad_mini/DEPLOYMENT.md

@@ -169,7 +169,7 @@ curl -X POST http://localhost:8080/trigger | jq .
 
 ### 日级 ROI 三日口径
 
-- 正式任务每天北京时间11:00执行；在创建数据库批次和发送消息前，必须确认 `loghubods.opengid_base_data` 已存在精确T-1分区，且小程序投流和公众号投流相关数据均非空。未就绪时禁止退回旧分区。
+- 生产正式任务每天北京时间09:00开始检查；在创建数据库批次和发送消息前，必须确认 `loghubods.opengid_base_data` 已存在精确T-1分区，且小程序投流和公众号投流相关数据均非空。未就绪时每30分钟检查一次，18:00做最后一次检查；截止仍未就绪时发送一次告警并停止当天轮询，禁止退回旧分区。目标日期已有正式批次时不重复计算或补发。
 - 日级 ROI 读取 T-1 至 T-3 三个连续完整数据日，数据深度过滤使用最新业务 SQL 的 `usersharedepth<=1`。
 - 正式阈值样本为连续三天每天首层 UV>200、成本>0 且 ROI 有效的小程序创意级和公众号实体；两类渠道独立计算，禁止合并分位点。公众号即转使用自身合格实体等权 P20；小程序按 T-1 实际成本软预算动态反推关停线和实体分位点。
 - 小程序广告级从 ODPS 原始数据按广告直接 `COUNT(DISTINCT mid)`，复用小程序三日动态关停线，但不重复进入创意成本预算样本池；低于关停线且广告 age>3 天时生成广告级关停建议。
@@ -185,7 +185,7 @@ curl -X POST http://localhost:8080/trigger | jq .
 - “审批选择”默认隐藏，需取消隐藏后审批；创意级“当前创意状态”只对关停建议只读腾讯状态，显示正常、已停止或读取失败，其他行留空。
 - 完整主报表保存后，流程会额外按“小程序投流”渠道的“代理名称”生成一代理一份调控建议工作簿；文件名为 `YYYYMMDD_代理名称_调控建议.xlsx`。代理版只包含小程序创意级和广告级三日汇总，广告级 Sheet 继续隐藏，不包含每日明细。包名、广告age、日均首层UV、建议说明、收入、预测总效率ROI、P20/P80/P30、排名、两个裂变系数、日均T0裂变人数/率、审批、执行状态和幂等键均不会写入代理文件；“评分”使用最新日实际ROI并保留两位小数，建议动作仍按内部策略生成并保留“关停创意/关停广告/扩量/观察”。代理表默认只落本地；开启 `ROI_AGENCY_WEBHOOK_ENABLED` 后，飞书应用先上传在线表，再由 `ROI_AGENCY_WEBHOOKS_JSON` 中精确匹配的代理机器人发送卡片。未配置代理直接跳过，不回退总群；完整 webhook 只能放在真实 `.env` 或密钥系统，数据库和日志只保存哈希指纹。
 - 可选内部测试任务每天北京时间15:30执行独立复算，只将“自动化投放”分表发送到“内部”Webhook，不发送完整主表、正式总群或代理群，也不生成可审批腾讯动作。缺少该分表时任务失败并告警。正式任务始终去重；内部测试使用 `ROI_INTERNAL_TEST_DEDUP_ENABLED` 控制，设为`0`时每次显式触发都发送新的测试批次。内部任务无startup入口，重启服务不会自动补发。
-- 11:00正式任务或15:30内部测试执行失败时，调度服务通过飞书应用向 `ROI_FAILURE_FEISHU_CHAT_ID` 发送红色告警卡片；未配置时回退 `FEISHU_OPERATOR_CHAT_ID`。告警失败只记日志，不会掩盖原任务失败。
+- 正式任务的数据未就绪只进入等待，不按失败重复告警；18:00仍未就绪时发送一次截止告警。正式计算或15:30内部测试实际执行失败时，调度服务通过飞书应用向 `ROI_FAILURE_FEISHU_CHAT_ID` 发送红色告警卡片；未配置时回退 `FEISHU_OPERATOR_CHAT_ID`。告警失败只记日志，不会掩盖原任务失败。
 
 离线复算不发送飞书：
 

examples/auto_put_ad_mini/PROJECT_STRUCTURE.md

@@ -16,7 +16,7 @@
 | **Flow A: 决策调控** | `execute_once.py` | 每日 1 次 | Agent 驱动：LLM 动态编排工具调用 → ROI 分析 → 出价调整/暂停广告 |
 | **Flow B: 创建投放** | `execute_creation_once.py` | 每日 1 次 | 硬编码 Pipeline：创建广告 → 准备创意 → 飞书审批 → 提交腾讯 |
 | **Flow C: 审核扫描** | `scan_creative_reviews.py` | 每 2 小时 | 硬编码循环：扫描创意审核结果 → 记录拒审事实 |
-| **Flow D: ROI 精算** | `roi_control/` + `run_daily_roi.py` | 每日 1 次 | 版本化日级 ROI 计算：动态ROI → T15成熟系数 → 策略规则 → 飞书行级审批 → 执行 |
+| **Flow D: ROI 精算** | `roi_control/` + `run_daily_roi.py` | 每日成功 1 批 | 生产09:00起每30分钟检查T-1来源、18:00截止；就绪后执行版本化日级 ROI 计算 → 飞书行级审批 → 执行 |
 
 **两种执行模式**：
 - **Flow A**：Agent 驱动 — `execute_once.py` 通过 Agent 框架启动 LLM，由 LLM 动态决定调用哪些工具、以何种顺序执行。system prompt 中嵌入了推荐的执行流程，但实际编排权在模型。
@@ -598,6 +598,7 @@ docker-compose.yml
 ### 8.3 生产时间线
 
 ```
+北京时间 09:00-18:00        →  Flow D: 每30分钟检查T-1来源，就绪后只执行一个正式ROI批次
 北京时间 10:00 (UTC 02:00)  →  Flow A: Agent 驱动决策调控（execute_once.py）
 北京时间 10:30 (UTC 02:30)  →  Flow B: 创建投放（execute_creation_once.py）
 北京时间 15:30 (UTC 07:30)  →  Flow D 内部测试报表（run_daily_roi.py --internal-test）

examples/auto_put_ad_mini/docs/manual_daily_roi_rerun.md

@@ -0,0 +1,93 @@
+# 生产日级 ROI 手动补跑
+
+正常情况下，日服务会从配置时间开始每30分钟检查一次 T-1 数据，并在18:00做最后一次检查；数据就绪后当天只运行和发送一个正式批次。本文仅适用于18:00截止后经人工确认仍需补跑，或自动调度异常的应急场景。
+
+## 执行目录
+
+先登录生产服务器，然后进入 Compose 配置所在目录：
+
+```bash
+cd /home/server
+```
+
+后续 `docker compose` 命令都在 `/home/server` 下执行。不要在 `/root` 或 `/` 下直接执行，否则会出现：
+
+```text
+no configuration file provided: not found
+```
+
+## 1. 确认日服务正在运行
+
+```bash
+docker compose ps ad_daily_service
+```
+
+应看到 `ad_daily_service` 为 `Up` 或 `healthy`。如果服务未运行，先查看启动日志，不要直接补跑：
+
+```bash
+docker compose logs --tail=200 ad_daily_service
+```
+
+## 2. 手动补跑正式日级 ROI
+
+执行前必须先确认目标日期没有已经计算或发送过的正式批次，避免把应急命令当成日常调度方式。
+
+下面整行命令可以直接复制执行：
+
+```bash
+docker compose exec -T ad_daily_service python -c 'from run_daily_service import run_daily_roi; run_daily_roi()'
+```
+
+该入口与生产定时任务一致，会：
+
+- 使用 `ad_daily_roi` MySQL advisory lock，避免同一任务并发执行；
+- 执行正式日级 ROI 计算；
+- 保存运行批次、实体快照和动作建议；
+- 生成并发送飞书 ROI 审批表。
+
+它不会因为补跑而绕过审批直接修改腾讯。腾讯动作仍以飞书审批表的逐行批准和生产开关为准。
+
+## 3. 查看本次执行日志
+
+另开一个终端，同样先进入 `/home/server`：
+
+```bash
+cd /home/server
+docker compose logs -f --tail=200 ad_daily_service
+```
+
+只查看最近 30 分钟日志：
+
+```bash
+docker compose logs --since=30m ad_daily_service
+```
+
+停止持续查看日志时按 `Ctrl+C`，不会停止容器或 ROI 任务。
+
+## 4. 常见异常
+
+### 上游分区仍未就绪
+
+任务会失败关闭，不会回退到更早分区。确认 T-1 数据完整后，重新执行第 2 步的同一条命令即可。
+
+### 提示锁未获得或任务正在运行
+
+说明已有日级 ROI 任务持有锁。不要并发重复执行，先等待现有任务结束并检查日志。
+
+### 容器处于 restarting
+
+先检查容器状态和错误日志：
+
+```bash
+docker inspect ad-daily-service --format 'status={{.State.Status}} exit={{.State.ExitCode}} restarts={{.RestartCount}} error={{.State.Error}}'
+docker compose logs --tail=200 ad_daily_service
+```
+
+容器恢复为正常运行后再补跑。
+
+## 不需要做的操作
+
+- 不需要修改 `DAILY_ROI_HOUR` 或 `DAILY_ROI_MINUTE`；
+- 不需要为了补跑而重启或重建服务；
+- 不要同时执行多条补跑命令；
+- 不要直接运行绕过日服务入口的腾讯写操作。

examples/auto_put_ad_mini/docs/unified_services_deployment.md

@@ -90,8 +90,11 @@ DAILY_REVIEW_RUN_ON_STARTUP=0
 DAILY_SYNC_DELIVERY_TEMPLATE=0
 
 DAILY_ROI_ENABLED=0
-DAILY_ROI_HOUR=11
+DAILY_ROI_HOUR=9
 DAILY_ROI_MINUTE=0
+DAILY_ROI_POLL_INTERVAL_MINUTES=30
+DAILY_ROI_POLL_CUTOFF_HOUR=18
+DAILY_ROI_POLL_CUTOFF_MINUTE=0
 DAILY_ROI_RUN_ON_STARTUP=0
 ROI_INTERNAL_TEST_ENABLED=0
 ROI_INTERNAL_TEST_HOUR=15
@@ -263,16 +266,16 @@ ROI 表格使用获得链接者可编辑权限。默认隐藏的黄色【审批
 创意级“当前创意状态”只对关停建议只读腾讯状态并显示正常、已停止或读取失败，其他行留空；“审批选择”默认隐藏。
 完整主报表写入成功后，服务额外按“小程序投流”渠道的“代理名称”生成一代理一份 `roi_agency_advice_v9` 工作簿，文件名为 `YYYYMMDD_代理名称_调控建议.xlsx`。代理版可见 Sheet 名为“小程序创意调控建议”，并保留默认隐藏的“小程序广告级三日汇总”，不包含每日明细；包名、广告age、日均首层UV、建议说明、收入、预测总效率ROI、阈值、排名、两个裂变系数、日均T0裂变人数/率、审批和执行审计列均物理删除。“评分”使用最新日实际ROI并保留两位小数，不参与代理动作计算；建议动作仍按内部策略生成，并明确显示“关停创意”或“关停广告”。代理表生成不改变主报表、数据库快照或主审批链接。
 
-正式任务每天北京时间11:00执行。任务固定以T-1为结束日，并在创建MySQL运行批次、生成报表和发送飞书前校验 `loghubods.opengid_base_data` 的精确T-1分区；小程序投流或公众号投流相关数据任一为空即失败关闭，禁止静默回退到更早分区。
+生产正式任务每天北京时间09:00开始检查。任务固定以T-1为结束日，并在创建MySQL运行批次、生成报表和发送飞书前校验 `loghubods.opengid_base_data` 的精确T-1分区；小程序投流或公众号投流相关数据任一为空时不启动计算，每30分钟重新检查一次。18:00执行最后一次检查：数据已就绪则执行，仍未就绪则发送一次红色截止告警并停止当天轮询，禁止静默回退到更早分区。目标日期已经存在任意正式 `roi_metric_run` 时直接跳过，不重复计算、不重复发送主表或代理表；内部测试批次不影响正式任务判断。
 
 总表继续由飞书应用按 `ROI_FEISHU_CHAT_ID` / `FEISHU_OPERATOR_CHAT_ID` 发布。代理表发布由 `ROI_AGENCY_WEBHOOK_ENABLED` 独立控制，完整机器人地址仅允许保存在密钥系统或真实 `.env` 的 `ROI_AGENCY_WEBHOOKS_JSON`，JSON key 使用代理简称。飞书应用先上传并开放在线表链接，再由对应代理机器人发送可点击卡片；未配置代理记录为跳过，不回退到总群。同一 `run_id + 代理 + 代理报表版本` 幂等审计，已上传表格在通知重试时复用，避免重复建表。机器人地址只保存哈希指纹，不能写入数据库或日志。
 
 `ROI_INTERNAL_TEST_ENABLED=1` 时，服务每天北京时间15:30注册独立内部任务。该任务复算后使用空管理账户集合，不生成可审批腾讯动作，只把“自动化投放”分表发送到 `ROI_AGENCY_WEBHOOKS_JSON` 的“内部”路由，不发送完整主表、正式总群或代理群；缺少该分表时任务失败并告警。它不注册startup任务，服务重启不会主动补发。正式任务始终去重；内部测试由 `ROI_INTERNAL_TEST_DEDUP_ENABLED` 控制，生产默认`1`，测试环境显式设为`0`时每次触发生成独立测试run并重新发送。
 
-11:00正式任务或15:30内部测试的子进程启动失败或非零退出时，调度服务通过飞书应用向 `ROI_FAILURE_FEISHU_CHAT_ID` 发送红色失败告警；该变量为空时回退 `FEISHU_OPERATOR_CHAT_ID`。失败告警本身发送失败只记录错误日志，不会吞掉或替换原任务异常。
+正式轮询的数据就绪检查不会把来源延迟当成任务失败，也不会每30分钟发送失败告警。18:00仍未就绪时只发送一次截止告警。正式任务实际启动后或15:30内部测试的子进程启动失败、非零退出时，调度服务通过飞书应用向 `ROI_FAILURE_FEISHU_CHAT_ID` 发送红色失败告警；该变量为空时回退 `FEISHU_OPERATOR_CHAT_ID`。失败告警本身发送失败只记录错误日志，不会吞掉或替换原任务异常。
 
 1. 首次部署先执行下方参数发布命令并完成回读校验。
-2. 设置 `DAILY_ROI_ENABLED=1`、`ROI_APPLY_ENABLED=0`、`ROI_SHEET_APPROVAL_ENABLED=0`，观察11:00的T-1来源校验、ODPS计算、数据库快照和可编辑飞书表。
+2. 设置 `DAILY_ROI_ENABLED=1`、`ROI_APPLY_ENABLED=0`、`ROI_SHEET_APPROVAL_ENABLED=0`，观察09:00起每30分钟的T-1来源校验；数据就绪后应只生成一份数据库快照和可编辑飞书表，18:00未就绪应只收到一次截止告警。
 3. 核对阈值、账户可执行范围、黄色审批列和动作数量。
 4. 再设置 `ROI_APPLY_ENABLED=1`、`ROI_SHEET_APPROVAL_ENABLED=1`，重建 `ad-control-service`；批准一条低风险测试行，核对腾讯回读、表格结果和群通知。
 5. 生产审批有效期默认 120 分钟；过期行不能执行。表格逐行选择“拒绝”不会产生腾讯写操作。
@@ -338,6 +341,12 @@ docker compose --env-file /dev/null logs --since=30m ad_control_service
 docker compose --env-file /dev/null logs --since=30m ad_daily_service
 ```
 
+`ad_daily_service` 启动日志应包含：
+
+```text
+Daily ROI polling configured start=09:00 interval=30m cutoff=18:00
+```
+
 应确认：
 
 - 两个容器均为 healthy。
@@ -345,7 +354,7 @@ docker compose --env-file /dev/null logs --since=30m ad_daily_service
 - 飞书只有一个 WebSocket 消费者。
 - 10:30 只出现一次创建任务。
 - 审核扫描每 2 小时最多一个实例。
-- 11:00正式任务只生成一个相同指标/策略版本的ROI批次，同日重跑不会重复发布；15:30内部测试是否去重由独立变量控制。
+- 09:00至18:00正式轮询只生成一个目标日期的正式ROI批次，已有正式批次后不再计算或补发；15:30内部测试是否去重由独立变量控制。
 - ROI 审批执行与实时 CPM 调控不并发写腾讯，基础出价调整后两边状态一致。
 
 回滚时将 `VERSION` 改为上一个镜像版本并重新部署。数据库新增表和列可以保留，

examples/auto_put_ad_mini/roi_control/config.py

@@ -77,6 +77,9 @@ class RoiConfig:
     sheet_approval_poll_seconds: int = 60
     report_hour: int = 11
     report_minute: int = 0
+    daily_poll_interval_minutes: int = 30
+    daily_poll_cutoff_hour: int = 18
+    daily_poll_cutoff_minute: int = 0
     internal_test_enabled: bool = False
     internal_test_hour: int = 15
     internal_test_minute: int = 30
@@ -114,6 +117,15 @@ class RoiConfig:
             ),
             report_hour=int(os.getenv("DAILY_ROI_HOUR", "11")),
             report_minute=int(os.getenv("DAILY_ROI_MINUTE", "0")),
+            daily_poll_interval_minutes=int(
+                os.getenv("DAILY_ROI_POLL_INTERVAL_MINUTES", "30")
+            ),
+            daily_poll_cutoff_hour=int(
+                os.getenv("DAILY_ROI_POLL_CUTOFF_HOUR", "18")
+            ),
+            daily_poll_cutoff_minute=int(
+                os.getenv("DAILY_ROI_POLL_CUTOFF_MINUTE", "0")
+            ),
             internal_test_enabled=env_flag("ROI_INTERNAL_TEST_ENABLED"),
             internal_test_hour=int(os.getenv("ROI_INTERNAL_TEST_HOUR", "15")),
             internal_test_minute=int(os.getenv("ROI_INTERNAL_TEST_MINUTE", "30")),
@@ -196,6 +208,24 @@ class RoiConfig:
             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.daily_poll_interval_minutes < 1
+            or self.daily_poll_interval_minutes > 60
+            or 60 % self.daily_poll_interval_minutes
+        ):
+            raise ValueError(
+                "DAILY_ROI_POLL_INTERVAL_MINUTES must be a positive divisor of 60"
+            )
+        if (
+            not 0 <= self.daily_poll_cutoff_hour <= 23
+            or not 0 <= self.daily_poll_cutoff_minute <= 59
+        ):
+            raise ValueError("DAILY_ROI_POLL_CUTOFF_HOUR/MINUTE is invalid")
+        if (self.daily_poll_cutoff_hour, self.daily_poll_cutoff_minute) <= (
+            self.report_hour,
+            self.report_minute,
+        ):
+            raise ValueError("DAILY_ROI poll cutoff must be later than report time")
         if not 0 <= self.internal_test_hour <= 23 or not 0 <= self.internal_test_minute <= 59:
             raise ValueError("ROI_INTERNAL_TEST_HOUR/MINUTE is invalid")
         if self.approval_ttl_minutes < 1:

examples/auto_put_ad_mini/roi_control/repository.py

@@ -202,6 +202,35 @@ def load_run(run_id: str) -> dict[str, Any] | None:
         connection.close()
 
 
+def load_formal_run_for_end_date(end_date: date) -> dict[str, Any] | None:
+    """Return an existing formal run for a data date, excluding internal tests."""
+
+    connection = connect()
+    try:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                """
+                SELECT * FROM roi_metric_run
+                WHERE end_date=%s
+                ORDER BY created_at DESC
+                """,
+                (end_date,),
+            )
+            for run in cursor.fetchall():
+                try:
+                    config = json.loads(run.get("config_json") or "{}")
+                except (TypeError, json.JSONDecodeError):
+                    config = {}
+                internal_test = config.get("internal_test")
+                if internal_test is None:
+                    internal_test = "_internal_test" in str(run.get("run_id") or "")
+                if not internal_test:
+                    return run
+            return None
+    finally:
+        connection.close()
+
+
 def replace_run_results(
     run_id: str,
     *,

examples/auto_put_ad_mini/run_daily_service.py

@@ -9,6 +9,7 @@ import logging
 import os
 import subprocess
 import sys
+from datetime import datetime, timedelta
 from pathlib import Path
 
 from apscheduler.schedulers.blocking import BlockingScheduler
@@ -38,7 +39,14 @@ sys.path.insert(0, str(HERE))
 from db.connection import get_connection  # noqa: E402
 from storage import advisory_lock, initialize_schema  # noqa: E402
 from roi_control.config import AgencyWebhookConfig, RoiConfig  # noqa: E402
+from roi_control.data_source import (  # noqa: E402
+    SHANGHAI,
+    SourceDataNotReadyError,
+    validate_source_ready,
+)
 from roi_control.feishu import RoiFeishuPublisher  # noqa: E402
+from roi_control.odps_client import ODPSClient  # noqa: E402
+from roi_control.repository import load_formal_run_for_end_date  # noqa: E402
 from tools.delivery_config import validate_region_dictionary  # noqa: E402
 
 
@@ -57,7 +65,12 @@ def _notify_roi_failure(
     extra_args: list[str] | None,
     error: str,
 ) -> None:
-    mode = "内部15:30测试" if "--internal-test" in (extra_args or []) else "正式11:00任务"
+    if "--internal-test" in (extra_args or []):
+        mode = "内部15:30测试"
+    else:
+        report_hour = int(os.getenv("DAILY_ROI_HOUR", "11"))
+        report_minute = int(os.getenv("DAILY_ROI_MINUTE", "0"))
+        mode = f"正式{report_hour:02d}:{report_minute:02d}起轮询任务"
     publisher: RoiFeishuPublisher | None = None
     try:
         publisher = RoiFeishuPublisher(require_chat_ids=False)
@@ -76,6 +89,31 @@ def _notify_roi_failure(
             publisher.close()
 
 
+def _notify_roi_source_cutoff(
+    *,
+    target_date: str,
+    cutoff_time: str,
+    error: str,
+) -> None:
+    publisher: RoiFeishuPublisher | None = None
+    try:
+        publisher = RoiFeishuPublisher(require_chat_ids=False)
+        publisher.send_service_alert(
+            title="日级ROI数据截止仍未就绪",
+            content=(
+                f"目标数据日期：**{target_date}**\n"
+                f"截止时间：**{cutoff_time}**\n"
+                f"原因：`{error[:500]}`\n"
+                "当天已停止自动轮询，不会使用旧分区，也不会生成或补发ROI报表。"
+            ),
+        )
+    except Exception as exc:
+        logger.error("Failed to send ROI source cutoff alert: %s", exc)
+    finally:
+        if publisher is not None:
+            publisher.close()
+
+
 def sync_enabled_delivery_templates() -> int:
     """Keep enabled DB templates aligned with the code's global delivery window."""
     serialized = json.dumps(TIME_SERIES_DEFAULT)
@@ -168,6 +206,106 @@ def run_daily_roi() -> None:
     )
 
 
+def _effective_shanghai_now(now: datetime | None = None) -> datetime:
+    current = now or datetime.now(SHANGHAI)
+    if current.tzinfo is None:
+        current = current.replace(tzinfo=SHANGHAI)
+    return current.astimezone(SHANGHAI)
+
+
+def _daily_roi_times(
+    config: RoiConfig,
+    now: datetime,
+) -> tuple[datetime, datetime]:
+    start = now.replace(
+        hour=config.report_hour,
+        minute=config.report_minute,
+        second=0,
+        microsecond=0,
+    )
+    cutoff = now.replace(
+        hour=config.daily_poll_cutoff_hour,
+        minute=config.daily_poll_cutoff_minute,
+        second=0,
+        microsecond=0,
+    )
+    return start, cutoff
+
+
+def _roi_poll_minutes(config: RoiConfig) -> str:
+    values = {
+        (config.report_minute + offset) % 60
+        for offset in range(0, 60, config.daily_poll_interval_minutes)
+    }
+    return ",".join(str(value) for value in sorted(values))
+
+
+def _attempt_daily_roi_when_ready(
+    config: RoiConfig,
+    *,
+    now: datetime,
+    at_cutoff: bool,
+) -> str:
+    target = (now - timedelta(days=1)).date()
+    existing = load_formal_run_for_end_date(target)
+    if existing:
+        logger.info(
+            "Skip daily ROI target=%s: formal run already exists run=%s status=%s",
+            target.strftime("%Y%m%d"),
+            existing.get("run_id"),
+            existing.get("status"),
+        )
+        return "ALREADY_RAN"
+
+    target_date = target.strftime("%Y%m%d")
+    client = ODPSClient(project=os.getenv("ODPS_PROJECT", "loghubods"))
+    try:
+        counts = validate_source_ready(client, target_date)
+    except SourceDataNotReadyError as exc:
+        if at_cutoff:
+            logger.error(
+                "Daily ROI source missed cutoff target=%s: %s",
+                target_date,
+                exc,
+            )
+            _notify_roi_source_cutoff(
+                target_date=target_date,
+                cutoff_time=(
+                    f"{config.daily_poll_cutoff_hour:02d}:"
+                    f"{config.daily_poll_cutoff_minute:02d}"
+                ),
+                error=str(exc),
+            )
+            return "CUTOFF_NOT_READY"
+        logger.info("Daily ROI source not ready target=%s: %s", target_date, exc)
+        return "WAITING_SOURCE"
+
+    logger.info("Daily ROI source ready target=%s counts=%s", target_date, counts)
+    run_daily_roi()
+    return "TRIGGERED"
+
+
+def poll_daily_roi(
+    config: RoiConfig,
+    *,
+    now: datetime | None = None,
+) -> str:
+    current = _effective_shanghai_now(now)
+    start, cutoff = _daily_roi_times(config, current)
+    if current < start or current >= cutoff:
+        return "OUTSIDE_WINDOW"
+    return _attempt_daily_roi_when_ready(config, now=current, at_cutoff=False)
+
+
+def finalize_daily_roi_at_cutoff(
+    config: RoiConfig,
+    *,
+    now: datetime | None = None,
+) -> str:
+    current = _effective_shanghai_now(now)
+    return _attempt_daily_roi_when_ready(config, now=current, at_cutoff=True)
+
+
 def run_internal_roi_test() -> None:
     _run_script(
         "run_daily_roi.py",
@@ -265,21 +403,43 @@ def main(argv: list[str] | None = None) -> None:
             misfire_grace_time=1800,
         )
     if roi_config.daily_enabled:
+        logger.info(
+            "Daily ROI polling configured start=%02d:%02d interval=%dm cutoff=%02d:%02d",
+            roi_config.report_hour,
+            roi_config.report_minute,
+            roi_config.daily_poll_interval_minutes,
+            roi_config.daily_poll_cutoff_hour,
+            roi_config.daily_poll_cutoff_minute,
+        )
         scheduler.add_job(
-            run_daily_roi,
+            poll_daily_roi,
             CronTrigger(
-                hour=roi_config.report_hour,
-                minute=roi_config.report_minute,
+                minute=_roi_poll_minutes(roi_config),
                 timezone="Asia/Shanghai",
             ),
             id="daily_roi",
-            name="日级ROI计算与逐行审批",
+            name="日级ROI数据就绪轮询",
+            args=[roi_config],
             max_instances=1,
             coalesce=True,
             misfire_grace_time=int(
                 os.getenv("DAILY_ROI_MISFIRE_GRACE_SECONDS", "3600")
             ),
         )
+        scheduler.add_job(
+            finalize_daily_roi_at_cutoff,
+            CronTrigger(
+                hour=roi_config.daily_poll_cutoff_hour,
+                minute=roi_config.daily_poll_cutoff_minute,
+                timezone="Asia/Shanghai",
+            ),
+            id="daily_roi_cutoff",
+            name="日级ROI数据就绪截止检查",
+            args=[roi_config],
+            max_instances=1,
+            coalesce=True,
+            misfire_grace_time=300,
+        )
     if roi_config.internal_test_enabled:
         scheduler.add_job(
             run_internal_roi_test,
@@ -310,9 +470,10 @@ def main(argv: list[str] | None = None) -> None:
         )
     if roi_config.daily_enabled and _env_flag("DAILY_ROI_RUN_ON_STARTUP", False):
         scheduler.add_job(
-            run_daily_roi,
+            poll_daily_roi,
             id="daily_roi_startup",
-            name="启动时ROI计算",
+            name="启动时ROI数据就绪检查",
+            args=[roi_config],
         )
     logger.info(
         "Daily service started jobs=%s",

examples/auto_put_ad_mini/test_daily_roi_polling.py

@@ -0,0 +1,272 @@
+import json
+import sys
+import unittest
+from datetime import date, datetime
+from pathlib import Path
+from unittest.mock import Mock, patch
+
+
+HERE = Path(__file__).resolve().parent
+sys.path.insert(0, str(HERE))
+
+import run_daily_service  # noqa: E402
+from roi_control import repository  # noqa: E402
+from roi_control.config import RoiConfig  # noqa: E402
+from roi_control.data_source import SHANGHAI, SourceDataNotReadyError  # noqa: E402
+
+
+class RoiPollingConfigTest(unittest.TestCase):
+    def test_default_polling_configuration_is_30_minutes_until_1800(self):
+        config = RoiConfig()
+        config.validate()
+
+        self.assertEqual(30, config.daily_poll_interval_minutes)
+        self.assertEqual((18, 0), (
+            config.daily_poll_cutoff_hour,
+            config.daily_poll_cutoff_minute,
+        ))
+        self.assertEqual("0,30", run_daily_service._roi_poll_minutes(config))
+
+    def test_poll_interval_must_divide_one_hour(self):
+        with self.assertRaisesRegex(ValueError, "positive divisor of 60"):
+            RoiConfig(daily_poll_interval_minutes=25).validate()
+
+    def test_cutoff_must_be_later_than_report_time(self):
+        with self.assertRaisesRegex(ValueError, "cutoff must be later"):
+            RoiConfig(
+                report_hour=18,
+                report_minute=0,
+                daily_poll_cutoff_hour=18,
+                daily_poll_cutoff_minute=0,
+            ).validate()
+
+
+class DailyRoiPollingTest(unittest.TestCase):
+    def setUp(self):
+        self.config = RoiConfig(
+            daily_enabled=True,
+            report_hour=9,
+            report_minute=0,
+            daily_poll_interval_minutes=30,
+            daily_poll_cutoff_hour=18,
+            daily_poll_cutoff_minute=0,
+        )
+        self.active_time = datetime(2026, 8, 14, 9, 30, tzinfo=SHANGHAI)
+        self.cutoff_time = datetime(2026, 8, 14, 18, 0, tzinfo=SHANGHAI)
+
+    def test_existing_formal_run_skips_source_check_and_report(self):
+        existing = {"run_id": "roi-existing", "status": "PENDING_APPROVAL"}
+        with (
+            patch.object(
+                run_daily_service,
+                "load_formal_run_for_end_date",
+                return_value=existing,
+            ),
+            patch.object(run_daily_service, "validate_source_ready") as validate,
+            patch.object(run_daily_service, "run_daily_roi") as run,
+        ):
+            result = run_daily_service.poll_daily_roi(
+                self.config,
+                now=self.active_time,
+            )
+
+        self.assertEqual("ALREADY_RAN", result)
+        validate.assert_not_called()
+        run.assert_not_called()
+
+    def test_cutoff_skips_existing_formal_run_without_alert(self):
+        existing = {"run_id": "roi-existing", "status": "COMPLETED"}
+        with (
+            patch.object(
+                run_daily_service,
+                "load_formal_run_for_end_date",
+                return_value=existing,
+            ),
+            patch.object(run_daily_service, "validate_source_ready") as validate,
+            patch.object(run_daily_service, "_notify_roi_source_cutoff") as notify,
+            patch.object(run_daily_service, "run_daily_roi") as run,
+        ):
+            result = run_daily_service.finalize_daily_roi_at_cutoff(
+                self.config,
+                now=self.cutoff_time,
+            )
+
+        self.assertEqual("ALREADY_RAN", result)
+        validate.assert_not_called()
+        notify.assert_not_called()
+        run.assert_not_called()
+
+    def test_missing_source_waits_without_failure_alert(self):
+        error = SourceDataNotReadyError("required_dt=20260813")
+        with (
+            patch.object(
+                run_daily_service,
+                "load_formal_run_for_end_date",
+                return_value=None,
+            ),
+            patch.object(run_daily_service, "ODPSClient"),
+            patch.object(
+                run_daily_service,
+                "validate_source_ready",
+                side_effect=error,
+            ),
+            patch.object(run_daily_service, "_notify_roi_source_cutoff") as notify,
+            patch.object(run_daily_service, "run_daily_roi") as run,
+        ):
+            result = run_daily_service.poll_daily_roi(
+                self.config,
+                now=self.active_time,
+            )
+
+        self.assertEqual("WAITING_SOURCE", result)
+        notify.assert_not_called()
+        run.assert_not_called()
+
+    def test_ready_source_triggers_formal_run_once(self):
+        with (
+            patch.object(
+                run_daily_service,
+                "load_formal_run_for_end_date",
+                return_value=None,
+            ),
+            patch.object(run_daily_service, "ODPSClient"),
+            patch.object(
+                run_daily_service,
+                "validate_source_ready",
+                return_value={"row_count": 10, "self_rows": 6, "gzh_rows": 4},
+            ) as validate,
+            patch.object(run_daily_service, "run_daily_roi") as run,
+        ):
+            result = run_daily_service.poll_daily_roi(
+                self.config,
+                now=self.active_time,
+            )
+
+        self.assertEqual("TRIGGERED", result)
+        validate.assert_called_once_with(unittest.mock.ANY, "20260813")
+        run.assert_called_once_with()
+
+    def test_cutoff_performs_final_check_and_alerts_when_still_missing(self):
+        error = SourceDataNotReadyError("required_dt=20260813")
+        with (
+            patch.object(
+                run_daily_service,
+                "load_formal_run_for_end_date",
+                return_value=None,
+            ),
+            patch.object(run_daily_service, "ODPSClient"),
+            patch.object(
+                run_daily_service,
+                "validate_source_ready",
+                side_effect=error,
+            ),
+            patch.object(run_daily_service, "_notify_roi_source_cutoff") as notify,
+            patch.object(run_daily_service, "run_daily_roi") as run,
+        ):
+            result = run_daily_service.finalize_daily_roi_at_cutoff(
+                self.config,
+                now=self.cutoff_time,
+            )
+
+        self.assertEqual("CUTOFF_NOT_READY", result)
+        notify.assert_called_once_with(
+            target_date="20260813",
+            cutoff_time="18:00",
+            error="required_dt=20260813",
+        )
+        run.assert_not_called()
+
+    def test_cutoff_runs_when_source_becomes_ready(self):
+        with (
+            patch.object(
+                run_daily_service,
+                "load_formal_run_for_end_date",
+                return_value=None,
+            ),
+            patch.object(run_daily_service, "ODPSClient"),
+            patch.object(
+                run_daily_service,
+                "validate_source_ready",
+                return_value={"row_count": 10, "self_rows": 6, "gzh_rows": 4},
+            ),
+            patch.object(run_daily_service, "run_daily_roi") as run,
+        ):
+            result = run_daily_service.finalize_daily_roi_at_cutoff(
+                self.config,
+                now=self.cutoff_time,
+            )
+
+        self.assertEqual("TRIGGERED", result)
+        run.assert_called_once_with()
+
+    def test_poll_does_nothing_outside_active_window(self):
+        for now in (
+            datetime(2026, 8, 14, 8, 59, tzinfo=SHANGHAI),
+            self.cutoff_time,
+            datetime(2026, 8, 14, 18, 30, tzinfo=SHANGHAI),
+        ):
+            with self.subTest(now=now), patch.object(
+                run_daily_service,
+                "load_formal_run_for_end_date",
+            ) as load_run:
+                result = run_daily_service.poll_daily_roi(self.config, now=now)
+                self.assertEqual("OUTSIDE_WINDOW", result)
+                load_run.assert_not_called()
+
+    def test_scheduler_registers_poll_and_cutoff_jobs(self):
+        scheduler = Mock()
+        scheduler.get_jobs.return_value = []
+        with (
+            patch.object(
+                run_daily_service.RoiConfig,
+                "from_env",
+                return_value=self.config,
+            ),
+            patch.object(run_daily_service, "validate_region_dictionary"),
+            patch.object(run_daily_service, "initialize_schema"),
+            patch.object(run_daily_service, "_env_flag", return_value=False),
+            patch.object(
+                run_daily_service,
+                "BlockingScheduler",
+                return_value=scheduler,
+            ),
+        ):
+            run_daily_service.main([])
+
+        jobs = {call.kwargs["id"]: call for call in scheduler.add_job.call_args_list}
+        self.assertIn("daily_roi", jobs)
+        self.assertIn("daily_roi_cutoff", jobs)
+        self.assertIs(jobs["daily_roi"].args[0], run_daily_service.poll_daily_roi)
+        self.assertIs(
+            jobs["daily_roi_cutoff"].args[0],
+            run_daily_service.finalize_daily_roi_at_cutoff,
+        )
+
+
+class FormalRunLookupTest(unittest.TestCase):
+    def test_internal_test_is_ignored_when_formal_run_exists(self):
+        cursor = Mock()
+        cursor.__enter__ = Mock(return_value=cursor)
+        cursor.__exit__ = Mock(return_value=False)
+        cursor.fetchall.return_value = [
+            {
+                "run_id": "roi-internal_internal_test",
+                "config_json": json.dumps({"internal_test": True}),
+            },
+            {
+                "run_id": "roi-formal",
+                "config_json": json.dumps({"internal_test": False}),
+            },
+        ]
+        connection = Mock()
+        connection.cursor.return_value = cursor
+
+        with patch.object(repository, "connect", return_value=connection):
+            result = repository.load_formal_run_for_end_date(date(2026, 8, 13))
+
+        self.assertEqual("roi-formal", result["run_id"])
+        connection.close.assert_called_once_with()
+
+
+if __name__ == "__main__":
+    unittest.main()

runtime.env.example

@@ -118,10 +118,13 @@ DAILY_RUN_ON_STARTUP=0
 DAILY_REVIEW_RUN_ON_STARTUP=0
 DAILY_SYNC_DELIVERY_TEMPLATE=0
 
-# 日级 ROI：正式任务统一为北京时间 11:00，写操作默认关闭。
+# 日级 ROI：生产09:00开始检查，未就绪时每30分钟重试，18:00最后检查后截止。
 DAILY_ROI_ENABLED=0
-DAILY_ROI_HOUR=11
+DAILY_ROI_HOUR=9
 DAILY_ROI_MINUTE=0
+DAILY_ROI_POLL_INTERVAL_MINUTES=30
+DAILY_ROI_POLL_CUTOFF_HOUR=18
+DAILY_ROI_POLL_CUTOFF_MINUTE=0
 DAILY_ROI_RUN_ON_STARTUP=0
 DAILY_ROI_MISFIRE_GRACE_SECONDS=3600
 DAILY_ROI_LOCK_NAME=ad_daily_roi