feat(roi): cap creative stops by spend budget

AGENTS.md

@@ -41,18 +41,18 @@
 
 - ROI 是指导渠道、账户、广告和创意动作的核心北极星指标,必须作为独立领域模块维护,不能散落在飞书、腾讯 API 或调度代码中。
 - ROI 指标计算必须保持纯数据输入/输出,不能依赖飞书审批、腾讯写操作或具体动作执行器。动作策略只能消费一个明确的 `metric_version`。
-- 当前日级 ROI 指标版本为 `north_star_roi_t15_v8`,策略版本为 `roi_policy_v12`,报表版本为 `roi_report_v36`,使用已发布参数 `20260712_A0-A15_v2`。T0 裂变人数取 `SUM(t0_fission_uv_root)`,不得使用旧字段 `t0裂变人数`。首层效率收入取 `SUM(效率收入)`,T0 实际裂变收入取当天的 `SUM(裂变效率收入)`;只预测 T1-T15 增量:`T0实际裂变收入 * (传播裂变系数 - 1)`,预测总收入为 `首层效率收入 + T0实际裂变收入 + 预测T1-T15裂变收入`,禁止继续读取已废弃的 `多层t0裂变收入`。
+- 当前日级 ROI 指标版本为 `north_star_roi_t15_v8`,策略版本为 `roi_policy_v13`,报表版本为 `roi_report_v37`,使用已发布参数 `20260712_A0-A15_v2`。T0 裂变人数取 `SUM(t0_fission_uv_root)`,不得使用旧字段 `t0裂变人数`。首层效率收入取 `SUM(效率收入)`,T0 实际裂变收入取当天的 `SUM(裂变效率收入)`;只预测 T1-T15 增量:`T0实际裂变收入 * (传播裂变系数 - 1)`,预测总收入为 `首层效率收入 + T0实际裂变收入 + 预测T1-T15裂变收入`,禁止继续读取已废弃的 `多层t0裂变收入`。
 - 小程序传播裂变系数按 `人群包+转化目标精确 -> 人群包回退 -> 转化目标回退 -> 渠道回退` 匹配;公众号按 `合作方+公众号精确 -> 合作方回退 -> 渠道回退` 匹配。样本不足的精确实体不能使用自身系数。
 - 小程序日级 ODPS 数据必须保留 `广告优化目标`;缺失目标只能进入人群包回退,不得默认成关键页面访问。企微 ROI 口径未复核,优先使用已发布的合作方 SQL 计算结果,合作方未匹配、样本不足或旧版本无企微参数时默认参考系数为 2.5 并明确标注;企微不进入阈值样本池且不得生成调控动作。
-- ROI 报表包含小程序创意级、小程序广告级和公众号的三日汇总与每日明细共 6 个主 Sheet,其中广告级两个 Sheet 默认隐藏。金额、UV 和人数显示三日总量/3;ROI 和裂变率使用三日汇总后的加权口径。可见尾部列按“当日效率ROI -> 预测总效率ROI -> 关停线（P20） -> 建议动作 -> 当前创意状态”排列;当前创意状态只对创意级关停建议只读腾讯并显示正常、已停止或读取失败,其他行留空。审批选择、整体三日排名百分位、创意前三20%标记、最新日UV、覆盖天数、动作原因、阈值样本状态和执行结果只保留为隐藏审计字段。“当日效率ROI”和“预测总效率ROI”均使用深红—黄—绿色阶,绿色代表表现好。普通数值显示两位小数,UV、人数、数量和广告年龄显示整数。汇总顺序为关停、扩量、其他中间 ROI、观察,第一条扩量行顶部使用粗线分隔。每日明细的 `dt` 必须位于第一列并按日期倒序。主 Sheet 冻结首行和前 7 列。
+- ROI 报表包含小程序创意级、小程序广告级和公众号的三日汇总与每日明细共 6 个主 Sheet,其中广告级两个 Sheet 默认隐藏。金额、UV 和人数显示三日总量/3;ROI 和裂变率使用三日汇总后的加权口径。可见尾部列按“当日效率ROI -> 预测总效率ROI -> 关停线 -> 扩量线（P80） -> 建议动作 -> 建议说明 -> 当前创意状态”排列;当前创意状态只对创意级关停建议只读腾讯并显示正常、已停止或读取失败,其他行留空。审批选择、整体三日排名百分位、最新日UV、覆盖天数、昨日成本、关停规则、预算选择状态、实际关停成本占比、动作原因、阈值样本状态和执行结果只保留为隐藏审计字段;关停线分位点只进入运行摘要和数据库快照,不写入业务 Sheet。“当日效率ROI”和“预测总效率ROI”均使用深红—黄—绿色阶,绿色代表表现好。普通数值显示两位小数,UV、人数、数量和广告年龄显示整数。汇总顺序为关停、扩量、其他中间 ROI、观察,第一条扩量行顶部使用粗线分隔。每日明细的 `dt` 必须位于第一列并按日期倒序。主 Sheet 冻结首行和前 7 列。
 - ROI 审批表应展示所有进入阈值样本池的小程序和公众号实体,以及最新日首层 UV>200 的补充观察实体;不能只展示有动作的数据。企微暂不进入本版报表、阈值和动作。正式样本按关停、扩量、其他中间 ROI、观察排列,补充观察置底并按最新日 UV 倒序。
 - 成熟参数是独立版本化只读资产。审计后的参数必须通过更新脚本发布到 MySQL 的 `roi_fission_parameter_release` 和 `roi_fission_parameter_value`;日级 ROI 只从数据库消费 `ROI_FISSION_PARAMETER_VERSION` 指定且通过内容哈希校验的版本,不能在日常任务中现场重算或静默回退到镜像文件。离线复算只能生成草稿,经恒等式、行数、匹配率和新旧结果对账后才能发布。
 - ROI 报表只上传一次,同一在线表链接发送到 ROI 通知群和 `FEISHU_OPERATOR_CHAT_ID` 投放审批群;群 ID 相同时必须去重。
 - ROI 公式、收入/成本归属、裂变口径或实体粒度发生语义变化时,必须升级指标版本并保存新快照;不得覆盖或重算成旧版本历史结果。
 - 策略阈值和动作语义使用独立 `policy_version`;指标版本与策略版本必须同时写入每个运行批次。
 - 每次运行必须保存完整实体快照、阈值配置、动作建议、可执行性原因和最终执行审计,不能只保存候选或飞书表格。
-- 日级 ROI 当前读取 T-1 至 T-3 的连续三日数据,并使用 `usersharedepth<=1`。连续三天每天首层 UV>200、成本>0 且 ROI 有效的小程序创意级和公众号实体合并后,按实体等权计算整体 P20 关停线。小程序广告级直接从 ODPS 原始数据按广告去重计算,复用同一关停线但不进入样本池。最新日首层 UV>200 但不满足三日条件的实体只标记为“观察”,不得生成腾讯写操作。
-- 当前低 ROI 使用创意级+公众号合格实体统一等权 P20;高 ROI 仅使用合格小程序创意实体等权 P80。非三日正式样本的小程序创意补充使用最新日 UV>200且预测 ROI≤0.20,或最新日 UV>500且预测 ROI≤实体等权 P30 的单日关停规则。创意级低质且广告 age>3 生成关停创意建议,创意级头部20%且广告 age≥3 生成扩量建议;小程序广告级低质且广告 age>3 时批准后暂停整个广告,公众号只通知参考。
+- 日级 ROI 当前读取 T-1 至 T-3 的连续三日数据,并使用 `usersharedepth<=1`。连续三天每天首层 UV>200、成本>0 且 ROI 有效的小程序创意级和公众号实体分别进入渠道独立样本池。公众号即转使用自身合格实体等权 P20;小程序按 T-1 创意级实际成本5%软预算动态反推关停线和实体分位点,广告级重复聚合成本不得进入分母。
+- 小程序三日持续低 ROI、单日绝对低 ROI、单日高 UV 低分位按 6:2:2 获得目标成本的60%/20%/20%基础额度,各组按对应预测 ROI 从低到高取连续前缀,未用额度只在剩余候选间软流转。目标占昨日总成本5%,软上限5.5%,绝对上限6%。非三日正式样本中最新日 UV>200且预测 ROI≤0.20 进入绝对低 ROI 组;ROI>0.20且最新日 UV>500进入高 UV 动态分位组,不再使用固定 P30。高 ROI 仍使用合格小程序创意实体等权 P80。所有小程序关停要求广告 age>3;公众号只通知参考。
 - 同广告永久基础出价 3 天内最多上调一次,且不超过首次纳管基础价的 2 倍。调整后必须同步实时 CPM 模块的基础出价,避免恢复旧值。
 - ROI 逐行审批有效期默认 120 分钟。腾讯写操作必须与实时调控共用数据库 advisory lock,执行前回读映射和状态,执行后再次回读校验;执行结果必须回写表格并发送飞书通知,通知失败只能重试通知,不能重复腾讯写操作。
 - `DAILY_ROI_ENABLED=1`、`ROI_APPLY_ENABLED=0` 只允许计算、快照、报表和审批预览;只有 `DAILY_ROI_ENABLED=1`、`ROI_APPLY_ENABLED=1`、`ROI_SHEET_APPROVAL_ENABLED=1` 时,表格批准后才允许自动执行腾讯写操作。

examples/auto_put_ad_mini/.env.example

@@ -47,16 +47,22 @@ ROI_APPROVAL_TTL_MINUTES=120
 ROI_SCALE_RATIO=1.10
 ROI_SCALE_COOLDOWN_DAYS=3
 ROI_MAX_BASE_RATIO=2.00
-# 连续三天合格的小程序创意+公众号实体等权整体P20。
+# 小程序使用T-1实际成本5%软预算；公众号即转保留独立实体等权P20。
 ROI_SELF_STOP_MIN_AGE=4
-ROI_STOP_QUANTILE=0.20
+ROI_SELF_STOP_COST_SOFT_LOWER_RATIO=0.045
+ROI_SELF_STOP_COST_TARGET_RATIO=0.05
+ROI_SELF_STOP_COST_SOFT_UPPER_RATIO=0.055
+ROI_SELF_STOP_COST_HARD_CAP_RATIO=0.06
+ROI_SELF_STOP_THREE_DAY_SHARE=0.60
+ROI_SELF_STOP_ONE_DAY_HARD_SHARE=0.20
+ROI_SELF_STOP_ONE_DAY_HIGH_UV_SHARE=0.20
+ROI_PARTNER_STOP_QUANTILE=0.20
 ROI_UP_QUANTILE=0.80
 ROI_OBSERVE_MIN_LATEST_UV=200
-# 非三日正式样本的小程序创意单日补充：UV>200且ROI<=0.20，或UV>500且ROI<=P30；关停仍要求广告age>3天。
+# 非三日正式样本：UV>200且ROI<=0.20进入绝对低ROI组；ROI>0.20且UV>500进入高UV动态分位组。
 ROI_ONE_DAY_MIN_UV=200
-ROI_ONE_DAY_P30_MIN_UV=500
+ROI_ONE_DAY_HIGH_UV_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自动去重。
 # 审批表链接为获得链接者可编辑，黄色审批列选择“批准”即自动执行。

