feat(roi): add agency webhook report delivery

examples/auto_put_ad_mini/.env.example

@@ -42,14 +42,24 @@ ROI_SCALE_RATIO=1.10
 ROI_SCALE_COOLDOWN_DAYS=3
 ROI_MAX_BASE_RATIO=2.00
 # 连续三天合格的小程序创意+公众号实体等权整体P20。
+ROI_SELF_STOP_MIN_AGE=4
 ROI_STOP_QUANTILE=0.20
 ROI_UP_QUANTILE=0.80
 ROI_OBSERVE_MIN_LATEST_UV=200
-ROI_FISSION_PARAMETER_VERSION=20260712_A0-A15_v1
+# 非三日正式样本的小程序创意单日补充：UV>200且ROI<=0.20，或UV>500且ROI<=P30；关停仍要求广告age>3天。
+ROI_ONE_DAY_MIN_UV=200
+ROI_ONE_DAY_P30_MIN_UV=500
+ROI_ONE_DAY_HARD_STOP_ROI=0.20
+ROI_ONE_DAY_STOP_QUANTILE=0.30
+ROI_FISSION_PARAMETER_VERSION=20260712_A0-A15_v2
 # ROI报表发送到此群；同时抄送 FEISHU_OPERATOR_CHAT_ID，重复ID自动去重。
 # 审批表链接为获得链接者可编辑，黄色审批列选择“批准”即自动执行。
 # 为空时依次复用 FEISHU_AD_PROJECT_CHAT_ID / RTC_COMMAND_CHAT_ID
 ROI_FEISHU_CHAT_ID=
+# 代理表由飞书应用上传后，通过各代理自定义机器人发送在线表卡片。
+# 默认关闭；JSON 的 key 使用代理简称（如“棱镜”），完整 webhook 只写入真实 .env/密钥系统。
+ROI_AGENCY_WEBHOOK_ENABLED=0
+ROI_AGENCY_WEBHOOKS_JSON={}
 
 RTC_START_HOUR=12
 RTC_STOP_HOUR=21

examples/auto_put_ad_mini/DEPLOYMENT.md

@@ -172,16 +172,17 @@ curl -X POST http://localhost:8080/trigger | jq .
 
 - 日级 ROI 读取 T-1 至 T-3 三个连续完整数据日，数据深度过滤使用最新业务 SQL 的 `usersharedepth<=1`。
 - 正式阈值样本为连续三天每天首层 UV>200、成本>0 且 ROI 有效的小程序创意级和公众号实体；两类实体合并后按实体等权计算整体 P20 关停线。
-- 小程序广告级从 ODPS 原始数据按广告直接 `COUNT(DISTINCT mid)`，复用同一 P20，但不重复进入样本池；低于关停线且广告 age 达标时生成广告级关停建议。
+- 小程序广告级从 ODPS 原始数据按广告直接 `COUNT(DISTINCT mid)`，复用同一 P20，但不重复进入样本池；低于关停线且广告 age>3 天时生成广告级关停建议。
 - 金额、UV 和人数在三日汇总表展示为三日总量/3；ROI 与裂变率使用三日总分子/总分母的加权口径。
-- 未进入正式样本、但最新日首层 UV>200 的实体放在各汇总 Sheet 末尾，仅观察、不参与 P20、不自动执行。
+- 未进入三日正式样本的小程序创意增加单日补充判断：最新日首层 UV>200 且预测总效率 ROI≤0.20，或最新日首层 UV>500 且预测总效率 ROI≤全部最新日 UV>500 小程序创意的实体等权 P30；命中且广告 age>3 天时建议关停，命中但 age≤3 天时观察。动作原因同时展示最新日判断值与仅供参考的三日预测 ROI。其余最新日首层 UV>200 的非正式实体继续置底观察，不参与三日 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/…”比率列固定显示两位小数。“传播裂变系数匹配”保留审计数据但默认隐藏。
+- “动作”可见列改为“建议动作”，其后展示“建议说明”；小程序创意级和广告级关停分别显示“关停创意”“关停广告”，但底层动作不变。创意关停按三日P20、单日硬线、单日P30三类原因分组，同类按 ROI 升序。底层无动作的正式中间区间在报表中显示为“观察”，但不生成可执行动作。阈值样本状态、执行状态和执行结果保留为隐藏审计字段。
+- 当前三日策略以创意级+公众号合格实体统一等权 P20 生成低 ROI 关停建议；以合格小程序创意实体等权 P80 识别头部20%，广告 age≥3 时生成扩量建议。所有小程序创意级和广告级关停均要求广告 age>3 天；创意级关停批准后只暂停对应动态创意，广告级关停批准后暂停整个广告。
+- 汇总表将三日总 T0 裂变人数 / 三日总首层 UV 的加权比例展示为“日均T0裂变率”。汇总顺序为关停、扩量、中间观察、条件不足观察；第一条扩量行和扩量后第一条观察行顶部均使用粗线分隔。“当日效率ROI”和“预测总效率ROI”均使用深红—黄—绿色阶，绿色代表表现好。普通数值显示两位小数，UV、人数、数量和广告年龄显示整数；两个“裂变系数-总裂变UV/…”比率列固定显示两位小数。“传播裂变系数匹配”保留审计数据但默认隐藏。
 - “审批选择”默认隐藏，需取消隐藏后审批；创意级“当前创意状态”只对关停建议只读腾讯状态，显示正常、已停止或读取失败，其他行留空。
+- 完整主报表保存后，流程会额外按“小程序投流”渠道的“代理名称”生成一代理一份调控建议工作簿；文件名为 `YYYYMMDD_代理名称_调控建议.xlsx`。代理版只包含小程序创意级和广告级三日汇总，广告级 Sheet 继续隐藏，不包含每日明细。包名、广告age、日均首层UV、建议说明、收入、预测总效率ROI、P20/P80/P30、排名、两个裂变系数、日均T0裂变人数/率、审批、执行状态和幂等键均不会写入代理文件；内部“当日效率ROI”仅改名为两位小数的“评分”展示，建议动作仍按预测总效率ROI计算并保留“关停创意/关停广告/扩量/观察”。代理表默认只落本地；开启 `ROI_AGENCY_WEBHOOK_ENABLED` 后，飞书应用先上传在线表，再由 `ROI_AGENCY_WEBHOOKS_JSON` 中精确匹配的代理机器人发送卡片。未配置代理直接跳过，不回退总群；完整 webhook 只能放在真实 `.env` 或密钥系统，数据库和日志只保存哈希指纹。
 
 离线复算不发送飞书：
 

examples/auto_put_ad_mini/docs/unified_services_deployment.md

@@ -82,6 +82,8 @@ ROI_SCALE_COOLDOWN_DAYS=3
 ROI_MAX_BASE_RATIO=2.00
 ROI_FISSION_PARAMETER_VERSION=20260712_A0-A15_v2
 ROI_FEISHU_CHAT_ID=oc_xxx