examples/auto_put_ad_mini/DEPLOYMENT.md

@@ -172,15 +172,16 @@ curl -X POST http://localhost:8080/trigger | jq .
 
 - 正式任务每天北京时间09:00执行；在创建数据库批次和发送消息前，必须确认 `loghubods.opengid_base_data` 已存在精确T-1分区，且小程序投流和公众号投流相关数据均非空。未就绪时禁止退回旧分区。
 - 日级 ROI 读取 T-1 至 T-3 三个连续完整数据日，数据深度过滤使用最新业务 SQL 的 `usersharedepth<=1`。
-- 正式阈值样本为连续三天每天首层 UV>200、成本>0 且 ROI 有效的小程序创意级和公众号实体；两类实体合并后按实体等权计算整体 P20 关停线。
-- 小程序广告级从 ODPS 原始数据按广告直接 `COUNT(DISTINCT mid)`，复用同一 P20，但不重复进入样本池；低于关停线且广告 age>3 天时生成广告级关停建议。
+- 正式阈值样本为连续三天每天首层 UV>200、成本>0 且 ROI 有效的小程序创意级和公众号实体；两类渠道独立计算，禁止合并分位点。公众号即转使用自身合格实体等权 P20；小程序按 T-1 实际成本软预算动态反推关停线和实体分位点。
+- 小程序广告级从 ODPS 原始数据按广告直接 `COUNT(DISTINCT mid)`，复用小程序三日动态关停线，但不重复进入创意成本预算样本池；低于关停线且广告 age>3 天时生成广告级关停建议。
 - 金额、UV 和人数在三日汇总表展示为三日总量/3；ROI 与裂变率使用三日总分子/总分母的加权口径。
-- 未进入三日正式样本的小程序创意增加单日补充判断：最新日首层 UV>200 且预测总效率 ROI≤0.20，或最新日首层 UV>500 且预测总效率 ROI≤全部最新日 UV>500 小程序创意的实体等权 P30；命中且广告 age>3 天时建议关停，命中但 age≤3 天时观察。动作原因同时展示最新日判断值与仅供参考的三日预测 ROI。其余最新日首层 UV>200 的非正式实体继续置底观察，不参与三日 P20。
+- 小程序创意关停以小程序创意级 T-1 实际成本为唯一成本分母，目标占比 5%、软上限 5.5%、绝对上限 6%。三日持续低 ROI、单日绝对低 ROI、单日高 UV 低分位按 6:2:2 获得 3%/1%/1%基础成本额度；各组按对应预测总效率 ROI 从低到高取连续前缀，未用额度只在仍满足候选条件的组间软流转，不为凑额度跳过更差创意或关停非候选。
+- 未进入三日正式样本的小程序创意增加单日补充判断：最新日首层 UV>200 且预测总效率 ROI≤0.20 进入单日绝对低 ROI 组；ROI>0.20 且最新日首层 UV>500 进入单日高 UV 组。两组不再使用固定 P30，均按 T-1 实际成本额度动态反推关停线和实体分位点。命中候选但未被预算选中、或广告 age≤3 天时只观察。动作原因同时展示最新日判断值、昨日成本与仅供参考的三日预测 ROI。
 - 报表包含小程序创意级、小程序广告级、公众号的三日汇总与每日明细共 6 个主 Sheet；广告级两个 Sheet 默认隐藏，企微暂不进入本版计算和报表。
 - 每日明细 Sheet 的 `dt` 位于第一列，整表按日期倒序排列。
-- 可见 ROI 列简化为“当日效率ROI”和“预测总效率ROI”；其后依次展示“关停线（P20）”和“扩量线（P80）”，公众号不参与扩量故扩量线留空；整体三日排名百分位保留为隐藏审计字段。
-- “动作”可见列改为“建议动作”，其后展示“建议说明”；小程序创意级和广告级关停分别显示“关停创意”“关停广告”，但底层动作不变。创意关停按三日P20、单日硬线、单日P30三类原因分组，同类按 ROI 升序。底层无动作的正式中间区间在报表中显示为“观察”，但不生成可执行动作。阈值样本状态、执行状态和执行结果保留为隐藏审计字段。
-- 当前三日策略以创意级+公众号合格实体统一等权 P20 生成低 ROI 关停建议；以合格小程序创意实体等权 P80 识别头部20%，广告 age≥3 时生成扩量建议。所有小程序创意级和广告级关停均要求广告 age>3 天；创意级关停批准后只暂停对应动态创意，广告级关停批准后暂停整个广告。
+- 可见 ROI 列简化为“当日效率ROI”和“预测总效率ROI”；其后依次展示动态“关停线”和“扩量线（P80）”，公众号不参与扩量故扩量线留空；整体三日排名百分位保留为隐藏审计字段，关停线分位点只进入运行摘要和数据库快照，不写入业务 Sheet。
+- “动作”可见列改为“建议动作”，其后展示“建议说明”；小程序创意级和广告级关停分别显示“关停创意”“关停广告”，但底层动作不变。创意关停按三日持续低 ROI、单日绝对低 ROI、单日高 UV 低分位三类原因分组，同类按对应 ROI 升序。底层无动作的正式中间区间在报表中显示为“观察”，但不生成可执行动作。昨日成本、关停规则、预算选择状态、实际关停成本占比、阈值样本状态和执行结果保留为隐藏审计字段。
+- 当前三日策略以小程序 T-1 成本预算动态线生成低 ROI 关停建议，公众号使用独立实体等权 P20；高 ROI 仍使用合格小程序创意实体等权 P80。所有小程序创意级和广告级关停均要求广告 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

@@ -85,6 +85,18 @@ ROI_APPROVAL_TTL_MINUTES=120
 ROI_SCALE_RATIO=1.10
 ROI_SCALE_COOLDOWN_DAYS=3
 ROI_MAX_BASE_RATIO=2.00
+ROI_SELF_STOP_COST_SOFT_LOWER_RATIO=0.045
+ROI_SELF_STOP_COST_TARGET_RATIO=0.05
+ROI_SELF_STOP_COST_SOFT_UPPER_RATIO=0.055
+ROI_SELF_STOP_COST_HARD_CAP_RATIO=0.06
+ROI_SELF_STOP_THREE_DAY_SHARE=0.60
+ROI_SELF_STOP_ONE_DAY_HARD_SHARE=0.20
+ROI_SELF_STOP_ONE_DAY_HIGH_UV_SHARE=0.20
+ROI_ONE_DAY_MIN_UV=200
+ROI_ONE_DAY_HIGH_UV_MIN_UV=500
+ROI_ONE_DAY_HARD_STOP_ROI=0.20
+ROI_PARTNER_STOP_QUANTILE=0.20
+ROI_UP_QUANTILE=0.80
 ROI_FISSION_PARAMETER_VERSION=20260712_A0-A15_v2
 ROI_FEISHU_CHAT_ID=oc_xxx
 ROI_AGENCY_WEBHOOK_ENABLED=0
@@ -172,7 +184,7 @@ docker compose --env-file /dev/null run --rm \
 
 日级 ROI 单独分两阶段启用：
 
-当前 `north_star_roi_t15_v8` 使用版本化传播裂变参数，策略版本为 `roi_policy_v12`，报表版本为 `roi_report_v36`。T0 裂变人数取 `SUM(t0_fission_uv_root)`，不再读取旧字段 `t0裂变人数`。小程序按人群包和转化目标、
+当前 `north_star_roi_t15_v8` 使用版本化传播裂变参数，策略版本为 `roi_policy_v13`，报表版本为 `roi_report_v37`。T0 裂变人数取 `SUM(t0_fission_uv_root)`，不再读取旧字段 `t0裂变人数`。小程序按人群包和转化目标、
 公众号按合作方和公众号匹配传播裂变系数；企微暂不进入本版报表、阈值和调控。首层效率收入读取 `效率收入`，T0 实际裂变收入读取当天的 `裂变效率收入`，
 并以 T0 实际裂变收入乘传播裂变系数预测完整裂变收入。日级任务不会现场重算 cohort 参数。部署前应保持
 `ROI_FISSION_PARAMETER_VERSION=20260712_A0-A15_v2`。参数必须先发布到 MySQL，
@@ -184,13 +196,14 @@ ROI 表格使用获得链接者可编辑权限。默认隐藏的黄色【审批
 
 审批表包含小程序创意级、小程序广告级和公众号的三日汇总与每日明细共 6 个主 Sheet，广告级两个 Sheet 默认隐藏。
 金额、UV 和人数展示三日总量/3，ROI 和裂变率使用三日汇总后的加权口径。连续三天每天
-首层 UV>200、成本>0 且 ROI 有效的小程序创意和公众号实体合并后按实体等权计算整体 P20；
-广告级直接从 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、覆盖天数、阈值样本状态和执行结果只保留为隐藏审计字段；底层动作原因通过“建议说明”展示。
+首层 UV>200、成本>0 且 ROI 有效的小程序创意和公众号实体分别建立渠道独立样本池。公众号即转使用自身合格实体等权 P20；小程序使用 T-1 创意级实际成本5%软预算动态反推关停线和实体分位点，禁止把广告级重复成本加入分母。
+广告级直接从 ODPS 原始数据按广告去重，复用小程序三日动态关停线但不进入创意成本预算样本池；低于关停线且广告 age>3 天时生成广告级关停建议，批准后暂停整个广告。
+小程序三日持续低 ROI、单日绝对低 ROI、单日高 UV 低分位按 6:2:2 获得目标成本的60%/20%/20%基础额度。三组分别按三日加权预测 ROI、T-1预测 ROI、T-1预测 ROI 从低到高取连续前缀；目标关停成本占昨日总成本5%，软上限5.5%，绝对上限6%，未用额度只在剩余候选间软流转。
+未进入三日正式样本的小程序创意中，最新日首层 UV>200 且预测总效率 ROI≤0.20 进入绝对低 ROI 组；ROI>0.20 且最新日首层 UV>500 进入高 UV 动态分位组，不再使用固定 P30。命中候选但未被预算选中或 age≤3 天时观察。动作原因展示最新日预测 ROI、T-1实际成本与仅供参考的三日预测 ROI。
+可见尾部列按“当日效率ROI → 预测总效率ROI → 关停线 → 扩量线（P80） → 建议动作 → 建议说明”排列；小程序创意级、广告级可见关停动作分别显示为“关停创意”“关停广告”，底层仍保存统一动作值供执行器消费。创意关停按三日持续低 ROI、单日绝对低 ROI、单日高 UV 低分位三类建议说明分组排序，同类内部按对应 ROI 升序。公众号不参与扩量，扩量线留空。
+整体三日排名百分位、是否低于三日关停线、最新日UV、覆盖天数、昨日成本、关停规则、预算选择状态、实际关停成本占比、阈值样本状态和执行结果只保留为隐藏审计字段；底层动作原因通过“建议说明”展示。
 每日明细 Sheet 的 `dt` 位于第一列并按日期倒序。主 Sheet 冻结首行和前 7 列，
-关停线列为“关停线（P20）”。所有小程序创意级和广告级关停统一要求广告 age>3 天。汇总表将三日总 T0 裂变人数 / 三日总首层 UV 的加权比例展示为“日均T0裂变率”。合格小程序创意实体等权 P80 为扩量线，处于头部20%且广告 age≥3 时生成扩量建议。汇总顺序为关停、扩量、中间观察、条件不足观察；第一条扩量行和扩量后第一条观察行顶部均使用粗线分隔。“当日效率ROI”和“预测总效率ROI”均使用深红—黄—绿色阶，绿色代表表现好。普通数值显示两位小数，UV、人数、数量和广告年龄显示整数；两个“裂变系数-总裂变UV/…”比率列固定显示两位小数。“传播裂变系数匹配”保留审计数据但默认隐藏。
+关停线列为动态“关停线”；“关停线分位点”只进入运行摘要和数据库快照，不写入业务 Sheet。所有小程序创意级和广告级关停统一要求广告 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生成，并明确显示“关停创意”或“关停广告”。代理表生成不改变主报表、数据库快照或主审批链接。
 

examples/auto_put_ad_mini/roi_control/config.py

@@ -91,9 +91,15 @@ class RoiConfig:
     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_high_uv_min_uv: float = 500
     one_day_hard_stop_roi: float = 0.20
-    one_day_stop_quantile: float = 0.30
+    self_stop_cost_soft_lower_ratio: float = 0.045
+    self_stop_cost_target_ratio: float = 0.05
+    self_stop_cost_soft_upper_ratio: float = 0.055
+    self_stop_cost_hard_cap_ratio: float = 0.06
+    self_stop_three_day_share: float = 0.60
+    self_stop_one_day_hard_share: float = 0.20
+    self_stop_one_day_high_uv_share: float = 0.20
     stop_quantile: float = 0.20
     up_quantile: float = 0.80
     stop_weight_cap_quantile: float = 0.95