+ROI_AGENCY_WEBHOOK_ENABLED=0
+ROI_AGENCY_WEBHOOKS_JSON={}
 ```
 
 多个操作人使用英文逗号分隔 `RTC_COMMAND_ALLOWED_OPEN_IDS`。
@@ -165,7 +167,7 @@ docker compose --env-file /dev/null run --rm \
 
 日级 ROI 单独分两阶段启用：
 
-当前 `north_star_roi_t15_v8` 使用版本化传播裂变参数，策略版本为 `roi_policy_v11`，报表版本为 `roi_report_v29`。T0 裂变人数取 `SUM(t0_fission_uv_root)`，不再读取旧字段 `t0裂变人数`。小程序按人群包和转化目标、
+当前 `north_star_roi_t15_v8` 使用版本化传播裂变参数，策略版本为 `roi_policy_v12`，报表版本为 `roi_report_v36`。T0 裂变人数取 `SUM(t0_fission_uv_root)`，不再读取旧字段 `t0裂变人数`。小程序按人群包和转化目标、
 公众号按合作方和公众号匹配传播裂变系数；企微暂不进入本版报表、阈值和调控。首层效率收入读取 `效率收入`，T0 实际裂变收入读取当天的 `裂变效率收入`，
 并以 T0 实际裂变收入乘传播裂变系数预测完整裂变收入。日级任务不会现场重算 cohort 参数。部署前应保持
 `ROI_FISSION_PARAMETER_VERSION=20260712_A0-A15_v2`。参数必须先发布到 MySQL，
@@ -178,13 +180,16 @@ ROI 表格使用获得链接者可编辑权限。默认隐藏的黄色【审批
 审批表包含小程序创意级、小程序广告级和公众号的三日汇总与每日明细共 6 个主 Sheet，广告级两个 Sheet 默认隐藏。
 金额、UV 和人数展示三日总量/3，ROI 和裂变率使用三日汇总后的加权口径。连续三天每天
 首层 UV>200、成本>0 且 ROI 有效的小程序创意和公众号实体合并后按实体等权计算整体 P20；
-广告级直接从 ODPS 原始数据按广告去重，复用 P20 但不进入样本池；低于关停线且广告 age 达标时生成广告级关停建议，批准后暂停整个广告。
-最新日首层 UV>200 但未满足三日条件的实体作为“观察”放在正式样本之后，不进入阈值和腾讯执行。
-可见尾部列按“当日效率ROI → 预测总效率ROI → 关停线（P20） → 扩量线（P80） → 建议动作”排列；公众号不参与扩量，扩量线留空。
-整体三日排名百分位、是否位于后20%、最新日UV、覆盖天数、动作原因、阈值样本状态和执行结果只保留为隐藏审计字段。
+广告级直接从 ODPS 原始数据按广告去重，复用 P20 但不进入样本池；低于关停线且广告 age>3 天时生成广告级关停建议，批准后暂停整个广告。
+未进入三日正式样本的小程序创意增加单日补充判断：最新日首层 UV>200 且预测总效率 ROI≤0.20，或最新日首层 UV>500 且预测总效率 ROI≤全部最新日 UV>500 小程序创意的实体等权 P30；命中且广告 age>3 天时建议关停，age≤3 天时观察。动作原因必须明确启用单日规则的前提，并同时展示最新日预测 ROI 与仅供参考的三日预测 ROI，避免把三日值误认为单日判断值。其余最新日首层 UV>200 但未满足三日条件的实体继续作为“观察”放在正式样本之后，不进入三日阈值。
+可见尾部列按“当日效率ROI → 预测总效率ROI → 关停线（P20） → 扩量线（P80） → 建议动作 → 建议说明”排列；小程序创意级、广告级可见关停动作分别显示为“关停创意”“关停广告”，底层仍保存统一动作值供执行器消费。创意关停按三日P20、单日硬线、单日P30三类建议说明分组排序，同类内部按 ROI 升序。公众号不参与扩量，扩量线留空。底层无动作的正式中间区间在报表中显示为“观察”并说明当前无需关停或扩量，但不生成可执行动作。
+整体三日排名百分位、是否位于后20%、最新日UV、覆盖天数、阈值样本状态和执行结果只保留为隐藏审计字段；底层动作原因通过“建议说明”展示。
 每日明细 Sheet 的 `dt` 位于第一列并按日期倒序。主 Sheet 冻结首行和前 7 列，
-关停线列为“关停线（P20）”。汇总表将三日总 T0 裂变人数 / 三日总首层 UV 的加权比例展示为“日均T0裂变率”。合格小程序创意实体等权 P80 为扩量线，处于头部20%且广告 age≥3 时生成扩量建议。汇总顺序为关停、扩量、其他中间 ROI、观察；第一条扩量行和扩量后第一条无动作行顶部均使用粗线分隔。“当日效率ROI”和“预测总效率ROI”均使用深红—黄—绿色阶，绿色代表表现好。普通数值显示两位小数，UV、人数、数量和广告年龄显示整数；两个“裂变系数-总裂变UV/…”比率列固定显示两位小数。“传播裂变系数匹配”保留审计数据但默认隐藏。
+关停线列为“关停线（P20）”。所有小程序创意级和广告级关停统一要求广告 age>3 天。汇总表将三日总 T0 裂变人数 / 三日总首层 UV 的加权比例展示为“日均T0裂变率”。合格小程序创意实体等权 P80 为扩量线，处于头部20%且广告 age≥3 时生成扩量建议。汇总顺序为关停、扩量、中间观察、条件不足观察；第一条扩量行和扩量后第一条观察行顶部均使用粗线分隔。“当日效率ROI”和“预测总效率ROI”均使用深红—黄—绿色阶，绿色代表表现好。普通数值显示两位小数，UV、人数、数量和广告年龄显示整数；两个“裂变系数-总裂变UV/…”比率列固定显示两位小数。“传播裂变系数匹配”保留审计数据但默认隐藏。
 创意级“当前创意状态”只对关停建议只读腾讯状态并显示正常、已停止或读取失败，其他行留空；“审批选择”默认隐藏。
+完整主报表写入成功后，服务额外按“小程序投流”渠道的“代理名称”生成一代理一份 `roi_agency_advice_v7` 工作簿，文件名为 `YYYYMMDD_代理名称_调控建议.xlsx`。代理版可见 Sheet 名为“小程序创意调控建议”，并保留默认隐藏的“小程序广告级三日汇总”，不包含每日明细；包名、广告age、日均首层UV、建议说明、收入、预测总效率ROI、阈值、排名、两个裂变系数、日均T0裂变人数/率、审批和执行审计列均物理删除。内部“当日效率ROI”仅以两位小数的“评分”列对外展示，不参与代理动作计算；建议动作仍按预测总效率ROI生成，并明确显示“关停创意”或“关停广告”。代理表生成不改变主报表、数据库快照或主审批链接。
+
+总表继续由飞书应用按 `ROI_FEISHU_CHAT_ID` / `FEISHU_OPERATOR_CHAT_ID` 发布。代理表发布由 `ROI_AGENCY_WEBHOOK_ENABLED` 独立控制，完整机器人地址仅允许保存在密钥系统或真实 `.env` 的 `ROI_AGENCY_WEBHOOKS_JSON`，JSON key 使用代理简称。飞书应用先上传并开放在线表链接，再由对应代理机器人发送可点击卡片；未配置代理记录为跳过，不回退到总群。同一 `run_id + 代理 + 代理报表版本` 幂等审计，已上传表格在通知重试时复用，避免重复建表。机器人地址只保存哈希指纹，不能写入数据库或日志。
 
 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/agency_delivery.py

@@ -0,0 +1,237 @@
+"""Upload per-agency ROI workbooks and notify exact Feishu bot routes."""
+
+from __future__ import annotations
+
+import hashlib
+import logging
+from datetime import datetime
+from pathlib import Path
+from typing import Any
+from zoneinfo import ZoneInfo
+
+import httpx
+
+from .config import AgencyWebhookConfig
+from .feishu import RoiFeishuPublisher
+from .repository import upsert_agency_delivery, update_agency_delivery
+
+
+SHANGHAI = ZoneInfo("Asia/Shanghai")
+logger = logging.getLogger("auto_put_ad_mini.roi_control.agency_delivery")
+
+
+def agency_route_key(agency_name: str) -> str:
+    """Resolve the configured short agency key without fuzzy matching."""
+
+    normalized = "".join(str(agency_name).split())
+    for prefix in ("小程序-代投-", "小程序-代投"):
+        if normalized.startswith(prefix):
+            return normalized[len(prefix) :]
+    return normalized
+
+
+def _file_sha256(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as handle:
+        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def _route_fingerprint(url: str) -> str:
+    return hashlib.sha256(url.encode("utf-8")).hexdigest()
+
+
+class AgencyWebhookNotifier:
+    def __init__(self, timeout: float = 20.0) -> None:
+        self.client = httpx.Client(timeout=timeout)
+
+    def close(self) -> None:
+        self.client.close()
+
+    def send(
+        self,
+        webhook_url: str,
+        *,
+        title: str,
+        sheet_url: str,
+        creative_rows: int,
+        ad_rows: int,
+    ) -> str:
+        card = {
+            "config": {"wide_screen_mode": True},
+            "header": {
+                "template": "blue",
+                "title": {"tag": "plain_text", "content": title},
+            },
+            "elements": [
+                {
+                    "tag": "div",
+                    "text": {
+                        "tag": "lark_md",
+                        "content": (
+                            f"创意建议：**{creative_rows}** 条\n"
+                            f"广告建议：**{ad_rows}** 条\n"
+                            "请点击下方按钮查看调控建议。"
+                        ),
+                    },
+                },
+                {
+                    "tag": "action",
+                    "actions": [
+                        {
+                            "tag": "button",
+                            "type": "primary",
+                            "text": {"tag": "plain_text", "content": "查看调控建议"},
+                            "url": sheet_url,
+                        }
+                    ],
+                },
+            ],
+        }
+        response = self.client.post(
+            webhook_url,
+            json={"msg_type": "interactive", "card": card},
+        )
+        response.raise_for_status()
+        payload = response.json()
+        code = payload.get("code", payload.get("StatusCode"))
+        if code != 0:
+            message = payload.get("msg", payload.get("StatusMessage", "unknown"))
+            raise RuntimeError(f"Feishu agency webhook rejected card: {message}")
+        return str(code)
+
+
+def publish_agency_reports(
+    *,
+    run_id: str,
+    reports: list[dict[str, object]],
+    config: AgencyWebhookConfig,
+    publisher: RoiFeishuPublisher,
+    notifier: AgencyWebhookNotifier | None = None,
+    now: datetime | None = None,
+) -> list[dict[str, object]]:
+    """Publish configured agency reports independently and return safe outcomes."""
+
+    if not config.enabled:
+        return []
+    routes = config.webhooks or {}
+    owned_notifier = notifier is None
+    sender = notifier or AgencyWebhookNotifier()
+    outcomes: list[dict[str, object]] = []
+    try:
+        for report in reports:
+            agency_name = str(report["agency_name"])
+            route_key = agency_route_key(agency_name)
+            webhook_url = routes.get(route_key)
+            path = Path(str(report["report"]))
+            delivery_id: int | None = None
+            try:
+                if not path.is_file():
+                    raise FileNotFoundError(path)
+                delivery = upsert_agency_delivery(
+                    {
+                        "run_id": run_id,
+                        "agency_name": agency_name,
+                        "agency_report_version": str(report["report_version"]),
+                        "file_path": str(path),
+                        "file_sha256": _file_sha256(path),
+                        "creative_rows": int(report.get("creative_rows") or 0),
+                        "ad_rows": int(report.get("ad_rows") or 0),
+                        "route_fingerprint": (
+                            _route_fingerprint(webhook_url) if webhook_url else None
+                        ),
+                    }
+                )
+                delivery_id = int(delivery["id"])
+                if delivery.get("status") == "SENT":
+                    outcomes.append(
+                        {
+                            "agency_name": agency_name,
+                            "status": "SENT",
+                            "reused": True,
+                        }
+                    )
+                    continue
+                if not webhook_url:
+                    update_agency_delivery(
+                        delivery_id,
+                        status="SKIPPED",
+                        error_message="未配置代理机器人",
+                    )
+                    outcomes.append(
+                        {
+                            "agency_name": agency_name,
+                            "status": "SKIPPED",
+                            "reason": "未配置代理机器人",
+                        }
+                    )
+                    continue
+                sheet_url = str(delivery.get("sheet_url") or "")
+                sheet_token = str(delivery.get("sheet_token") or "")
+                if not sheet_url or not sheet_token:
+                    imported = publisher.upload_workbook(path)
+                    sheet_url = imported["url"]
+                    sheet_token = imported["sheet_token"]
+                    update_agency_delivery(
+                        delivery_id,
+                        status="UPLOADED",
+                        sheet_token=sheet_token,
+                        sheet_url=sheet_url,
+                    )
+                response_code = sender.send(
+                    webhook_url,
+                    title=path.stem,
+                    sheet_url=sheet_url,
+                    creative_rows=int(report.get("creative_rows") or 0),
+                    ad_rows=int(report.get("ad_rows") or 0),
+                )
+                update_agency_delivery(
+                    delivery_id,
+                    status="SENT",
+                    response_code=response_code,
+                    sent_at=now or datetime.now(SHANGHAI),
+                    increment_attempt=True,
+                )
+                outcomes.append(
+                    {
+                        "agency_name": agency_name,
+                        "status": "SENT",
+                        "sheet_url": sheet_url,
+                        "reused": False,
+                    }
+                )
+            except Exception as exc:
+                safe_error = str(exc)
+                if webhook_url:
+                    safe_error = safe_error.replace(webhook_url, "<redacted>")
+                if delivery_id is not None:
+                    try:
+                        update_agency_delivery(
+                            delivery_id,
+                            status="FAILED",
+                            error_message=safe_error,
+                            increment_attempt=True,
+                        )
+                    except Exception as audit_exc:
+                        logger.error(
+                            "Agency ROI delivery audit failed agency=%s: %s",
+                            agency_name,
+                            str(audit_exc),
+                        )
+                logger.error(
+                    "Agency ROI delivery failed agency=%s: %s",
+                    agency_name,
+                    safe_error,
+                )
+                outcomes.append(
+                    {
+                        "agency_name": agency_name,
+                        "status": "FAILED",
+                        "error": safe_error,
+                    }
+                )
+    finally:
+        if owned_notifier:
+            sender.close()
+    return outcomes

examples/auto_put_ad_mini/roi_control/config.py

@@ -2,9 +2,13 @@
 
 from __future__ import annotations
 
+import hashlib
+import json
+import math
 import os
 from dataclasses import asdict, dataclass
 from decimal import Decimal
+from typing import Mapping
 
 
 def env_flag(name: str, default: bool = False) -> bool:
@@ -14,6 +18,57 @@ def env_flag(name: str, default: bool = False) -> bool:
     return raw.strip().lower() in {"1", "true", "yes", "on"}
 
 
+AGENCY_WEBHOOK_PREFIX = "https://open.feishu.cn/open-apis/bot/v2/hook/"
+
+
+@dataclass(frozen=True)
+class AgencyWebhookConfig:
+    """Secret-backed routing for per-agency ROI report notifications."""
+
+    enabled: bool = False
+    webhooks: Mapping[str, str] | None = None
+
+    @classmethod
+    def from_env(cls) -> "AgencyWebhookConfig":
+        enabled = env_flag("ROI_AGENCY_WEBHOOK_ENABLED")
+        if not enabled:
+            return cls(enabled=False, webhooks={})
+        raw = os.getenv("ROI_AGENCY_WEBHOOKS_JSON", "").strip()
+        if not raw:
+            raise ValueError(
+                "ROI_AGENCY_WEBHOOK_ENABLED=1 requires ROI_AGENCY_WEBHOOKS_JSON"
+            )
+        try:
+            values = json.loads(raw)
+        except json.JSONDecodeError as exc:
+            raise ValueError("ROI_AGENCY_WEBHOOKS_JSON must be valid JSON") from exc
+        if not isinstance(values, dict):
+            raise ValueError("ROI_AGENCY_WEBHOOKS_JSON must be a JSON object")
+        webhooks: dict[str, str] = {}
+        for raw_name, raw_url in values.items():
+            name = str(raw_name).strip()
+            url = str(raw_url).strip()
+            if not name:
+                raise ValueError("ROI agency webhook name cannot be empty")
+            if not url.startswith(AGENCY_WEBHOOK_PREFIX):
+                raise ValueError(f"Invalid Feishu webhook for agency: {name}")
+            if name in webhooks:
+                raise ValueError(f"Duplicate ROI agency webhook: {name}")
+            webhooks[name] = url
+        return cls(enabled=True, webhooks=webhooks)
+
+    def snapshot(self) -> dict[str, object]:
+        routes = self.webhooks or {}
+        return {
+            "enabled": self.enabled,
+            "agencies": sorted(routes),
+            "route_fingerprints": {
+                name: hashlib.sha256(url.encode("utf-8")).hexdigest()[:12]
+                for name, url in sorted(routes.items())
+            },
+        }
+
+
 @dataclass(frozen=True)
 class RoiConfig:
     daily_enabled: bool = False
@@ -26,11 +81,15 @@ class RoiConfig:
     scale_ratio: Decimal = Decimal("1.10")
     scale_cooldown_days: int = 3
     max_base_ratio: Decimal = Decimal("2.00")
-    self_stop_min_age: int = 5
+    self_stop_min_age: int = 4
     self_up_min_age: int = 3
     self_min_daily_uv: float = 200
     partner_min_daily_uv: float = 200
     observe_min_latest_uv: float = 200
+    one_day_min_uv: float = 200
+    one_day_p30_min_uv: float = 500
+    one_day_hard_stop_roi: float = 0.20
+    one_day_stop_quantile: float = 0.30
     stop_quantile: float = 0.20
     up_quantile: float = 0.80
     stop_weight_cap_quantile: float = 0.95
@@ -52,7 +111,7 @@ class RoiConfig:
             scale_ratio=Decimal(os.getenv("ROI_SCALE_RATIO", "1.10")),
             scale_cooldown_days=int(os.getenv("ROI_SCALE_COOLDOWN_DAYS", "3")),
             max_base_ratio=Decimal(os.getenv("ROI_MAX_BASE_RATIO", "2.00")),
-            self_stop_min_age=int(os.getenv("ROI_SELF_STOP_MIN_AGE", "5")),
+            self_stop_min_age=int(os.getenv("ROI_SELF_STOP_MIN_AGE", "4")),
             self_up_min_age=int(os.getenv("ROI_SELF_UP_MIN_AGE", "3")),
             self_min_daily_uv=float(
                 os.getenv(
@@ -69,6 +128,16 @@ class RoiConfig:
             observe_min_latest_uv=float(
                 os.getenv("ROI_OBSERVE_MIN_LATEST_UV", "200")
             ),
+            one_day_min_uv=float(os.getenv("ROI_ONE_DAY_MIN_UV", "200")),
+            one_day_p30_min_uv=float(
+                os.getenv("ROI_ONE_DAY_P30_MIN_UV", "500")
+            ),
+            one_day_hard_stop_roi=float(
+                os.getenv("ROI_ONE_DAY_HARD_STOP_ROI", "0.20")
+            ),
+            one_day_stop_quantile=float(
+                os.getenv("ROI_ONE_DAY_STOP_QUANTILE", "0.30")
+            ),
             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(
@@ -103,8 +172,20 @@ class RoiConfig:
             self.self_min_daily_uv,
             self.partner_min_daily_uv,
             self.observe_min_latest_uv,
+            self.one_day_min_uv,
+            self.one_day_p30_min_uv,
         ) <= 0:
             raise ValueError("ROI UV thresholds must be positive")
+        if self.one_day_p30_min_uv <= self.one_day_min_uv:
+            raise ValueError(
+                "ROI_ONE_DAY_P30_MIN_UV must be greater than ROI_ONE_DAY_MIN_UV"
+            )
+        if not math.isfinite(self.one_day_hard_stop_roi):
+            raise ValueError("ROI_ONE_DAY_HARD_STOP_ROI must be finite")
+        if not 0 < self.one_day_stop_quantile < 1:
+            raise ValueError(
+                "ROI_ONE_DAY_STOP_QUANTILE must satisfy 0 < value < 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:

examples/auto_put_ad_mini/roi_control/feishu.py

@@ -263,31 +263,40 @@ class RoiFeishuPublisher:
         summary: str,
         requires_approval: bool,
     ) -> dict[str, str]:
-        if not path.is_file():
-            raise FileNotFoundError(path)
+        imported = self.upload_workbook(path)
         token = self._token()
-        file_token = self._upload(token, path)
-        ticket = self._import_sheet(token, file_token, path)
-        result = self._wait_import(token, ticket)
-        url = str(result.get("url") or "")
-        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_editable_link(token, sheet_token)
         message_ids = [
             self._send_card(
                 token,
                 run_id=run_id,
                 batch_name=batch_name,
                 chat_id=chat_id,
-                url=url,
+                url=imported["url"],
                 summary=summary,
                 requires_approval=requires_approval,
             )
             for chat_id in self.chat_ids
         ]
+        return {
+            **imported,
+            "message_id": message_ids[0],
+        }
+
+    def upload_workbook(self, path: Path) -> dict[str, str]:
+        """Upload one workbook as an editable online sheet without notifying chats."""
+
+        if not path.is_file():
+            raise FileNotFoundError(path)
+        token = self._token()
+        file_token = self._upload(token, path)
+        ticket = self._import_sheet(token, file_token, path)
+        result = self._wait_import(token, ticket)
+        url = str(result.get("url") or "")
+        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_editable_link(token, sheet_token)
         return {
             "url": url,
             "sheet_token": sheet_token,
-            "message_id": message_ids[0],
         }

examples/auto_put_ad_mini/roi_control/reporting.py

@@ -28,8 +28,9 @@ OBSERVE_FILL = PatternFill("solid", fgColor="FFF2CC")
 APPROVAL_FILL = PatternFill("solid", fgColor="FFD966")
 APPROVAL_HEADER_FILL = PatternFill("solid", fgColor="BF9000")
 
-REPORT_VERSION = "roi_report_v29"
-REPORT_RUN_SUFFIX = "r29"
+REPORT_VERSION = "roi_report_v36"
+REPORT_RUN_SUFFIX = "r36"
+AGENCY_REPORT_VERSION = "roi_agency_advice_v7"
 
 T0_FISSION_MULTIPLIER_COLUMN = "裂变系数-总裂变UV/T0裂变UV"
 TOTAL_FISSION_TO_FIRST_UV_COLUMN = DISPLAY_TOTAL_TO_FIRST_COLUMN
@@ -40,6 +41,10 @@ SUMMARY_SHEETS = {
     ENTITY_SELF_AD: "小程序广告级三日汇总",
     ENTITY_GZH: "公众号三日汇总",
 }
+AGENCY_SUMMARY_SHEETS = {
+    ENTITY_SELF: "小程序创意调控建议",
+    ENTITY_SELF_AD: SUMMARY_SHEETS[ENTITY_SELF_AD],
+}
 DAILY_SHEETS = {
     ENTITY_SELF: "小程序创意级每日明细",
     ENTITY_SELF_AD: "小程序广告级每日明细",
@@ -76,6 +81,27 @@ ENTITY_DIMENSIONS = {
     ),
     ENTITY_GZH: ("渠道", "合作方名", "公众号名"),
 }
+AGENCY_ENTITY_DIMENSIONS = {
+    ENTITY_SELF: (
+        "渠道",
+        "代理名称",
+        "账号id",
+        "账号名称",
+        "广告id",
+        "广告名称",
+        "广告优化目标",
+        "创意id",
+    ),
+    ENTITY_SELF_AD: (
+        "渠道",
+        "代理名称",
+        "账号id",
+        "账号名称",
+        "广告id",
+        "广告名称",
+        "广告优化目标",
+    ),
+}
 
 SUMMARY_METRICS = (
     "日均首层UV",
@@ -92,6 +118,7 @@ SUMMARY_METRICS = (
     "关停线（P20）",
     "扩量线（P80）",
     "建议动作",
+    "建议说明",
 )
 
 DAILY_METRICS = (
@@ -108,6 +135,12 @@ DAILY_METRICS = (
     TOTAL_FISSION_TO_FIRST_UV_COLUMN,
 )
 
+AGENCY_SUMMARY_METRICS = (
+    "日均成本",
+    "评分",
+    "建议动作",
+)
+
 
 def _visible_columns(
     sheet_name: str,
@@ -136,6 +169,27 @@ def _ensure_columns(frame: pd.DataFrame, columns: Sequence[str]) -> pd.DataFrame
     return result
 
 
+def _canonical_agency_name(value: object) -> str:
+    if pd.isna(value):
+        return ""
+    return re.sub(r"\s*-\s*", "-", str(value).strip())
+
+
+def _safe_filename_component(value: str) -> str:
+    return re.sub(r'[\\/:*?"<>|]', "_", value).strip(" .") or "未命名代理"
+
+
+def _agency_summary_frame(rows: pd.DataFrame, entity_type: str) -> pd.DataFrame:
+    frame = _summary_frame(rows, entity_type)
+    frame["评分"] = pd.to_numeric(frame["当日效率ROI"], errors="coerce")
+    columns = list(AGENCY_ENTITY_DIMENSIONS[entity_type]) + list(
+        AGENCY_SUMMARY_METRICS
+    )
+    if entity_type == ENTITY_SELF:
+        columns.append("当前创意状态")
+    return _ensure_columns(frame, columns)[columns].copy()
+
+
 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": "渠道"})
@@ -178,6 +232,15 @@ def _summary_frame(rows: pd.DataFrame, entity_type: str) -> pd.DataFrame:
         else np.nan
     )
     subset["建议动作"] = subset.get("动作", "")
+    if entity_type == ENTITY_SELF:
+        subset["建议动作"] = subset["建议动作"].replace(
+            {"关停": "关停创意"}
+        )
+    elif entity_type == ENTITY_SELF_AD:
+        subset["建议动作"] = subset["建议动作"].replace(
+            {"关停": "关停广告"}
+        )
+    subset["建议说明"] = subset.get("动作原因", "")
     visible = _visible_columns(SUMMARY_SHEETS[entity_type])
     required = list(visible)
     if entity_type in (ENTITY_SELF, ENTITY_SELF_AD):
@@ -191,10 +254,29 @@ def _summary_frame(rows: pd.DataFrame, entity_type: str) -> pd.DataFrame:
         subset["_动作排序"] = (
             subset["建议动作"]
             .fillna("")
-            .map({"关停": 0, "扩量": 1, "": 2, "观察": 3})
+            .map(
+                {
+                    "关停": 0,
+                    "关停创意": 0,
+                    "关停广告": 0,
+                    "扩量": 1,
+                    "": 2,
+                    "观察": 3,
+                }
+            )
             .fillna(4)
             .astype(int)
         )
+        reason = subset["建议说明"].fillna("").astype(str)
+        subset["_说明排序"] = np.select(
+            [
+                reason.str.contains("三日加权平均效率ROI≤", regex=False),
+                reason.str.contains("单日硬关停线", regex=False),
+                reason.str.contains("单日实体等权P30", regex=False),
+            ],
+            [0, 1, 2],
+            default=3,
+        )
         roi_sort = pd.to_numeric(
             subset["三日加权平均效率ROI"], errors="coerce"
         ).fillna(np.inf)
@@ -211,10 +293,38 @@ def _summary_frame(rows: pd.DataFrame, entity_type: str) -> pd.DataFrame:
         subset.loc[observation, "_ROI排序"] = np.inf
         subset.loc[observation, "_成本排序"] = 0
         subset = subset.sort_values(
-            ["_观察排序", "_动作排序", "_ROI排序", "_成本排序", "_UV排序"],
-            ascending=[True, True, True, False, False],
+            [
+                "_观察排序",
+                "_动作排序",
+                "_说明排序",
+                "_ROI排序",
+                "_成本排序",
+                "_UV排序",
+            ],
+            ascending=[True, True, True, True, False, False],
             kind="stable",
-        ).drop(columns=["_观察排序", "_动作排序", "_ROI排序", "_成本排序", "_UV排序"])
+        ).drop(
+            columns=[
+                "_观察排序",
+                "_动作排序",
+                "_说明排序",
+                "_ROI排序",
+                "_成本排序",
+                "_UV排序",
+            ]
+        )
+    neutral = subset["建议动作"].fillna("").eq("")
+    subset.loc[neutral, "建议动作"] = "观察"
+    if entity_type == ENTITY_SELF:
+        neutral_reason = (
+            "预测总效率ROI位于关停线（P20）与扩量线（P80）之间，"
+            "当前无需关停或扩量"
+        )
+    elif entity_type == ENTITY_SELF_AD:
+        neutral_reason = "预测总效率ROI高于关停线（P20），当前无需关停整个广告"
+    else:
+        neutral_reason = "预测总效率ROI高于关停线（P20），当前仅观察"
+    subset.loc[neutral, "建议说明"] = neutral_reason
     hidden = [column for column in subset.columns if column not in visible]
     return subset[visible + hidden]
 
@@ -377,16 +487,16 @@ def _format_sheet(
         )
         if first_scale_row:
             add_top_separator(first_scale_row)
-            first_no_action_row = next(
+            first_observe_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 "")
+                    if ws.cell(row_number, action_column).value == "观察"
                 ),
                 None,
             )
-            if first_no_action_row:
-                add_top_separator(first_no_action_row)
+            if first_observe_row:
+                add_top_separator(first_observe_row)
 
     for roi_header in ("当日效率ROI", "预测总效率ROI"):
         roi_column = headers.get(roi_header)
@@ -437,11 +547,15 @@ def _write_summary_sheet(
         ("阈值样本数", threshold.get("阈值样本数")),
         ("创意扩量线（P80）", threshold.get("t_up")),
         ("扩量样本数", threshold.get("扩量样本数")),
+        ("单日关停线（P30）", threshold.get("t_one_day_stop")),
+        ("单日P30样本数", threshold.get("单日P30样本数")),
         ("阈值样本", "连续三天每天首层UV>200、成本>0且ROI有效的小程序创意和公众号实体"),
-        ("广告级", "直接按广告去重计算，复用统一P20但不进入样本池；低于关停线且广告age达标时，审批后暂停整个广告"),
+        ("关停年龄门槛", "所有小程序创意级和广告级关停均要求广告age>3天"),
+        ("广告级", "直接按广告去重计算，复用统一P20但不进入样本池；低于关停线且广告age>3天时，审批后暂停整个广告"),
         ("日均字段", "三日总量/3，缺失日按0"),
         ("ROI与裂变率", "三日汇总分子/三日汇总分母的加权口径"),
-        ("补充观察", "非正式样本中最新日首层UV>200，置于汇总表末尾且不执行"),
+        ("单日补充规则", "非三日正式样本的小程序创意：最新日UV>200且ROI≤0.20，或UV>500且ROI≤实体等权P30；广告age>3天时建议关停，否则观察"),
+        ("补充观察", "其余非正式样本中最新日首层UV>200，置于汇总表末尾且不执行"),
         ("配置快照", json.dumps(dict(config), ensure_ascii=False, default=str)),
     ]
     for row in rows:
@@ -507,3 +621,54 @@ def write_workbook(
     _write_summary_sheet(workbook, thresholds, expected_dates, config)
     output_path.parent.mkdir(parents=True, exist_ok=True)
     workbook.save(output_path)
+
+
+def write_agency_workbooks(
+    rows: pd.DataFrame,
+    output_dir: Path,
+    report_date: str,
+) -> list[dict[str, object]]:
+    """Create one miniapp control-advice workbook per agency."""
+
+    miniapp = rows[
+        rows["entity_type"].isin([ENTITY_SELF, ENTITY_SELF_AD])
+        & rows["channel"].astype(str).str.startswith("小程序投流")
+    ].copy()
+    miniapp["_代理规范名"] = miniapp["代理名称"].map(_canonical_agency_name)
+    agencies = sorted(name for name in miniapp["_代理规范名"].unique() if name)
+    output_dir.mkdir(parents=True, exist_ok=True)
+    outputs: list[dict[str, object]] = []
+
+    for agency_name in agencies:
+        agency_rows = miniapp[miniapp["_代理规范名"].eq(agency_name)].drop(
+            columns=["_代理规范名"]
+        )
+        workbook = Workbook()
+        workbook.remove(workbook.active)
+
+        for entity_type in (ENTITY_SELF, ENTITY_SELF_AD):
+            summary_name = AGENCY_SUMMARY_SHEETS[entity_type]
+            summary = _agency_summary_frame(agency_rows, entity_type)
+            summary_ws = workbook.create_sheet(summary_name)
+            _write_dataframe(summary_ws, summary)
+            _format_sheet(summary_ws, list(summary.columns), freeze_panes="H2")
+            if entity_type == ENTITY_SELF_AD:
+                summary_ws.sheet_state = "hidden"
+
+        filename = f"{report_date}_{_safe_filename_component(agency_name)}_调控建议.xlsx"
+        output_path = output_dir / filename
+        workbook.save(output_path)
+        outputs.append(
+            {
+                "agency_name": agency_name,
+                "report_version": AGENCY_REPORT_VERSION,
+                "report": str(output_path),
+                "creative_rows": int(
+                    agency_rows["entity_type"].eq(ENTITY_SELF).sum()
+                ),
+                "ad_rows": int(
+                    agency_rows["entity_type"].eq(ENTITY_SELF_AD).sum()
+                ),
+            }
+        )
+    return outputs

examples/auto_put_ad_mini/roi_control/repository.py

@@ -354,6 +354,125 @@ def mark_failed(run_id: str, error_message: str) -> None:
         connection.close()
 
 
+def upsert_agency_delivery(record: dict[str, Any]) -> dict[str, Any]:
+    """Create or refresh one idempotent agency delivery record."""
+
+    connection = connect()
+    try:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                """
+                INSERT INTO roi_agency_delivery
+                    (run_id, agency_name, agency_report_version, file_path,
+                     file_sha256, creative_rows, ad_rows,
+                     route_fingerprint, status)
+                VALUES (%s,%s,%s,%s,%s,%s,%s,%s,'PENDING')
+                ON DUPLICATE KEY UPDATE
+                    status=CASE
+                        WHEN status='SENT' THEN status
+                        WHEN NOT (route_fingerprint <=> VALUES(route_fingerprint))
+                            THEN 'PENDING'
+                        ELSE status
+                    END,
+                    error_message=CASE
+                        WHEN NOT (route_fingerprint <=> VALUES(route_fingerprint))
+                            THEN NULL
+                        ELSE error_message
+                    END,
+                    file_path=IF(status='SENT', file_path, VALUES(file_path)),
+                    file_sha256=IF(status='SENT', file_sha256, VALUES(file_sha256)),
+                    creative_rows=VALUES(creative_rows),
+                    ad_rows=VALUES(ad_rows),
+                    route_fingerprint=VALUES(route_fingerprint)
+                """,
+                (
+                    record["run_id"],
+                    record["agency_name"],
+                    record["agency_report_version"],
+                    record["file_path"],
+                    record["file_sha256"],
+                    int(record.get("creative_rows") or 0),
+                    int(record.get("ad_rows") or 0),
+                    record.get("route_fingerprint"),
+                ),
+            )
+            cursor.execute(
+                """
+                SELECT * FROM roi_agency_delivery
+                WHERE run_id=%s AND agency_name=%s AND agency_report_version=%s
+                """,
+                (
+                    record["run_id"],
+                    record["agency_name"],
+                    record["agency_report_version"],
+                ),
+            )
+            return cursor.fetchone() or {}
+    finally:
+        connection.close()
+
+
+def load_agency_deliveries(run_id: str) -> list[dict[str, Any]]:
+    connection = connect()
+    try:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                """
+                SELECT * FROM roi_agency_delivery
+                WHERE run_id=%s
+                ORDER BY agency_name
+                """,
+                (run_id,),
+            )
+            return list(cursor.fetchall())
+    finally:
+        connection.close()
+
+
+def update_agency_delivery(
+    delivery_id: int,
+    *,
+    status: str,
+    sheet_token: str | None = None,
+    sheet_url: str | None = None,
+    response_code: str | None = None,
+    error_message: str | None = None,
+    sent_at: datetime | None = None,
+    increment_attempt: bool = False,
+) -> None:
+    allowed_statuses = {"PENDING", "UPLOADED", "SENT", "SKIPPED", "FAILED"}
+    if status not in allowed_statuses:
+        raise ValueError(f"Unsupported agency delivery status: {status}")
+    connection = connect()
+    try:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                """
+                UPDATE roi_agency_delivery
+                SET status=%s,
+                    sheet_token=COALESCE(%s, sheet_token),
+                    sheet_url=COALESCE(%s, sheet_url),
+                    response_code=%s,
+                    error_message=%s,
+                    sent_at=COALESCE(%s, sent_at),
+                    attempt_count=attempt_count+%s
+                WHERE id=%s
+                """,
+                (
+                    status,
+                    sheet_token,
+                    sheet_url,
+                    response_code,
+                    error_message[:4000] if error_message else None,
+                    sent_at,
+                    1 if increment_attempt else 0,
+                    delivery_id,
+                ),
+            )
+    finally:
+        connection.close()
+
+
 def reject_run(run_id: str, *, sender_open_id: str, now: datetime) -> dict[str, Any]:
     connection = connect()
     connection.autocommit(False)

examples/auto_put_ad_mini/roi_control/rules.py

@@ -1,4 +1,4 @@
-"""Versioned three-day P20 stop and creative-level P80 scale policy."""
+"""Versioned three-day policy with a latest-day creative stop supplement."""
 
 from __future__ import annotations
 
@@ -17,17 +17,21 @@ from .metrics import (
 )
 
 
-POLICY_VERSION = "roi_policy_v11"
-POLICY_RUN_SUFFIX = "p11"
+POLICY_VERSION = "roi_policy_v12"
+POLICY_RUN_SUFFIX = "p12"
 
 
 @dataclass(frozen=True)
 class RuleConfig:
-    self_stop_min_age: int = 5
+    self_stop_min_age: int = 4
     self_up_min_age: int = 3
     self_min_daily_uv: float = 200
     partner_min_daily_uv: float = 200
     observe_min_latest_uv: float = 200
+    one_day_min_uv: float = 200
+    one_day_p30_min_uv: float = 500
+    one_day_hard_stop_roi: float = 0.20
+    one_day_stop_quantile: float = 0.30
     stop_quantile: float = 0.20
     up_quantile: float = 0.80
 
@@ -101,6 +105,44 @@ def observation_mask(
     )
 
 
+def one_day_p30_pool_mask(
+    summary: pd.DataFrame,
+    config: RuleConfig,
+    expected_dates: list[str],
+) -> pd.Series:
+    latest = expected_dates[-1]
+    latest_roi = pd.to_numeric(summary[f"ROI_{latest}"], errors="coerce")
+    return (
+        summary["entity_type"].eq(ENTITY_SELF)
+        & pd.to_numeric(summary[f"首层UV_{latest}"], errors="coerce").gt(
+            config.one_day_p30_min_uv
+        )
+        & pd.to_numeric(summary[f"成本_{latest}"], errors="coerce").gt(0)
+        & np.isfinite(latest_roi)
+    )
+
+
+def one_day_supplement_mask(
+    summary: pd.DataFrame,
+    config: RuleConfig,
+    expected_dates: list[str],
+) -> pd.Series:
+    latest = expected_dates[-1]
+    latest_roi = pd.to_numeric(summary[f"ROI_{latest}"], errors="coerce")
+    formal_creative = entity_eligibility_mask(
+        summary, ENTITY_SELF, config, expected_dates
+    )
+    return (
+        summary["entity_type"].eq(ENTITY_SELF)
+        & ~formal_creative
+        & pd.to_numeric(summary[f"首层UV_{latest}"], errors="coerce").gt(
+            config.one_day_min_uv
+        )
+        & pd.to_numeric(summary[f"成本_{latest}"], errors="coerce").gt(0)
+        & np.isfinite(latest_roi)
+    )
+
+
 def compute_global_threshold(
     summary: pd.DataFrame,
     expected_dates: list[str],
@@ -123,6 +165,20 @@ def compute_global_threshold(
         if not creative_sample.empty
         else np.nan
     )
+    latest = expected_dates[-1]
+    one_day_pool = pd.to_numeric(
+        summary.loc[
+            one_day_p30_pool_mask(summary, config, expected_dates),
+            f"ROI_{latest}",
+        ],
+        errors="coerce",
+    )
+    one_day_pool = one_day_pool[np.isfinite(one_day_pool)]
+    t_one_day_stop = (
+        float(one_day_pool.quantile(config.one_day_stop_quantile))
+        if not one_day_pool.empty
+        else np.nan
+    )
     return pd.DataFrame(
         [
             {
@@ -131,8 +187,13 @@ def compute_global_threshold(
                 "渠道": "小程序创意级+公众号",
                 "t_stop": float(sample.quantile(config.stop_quantile)),
                 "t_up": t_up,
+                "t_one_day_stop": t_one_day_stop,
                 "阈值样本数": int(len(sample)),
                 "扩量样本数": int(len(creative_sample)),
+                "单日P30样本数": int(len(one_day_pool)),
+                "单日硬关停ROI": config.one_day_hard_stop_roi,
+                "单日最低UV": config.one_day_min_uv,
+                "单日P30最低UV": config.one_day_p30_min_uv,
                 "关停线口径": f"合格实体等权P{int(config.stop_quantile * 100)}",
                 "扩量线口径": f"合格小程序创意实体等权P{int(config.up_quantile * 100)}",
                 "广告级是否入池": "否_仅复用统一关停线",
@@ -160,12 +221,18 @@ def apply_actions(
     result["动作原因"] = ""
     result["t_stop"] = float(thresholds.iloc[0]["t_stop"])
     result["t_up"] = float(thresholds.iloc[0]["t_up"])
+    result["t_one_day_stop"] = float(
+        thresholds.iloc[0]["t_one_day_stop"]
+    )
 
     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)
+    one_day_supplement = one_day_supplement_mask(
+        result, config, expected_dates
+    )
     sample_roi = result.loc[threshold_eligible, "ROI"]
     creative_eligible = entity_eligibility_mask(
         result, ENTITY_SELF, config, expected_dates
@@ -196,6 +263,61 @@ def apply_actions(
 
     for index, row in result.iterrows():
         entity_type = str(row["entity_type"])
+        if bool(one_day_supplement.loc[index]):
+            latest = expected_dates[-1]
+            latest_uv = float(row[f"首层UV_{latest}"])
+            latest_roi = float(row[f"ROI_{latest}"])
+            three_day_roi = float(row["ROI"])
+            raw_age = row.get("广告age")
+            age = int(raw_age) if pd.notna(raw_age) else 0
+            decision_context = (
+                "未满足连续三天每天首层UV>200、成本>0且ROI有效，"
+                "启用单日补充规则；"
+            )
+            three_day_context = (
+                f"；三日预测总效率ROI={three_day_roi:.2f}仅作参考，"
+                "不参与本次单日判断"
+            )
+            stop_reason = ""
+            if latest_roi <= config.one_day_hard_stop_roi:
+                stop_reason = (
+                    f"最新日首层UV={latest_uv:.0f}>{config.one_day_min_uv:g}，"
+                    f"最新日预测总效率ROI={latest_roi:.2f}≤单日硬关停线"
+                    f"{config.one_day_hard_stop_roi:.2f}"
+                )
+            elif (
+                latest_uv > config.one_day_p30_min_uv
+                and np.isfinite(float(row["t_one_day_stop"]))
+                and latest_roi <= float(row["t_one_day_stop"])
+            ):
+                stop_reason = (
+                    f"最新日首层UV={latest_uv:.0f}>{config.one_day_p30_min_uv:g}，"
+                    f"最新日预测总效率ROI={latest_roi:.2f}≤单日实体等权P30关停线"
+                    f"{float(row['t_one_day_stop']):.2f}"
+                )
+
+            if stop_reason and age >= config.self_stop_min_age:
+                result.at[index, "动作"] = "关停"
+                result.at[index, "动作原因"] = (
+                    f"{decision_context}{stop_reason}{three_day_context}；"
+                    f"广告age={age}>{config.self_stop_min_age - 1}天，"
+                    "建议审批后暂停动态创意"
+                )
+            elif stop_reason:
+                result.at[index, "动作"] = "观察"
+                result.at[index, "动作原因"] = (
+                    f"{decision_context}{stop_reason}{three_day_context}；"
+                    f"广告age={age}≤{config.self_stop_min_age - 1}天，暂不关停"
+                )
+            else:
+                result.at[index, "动作"] = "观察"
+                result.at[index, "动作原因"] = (
+                    f"{decision_context}"
+                    f"最新日首层UV={latest_uv:.0f}>{config.one_day_min_uv:g}，"
+                    f"最新日预测总效率ROI={latest_roi:.2f}未命中单日关停规则"
+                    f"{three_day_context}；建议继续观察"
+                )
+            continue
         if bool(observe_only.loc[index]):
             result.at[index, "动作"] = "观察"
             result.at[index, "动作原因"] = (
@@ -211,13 +333,13 @@ def apply_actions(
                     result.at[index, "动作"] = "关停"
                     result.at[index, "动作原因"] = (
                         f"广告级三日加权平均效率ROI≤统一实体等权P20，"
-                        f"广告age≥{config.self_stop_min_age}天；审批后暂停整个广告"
+                        f"广告age>{config.self_stop_min_age - 1}天；审批后暂停整个广告"
                     )
                 else:
                     result.at[index, "动作"] = "观察"
                     result.at[index, "动作原因"] = (
-                        f"广告级三日加权平均效率ROI≤统一实体等权P20，但广告age<"
-                        f"{config.self_stop_min_age}天"
+                        f"广告级三日加权平均效率ROI≤统一实体等权P20，但广告age≤"
+                        f"{config.self_stop_min_age - 1}天"
                     )
             continue
         if not bool(threshold_eligible.loc[index]):
@@ -230,13 +352,13 @@ def apply_actions(
                     result.at[index, "动作"] = "关停"
                     result.at[index, "动作原因"] = (
                         f"三日加权平均效率ROI≤统一实体等权P20，"
-                        f"广告age≥{config.self_stop_min_age}天；审批后仅暂停动态创意"
+                        f"广告age>{config.self_stop_min_age - 1}天；审批后仅暂停动态创意"
                     )
                 else:
                     result.at[index, "动作"] = "观察"
                     result.at[index, "动作原因"] = (
-                        f"三日加权平均效率ROI≤统一实体等权P20，但广告age<"
-                        f"{config.self_stop_min_age}天"
+                        f"三日加权平均效率ROI≤统一实体等权P20，但广告age≤"
+                        f"{config.self_stop_min_age - 1}天"
                     )
             elif bool(row["是否位于创意三日ROI前20%"]):
                 if age >= config.self_up_min_age:
@@ -277,10 +399,14 @@ def evaluate_rules(
     evaluated = apply_actions(summary, thresholds, 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)
+    one_day_supplement = one_day_supplement_mask(evaluated, config, dates)
+    observe_only = observation_mask(evaluated, config, dates) & ~one_day_supplement
     evaluated["阈值样本状态"] = "未达到三日正式样本门槛"
     evaluated.loc[formal_threshold, "阈值样本状态"] = "进入三日统一阈值样本池"
     evaluated.loc[formal_ad, "阈值样本状态"] = "广告级三日合格_不进入阈值样本池"
+    evaluated.loc[one_day_supplement, "阈值样本状态"] = (
+        "单日补充决策_昨日UV>200"
+    )
     evaluated.loc[observe_only, "阈值样本状态"] = "补充观察_昨日UV>200"
     candidates = evaluated[evaluated["动作"].ne("")].copy()
     return candidates, thresholds, evaluated

examples/auto_put_ad_mini/roi_control/service.py

@@ -14,7 +14,8 @@ from zoneinfo import ZoneInfo
 from storage import initialize_schema, load_managed_accounts
 from tencent_client import ACTIVE_STATUS, TencentClient
 
-from .config import RoiConfig
+from .agency_delivery import publish_agency_reports
+from .config import AgencyWebhookConfig, RoiConfig
 from .data_source import (
     ODPSClient,
     date_window,
@@ -30,7 +31,12 @@ from .fission_multiplier import (
     parameters_from_database,
 )
 from .policy import annotate_execution
-from .reporting import REPORT_RUN_SUFFIX, REPORT_VERSION, write_workbook
+from .reporting import (
+    REPORT_RUN_SUFFIX,
+    REPORT_VERSION,
+    write_agency_workbooks,
+    write_workbook,
+)
 from .rules import (
     POLICY_RUN_SUFFIX,
     POLICY_VERSION,
@@ -40,6 +46,7 @@ from .rules import (
 from .repository import (
     FINAL_STATUSES,
     create_or_load_run,
+    load_agency_deliveries,
     load_fission_parameter_release,
     mark_failed,
     mark_published,
@@ -51,6 +58,16 @@ SHANGHAI = ZoneInfo("Asia/Shanghai")
 logger = logging.getLogger("auto_put_ad_mini.roi_control")
 
 
+def _redact_agency_webhooks(
+    error: Exception,
+    config: AgencyWebhookConfig,
+) -> str:
+    message = str(error)
+    for webhook_url in (config.webhooks or {}).values():
+        message = message.replace(webhook_url, "<redacted>")
+    return message
+
+
 def _annotate_current_creative_status(
     rows,
     tencent: TencentClient | None = None,
@@ -105,6 +122,10 @@ def _rule_config(config: RoiConfig) -> RuleConfig:
         self_min_daily_uv=config.self_min_daily_uv,
         partner_min_daily_uv=config.partner_min_daily_uv,
         observe_min_latest_uv=config.observe_min_latest_uv,
+        one_day_min_uv=config.one_day_min_uv,
+        one_day_p30_min_uv=config.one_day_p30_min_uv,
+        one_day_hard_stop_roi=config.one_day_hard_stop_roi,
+        one_day_stop_quantile=config.one_day_stop_quantile,
         stop_quantile=config.stop_quantile,
         up_quantile=config.up_quantile,
     )
@@ -162,6 +183,7 @@ def run_daily_roi(
     """Compute, snapshot, report, and optionally publish one ROI batch."""
 
     config = RoiConfig.from_env()
+    agency_webhook_config = AgencyWebhookConfig.from_env()
     initialize_schema()
     fission_version = os.getenv(
         "ROI_FISSION_PARAMETER_VERSION",
@@ -173,6 +195,7 @@ def run_daily_roi(
     run_config = config.snapshot()
     run_config["fission_multiplier"] = fission_parameters.snapshot()
     run_config["report_version"] = REPORT_VERSION
+    run_config["agency_webhook"] = agency_webhook_config.snapshot()
     if source_revision:
         run_config["source_revision"] = source_revision
     client = ODPSClient(project=os.getenv("ODPS_PROJECT", "loghubods"))
@@ -204,12 +227,45 @@ def run_daily_roi(
         run.get("status") == "COMPUTED" and not send_feishu
     ):
         logger.info("Reuse ROI run=%s status=%s", run_id, run.get("status"))
-        return {
+        reused_result: dict[str, Any] = {
             "run_id": run_id,
             "status": run.get("status"),
             "reused": True,
             "sheet_url": run.get("sheet_url"),
         }
+        if send_feishu and agency_webhook_config.enabled:
+            stored_deliveries = load_agency_deliveries(run_id)
+            if stored_deliveries:
+                retry_reports = [
+                    {
+                        "agency_name": row["agency_name"],
+                        "report_version": row["agency_report_version"],
+                        "report": row["file_path"],
+                        "creative_rows": row.get("creative_rows") or 0,
+                        "ad_rows": row.get("ad_rows") or 0,
+                    }
+                    for row in stored_deliveries
+                ]
+                publisher = RoiFeishuPublisher()
+                try:
+                    try:
+                        reused_result["agency_deliveries"] = publish_agency_reports(
+                            run_id=run_id,
+                            reports=retry_reports,
+                            config=agency_webhook_config,
+                            publisher=publisher,
+                            now=now,
+                        )
+                    except Exception as exc:
+                        safe_error = _redact_agency_webhooks(
+                            exc,
+                            agency_webhook_config,
+                        )
+                        logger.error("Agency ROI retry failed: %s", safe_error)
+                        reused_result["agency_delivery_error"] = safe_error
+                finally:
+                    publisher.close()
+        return reused_result
 
     try:
         logger.info(
@@ -267,6 +323,7 @@ def run_daily_roi(
                 [
                     "进入三日统一阈值样本池",
                     "广告级三日合格_不进入阈值样本池",
+                    "单日补充决策_昨日UV>200",
                     "补充观察_昨日UV>200",
                 ]
             )
@@ -299,6 +356,12 @@ def run_daily_roi(
             run_config,
             fission_match_summary,
         )
+        agency_report_date = (now or datetime.now(SHANGHAI)).strftime("%Y%m%d")
+        agency_reports = write_agency_workbooks(
+            report_rows,
+            output_dir / f"{agency_report_date}_调控建议",
+            agency_report_date,
+        )
 
         result: dict[str, Any] = {
             "run_id": run_id,
@@ -316,6 +379,7 @@ def run_daily_roi(
                 "records"
             ),
             "report": str(output_path),
+            "agency_reports": agency_reports,
         }
         if not send_feishu:
             return result
@@ -337,22 +401,37 @@ def run_daily_roi(
                 summary=summary_text,
                 requires_approval=bool(actions),
             )
+            expires_at = (
+                (now or datetime.now(SHANGHAI))
+                + timedelta(minutes=config.approval_ttl_minutes)
+                if actions
+                else None
+            )
+            mark_published(
+                run_id,
+                sheet_token=published["sheet_token"],
+                sheet_url=published["url"],
+                message_id=published["message_id"],
+                expires_at=expires_at,
+                requires_approval=bool(actions),
+            )
+            try:
+                result["agency_deliveries"] = publish_agency_reports(
+                    run_id=run_id,
+                    reports=agency_reports,
+                    config=agency_webhook_config,
+                    publisher=publisher,
+                    now=now,
+                )
+            except Exception as exc:
+                safe_error = _redact_agency_webhooks(
+                    exc,
+                    agency_webhook_config,
+                )
+                logger.error("Agency ROI delivery phase failed: %s", safe_error)
+                result["agency_delivery_error"] = safe_error
         finally:
             publisher.close()
-        expires_at = (
-            (now or datetime.now(SHANGHAI))
-            + timedelta(minutes=config.approval_ttl_minutes)
-            if actions
-            else None
-        )
-        mark_published(
-            run_id,
-            sheet_token=published["sheet_token"],
-            sheet_url=published["url"],
-            message_id=published["message_id"],
-            expires_at=expires_at,
-            requires_approval=bool(actions),
-        )
         result.update(
             {
                 "status": "PENDING_APPROVAL" if actions else "COMPLETED",

examples/auto_put_ad_mini/test_roi_agency_delivery.py

@@ -0,0 +1,225 @@
+import json
+import os
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import Mock, patch
+
+import httpx
+
+from roi_control.agency_delivery import (
+    AgencyWebhookNotifier,
+    agency_route_key,
+    publish_agency_reports,
+)
+from roi_control.config import AgencyWebhookConfig
+from roi_control.feishu import RoiFeishuPublisher
+from roi_control.service import _redact_agency_webhooks
+
+
+WEBHOOK = "https://open.feishu.cn/open-apis/bot/v2/hook/test-route"
+
+
+class FakePublisher:
+    def __init__(self):
+        self.paths = []
+
+    def upload_workbook(self, path):
+        self.paths.append(path)
+        return {"url": "https://example.test/sheet", "sheet_token": "sheet-1"}
+
+
+class FakeNotifier:
+    def __init__(self):
+        self.calls = []
+
+    def send(self, webhook_url, **values):
+        self.calls.append((webhook_url, values))
+        return "0"
+
+
+class AgencyDeliveryTest(unittest.TestCase):
+    def test_config_snapshot_never_contains_webhook_secret(self):
+        env = {
+            "ROI_AGENCY_WEBHOOK_ENABLED": "1",
+            "ROI_AGENCY_WEBHOOKS_JSON": json.dumps({"棱镜": WEBHOOK}),
+        }
+        with patch.dict(os.environ, env, clear=False):
+            config = AgencyWebhookConfig.from_env()
+        snapshot = config.snapshot()
+        self.assertEqual(snapshot["agencies"], ["棱镜"])
+        self.assertNotIn(WEBHOOK, json.dumps(snapshot, ensure_ascii=False))
+
+    def test_disabled_config_does_not_parse_secret_json(self):
+        with patch.dict(
+            os.environ,
+            {
+                "ROI_AGENCY_WEBHOOK_ENABLED": "0",
+                "ROI_AGENCY_WEBHOOKS_JSON": "not-json",
+            },
+            clear=False,
+        ):
+            config = AgencyWebhookConfig.from_env()
+        self.assertFalse(config.enabled)
+        self.assertEqual(config.webhooks, {})
+
+    def test_route_key_is_exact_and_only_removes_known_prefix(self):
+        self.assertEqual(agency_route_key("小程序-代投-棱镜"), "棱镜")
+        self.assertEqual(agency_route_key("小程序 -代投-贝湉"), "贝湉")
+        self.assertEqual(agency_route_key("自动化投放"), "自动化投放")
+
+    def test_service_error_redaction_removes_webhook_secret(self):
+        config = AgencyWebhookConfig(enabled=True, webhooks={"棱镜": WEBHOOK})
+        error = RuntimeError(f"POST {WEBHOOK} failed")
+        redacted = _redact_agency_webhooks(error, config)
+        self.assertNotIn(WEBHOOK, redacted)
+        self.assertIn("<redacted>", redacted)
+
+    @patch("roi_control.agency_delivery.update_agency_delivery")
+    @patch("roi_control.agency_delivery.upsert_agency_delivery")
+    def test_only_configured_agency_uploads_and_sends(
+        self,
+        upsert,
+        update,
+    ):
+        upsert.side_effect = [
+            {"id": 1, "status": "PENDING"},
+            {"id": 2, "status": "PENDING"},
+        ]
+        publisher = FakePublisher()
+        notifier = FakeNotifier()
+        with tempfile.TemporaryDirectory() as directory:
+            route_path = Path(directory) / "20260804_棱镜_调控建议.xlsx"
+            route_path.write_bytes(b"xlsx")
+            skipped_path = Path(directory) / "20260804_自动化投放_调控建议.xlsx"
+            skipped_path.write_bytes(b"xlsx")
+            reports = [
+                {
+                    "agency_name": "小程序-代投-棱镜",
+                    "report_version": "roi_agency_advice_v7",
+                    "report": str(route_path),
+                    "creative_rows": 10,
+                    "ad_rows": 3,
+                },
+                {
+                    "agency_name": "自动化投放",
+                    "report_version": "roi_agency_advice_v7",
+                    "report": str(skipped_path),
+                    "creative_rows": 1,
+                    "ad_rows": 1,
+                },
+            ]
+            outcomes = publish_agency_reports(
+                run_id="run-1",
+                reports=reports,
+                config=AgencyWebhookConfig(enabled=True, webhooks={"棱镜": WEBHOOK}),
+                publisher=publisher,
+                notifier=notifier,
+            )
+
+        self.assertEqual([row["status"] for row in outcomes], ["SENT", "SKIPPED"])
+        self.assertEqual(len(publisher.paths), 1)
+        self.assertEqual(len(notifier.calls), 1)
+        self.assertEqual(notifier.calls[0][0], WEBHOOK)
+        self.assertEqual(notifier.calls[0][1]["creative_rows"], 10)
+        statuses = [call.kwargs["status"] for call in update.call_args_list]
+        self.assertEqual(statuses, ["UPLOADED", "SENT", "SKIPPED"])
+        self.assertNotIn(WEBHOOK, str(upsert.call_args_list))
+
+    @patch("roi_control.agency_delivery.update_agency_delivery")
+    @patch("roi_control.agency_delivery.upsert_agency_delivery")
+    def test_uploaded_sheet_is_reused_when_notification_retries(
+        self,
+        upsert,
+        update,
+    ):
+        upsert.return_value = {
+            "id": 3,
+            "status": "FAILED",
+            "sheet_url": "https://example.test/existing-sheet",
+            "sheet_token": "existing-token",
+        }
+        publisher = FakePublisher()
+        notifier = FakeNotifier()
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / "20260804_棱镜_调控建议.xlsx"
+            path.write_bytes(b"xlsx")
+            outcomes = publish_agency_reports(
+                run_id="run-2",
+                reports=[
+                    {
+                        "agency_name": "小程序-代投-棱镜",
+                        "report_version": "roi_agency_advice_v7",
+                        "report": str(path),
+                        "creative_rows": 10,
+                        "ad_rows": 3,
+                    }
+                ],
+                config=AgencyWebhookConfig(
+                    enabled=True,
+                    webhooks={"棱镜": WEBHOOK},
+                ),
+                publisher=publisher,
+                notifier=notifier,
+            )
+        self.assertEqual(outcomes[0]["status"], "SENT")
+        self.assertEqual(publisher.paths, [])
+        self.assertEqual(
+            notifier.calls[0][1]["sheet_url"],
+            "https://example.test/existing-sheet",
+        )
+        self.assertEqual(update.call_args.kwargs["status"], "SENT")
+
+    def test_webhook_card_contains_sheet_button(self):
+        captured = {}
+
+        def handler(request):
+            captured.update(json.loads(request.content))
+            return httpx.Response(200, json={"StatusCode": 0, "StatusMessage": "success"})
+
+        notifier = AgencyWebhookNotifier()
+        notifier.client.close()
+        notifier.client = httpx.Client(transport=httpx.MockTransport(handler))
+        try:
+            code = notifier.send(
+                WEBHOOK,
+                title="20260804_棱镜_调控建议",
+                sheet_url="https://example.test/sheet",
+                creative_rows=10,
+                ad_rows=3,
+            )
+        finally:
+            notifier.close()
+        self.assertEqual(code, "0")
+        self.assertEqual(captured["msg_type"], "interactive")
+        button = captured["card"]["elements"][1]["actions"][0]
+        self.assertEqual(button["url"], "https://example.test/sheet")
+
+    def test_main_chat_publish_still_uploads_once_and_notifies_each_chat(self):
+        publisher = RoiFeishuPublisher.__new__(RoiFeishuPublisher)
+        publisher.chat_ids = ["chat-1", "chat-2"]
+        publisher.upload_workbook = Mock(
+            return_value={
+                "url": "https://example.test/main-sheet",
+                "sheet_token": "main-token",
+            }
+        )
+        publisher._token = Mock(return_value="tenant-token")
+        publisher._send_card = Mock(side_effect=["message-1", "message-2"])
+
+        result = publisher.publish(
+            Path("main.xlsx"),
+            run_id="run-main",
+            batch_name="batch-main",
+            summary="summary",
+            requires_approval=True,
+        )
+
+        publisher.upload_workbook.assert_called_once_with(Path("main.xlsx"))
+        self.assertEqual(publisher._send_card.call_count, 2)
+        self.assertEqual(result["message_id"], "message-1")
+        self.assertEqual(result["sheet_token"], "main-token")
+
+
+if __name__ == "__main__":
+    unittest.main()

examples/auto_put_ad_mini/test_roi_control_metrics.py

@@ -19,11 +19,13 @@ from roi_control.metrics import (
     prepare_daily_metrics,
 )
 from roi_control.reporting import (
+    AGENCY_SUMMARY_SHEETS,
     DAILY_SHEETS,
     SUMMARY_SHEETS,
     _daily_frame,
     _summary_frame,
     _visible_columns,
+    write_agency_workbooks,
     write_workbook,
 )
 from roi_control.rules import evaluate_rules as _evaluate_rules
@@ -104,8 +106,8 @@ class RoiThreeDayRulesTest(unittest.TestCase):
         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)
+        self.assertIn("p12", run_id)
+        self.assertIn("r36", run_id)
 
     def test_current_creative_status_only_reads_stop_decisions(self):
         rows = pd.DataFrame(
@@ -193,11 +195,72 @@ class RoiThreeDayRulesTest(unittest.TestCase):
         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["阈值样本状态"], "单日补充决策_昨日UV>200")
         self.assertEqual(target["动作"], "观察")
         self.assertEqual(target["日均首层UV"], 100)
         self.assertTrue(candidates["广告id"].eq("latest-only").any())
 
+    def test_one_day_supplement_requires_ad_age_over_three_days(self):
+        daily = self.build_daily()
+        supplemental = [
+            row(ENTITY_SELF, "one-day-hard", DATES[-1], 0.2, uv=300),
+            row(ENTITY_SELF, "one-day-p30", DATES[-1], 0.5, uv=600),
+            row(ENTITY_SELF, "one-day-young", DATES[-1], 0.6, uv=600),
+            row(ENTITY_SELF, "one-day-mid", DATES[-1], 8.0, uv=600),
+            row(ENTITY_SELF, "one-day-high", DATES[-1], 9.0, uv=600),
+        ]
+        daily = pd.concat([daily, pd.DataFrame(supplemental)], ignore_index=True)
+        ages = pd.concat(
+            [
+                self.ages(),
+                pd.DataFrame(
+                    {
+                        "广告id": [
+                            "one-day-hard",
+                            "one-day-p30",
+                            "one-day-young",
+                            "one-day-mid",
+                            "one-day-high",
+                        ],
+                        "广告age": [4, 4, 3, 4, 4],
+                    }
+                ),
+            ],
+            ignore_index=True,
+        )
+        _, thresholds, summary = evaluate_rules(daily, DATES, ages)
+        self.assertEqual(int(thresholds.iloc[0]["单日P30样本数"]), 8)
+        self.assertAlmostEqual(float(thresholds.iloc[0]["t_one_day_stop"]), 0.64)
+        targets = summary.set_index("广告id")
+        self.assertEqual(targets.loc["one-day-hard", "动作"], "关停")
+        self.assertIn("单日硬关停线", targets.loc["one-day-hard", "动作原因"])
+        self.assertIn("启用单日补充规则", targets.loc["one-day-hard", "动作原因"])
+        self.assertIn("最新日预测总效率ROI", targets.loc["one-day-hard", "动作原因"])
+        self.assertIn("三日预测总效率ROI", targets.loc["one-day-hard", "动作原因"])
+        self.assertIn("不参与本次单日判断", targets.loc["one-day-hard", "动作原因"])
+        self.assertIn("广告age=4>3天", targets.loc["one-day-hard", "动作原因"])
+        self.assertEqual(targets.loc["one-day-p30", "动作"], "关停")
+        self.assertIn("单日实体等权P30", targets.loc["one-day-p30", "动作原因"])
+        self.assertEqual(targets.loc["one-day-young", "动作"], "观察")
+        self.assertIn("广告age=3≤3天", targets.loc["one-day-young", "动作原因"])
+        self.assertEqual(targets.loc["one-day-mid", "动作"], "观察")
+        self.assertIn("未命中单日关停规则", targets.loc["one-day-mid", "动作原因"])
+        creative_frame = _summary_frame(summary, ENTITY_SELF)
+        creative_stops = creative_frame[
+            creative_frame["建议动作"].eq("关停创意")
+        ]
+        self.assertEqual(
+            creative_stops["广告id"].tolist(),
+            ["creative-ad-0", "one-day-hard", "one-day-p30"],
+        )
+        self.assertTrue(creative_stops["动作"].eq("关停").all())
+        self.assertTrue(
+            summary[
+                summary["entity_type"].isin([ENTITY_SELF, ENTITY_SELF_AD])
+                & summary["动作"].eq("关停")
+            ]["广告age"].ge(4).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]
@@ -207,6 +270,123 @@ class RoiThreeDayRulesTest(unittest.TestCase):
         self.assertAlmostEqual(result["预测全链路效率收入"], 50 + 30 * multiplier)
         self.assertAlmostEqual(result["ROI"], (50 + 30 * multiplier) / 100)
 
+    def test_agency_workbooks_filter_and_physically_remove_sensitive_columns(self):
+        _, _, summary = evaluate_rules(self.build_daily(), DATES, self.ages())
+        agency_rows = summary.copy()
+        creative_indexes = agency_rows[
+            agency_rows["entity_type"].eq(ENTITY_SELF)
+        ].index.tolist()
+        ad_indexes = agency_rows[
+            agency_rows["entity_type"].eq(ENTITY_SELF_AD)
+        ].index.tolist()
+        agency_rows.loc[creative_indexes, "代理名称"] = "代理B"
+        agency_rows.loc[ad_indexes, "代理名称"] = "代理B"
+        agency_rows.loc[creative_indexes[0], "代理名称"] = "小程序-代投-贝湉"
+        agency_rows.loc[creative_indexes[1], "代理名称"] = "小程序 -代投-贝湉"
+        agency_rows.loc[ad_indexes[0], "代理名称"] = "小程序-代投-贝湉"
+        original = agency_rows.copy(deep=True)
+
+        with tempfile.TemporaryDirectory() as directory:
+            outputs = write_agency_workbooks(
+                agency_rows,
+                Path(directory),
+                "20260803",
+            )
+            self.assertEqual(
+                [row["agency_name"] for row in outputs],
+                ["代理B", "小程序-代投-贝湉"],
+            )
+            pd.testing.assert_frame_equal(agency_rows, original)
+            bay = next(
+                row for row in outputs if row["agency_name"] == "小程序-代投-贝湉"
+            )
+            self.assertEqual(bay["creative_rows"], 2)
+            self.assertEqual(bay["ad_rows"], 1)
+            self.assertEqual(
+                Path(bay["report"]).name,
+                "20260803_小程序-代投-贝湉_调控建议.xlsx",
+            )
+
+            forbidden_fragments = (
+                "ROI",
+                "收入",
+                "关停线",
+                "扩量线",
+                "排名",
+                "是否位于",
+                "t_stop",
+                "t_up",
+                "审批",
+                "执行",
+                "幂等键",
+            )
+            for output in outputs:
+                workbook = load_workbook(output["report"], read_only=False)
+                expected_sheets = {
+                    AGENCY_SUMMARY_SHEETS[ENTITY_SELF],
+                    AGENCY_SUMMARY_SHEETS[ENTITY_SELF_AD],
+                }
+                self.assertEqual(set(workbook.sheetnames), expected_sheets)
+                self.assertEqual(
+                    workbook[AGENCY_SUMMARY_SHEETS[ENTITY_SELF_AD]].sheet_state,
+                    "hidden",
+                )
+                for sheet_name in expected_sheets:
+                    sheet = workbook[sheet_name]
+                    headers = [cell.value for cell in sheet[1]]
+                    for removed in (
+                        "包名",
+                        "广告age",
+                        "日均首层UV",
+                        "建议说明",
+                        "裂变系数-总裂变UV/T0裂变UV",
+                        "裂变系数-总裂变UV/首层UV",
+                        "日均T0裂变人数",
+                        "日均T0裂变率",
+                    ):
+                        self.assertNotIn(removed, headers)
+                    self.assertFalse(
+                        any(
+                            fragment in str(header)
+                            for header in headers
+                            for fragment in forbidden_fragments
+                        )
+                    )
+                    self.assertTrue(
+                        all(
+                            not sheet.column_dimensions[cell.column_letter].hidden
+                            for cell in sheet[1]
+                        )
+                    )
+                creative_headers = [
+                    cell.value
+                    for cell in workbook[AGENCY_SUMMARY_SHEETS[ENTITY_SELF]][1]
+                ]
+                self.assertEqual(
+                    creative_headers,
+                    [
+                        "渠道",
+                        "代理名称",
+                        "账号id",
+                        "账号名称",
+                        "广告id",
+                        "广告名称",
+                        "广告优化目标",
+                        "创意id",
+                        "日均成本",
+                        "评分",
+                        "建议动作",
+                        "当前创意状态",
+                    ],
+                )
+                creative = workbook[AGENCY_SUMMARY_SHEETS[ENTITY_SELF]]
+                score_column = creative_headers.index("评分") + 1
+                self.assertEqual(
+                    creative.cell(2, score_column).number_format,
+                    "0.00",
+                )
+                self.assertNotIn("当日效率ROI", creative_headers)
+
     def test_summary_and_daily_report_frames(self):
         daily = self.build_daily()
         daily = pd.concat(
@@ -224,18 +404,25 @@ class RoiThreeDayRulesTest(unittest.TestCase):
         _, 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")
+        self.assertEqual(frame.iloc[-1]["阈值样本状态"], "单日补充决策_昨日UV>200")
+        observations = frame[
+            frame["阈值样本状态"].eq("单日补充决策_昨日UV>200")
+        ]
+        self.assertEqual(observations.iloc[0]["广告id"], "latest-only-low")
         formal_actions = frame[
-            ~frame["阈值样本状态"].eq("补充观察_昨日UV>200")
+            ~frame["阈值样本状态"].eq("单日补充决策_昨日UV>200")
         ]["建议动作"].tolist()
-        action_rank = {"关停": 0, "扩量": 1, "": 2, "观察": 3}
+        action_rank = {"关停创意": 0, "扩量": 1, "观察": 2}
         self.assertEqual(
             [action_rank[action] for action in formal_actions],
             sorted(action_rank[action] for action in formal_actions),
         )
         self.assertIn("日均总预估效率收入", frame.columns)
+        neutral = frame[frame["动作"].eq("")]
+        self.assertTrue(neutral["建议动作"].eq("观察").all())
+        self.assertTrue(
+            neutral["建议说明"].str.contains("当前无需关停或扩量").all()
+        )
         self.assertIn("日均T0裂变率", frame.columns)
         self.assertNotIn("三日加权平均T0裂变率", _visible_columns(SUMMARY_SHEETS[ENTITY_SELF]))
         self.assertIn("当日效率ROI", frame.columns)
@@ -250,13 +437,14 @@ class RoiThreeDayRulesTest(unittest.TestCase):
         for removed in ("动作", "动作原因", "阈值样本状态", "执行状态", "执行结果"):
             self.assertNotIn(removed, visible)
         self.assertEqual(
-            visible[visible.index("当日效率ROI") : visible.index("建议动作") + 1],
+            visible[visible.index("当日效率ROI") : visible.index("建议说明") + 1],
             [
                 "当日效率ROI",
                 "预测总效率ROI",
                 "关停线（P20）",
                 "扩量线（P80）",
                 "建议动作",
+                "建议说明",
             ],
         )
         self.assertIn("关停线（P20）", frame.columns)
@@ -344,20 +532,21 @@ class RoiThreeDayRulesTest(unittest.TestCase):
                 ).border.top.style,
                 "medium",
             )
-            first_no_action_row = next(
+            first_observe_row = next(
                 row_number
                 for row_number in range(first_scale_row + 1, creative_sheet.max_row + 1)
-                if not creative_sheet.cell(
+                if creative_sheet.cell(
                     row_number, creative_headers["建议动作"]
                 ).value
+                == "观察"
             )
             self.assertEqual(
-                creative_sheet.cell(first_no_action_row, 1).border.top.style,
+                creative_sheet.cell(first_observe_row, 1).border.top.style,
                 "medium",
             )
             self.assertEqual(
                 creative_sheet.cell(
-                    first_no_action_row, creative_headers["建议动作"]
+                    first_observe_row, creative_headers["建议动作"]
                 ).border.top.style,
                 "medium",
             )
@@ -416,6 +605,14 @@ class RoiThreeDayRulesTest(unittest.TestCase):
                 run_summary.cell(summary_rows["扩量样本数"], 2).number_format,
                 "0",
             )
+            self.assertEqual(
+                run_summary.cell(summary_rows["单日关停线（P30）"], 2).number_format,
+                "0.00",
+            )
+            self.assertEqual(
+                run_summary.cell(summary_rows["单日P30样本数"], 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)

examples/tencent_realtime_control/schema.sql

@@ -201,6 +201,30 @@ CREATE TABLE IF NOT EXISTS roi_action_item (
     KEY idx_roi_action_creative (account_id, dynamic_creative_id)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='ROI审批后腾讯动作与审计';
 
+CREATE TABLE IF NOT EXISTS roi_agency_delivery (
+    id BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY,
+    run_id VARCHAR(64) NOT NULL,
+    agency_name VARCHAR(255) NOT NULL,
+    agency_report_version VARCHAR(64) NOT NULL,
+    file_path TEXT NOT NULL,
+    file_sha256 CHAR(64) NOT NULL,
+    creative_rows INT NOT NULL DEFAULT 0,
+    ad_rows INT NOT NULL DEFAULT 0,
+    route_fingerprint CHAR(64) DEFAULT NULL,
+    sheet_token VARCHAR(128) DEFAULT NULL,
+    sheet_url TEXT DEFAULT NULL,
+    status VARCHAR(32) NOT NULL DEFAULT 'PENDING',
+    attempt_count INT NOT NULL DEFAULT 0,
+    response_code VARCHAR(64) DEFAULT NULL,
+    error_message TEXT DEFAULT NULL,
+    sent_at DATETIME 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_agency_delivery
+        (run_id, agency_name, agency_report_version),
+    KEY idx_roi_agency_delivery_status (run_id, status)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='ROI代理表上传与机器人通知审计';
+
 CREATE TABLE IF NOT EXISTS realtime_control_action_log (
     id BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY,
     run_id VARCHAR(36) NOT NULL,

examples/tencent_realtime_control/storage.py

@@ -170,6 +170,10 @@ def initialize_schema() -> None:
                     "result_notified_at": "DATETIME DEFAULT NULL",
                     "notification_error": "TEXT DEFAULT NULL",
                 },
+                "roi_agency_delivery": {
+                    "creative_rows": "INT NOT NULL DEFAULT 0",
+                    "ad_rows": "INT NOT NULL DEFAULT 0",
+                },
             }
             for table_name, columns in roi_migrations.items():
                 cursor.execute(