@@ -140,16 +146,42 @@ class RoiConfig:
                 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_high_uv_min_uv=float(
+                os.getenv(
+                    "ROI_ONE_DAY_HIGH_UV_MIN_UV",
+                    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")
+            self_stop_cost_soft_lower_ratio=float(
+                os.getenv("ROI_SELF_STOP_COST_SOFT_LOWER_RATIO", "0.045")
+            ),
+            self_stop_cost_target_ratio=float(
+                os.getenv("ROI_SELF_STOP_COST_TARGET_RATIO", "0.05")
+            ),
+            self_stop_cost_soft_upper_ratio=float(
+                os.getenv("ROI_SELF_STOP_COST_SOFT_UPPER_RATIO", "0.055")
+            ),
+            self_stop_cost_hard_cap_ratio=float(
+                os.getenv("ROI_SELF_STOP_COST_HARD_CAP_RATIO", "0.06")
+            ),
+            self_stop_three_day_share=float(
+                os.getenv("ROI_SELF_STOP_THREE_DAY_SHARE", "0.60")
+            ),
+            self_stop_one_day_hard_share=float(
+                os.getenv("ROI_SELF_STOP_ONE_DAY_HARD_SHARE", "0.20")
+            ),
+            self_stop_one_day_high_uv_share=float(
+                os.getenv("ROI_SELF_STOP_ONE_DAY_HIGH_UV_SHARE", "0.20")
+            ),
+            stop_quantile=float(
+                os.getenv(
+                    "ROI_PARTNER_STOP_QUANTILE",
+                    os.getenv("ROI_STOP_QUANTILE", "0.20"),
+                )
             ),
-            stop_quantile=float(os.getenv("ROI_STOP_QUANTILE", "0.20")),
             up_quantile=float(os.getenv("ROI_UP_QUANTILE", "0.80")),
             stop_weight_cap_quantile=float(
                 os.getenv("ROI_STOP_WEIGHT_CAP_QUANTILE", "0.95")
@@ -186,24 +218,40 @@ class RoiConfig:
             self.partner_min_daily_uv,
             self.observe_min_latest_uv,
             self.one_day_min_uv,
-            self.one_day_p30_min_uv,
+            self.one_day_high_uv_min_uv,
         ) <= 0:
             raise ValueError("ROI UV thresholds must be positive")
-        if self.one_day_p30_min_uv <= self.one_day_min_uv:
+        if self.one_day_high_uv_min_uv <= self.one_day_min_uv:
             raise ValueError(
-                "ROI_ONE_DAY_P30_MIN_UV must be greater than ROI_ONE_DAY_MIN_UV"
+                "ROI_ONE_DAY_HIGH_UV_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:
+        if not (
+            0 <= self.self_stop_cost_soft_lower_ratio
+            <= self.self_stop_cost_target_ratio
+            <= self.self_stop_cost_soft_upper_ratio
+            <= self.self_stop_cost_hard_cap_ratio
+            <= 1
+        ):
             raise ValueError(
-                "ROI_ONE_DAY_STOP_QUANTILE must satisfy 0 < value < 1"
+                "ROI self stop cost ratios must satisfy 0 <= soft lower <= target "
+                "<= soft upper <= hard cap <= 1"
             )
+        shares = (
+            self.self_stop_three_day_share,
+            self.self_stop_one_day_hard_share,
+            self.self_stop_one_day_high_uv_share,
+        )
+        if min(shares) < 0 or not math.isclose(sum(shares), 1.0, abs_tol=1e-9):
+            raise ValueError("ROI self stop allocation shares must be non-negative and sum to 1")
         if not 0 < self.stop_quantile < 1:
-            raise ValueError("ROI_STOP_QUANTILE must satisfy 0 < value < 1")
+            raise ValueError(
+                "ROI_PARTNER_STOP_QUANTILE must satisfy 0 < value < 1"
+            )
         if not self.stop_quantile < self.up_quantile < 1:
             raise ValueError(
-                "ROI_UP_QUANTILE must satisfy ROI_STOP_QUANTILE < value < 1"
+                "ROI_UP_QUANTILE must satisfy ROI_PARTNER_STOP_QUANTILE < value < 1"
             )
         if not 0 < self.stop_weight_cap_quantile <= 1:
             raise ValueError(

examples/auto_put_ad_mini/roi_control/policy.py

@@ -21,6 +21,20 @@ ACTION_SCALE_BID = "SCALE_BID"
 MODE_ACTIONABLE = "ACTIONABLE"
 MODE_NOTIFY_ONLY = "NOTIFY_ONLY"
 
+POLICY_AUDIT_TEXT_FIELDS = (
+    "阈值样本状态",
+    "关停规则",
+    "关停预算选择状态",
+)
+POLICY_AUDIT_NUMBER_FIELDS = (
+    "昨日成本",
+    "小程序昨日总成本",
+    "小程序实际关停成本占比",
+    "关停线分位点",
+    "整体三日ROI排名百分位",
+    "创意三日ROI排名百分位",
+)
+
 
 def safe_int(value: Any) -> int | None:
     if value is None or value == "":
@@ -166,6 +180,14 @@ def annotate_execution(
                     row.get("调控参与状态") or ""
                 ),
                 "roi_method": str(row.get("ROI计算口径") or ""),
+                **{
+                    column: str(row.get(column) or "")
+                    for column in POLICY_AUDIT_TEXT_FIELDS
+                },
+                **{
+                    column: _finite(row.get(column))
+                    for column in POLICY_AUDIT_NUMBER_FIELDS
+                },
             }
         )
         snapshots.append(

examples/auto_put_ad_mini/roi_control/reporting.py

@@ -28,8 +28,8 @@ OBSERVE_FILL = PatternFill("solid", fgColor="FFF2CC")
 APPROVAL_FILL = PatternFill("solid", fgColor="FFD966")
 APPROVAL_HEADER_FILL = PatternFill("solid", fgColor="BF9000")
 
-REPORT_VERSION = "roi_report_v36"
-REPORT_RUN_SUFFIX = "r36"
+REPORT_VERSION = "roi_report_v37"
+REPORT_RUN_SUFFIX = "r37"
 AGENCY_REPORT_VERSION = "roi_agency_advice_v7"
 
 T0_FISSION_MULTIPLIER_COLUMN = "裂变系数-总裂变UV/T0裂变UV"
@@ -115,7 +115,7 @@ SUMMARY_METRICS = (
     TOTAL_FISSION_TO_FIRST_UV_COLUMN,
     "当日效率ROI",
     "预测总效率ROI",
-    "关停线（P20）",
+    "关停线",
     "扩量线（P80）",
     "建议动作",
     "建议说明",
@@ -141,6 +141,8 @@ AGENCY_SUMMARY_METRICS = (
     "建议动作",
 )
 
+REPORT_EXCLUDED_COLUMNS = {"关停线分位点"}
+
 
 def _visible_columns(
     sheet_name: str,
@@ -223,9 +225,12 @@ def _summary_frame(rows: pd.DataFrame, entity_type: str) -> pd.DataFrame:
     subset[TOTAL_FISSION_TO_FIRST_UV_COLUMN] = pd.to_numeric(
         subset.get(DISPLAY_TOTAL_TO_FIRST_COLUMN), errors="coerce"
     )
-    subset["关停线（P20）"] = pd.to_numeric(
+    subset["关停线"] = pd.to_numeric(
         subset.get("t_stop"), errors="coerce"
     )
+    subset["关停线分位点"] = pd.to_numeric(
+        subset.get("关停线分位点"), errors="coerce"
+    )
     subset["扩量线（P80）"] = (
         pd.to_numeric(subset.get("t_up"), errors="coerce")
         if entity_type == ENTITY_SELF
@@ -270,9 +275,9 @@ def _summary_frame(rows: pd.DataFrame, entity_type: str) -> pd.DataFrame:
         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),
+                reason.str.contains("三日加权平均预测总效率ROI", regex=False),
+                reason.str.contains("绝对低ROI候选线", regex=False),
+                reason.str.contains("单日高UV动态分位候选", regex=False),
             ],
             [0, 1, 2],
             default=3,
@@ -317,15 +322,19 @@ def _summary_frame(rows: pd.DataFrame, entity_type: str) -> pd.DataFrame:
     subset.loc[neutral, "建议动作"] = "观察"
     if entity_type == ENTITY_SELF:
         neutral_reason = (
-            "预测总效率ROI位于关停线（P20）与扩量线（P80）之间，"
+            "预测总效率ROI位于动态关停线与扩量线（P80）之间，"
             "当前无需关停或扩量"
         )
     elif entity_type == ENTITY_SELF_AD:
-        neutral_reason = "预测总效率ROI高于关停线（P20），当前无需关停整个广告"
+        neutral_reason = "预测总效率ROI高于小程序动态关停线，当前无需关停整个广告"
     else:
-        neutral_reason = "预测总效率ROI高于关停线（P20），当前仅观察"
+        neutral_reason = "预测总效率ROI高于公众号独立关停线，当前仅观察"
     subset.loc[neutral, "建议说明"] = neutral_reason
-    hidden = [column for column in subset.columns if column not in visible]
+    hidden = [
+        column
+        for column in subset.columns
+        if column not in visible and column not in REPORT_EXCLUDED_COLUMNS
+    ]
     return subset[visible + hidden]
 
 
@@ -396,6 +405,8 @@ def _write_dataframe(ws, frame: pd.DataFrame) -> None:
 
 def _number_format_for_header(header: object) -> str:
     name = re.sub(r"_\d{8}$", "", str(header or ""))
+    if "占比" in name or "分位点" in name:
+        return "0.00%"
     if name in {
         T0_FISSION_MULTIPLIER_COLUMN,
         TOTAL_FISSION_TO_FIRST_UV_COLUMN,
@@ -538,23 +549,62 @@ def _write_summary_sheet(
     config: Mapping[str, object],
 ) -> None:
     ws = workbook.create_sheet("运行摘要")
-    threshold = thresholds.iloc[0].to_dict() if not thresholds.empty else {}
+    by_type = (
+        thresholds.set_index("entity_type").to_dict("index")
+        if not thresholds.empty
+        else {}
+    )
+    self_threshold = by_type.get(ENTITY_SELF, {})
+    gzh_threshold = by_type.get(ENTITY_GZH, {})
     rows = [
         ("报表版本", REPORT_VERSION),
         ("统计窗口", f"{expected_dates[0]} 至 {expected_dates[-1]}"),
         ("数据深度口径", "usersharedepth<=1，与最新业务SQL一致"),
-        ("关停线（P20）", threshold.get("t_stop")),
-        ("阈值样本数", threshold.get("阈值样本数")),
-        ("创意扩量线（P80）", threshold.get("t_up")),
-        ("扩量样本数", threshold.get("扩量样本数")),
-        ("单日关停线（P30）", threshold.get("t_one_day_stop")),
-        ("单日P30样本数", threshold.get("单日P30样本数")),
-        ("阈值样本", "连续三天每天首层UV>200、成本>0且ROI有效的小程序创意和公众号实体"),
+        ("小程序昨日总成本", self_threshold.get("小程序昨日总成本")),
+        ("小程序目标关停成本", self_threshold.get("目标关停成本")),
+        ("小程序实际关停成本", self_threshold.get("实际关停成本")),
+        ("小程序实际关停成本占比", self_threshold.get("实际关停成本占比")),
+        ("小程序关停成本预算状态", self_threshold.get("关停成本预算状态")),
+        ("小程序三日关停线", self_threshold.get("三日关停线")),
+        ("小程序三日关停线分位点", self_threshold.get("三日关停线分位点")),
+        (
+            "小程序单日绝对低ROI关停线",
+            self_threshold.get("单日绝对低ROI关停线"),
+        ),
+        (
+            "小程序单日绝对低ROI关停线分位点",
+            self_threshold.get("单日绝对低ROI关停线分位点"),
+        ),
+        ("小程序单日高UV关停线", self_threshold.get("单日高UV关停线")),
+        (
+            "小程序单日高UV关停线分位点",
+            self_threshold.get("单日高UV关停线分位点"),
+        ),
+        ("小程序三日实际关停成本", self_threshold.get("三日实际关停成本")),
+        (
+            "小程序单日绝对低ROI实际关停成本",
+            self_threshold.get("单日绝对低ROI实际关停成本"),
+        ),
+        (
+            "小程序单日高UV实际关停成本",
+            self_threshold.get("单日高UV实际关停成本"),
+        ),
+        ("小程序阈值样本数", self_threshold.get("阈值样本数")),
+        ("创意扩量线（P80）", self_threshold.get("t_up")),
+        ("扩量样本数", self_threshold.get("扩量样本数")),
+        ("公众号独立关停线", gzh_threshold.get("t_stop")),
+        ("公众号关停线分位点", gzh_threshold.get("关停线分位点")),
+        ("公众号阈值样本数", gzh_threshold.get("阈值样本数")),
+        (
+            "阈值样本",
+            "小程序与公众号按渠道独立；连续三天每天首层UV>200、成本>0且ROI有效",
+        ),
         ("关停年龄门槛", "所有小程序创意级和广告级关停均要求广告age>3天"),
-        ("广告级", "直接按广告去重计算，复用统一P20但不进入样本池；低于关停线且广告age>3天时，审批后暂停整个广告"),
+        ("广告级", "直接按广告去重计算，复用小程序三日动态关停线但不进入成本预算样本池；低于关停线且广告age>3天时，审批后暂停整个广告"),
         ("日均字段", "三日总量/3，缺失日按0"),
         ("ROI与裂变率", "三日汇总分子/三日汇总分母的加权口径"),
-        ("单日补充规则", "非三日正式样本的小程序创意：最新日UV>200且ROI≤0.20，或UV>500且ROI≤实体等权P30；广告age>3天时建议关停，否则观察"),
+        ("小程序成本软预算", "使用T-1实际成本，目标占小程序创意级昨日总成本5%；三日持续低ROI、单日绝对低ROI、单日高UV低分位按6:2:2基础额度分配，未用额度只在低质候选间流转"),
+        ("单日补充规则", "非三日正式样本的小程序创意：最新日UV>200且ROI≤0.20进入绝对低ROI组；ROI>0.20且UV>500进入高UV组；两组均按ROI从低到高和昨日成本预算动态截断"),
         ("补充观察", "其余非正式样本中最新日首层UV>200，置于汇总表末尾且不执行"),
         ("配置快照", json.dumps(dict(config), ensure_ascii=False, default=str)),
     ]
@@ -570,7 +620,10 @@ def _write_summary_sheet(
         if isinstance(value_cell.value, (int, float)) and not isinstance(
             value_cell.value, bool
         ):
-            value_cell.number_format = "0" if label.endswith("数") else "0.00"
+            if "占比" in label or "分位点" in label:
+                value_cell.number_format = "0.00%"
+            else:
+                value_cell.number_format = "0" if label.endswith("数") else "0.00"
 
 
 def write_workbook(

examples/auto_put_ad_mini/roi_control/rules.py

@@ -13,12 +13,18 @@ from .metrics import (
     ENTITY_GZH,
     ENTITY_SELF,
     ENTITY_SELF_AD,
+    GZH_CHANNEL,
+    SELF_CHANNEL,
     compute_roi_summary,
 )
 
 
-POLICY_VERSION = "roi_policy_v12"
-POLICY_RUN_SUFFIX = "p12"
+POLICY_VERSION = "roi_policy_v13"
+POLICY_RUN_SUFFIX = "p13"
+
+STOP_RULE_THREE_DAY = "三日持续低ROI"
+STOP_RULE_ONE_DAY_HARD = "单日绝对低ROI"
+STOP_RULE_ONE_DAY_HIGH_UV = "单日高UV低分位"
 
 
 @dataclass(frozen=True)
@@ -29,9 +35,15 @@ class RuleConfig:
     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_high_uv_min_uv: float = 500
     one_day_hard_stop_roi: float = 0.20
-    one_day_stop_quantile: float = 0.30
+    self_stop_cost_soft_lower_ratio: float = 0.045
+    self_stop_cost_target_ratio: float = 0.05
+    self_stop_cost_soft_upper_ratio: float = 0.055
+    self_stop_cost_hard_cap_ratio: float = 0.06
+    self_stop_three_day_share: float = 0.60
+    self_stop_one_day_hard_share: float = 0.20
+    self_stop_one_day_high_uv_share: float = 0.20
     stop_quantile: float = 0.20
     up_quantile: float = 0.80
 
@@ -72,7 +84,7 @@ def threshold_eligibility_mask(
     config: RuleConfig,
     expected_dates: list[str],
 ) -> pd.Series:
-    """Only creative-level miniapp and official accounts enter global P20."""
+    """Return formal creative-level samples for channel-independent lines."""
 
     return entity_eligibility_mask(
         summary, ENTITY_SELF, config, expected_dates
@@ -105,7 +117,7 @@ def observation_mask(
     )
 
 
-def one_day_p30_pool_mask(
+def one_day_high_uv_pool_mask(
     summary: pd.DataFrame,
     config: RuleConfig,
     expected_dates: list[str],
@@ -115,7 +127,7 @@ def one_day_p30_pool_mask(
     return (
         summary["entity_type"].eq(ENTITY_SELF)
         & pd.to_numeric(summary[f"首层UV_{latest}"], errors="coerce").gt(
-            config.one_day_p30_min_uv
+            config.one_day_high_uv_min_uv
         )
         & pd.to_numeric(summary[f"成本_{latest}"], errors="coerce").gt(0)
         & np.isfinite(latest_roi)
@@ -143,61 +155,90 @@ def one_day_supplement_mask(
     )
 
 
-def compute_global_threshold(
+def compute_channel_thresholds(
     summary: pd.DataFrame,
     expected_dates: list[str],
     config: RuleConfig,
 ) -> pd.DataFrame:
-    eligible = threshold_eligibility_mask(summary, config, expected_dates)
-    sample = pd.to_numeric(summary.loc[eligible, "ROI"], errors="coerce")
-    sample = sample[np.isfinite(sample)]
-    if sample.empty:
-        raise ValueError("最近三日没有满足每日UV和成本门槛的统一阈值样本")
-    creative_eligible = entity_eligibility_mask(
+    self_eligible = entity_eligibility_mask(
         summary, ENTITY_SELF, config, expected_dates
     )
-    creative_sample = pd.to_numeric(
-        summary.loc[creative_eligible, "ROI"], errors="coerce"
+    gzh_eligible = entity_eligibility_mask(
+        summary, ENTITY_GZH, config, expected_dates
+    )
+    self_sample = pd.to_numeric(
+        summary.loc[self_eligible, "ROI"], errors="coerce"
     )
-    creative_sample = creative_sample[np.isfinite(creative_sample)]
+    self_sample = self_sample[np.isfinite(self_sample)]
+    gzh_sample = pd.to_numeric(
+        summary.loc[gzh_eligible, "ROI"], errors="coerce"
+    )
+    gzh_sample = gzh_sample[np.isfinite(gzh_sample)]
+    if self_sample.empty and gzh_sample.empty:
+        raise ValueError("最近三日没有满足每日UV和成本门槛的渠道独立阈值样本")
+
     t_up = (
-        float(creative_sample.quantile(config.up_quantile))
-        if not creative_sample.empty
+        float(self_sample.quantile(config.up_quantile))
+        if not self_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
+    gzh_stop = (
+        float(gzh_sample.quantile(config.stop_quantile))
+        if not gzh_sample.empty
         else np.nan
     )
+    latest = expected_dates[-1]
+    high_uv_pool = one_day_high_uv_pool_mask(summary, config, expected_dates)
     return pd.DataFrame(
         [
             {
-                "统计窗口": f"{expected_dates[0]} 至 {expected_dates[-1]}",
-                "entity_type": "global",
-                "渠道": "小程序创意级+公众号",
-                "t_stop": float(sample.quantile(config.stop_quantile)),
+                "统计窗口": f"{expected_dates[0]} 至 {latest}",
+                "entity_type": ENTITY_SELF,
+                "渠道": SELF_CHANNEL,
+                "t_stop": np.nan,
                 "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)),
+                "t_one_day_hard_stop": np.nan,
+                "t_one_day_stop": np.nan,
+                "关停线分位点": np.nan,
+                "单日绝对低ROI关停线分位点": np.nan,
+                "单日高UV关停线分位点": np.nan,
+                "阈值样本数": int(len(self_sample)),
+                "扩量样本数": int(len(self_sample)),
+                "单日高UV样本数": int(high_uv_pool.sum()),
                 "单日硬关停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)}",
-                "广告级是否入池": "否_仅复用统一关停线",
-            }
+                "单日高UV最低UV": config.one_day_high_uv_min_uv,
+                "关停线口径": "T-1实际成本5%软预算动态线",
+                "扩量线口径": (
+                    f"合格小程序创意实体等权P{int(config.up_quantile * 100)}"
+                ),
+                "广告级是否入池": "否_仅复用小程序三日动态关停线",
+            },
+            {
+                "统计窗口": f"{expected_dates[0]} 至 {latest}",
+                "entity_type": ENTITY_GZH,
+                "渠道": GZH_CHANNEL,
+                "t_stop": gzh_stop,
+                "t_up": np.nan,
+                "t_one_day_hard_stop": np.nan,
+                "t_one_day_stop": np.nan,
+                "关停线分位点": (
+                    config.stop_quantile if np.isfinite(gzh_stop) else np.nan
+                ),
+                "单日绝对低ROI关停线分位点": np.nan,
+                "单日高UV关停线分位点": np.nan,
+                "阈值样本数": int(len(gzh_sample)),
+                "扩量样本数": 0,
+                "单日高UV样本数": 0,
+                "单日硬关停ROI": np.nan,
+                "单日最低UV": np.nan,
+                "单日高UV最低UV": np.nan,
+                "关停线口径": (
+                    f"公众号合格实体等权P{int(config.stop_quantile * 100)}"
+                ),
+                "扩量线口径": "不适用",
+                "广告级是否入池": "不适用",
+            },
         ]
     )
 
@@ -210,20 +251,268 @@ def _sample_percentile(values: pd.Series, target: float) -> float:
     return float((clean <= target).mean())
 
 
+def _ordered_candidate_indices(
+    result: pd.DataFrame,
+    mask: pd.Series,
+    roi_column: str,
+    cost_column: str,
+) -> list[object]:
+    roi = pd.to_numeric(result[roi_column], errors="coerce")
+    cost = pd.to_numeric(result[cost_column], errors="coerce")
+    eligible = mask & np.isfinite(roi) & cost.gt(0)
+    return result.loc[eligible].assign(_decision_roi=roi.loc[eligible]).sort_values(
+        "_decision_roi",
+        ascending=True,
+        kind="stable",
+    ).index.tolist()
+
+
+def _select_prefix_within_budget(
+    result: pd.DataFrame,
+    ordered: list[object],
+    cost_column: str,
+    budget: float,
+) -> list[object]:
+    selected: list[object] = []
+    cost = 0.0
+    for index in ordered:
+        candidate_cost = float(result.at[index, cost_column])
+        if cost + candidate_cost > budget + 1e-9:
+            break
+        selected.append(index)
+        cost += candidate_cost
+    return selected
+
+
+def _selection_line(
+    result: pd.DataFrame,
+    ordered: list[object],
+    selected: set[object],
+    roi_column: str,
+) -> tuple[float, float]:
+    selected_indices = [index for index in ordered if index in selected]
+    if not selected_indices:
+        return np.nan, np.nan
+    line = float(
+        pd.to_numeric(
+            result.loc[selected_indices, roi_column], errors="coerce"
+        ).max()
+    )
+    sample = pd.to_numeric(result.loc[ordered, roi_column], errors="coerce")
+    return line, _sample_percentile(sample, line)
+
+
+def _allocate_self_stop_budget(
+    result: pd.DataFrame,
+    config: RuleConfig,
+    expected_dates: list[str],
+    formal_creative: pd.Series,
+    one_day_supplement: pd.Series,
+) -> tuple[pd.Series, pd.Series, dict[str, float | str]]:
+    latest = expected_dates[-1]
+    cost_column = f"成本_{latest}"
+    latest_roi_column = f"ROI_{latest}"
+    latest_cost = pd.to_numeric(result[cost_column], errors="coerce").fillna(0.0)
+    latest_roi = pd.to_numeric(result[latest_roi_column], errors="coerce")
+    latest_uv = pd.to_numeric(result[f"首层UV_{latest}"], errors="coerce")
+    age = pd.to_numeric(result.get("广告age"), errors="coerce").fillna(0)
+    actionable_age = age.ge(config.self_stop_min_age)
+
+    three_day_mask = formal_creative & actionable_age
+    one_day_hard_mask = (
+        one_day_supplement
+        & actionable_age
+        & latest_roi.le(config.one_day_hard_stop_roi)
+    )
+    one_day_high_uv_mask = (
+        one_day_supplement
+        & actionable_age
+        & latest_uv.gt(config.one_day_high_uv_min_uv)
+        & latest_roi.gt(config.one_day_hard_stop_roi)
+    )
+    masks = {
+        STOP_RULE_THREE_DAY: three_day_mask,
+        STOP_RULE_ONE_DAY_HARD: one_day_hard_mask,
+        STOP_RULE_ONE_DAY_HIGH_UV: one_day_high_uv_mask,
+    }
+    roi_columns = {
+        STOP_RULE_THREE_DAY: "ROI",
+        STOP_RULE_ONE_DAY_HARD: latest_roi_column,
+        STOP_RULE_ONE_DAY_HIGH_UV: latest_roi_column,
+    }
+    shares = {
+        STOP_RULE_THREE_DAY: config.self_stop_three_day_share,
+        STOP_RULE_ONE_DAY_HARD: config.self_stop_one_day_hard_share,
+        STOP_RULE_ONE_DAY_HIGH_UV: config.self_stop_one_day_high_uv_share,
+    }
+    all_self = result["entity_type"].eq(ENTITY_SELF)
+    total_cost = float(latest_cost.loc[all_self & latest_cost.gt(0)].sum())
+    target_cost = total_cost * config.self_stop_cost_target_ratio
+    soft_lower_cost = total_cost * config.self_stop_cost_soft_lower_ratio
+    soft_upper_cost = total_cost * config.self_stop_cost_soft_upper_ratio
+    hard_cap_cost = total_cost * config.self_stop_cost_hard_cap_ratio
+
+    ordered = {
+        rule: _ordered_candidate_indices(
+            result,
+            mask,
+            roi_columns[rule],
+            cost_column,
+        )
+        for rule, mask in masks.items()
+    }
+    selected: set[object] = set()
+    for rule in (
+        STOP_RULE_THREE_DAY,
+        STOP_RULE_ONE_DAY_HARD,
+        STOP_RULE_ONE_DAY_HIGH_UV,
+    ):
+        selected.update(
+            _select_prefix_within_budget(
+                result,
+                ordered[rule],
+                cost_column,
+                target_cost * shares[rule],
+            )
+        )
+
+    selected_cost = float(latest_cost.loc[list(selected)].sum()) if selected else 0.0
+    for rule in (
+        STOP_RULE_ONE_DAY_HARD,
+        STOP_RULE_THREE_DAY,
+        STOP_RULE_ONE_DAY_HIGH_UV,
+    ):
+        for index in ordered[rule]:
+            if index in selected:
+                continue
+            candidate_cost = float(latest_cost.loc[index])
+            proposed = selected_cost + candidate_cost
+            if proposed <= target_cost + 1e-9:
+                selected.add(index)
+                selected_cost = proposed
+                continue
+            if (
+                proposed <= soft_upper_cost + 1e-9
+                and abs(proposed - target_cost) < abs(selected_cost - target_cost)
+            ):
+                selected.add(index)
+                selected_cost = proposed
+            break
+
+    # The normal selection stays within the 5.5% soft upper bound. If discrete
+    # creative costs still leave the result below the 4.5% soft lower bound,
+    # allow the next uninterrupted ROI prefix to approach 5%, but never exceed
+    # the 6% absolute cap.
+    if selected_cost < soft_lower_cost:
+        for rule in (
+            STOP_RULE_ONE_DAY_HARD,
+            STOP_RULE_THREE_DAY,
+            STOP_RULE_ONE_DAY_HIGH_UV,
+        ):
+            for index in ordered[rule]:
+                if index in selected:
+                    continue
+                candidate_cost = float(latest_cost.loc[index])
+                proposed = selected_cost + candidate_cost
+                if (
+                    proposed <= hard_cap_cost + 1e-9
+                    and abs(proposed - target_cost)
+                    < abs(selected_cost - target_cost)
+                ):
+                    selected.add(index)
+                    selected_cost = proposed
+                break
+
+    if selected_cost > hard_cap_cost + 1e-9:
+        raise ValueError("小程序关停候选成本超过绝对上限")
+
+    selected_mask = pd.Series(result.index.isin(selected), index=result.index)
+    candidate_rule = pd.Series("", index=result.index, dtype="object")
+    for rule, mask in masks.items():
+        candidate_rule.loc[mask] = rule
+
+    lines: dict[str, float] = {}
+    for rule in ordered:
+        line, percentile = _selection_line(
+            result,
+            ordered[rule],
+            selected,
+            roi_columns[rule],
+        )
+        lines[f"{rule}_line"] = line
+        lines[f"{rule}_percentile"] = percentile
+        selected_indices = [index for index in ordered[rule] if index in selected]
+        lines[f"{rule}_cost"] = (
+            float(latest_cost.loc[selected_indices].sum())
+            if selected_indices
+            else 0.0
+        )
+
+    stats = {
+        "小程序昨日总成本": total_cost,
+        "软下限关停成本": soft_lower_cost,
+        "目标关停成本": target_cost,
+        "软上限关停成本": soft_upper_cost,
+        "绝对上限关停成本": hard_cap_cost,
+        "实际关停成本": selected_cost,
+        "实际关停成本占比": selected_cost / total_cost if total_cost > 0 else 0.0,
+        "关停成本预算状态": (
+            "正常范围"
+            if soft_lower_cost <= selected_cost <= soft_upper_cost
+            else "低于软下限"
+            if selected_cost < soft_lower_cost
+            else "高于软上限"
+        ),
+        "三日基础预算成本": target_cost * config.self_stop_three_day_share,
+        "单日绝对低ROI基础预算成本": (
+            target_cost * config.self_stop_one_day_hard_share
+        ),
+        "单日高UV基础预算成本": (
+            target_cost * config.self_stop_one_day_high_uv_share
+        ),
+        "三日候选数": len(ordered[STOP_RULE_THREE_DAY]),
+        "单日绝对低ROI候选数": len(ordered[STOP_RULE_ONE_DAY_HARD]),
+        "单日高UV候选数": len(ordered[STOP_RULE_ONE_DAY_HIGH_UV]),
+        "三日实际关停成本": lines[f"{STOP_RULE_THREE_DAY}_cost"],
+        "单日绝对低ROI实际关停成本": lines[f"{STOP_RULE_ONE_DAY_HARD}_cost"],
+        "单日高UV实际关停成本": lines[f"{STOP_RULE_ONE_DAY_HIGH_UV}_cost"],
+        "三日关停线": lines[f"{STOP_RULE_THREE_DAY}_line"],
+        "三日关停线分位点": lines[f"{STOP_RULE_THREE_DAY}_percentile"],
+        "单日绝对低ROI关停线": lines[f"{STOP_RULE_ONE_DAY_HARD}_line"],
+        "单日绝对低ROI关停线分位点": lines[
+            f"{STOP_RULE_ONE_DAY_HARD}_percentile"
+        ],
+        "单日高UV关停线": lines[f"{STOP_RULE_ONE_DAY_HIGH_UV}_line"],
+        "单日高UV关停线分位点": lines[
+            f"{STOP_RULE_ONE_DAY_HIGH_UV}_percentile"
+        ],
+    }
+    return selected_mask, candidate_rule, stats
+
+
 def apply_actions(
     summary: pd.DataFrame,
     thresholds: pd.DataFrame,
     config: RuleConfig,
     expected_dates: list[str],
-) -> pd.DataFrame:
+) -> tuple[pd.DataFrame, pd.DataFrame]:
     result = summary.copy()
     result["动作"] = ""
     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"]
+    thresholds = thresholds.copy()
+    threshold_by_type = thresholds.set_index("entity_type")
+    self_threshold = threshold_by_type.loc[ENTITY_SELF]
+    gzh_threshold = threshold_by_type.loc[ENTITY_GZH]
+    result["t_stop"] = np.nan
+    result.loc[result["entity_type"].eq(ENTITY_GZH), "t_stop"] = float(
+        gzh_threshold["t_stop"]
     )
+    result["t_up"] = np.nan
+    result.loc[
+        result["entity_type"].isin([ENTITY_SELF, ENTITY_SELF_AD]), "t_up"
+    ] = float(self_threshold["t_up"])
+    result["t_one_day_stop"] = np.nan
+    result["关停线分位点"] = np.nan
 
     threshold_eligible = threshold_eligibility_mask(result, config, expected_dates)
     ad_eligible = entity_eligibility_mask(
@@ -233,38 +522,102 @@ def apply_actions(
     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
     )
-    creative_sample_roi = result.loc[creative_eligible, "ROI"]
-    result["整体三日ROI排名百分位"] = result["ROI"].apply(
-        lambda value: _sample_percentile(sample_roi, float(value))
-        if pd.notna(value)
-        else np.nan
+    gzh_eligible = entity_eligibility_mask(
+        result, ENTITY_GZH, config, expected_dates
     )
-    result["是否位于三日ROI后20%"] = False
+    self_sample_roi = result.loc[creative_eligible, "ROI"]
+    gzh_sample_roi = result.loc[gzh_eligible, "ROI"]
+    creative_sample_roi = result.loc[creative_eligible, "ROI"]
+    result["整体三日ROI排名百分位"] = np.nan
+    for mask, sample in (
+        (result["entity_type"].isin([ENTITY_SELF, ENTITY_SELF_AD]), self_sample_roi),
+        (result["entity_type"].eq(ENTITY_GZH), gzh_sample_roi),
+    ):
+        result.loc[mask, "整体三日ROI排名百分位"] = result.loc[mask, "ROI"].apply(
+            lambda value: _sample_percentile(sample, float(value))
+            if pd.notna(value)
+            else np.nan
+        )
+    result["是否低于三日关停线"] = False
     result["创意三日ROI排名百分位"] = result["ROI"].apply(
         lambda value: _sample_percentile(creative_sample_roi, float(value))
         if pd.notna(value)
         else np.nan
     )
     result["是否位于创意三日ROI前20%"] = False
-    comparable = threshold_eligible | ad_eligible
-    result.loc[comparable, "是否位于三日ROI后20%"] = (
-        pd.to_numeric(result.loc[comparable, "ROI"], errors="coerce")
-        <= result.loc[comparable, "t_stop"]
-    )
-    if np.isfinite(float(result["t_up"].iloc[0])):
+    if np.isfinite(float(self_threshold["t_up"])):
         result.loc[creative_eligible, "是否位于创意三日ROI前20%"] = (
             pd.to_numeric(result.loc[creative_eligible, "ROI"], errors="coerce")
             >= result.loc[creative_eligible, "t_up"]
         )
 
+    selected_stop, candidate_rule, budget_stats = _allocate_self_stop_budget(
+        result,
+        config,
+        expected_dates,
+        creative_eligible,
+        one_day_supplement,
+    )
+    self_row = thresholds["entity_type"].eq(ENTITY_SELF)
+    threshold_updates = {
+        **budget_stats,
+        "t_stop": budget_stats["三日关停线"],
+        "t_one_day_hard_stop": budget_stats["单日绝对低ROI关停线"],
+        "t_one_day_stop": budget_stats["单日高UV关停线"],
+        "关停线分位点": budget_stats["三日关停线分位点"],
+        "单日绝对低ROI关停线分位点": budget_stats[
+            "单日绝对低ROI关停线分位点"
+        ],
+        "单日高UV关停线分位点": budget_stats["单日高UV关停线分位点"],
+    }
+    for column, value in threshold_updates.items():
+        thresholds.loc[self_row, column] = value
+
+    latest = expected_dates[-1]
+    result["昨日成本"] = np.where(
+        result["entity_type"].eq(ENTITY_SELF),
+        pd.to_numeric(result[f"成本_{latest}"], errors="coerce"),
+        np.nan,
+    )
+    result["关停规则"] = candidate_rule
+    result["关停预算选择状态"] = ""
+    result.loc[candidate_rule.ne(""), "关停预算选择状态"] = "预算未选中"
+    result.loc[selected_stop, "关停预算选择状态"] = "预算已选中"
+    result["小程序昨日总成本"] = budget_stats["小程序昨日总成本"]
+    result["小程序实际关停成本占比"] = budget_stats["实际关停成本占比"]
+    formal_line = budget_stats["三日关停线"]
+    formal_percentile = budget_stats["三日关停线分位点"]
+    self_or_ad = result["entity_type"].isin([ENTITY_SELF, ENTITY_SELF_AD])
+    result.loc[self_or_ad, "t_stop"] = formal_line
+    result.loc[self_or_ad, "关停线分位点"] = formal_percentile
+    hard_rows = candidate_rule.eq(STOP_RULE_ONE_DAY_HARD)
+    high_uv_rows = candidate_rule.eq(STOP_RULE_ONE_DAY_HIGH_UV)
+    result.loc[hard_rows, "t_stop"] = budget_stats["单日绝对低ROI关停线"]
+    result.loc[hard_rows, "关停线分位点"] = budget_stats[
+        "单日绝对低ROI关停线分位点"
+    ]
+    result.loc[high_uv_rows, "t_stop"] = budget_stats["单日高UV关停线"]
+    result.loc[high_uv_rows, "t_one_day_stop"] = budget_stats[
+        "单日高UV关停线"
+    ]
+    result.loc[high_uv_rows, "关停线分位点"] = budget_stats[
+        "单日高UV关停线分位点"
+    ]
+    result.loc[selected_stop & creative_eligible, "是否低于三日关停线"] = True
+    result.loc[gzh_eligible, "关停线分位点"] = float(
+        gzh_threshold["关停线分位点"]
+    )
+    result.loc[gzh_eligible, "是否低于三日关停线"] = (
+        pd.to_numeric(result.loc[gzh_eligible, "ROI"], errors="coerce")
+        <= float(gzh_threshold["t_stop"])
+    )
+
     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"])
@@ -282,33 +635,39 @@ def apply_actions(
             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"最新日预测总效率ROI={latest_roi:.2f}≤绝对低ROI候选线"
                     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"])
+                latest_uv > config.one_day_high_uv_min_uv
             ):
                 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}"
+                    f"最新日首层UV={latest_uv:.0f}>{config.one_day_high_uv_min_uv:g}，"
+                    f"进入单日高UV动态分位候选，最新日预测总效率ROI={latest_roi:.2f}"
                 )
 
-            if stop_reason and age >= config.self_stop_min_age:
+            if bool(selected_stop.loc[index]):
                 result.at[index, "动作"] = "关停"
                 result.at[index, "动作原因"] = (
                     f"{decision_context}{stop_reason}{three_day_context}；"
                     f"广告age={age}>{config.self_stop_min_age - 1}天，"
+                    f"昨日成本={float(row[f'成本_{latest}']):.2f}元，"
+                    f"按5%成本软预算选中（{candidate_rule.loc[index]}）；"
                     "建议审批后暂停动态创意"
                 )
-            elif stop_reason:
+            elif 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"昨日成本={float(row[f'成本_{latest}']):.2f}元，"
+                    "满足低质候选条件但未被5%成本软预算选中；建议观察"
+                )
             else:
                 result.at[index, "动作"] = "观察"
                 result.at[index, "动作原因"] = (
@@ -322,23 +681,25 @@ def apply_actions(
             result.at[index, "动作"] = "观察"
             result.at[index, "动作原因"] = (
                 f"最新日首层UV>{config.observe_min_latest_uv:g}，但未满足连续三天"
-                "每天首层UV>200、成本>0且ROI有效；仅置底展示，不进入P20和自动执行"
+                "每天首层UV>200、成本>0且ROI有效；仅置底展示，不进入正式阈值和自动执行"
             )
             continue
         if bool(ad_eligible.loc[index]):
-            if float(row["ROI"]) <= float(row["t_stop"]):
+            if np.isfinite(float(row["t_stop"])) and float(row["ROI"]) <= float(
+                row["t_stop"]
+            ):
                 raw_age = row.get("广告age")
                 age = int(raw_age) if pd.notna(raw_age) else 0
                 if age >= config.self_stop_min_age:
                     result.at[index, "动作"] = "关停"
                     result.at[index, "动作原因"] = (
-                        f"广告级三日加权平均效率ROI≤统一实体等权P20，"
+                        "广告级三日加权平均效率ROI≤小程序三日动态关停线，"
                         f"广告age>{config.self_stop_min_age - 1}天；审批后暂停整个广告"
                     )
                 else:
                     result.at[index, "动作"] = "观察"
                     result.at[index, "动作原因"] = (
-                        f"广告级三日加权平均效率ROI≤统一实体等权P20，但广告age≤"
+                        "广告级三日加权平均效率ROI≤小程序三日动态关停线，但广告age≤"
                         f"{config.self_stop_min_age - 1}天"
                     )
             continue
@@ -347,19 +708,24 @@ def apply_actions(
         if entity_type == ENTITY_SELF:
             raw_age = row.get("广告age")
             age = int(raw_age) if pd.notna(raw_age) else 0
-            if float(row["ROI"]) <= float(row["t_stop"]):
-                if age >= config.self_stop_min_age:
-                    result.at[index, "动作"] = "关停"
-                    result.at[index, "动作原因"] = (
-                        f"三日加权平均效率ROI≤统一实体等权P20，"
-                        f"广告age>{config.self_stop_min_age - 1}天；审批后仅暂停动态创意"
-                    )
-                else:
-                    result.at[index, "动作"] = "观察"
-                    result.at[index, "动作原因"] = (
-                        f"三日加权平均效率ROI≤统一实体等权P20，但广告age≤"
-                        f"{config.self_stop_min_age - 1}天"
-                    )
+            if bool(selected_stop.loc[index]):
+                result.at[index, "动作"] = "关停"
+                result.at[index, "动作原因"] = (
+                    "三日加权平均预测总效率ROI按从低到高排序，"
+                    f"昨日成本={float(row[f'成本_{latest}']):.2f}元，"
+                    "按5%成本软预算的60%基础额度或结余额度选中；"
+                    f"广告age>{config.self_stop_min_age - 1}天；审批后仅暂停动态创意"
+                )
+            elif (
+                np.isfinite(float(row["t_stop"]))
+                and float(row["ROI"]) <= float(row["t_stop"])
+                and age < config.self_stop_min_age
+            ):
+                result.at[index, "动作"] = "观察"
+                result.at[index, "动作原因"] = (
+                    "三日加权平均预测总效率ROI低于小程序动态关停线，但广告age≤"
+                    f"{config.self_stop_min_age - 1}天"
+                )
             elif bool(row["是否位于创意三日ROI前20%"]):
                 if age >= config.self_up_min_age:
                     result.at[index, "动作"] = "扩量"
@@ -373,12 +739,17 @@ def apply_actions(
                         f"三日加权平均效率ROI≥合格创意实体等权P80，但广告age<"
                         f"{config.self_up_min_age}天"
                     )
-        elif entity_type == ENTITY_GZH and float(row["ROI"]) <= float(row["t_stop"]):
+        elif (
+            entity_type == ENTITY_GZH
+            and np.isfinite(float(row["t_stop"]))
+            and float(row["ROI"]) <= float(row["t_stop"])
+        ):
             result.at[index, "动作"] = "关停"
             result.at[index, "动作原因"] = (
-                "三日加权平均效率ROI≤统一实体等权P20；公众号当前仅通知参考"
+                "三日加权平均效率ROI≤公众号独立实体等权关停线；"
+                "公众号当前仅通知参考"
             )
-    return result
+    return result, thresholds
 
 
 def evaluate_rules(
@@ -395,14 +766,16 @@ def evaluate_rules(
         ad_age,
         fission_parameters=fission_parameters,
     )
-    thresholds = compute_global_threshold(summary, dates, config)
-    evaluated = apply_actions(summary, thresholds, config, dates)
+    thresholds = compute_channel_thresholds(summary, dates, config)
+    evaluated, thresholds = 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)
     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_threshold, "阈值样本状态"] = (
+        "进入三日渠道独立阈值样本池"
+    )
     evaluated.loc[formal_ad, "阈值样本状态"] = "广告级三日合格_不进入阈值样本池"
     evaluated.loc[one_day_supplement, "阈值样本状态"] = (
         "单日补充决策_昨日UV>200"

examples/auto_put_ad_mini/roi_control/service.py

@@ -161,9 +161,15 @@ def _rule_config(config: RoiConfig) -> RuleConfig:
         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_high_uv_min_uv=config.one_day_high_uv_min_uv,
         one_day_hard_stop_roi=config.one_day_hard_stop_roi,
-        one_day_stop_quantile=config.one_day_stop_quantile,
+        self_stop_cost_soft_lower_ratio=config.self_stop_cost_soft_lower_ratio,
+        self_stop_cost_target_ratio=config.self_stop_cost_target_ratio,
+        self_stop_cost_soft_upper_ratio=config.self_stop_cost_soft_upper_ratio,
+        self_stop_cost_hard_cap_ratio=config.self_stop_cost_hard_cap_ratio,
+        self_stop_three_day_share=config.self_stop_three_day_share,
+        self_stop_one_day_hard_share=config.self_stop_one_day_hard_share,
+        self_stop_one_day_high_uv_share=config.self_stop_one_day_high_uv_share,
         stop_quantile=config.stop_quantile,
         up_quantile=config.up_quantile,
     )
@@ -383,7 +389,7 @@ def run_daily_roi(
         report_rows = annotated[
             annotated["阈值样本状态"].isin(
                 [
-                    "进入三日统一阈值样本池",
+                    "进入三日渠道独立阈值样本池",
                     "广告级三日合格_不进入阈值样本池",
                     "单日补充决策_昨日UV>200",
                     "补充观察_昨日UV>200",
@@ -396,7 +402,7 @@ def run_daily_roi(
             row["run_id"] = run_id
         threshold_record = {
             "统计窗口": f"{start_date} 至 {end_date}",
-            "整体三日关停线": thresholds.to_dict("records"),
+            "渠道独立关停线与小程序成本预算": thresholds.to_dict("records"),
         }
         replace_run_results(
             run_id,

examples/auto_put_ad_mini/test_roi_control_metrics.py

@@ -154,8 +154,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("p12", run_id)
-        self.assertIn("r36", run_id)
+        self.assertIn("p13", run_id)
+        self.assertIn("r37", run_id)
 
     def test_current_creative_status_only_reads_stop_decisions(self):
         rows = pd.DataFrame(
@@ -196,17 +196,26 @@ class RoiThreeDayRulesTest(unittest.TestCase):
         self.assertIn("usersharedepth <= 1", sql)
         self.assertNotIn("usersharedepth = '0'", sql)
 
-    def test_global_p20_and_creative_only_p80_scale_actions(self):
+    def test_channel_independent_stop_samples_and_creative_only_p80_scale_actions(self):
         candidates, thresholds, summary = evaluate_rules(
             self.build_daily(), DATES, self.ages()
         )
-        self.assertEqual(len(thresholds), 1)
-        self.assertAlmostEqual(float(thresholds.iloc[0]["t_stop"]), 0.7)
-        self.assertAlmostEqual(float(thresholds.iloc[0]["t_up"]), 3.4)
-        self.assertEqual(int(thresholds.iloc[0]["阈值样本数"]), 8)
-        self.assertEqual(int(thresholds.iloc[0]["扩量样本数"]), 4)
-        self.assertEqual(thresholds.iloc[0]["关停线口径"], "合格实体等权P20")
-        self.assertEqual(thresholds.iloc[0]["扩量线口径"], "合格小程序创意实体等权P80")
+        self.assertEqual(len(thresholds), 2)
+        by_type = thresholds.set_index("entity_type")
+        self.assertTrue(pd.isna(by_type.loc[ENTITY_SELF, "t_stop"]))
+        self.assertAlmostEqual(float(by_type.loc[ENTITY_GZH, "t_stop"]), 1.4)
+        self.assertAlmostEqual(float(by_type.loc[ENTITY_SELF, "t_up"]), 3.4)
+        self.assertEqual(int(by_type.loc[ENTITY_SELF, "阈值样本数"]), 4)
+        self.assertEqual(int(by_type.loc[ENTITY_GZH, "阈值样本数"]), 4)
+        self.assertEqual(int(by_type.loc[ENTITY_SELF, "扩量样本数"]), 4)
+        self.assertEqual(
+            by_type.loc[ENTITY_SELF, "关停线口径"],
+            "T-1实际成本5%软预算动态线",
+        )
+        self.assertEqual(
+            by_type.loc[ENTITY_SELF, "扩量线口径"],
+            "合格小程序创意实体等权P80",
+        )
         scale_rows = candidates[candidates["动作"].eq("扩量")]
         self.assertEqual(len(scale_rows), 1)
         self.assertEqual(scale_rows.iloc[0]["entity_type"], ENTITY_SELF)
@@ -214,7 +223,169 @@ class RoiThreeDayRulesTest(unittest.TestCase):
         self.assertTrue(scale_rows.iloc[0]["是否位于创意三日ROI前20%"])
         self.assertTrue((summary["覆盖天数"] == 3).all())
         pool = summary[summary["entity_type"].isin([ENTITY_SELF, ENTITY_GZH])]
-        self.assertTrue(pool["阈值样本状态"].eq("进入三日统一阈值样本池").all())
+        self.assertTrue(
+            pool["阈值样本状态"].eq("进入三日渠道独立阈值样本池").all()
+        )
+
+    def test_self_stop_budget_uses_latest_cost_with_six_two_two_soft_allocation(self):
+        rows = []
+        formal = [
+            ("formal-1", 0.10, 30.0),
+            ("formal-2", 0.20, 30.0),
+            ("formal-3", 1.00, 500.0),
+            ("formal-4", 2.00, 500.0),
+        ]
+        for entity_id, roi, latest_cost in formal:
+            for dt in DATES:
+                rows.append(
+                    row(
+                        ENTITY_SELF,
+                        entity_id,
+                        dt,
+                        roi,
+                        cost=latest_cost if dt == DATES[-1] else 100.0,
+                    )
+                )
+        rows.extend(
+            [
+                row(ENTITY_SELF, "hard-1", DATES[-1], 0.05, uv=300, cost=20.0),
+                row(ENTITY_SELF, "hard-2", DATES[-1], 0.10, uv=300, cost=200.0),
+                row(ENTITY_SELF, "rank-1", DATES[-1], 0.30, uv=600, cost=20.0),
+                row(ENTITY_SELF, "rank-2", DATES[-1], 0.40, uv=600, cost=700.0),
+            ]
+        )
+        ages = pd.DataFrame(
+            {
+                "广告id": [item[0] for item in formal]
+                + ["hard-1", "hard-2", "rank-1", "rank-2"],
+                "广告age": [10] * 8,
+            }
+        )
+
+        _, thresholds, summary = evaluate_rules(pd.DataFrame(rows), DATES, ages)
+        self_threshold = thresholds.set_index("entity_type").loc[ENTITY_SELF]
+        selected = summary[
+            summary["entity_type"].eq(ENTITY_SELF) & summary["动作"].eq("关停")
+        ]
+        selected_ids = set(selected["广告id"])
+        latest_cost = selected[f"成本_{DATES[-1]}"].sum()
+
+        self.assertEqual(
+            selected_ids,
+            {"formal-1", "formal-2", "hard-1", "rank-1"},
+        )
+        self.assertAlmostEqual(float(latest_cost), 100.0)
+        self.assertAlmostEqual(float(self_threshold["小程序昨日总成本"]), 2000.0)
+        self.assertAlmostEqual(float(self_threshold["目标关停成本"]), 100.0)
+        self.assertAlmostEqual(float(self_threshold["实际关停成本"]), 100.0)
+        self.assertAlmostEqual(float(self_threshold["实际关停成本占比"]), 0.05)
+        self.assertEqual(self_threshold["关停成本预算状态"], "正常范围")
+        self.assertAlmostEqual(float(self_threshold["三日基础预算成本"]), 60.0)
+        self.assertAlmostEqual(float(self_threshold["单日绝对低ROI基础预算成本"]), 20.0)
+        self.assertAlmostEqual(float(self_threshold["单日高UV基础预算成本"]), 20.0)
+        self.assertAlmostEqual(float(self_threshold["三日关停线"]), 0.20)
+        self.assertAlmostEqual(float(self_threshold["三日关停线分位点"]), 0.50)
+        self.assertAlmostEqual(float(self_threshold["单日绝对低ROI关停线"]), 0.05)
+        self.assertAlmostEqual(float(self_threshold["单日高UV关停线"]), 0.30)
+        self.assertEqual(
+            selected.groupby("关停规则")[f"成本_{DATES[-1]}"].sum().to_dict(),
+            {
+                "三日持续低ROI": 60.0,
+                "单日绝对低ROI": 20.0,
+                "单日高UV低分位": 20.0,
+            },
+        )
+
+    def test_self_and_official_account_stop_lines_are_independent(self):
+        daily = self.build_daily()
+        latest_costs = {
+            "creative-ad-0": 10.0,
+            "creative-ad-1": 330.0,
+            "creative-ad-2": 330.0,
+            "creative-ad-3": 330.0,
+        }
+        for entity_id, cost in latest_costs.items():
+            mask = (
+                daily["entity_type"].eq(ENTITY_SELF)
+                & daily["广告id"].eq(entity_id)
+                & daily["dt"].eq(DATES[-1])
+            )
+            roi = float(daily.loc[mask, "效率收入"].iloc[0]) / float(
+                daily.loc[mask, "成本"].iloc[0]
+            )
+            daily.loc[mask, "成本"] = cost
+            daily.loc[mask, "效率收入"] = roi * cost
+
+        _, thresholds, summary = evaluate_rules(daily, DATES, self.ages())
+        by_type = thresholds.set_index("entity_type")
+
+        self.assertEqual(set(by_type.index), {ENTITY_SELF, ENTITY_GZH})
+        self.assertAlmostEqual(float(by_type.loc[ENTITY_SELF, "t_stop"]), 0.10)
+        self.assertAlmostEqual(
+            float(by_type.loc[ENTITY_SELF, "关停线分位点"]),
+            0.25,
+        )
+        self.assertAlmostEqual(float(by_type.loc[ENTITY_GZH, "t_stop"]), 1.40)
+        self.assertAlmostEqual(
+            float(by_type.loc[ENTITY_GZH, "关停线分位点"]),
+            0.20,
+        )
+        self_rows = summary[summary["entity_type"].eq(ENTITY_SELF)]
+        gzh_rows = summary[summary["entity_type"].eq(ENTITY_GZH)]
+        self.assertTrue(self_rows["t_stop"].eq(0.10).all())
+        self.assertTrue(
+            gzh_rows["t_stop"].apply(
+                lambda value: abs(float(value) - 1.40) < 1e-9
+            ).all()
+        )
+
+    def test_unused_group_budget_rolls_to_remaining_low_roi_candidate(self):
+        rows = []
+        formal = [
+            ("formal-roll-1", 0.10, 60.0),
+            ("formal-roll-2", 0.20, 20.0),
+            ("formal-roll-3", 1.00, 900.0),
+        ]
+        for entity_id, roi, latest_cost in formal:
+            for dt in DATES:
+                rows.append(
+                    row(
+                        ENTITY_SELF,
+                        entity_id,
+                        dt,
+                        roi,
+                        cost=latest_cost if dt == DATES[-1] else 100.0,
+                    )
+                )
+        rows.extend(
+            [
+                row(ENTITY_SELF, "rank-roll-1", DATES[-1], 0.30, uv=600, cost=20),
+                row(ENTITY_SELF, "rank-roll-2", DATES[-1], 0.40, uv=600, cost=1000),
+            ]
+        )
+        ages = pd.DataFrame(
+            {
+                "广告id": [item[0] for item in formal]
+                + ["rank-roll-1", "rank-roll-2"],
+                "广告age": [10] * 5,
+            }
+        )
+
+        _, thresholds, summary = evaluate_rules(pd.DataFrame(rows), DATES, ages)
+        selected = summary[
+            summary["entity_type"].eq(ENTITY_SELF) & summary["动作"].eq("关停")
+        ]
+        self_threshold = thresholds.set_index("entity_type").loc[ENTITY_SELF]
+
+        self.assertEqual(
+            set(selected["广告id"]),
+            {"formal-roll-1", "formal-roll-2", "rank-roll-1"},
+        )
+        self.assertAlmostEqual(float(self_threshold["实际关停成本"]), 100.0)
+        self.assertAlmostEqual(float(self_threshold["三日基础预算成本"]), 60.0)
+        self.assertAlmostEqual(float(self_threshold["三日实际关停成本"]), 80.0)
+        self.assertAlmostEqual(float(self_threshold["单日高UV实际关停成本"]), 20.0)
+        self.assertEqual(self_threshold["关停成本预算状态"], "正常范围")
 
     def test_creative_p80_scale_requires_ad_age_at_least_three_days(self):
         ages = self.ages()
@@ -225,7 +396,15 @@ class RoiThreeDayRulesTest(unittest.TestCase):
         self.assertIn("广告age<3天", target["动作原因"])
 
     def test_ad_level_reuses_threshold_without_entering_sample_and_can_stop(self):
-        _, thresholds, summary = evaluate_rules(self.build_daily(), DATES, self.ages())
+        daily = self.build_daily()
+        low_creative = daily["entity_type"].eq(ENTITY_SELF) & daily["广告id"].eq(
+            "creative-ad-0"
+        )
+        latest_low = low_creative & daily["dt"].eq(DATES[-1])
+        daily.loc[latest_low, ["成本", "效率收入"]] = [10.0, 1.0]
+        low_ad = daily["entity_type"].eq(ENTITY_SELF_AD) & daily["广告id"].eq("ad-0")
+        daily.loc[low_ad, "效率收入"] = daily.loc[low_ad, "成本"] * 0.05
+        _, thresholds, summary = evaluate_rules(daily, DATES, self.ages())
         ad_rows = summary[summary["entity_type"].eq(ENTITY_SELF_AD)]
         self.assertTrue(
             ad_rows["阈值样本状态"].eq("广告级三日合格_不进入阈值样本池").all()
@@ -233,7 +412,10 @@ class RoiThreeDayRulesTest(unittest.TestCase):
         actions = dict(zip(ad_rows["广告id"], ad_rows["动作"]))
         self.assertEqual(actions["ad-0"], "关停")
         self.assertEqual(actions["ad-1"], "")
-        self.assertTrue(ad_rows["t_stop"].eq(float(thresholds.iloc[0]["t_stop"])).all())
+        self_line = float(
+            thresholds.set_index("entity_type").loc[ENTITY_SELF, "t_stop"]
+        )
+        self.assertTrue(ad_rows["t_stop"].eq(self_line).all())
         self.assertTrue(ad_rows["调控参与状态"].str.contains("审批后可暂停广告").all())
 
     def test_latest_day_uv_over_200_is_appended_with_fixed_three_day_average(self):
@@ -251,11 +433,11 @@ class RoiThreeDayRulesTest(unittest.TestCase):
     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-hard", DATES[-1], 0.2, uv=300, cost=10),
+            row(ENTITY_SELF, "one-day-p30", DATES[-1], 0.5, uv=600, cost=10),
+            row(ENTITY_SELF, "one-day-young", DATES[-1], 0.6, uv=600, cost=10),
             row(ENTITY_SELF, "one-day-mid", DATES[-1], 8.0, uv=600),
-            row(ENTITY_SELF, "one-day-high", DATES[-1], 9.0, uv=600),
+            row(ENTITY_SELF, "one-day-high", DATES[-1], 9.0, uv=600, cost=900),
         ]
         daily = pd.concat([daily, pd.DataFrame(supplemental)], ignore_index=True)
         ages = pd.concat(
@@ -277,29 +459,33 @@ class RoiThreeDayRulesTest(unittest.TestCase):
             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)
+        self_threshold = thresholds.set_index("entity_type").loc[ENTITY_SELF]
+        self.assertEqual(int(self_threshold["单日高UV样本数"]), 8)
+        self.assertAlmostEqual(float(self_threshold["t_one_day_stop"]), 0.5)
         targets = summary.set_index("广告id")
         self.assertEqual(targets.loc["one-day-hard", "动作"], "关停")
-        self.assertIn("单日硬关停线", targets.loc["one-day-hard", "动作原因"])
+        self.assertIn("绝对低ROI候选线", 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.assertIn("单日高UV动态分位候选", 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", "动作原因"])
+        self.assertIn(
+            "未被5%成本软预算选中",
+            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"],
+            ["one-day-hard", "one-day-p30"],
         )
         self.assertTrue(creative_stops["动作"].eq("关停").all())
         self.assertTrue(
@@ -499,17 +685,22 @@ class RoiThreeDayRulesTest(unittest.TestCase):
             [
                 "当日效率ROI",
                 "预测总效率ROI",
-                "关停线（P20）",
+                "关停线",
                 "扩量线（P80）",
                 "建议动作",
                 "建议说明",
             ],
         )
-        self.assertIn("关停线（P20）", frame.columns)
-        self.assertTrue(frame["扩量线（P80）"].eq(float(thresholds.iloc[0]["t_up"])).all())
+        self.assertIn("关停线", frame.columns)
+        self.assertNotIn("关停线分位点", frame.columns)
+        self.assertNotIn("关停线分位点", visible)
+        self_threshold = thresholds.set_index("entity_type").loc[ENTITY_SELF]
+        self.assertTrue(
+            frame["扩量线（P80）"].eq(float(self_threshold["t_up"])).all()
+        )
         gzh_frame = _summary_frame(summary, ENTITY_GZH)
         self.assertTrue(gzh_frame["扩量线（P80）"].isna().all())
-        self.assertIn("是否位于三日ROI后20%", frame.columns)
+        self.assertIn("是否低于三日关停线", frame.columns)
         self.assertNotIn("关停线（P25）", frame.columns)
         self.assertNotIn("整体实体等权P25关停线", frame.columns)
         self.assertEqual(set(detail["dt"]), set(DATES))
@@ -664,12 +855,16 @@ class RoiThreeDayRulesTest(unittest.TestCase):
                 "0",
             )
             self.assertEqual(
-                run_summary.cell(summary_rows["单日关停线（P30）"], 2).number_format,
+                run_summary.cell(
+                    summary_rows["小程序单日高UV关停线"], 2
+                ).number_format,
                 "0.00",
             )
             self.assertEqual(
-                run_summary.cell(summary_rows["单日P30样本数"], 2).number_format,
-                "0",
+                run_summary.cell(
+                    summary_rows["小程序实际关停成本占比"], 2
+                ).number_format,
+                "0.00%",
             )
             ad_sheet = workbook[SUMMARY_SHEETS[ENTITY_SELF_AD]]
             ad_headers = [cell.value for cell in ad_sheet[1]]

examples/auto_put_ad_mini/test_roi_control_policy.py

@@ -40,6 +40,13 @@ class RoiControlPolicyTest(unittest.TestCase):
                     "创意id": "2001",
                     "动作": "关停",
                     "动作原因": "低ROI",
+                    "阈值样本状态": "进入三日渠道独立阈值样本池",
+                    "关停规则": "三日持续低ROI",
+                    "关停预算选择状态": "预算已选中",
+                    "昨日成本": 123.45,
+                    "小程序昨日总成本": 2000.0,
+                    "小程序实际关停成本占比": 0.05,
+                    "关停线分位点": 0.12,
                 },
                 {
                     "entity_type": ENTITY_SELF,
@@ -103,6 +110,11 @@ class RoiControlPolicyTest(unittest.TestCase):
         self.assertEqual(annotated.iloc[2]["审批选择"], "不可执行")
         self.assertEqual(annotated.iloc[4]["执行模式"], MODE_ACTIONABLE)
         self.assertEqual(annotated.iloc[4]["审批选择"], "")
+        first_audit = snapshots[0]["daily_metrics_json"]
+        self.assertEqual(first_audit["关停规则"], "三日持续低ROI")
+        self.assertEqual(first_audit["关停预算选择状态"], "预算已选中")
+        self.assertEqual(first_audit["昨日成本"], 123.45)
+        self.assertEqual(first_audit["小程序实际关停成本占比"], 0.05)
         ad_action = next(
             action for action in actions if action["action_type"] == ACTION_PAUSE_AD
         )