feat(roi-control): add daily ROI approval workflow

AGENTS.md

@@ -17,16 +17,30 @@
 ## 生产服务架构
 
 - 生产使用同一个完整 `ad-put-agent` 镜像启动两个容器:`ad-control-service` 和 `ad-daily-service`。
-- `ad-control-service` 负责唯一飞书 WebSocket、运营暂停/停止/恢复和每 10 分钟实时 CPM 调控。
+- `ad-control-service` 负责唯一飞书 WebSocket、运营暂停/停止/恢复、ROI 整批审批执行和每 10 分钟实时 CPM 调控。
 - 实时调控账户范围是历史 `ad_creation_account_config` 与启用白名单的交集;飞书关闭创建配置只停止新建/补创意,不能让既有广告退出实时管理。
-- `ad-daily-service` 每天 10:30 调用现有广告/创意创建流程,每 2 小时扫描腾讯创意正式审核结果。
+- `ad-daily-service` 每天 10:30 调用现有广告/创意创建流程,每 2 小时扫描腾讯创意正式审核结果,并可在每天 11:00 计算和发布日级 ROI 批次。
 - 旧 `server.py`、`execute_once.py` 和 `run.py` 的模型调控链路不再作为生产入口,但暂不删除历史代码。
 - 飞书生产控制命令使用确定性解析,不能让模型直接决定账户范围或执行腾讯写操作。
 - `暂停` 表示仅暂停到下一投放日 06:00;`停止` 表示持续停止;`恢复` 只解除运营暂停,不能解除仍生效的 CPM 暂停。
 - 所有飞书触发的腾讯写操作都必须二次确认。群聊命令必须来自配置群并 @机器人,发送人必须在允许列表。
 - 运营暂停状态和 CPM 暂停状态必须分开持久化。原本人工暂停的广告不能被系统认领或自动开启;腾讯后台人工重新开启时以人工操作为准。
 - 实时控制和飞书写操作必须共用 MySQL advisory lock,并执行腾讯写后回读校验。
-- `RTC_APPLY_ENABLED` 默认关闭;生产切换前必须先完成 dry-run。日级 ROI 和预算节奏当前只保留职责边界,不能自动执行。
+- `RTC_APPLY_ENABLED`、`DAILY_ROI_ENABLED`、`ROI_APPLY_ENABLED` 默认关闭;生产切换前必须先完成 dry-run 和只通知验证。
+
+## ROI 北极星指标与日级调控
+
+- ROI 是指导渠道、账户、广告和创意动作的核心北极星指标,必须作为独立领域模块维护,不能散落在飞书、腾讯 API 或调度代码中。
+- ROI 指标计算必须保持纯数据输入/输出,不能依赖飞书审批、腾讯写操作或具体动作执行器。动作策略只能消费一个明确的 `metric_version`。
+- ROI 公式、收入/成本归属、裂变口径或实体粒度发生语义变化时,必须升级指标版本并保存新快照;不得覆盖或重算成旧版本历史结果。
+- 策略阈值和动作语义使用独立 `policy_version`;指标版本与策略版本必须同时写入每个运行批次。
+- 每次运行必须保存完整实体快照、阈值配置、动作建议、可执行性原因和最终执行审计,不能只保存候选或飞书表格。
+- 日级 ROI 当前读取 T-1 至 T-3 的连续三日数据。所有渠道都进入指标和通知,只有当前自动化腾讯账户允许在整批审批后执行。
+- 当前低 ROI 动作只暂停 `dynamic_creative_id`,不能暂停整个广告;高 ROI 动作只调整广告永久基础出价,默认上调 10%。
+- 同广告永久基础出价 3 天内最多上调一次,且不超过首次纳管基础价的 2 倍。调整后必须同步实时 CPM 模块的基础出价,避免恢复旧值。
+- ROI 整批审批有效期默认 120 分钟。腾讯写操作必须与实时调控共用数据库 advisory lock,执行前回读映射和状态,执行后再次回读校验。
+- `DAILY_ROI_ENABLED=1`、`ROI_APPLY_ENABLED=0` 只允许计算、快照、报表和审批预览;只有两个开关都为 1 时,审批通过后才允许执行腾讯写操作。
+- 禁止配置 `DAILY_ROI_ENABLED=0`、`ROI_APPLY_ENABLED=1`;服务启动时必须拒绝这种不完整配置。
 
 ## 模块 B 创意创建规则
 

examples/auto_put_ad_mini/.env.example

@@ -28,8 +28,21 @@ DAILY_RUN_ON_STARTUP=0
 DAILY_REVIEW_RUN_ON_STARTUP=0
 DAILY_SYNC_DELIVERY_TEMPLATE=0
 
-RTC_START_HOUR=6
-RTC_STOP_HOUR=20
+# 日级 ROI 北极星指标与整批审批。默认只关闭,分阶段启用。
+DAILY_ROI_ENABLED=0
+DAILY_ROI_HOUR=11
+DAILY_ROI_MINUTE=0
+DAILY_ROI_RUN_ON_STARTUP=0
+ROI_APPLY_ENABLED=0
+ROI_APPROVAL_TTL_MINUTES=120
+ROI_SCALE_RATIO=1.10
+ROI_SCALE_COOLDOWN_DAYS=3
+ROI_MAX_BASE_RATIO=2.00
+# 为空时依次复用 RTC_COMMAND_CHAT_ID / FEISHU_AD_PROJECT_CHAT_ID
+ROI_FEISHU_CHAT_ID=
+
+RTC_START_HOUR=12
+RTC_STOP_HOUR=21
 RTC_POLL_SECONDS=600
 RTC_HIGH_CPM=250
 RTC_LOW_CPM=190

examples/auto_put_ad_mini/docs/unified_services_deployment.md

@@ -61,10 +61,21 @@ DAILY_REVIEW_INTERVAL_HOURS=2
 DAILY_RUN_ON_STARTUP=0
 DAILY_REVIEW_RUN_ON_STARTUP=0
 DAILY_SYNC_DELIVERY_TEMPLATE=0
+
+DAILY_ROI_ENABLED=0
+DAILY_ROI_HOUR=11
+DAILY_ROI_MINUTE=0
+DAILY_ROI_RUN_ON_STARTUP=0
+ROI_APPLY_ENABLED=0
+ROI_APPROVAL_TTL_MINUTES=120
+ROI_SCALE_RATIO=1.10
+ROI_SCALE_COOLDOWN_DAYS=3
+ROI_MAX_BASE_RATIO=2.00
+ROI_FEISHU_CHAT_ID=oc_xxx
 ```
 
 多个操作人使用英文逗号分隔 `RTC_COMMAND_ALLOWED_OPEN_IDS`。
-代码默认 `RTC_APPLY_ENABLED=0`，日级创建、审核扫描和模板同步也默认关闭；
+代码默认 `RTC_APPLY_ENABLED=0`，日级创建、审核扫描、ROI 和模板同步也默认关闭；
 缺少显式生产配置时不会创建广告或修改腾讯。
 
 ## 3. ECS 部署脚本
@@ -134,13 +145,33 @@ docker compose --env-file /dev/null run --rm \
 ## 5. 生产切换
 
 1. 确认旧 `auto_put_ad_mini/server.py` 和旧实时容器不再运行。
-2. 设置 `RTC_APPLY_ENABLED=1`，保持日级服务关闭。
+2. 设置 `RTC_APPLY_ENABLED=1`，保持日级创建和 ROI 服务关闭。
 3. 重新 `docker compose up -d`，观察一个完整实时周期。
 4. 用单账户命令验证暂停、恢复、腾讯回读和飞书结果。
 5. 设置 `DAILY_REVIEW_ENABLED=1`、`DAILY_SYNC_DELIVERY_TEMPLATE=1`。
 6. 审核扫描稳定后设置 `DAILY_CREATION_ENABLED=1`。
 7. 保持 `DAILY_RUN_ON_STARTUP=0`，防止部署时额外触发完整创建。
 
+日级 ROI 单独分两阶段启用：
+
+1. 设置 `DAILY_ROI_ENABLED=1`、`ROI_APPLY_ENABLED=0`，观察 11:00 的 ODPS 计算、数据库快照、只读飞书表和整批审批提示。
+2. 使用 `确认 roi_<run_id>` 验证控制服务明确提示写操作关闭。
+3. 核对阈值、账户可执行范围和动作数量后，再设置 `ROI_APPLY_ENABLED=1`。
+4. 生产审批有效期默认 120 分钟；过期批次不能执行。拒绝整批使用 `拒绝 roi_<run_id>`。
+5. 不允许 `DAILY_ROI_ENABLED=0`、`ROI_APPLY_ENABLED=1`，两个服务都会在启动阶段拒绝该配置。
+
+首次启用前可手工运行一个“不发飞书、不写腾讯”的计算批次。该命令会读取 ODPS、
+初始化/写入 ROI 数据库快照并在挂载目录生成本地 Excel：
+
+```bash
+docker compose run --rm ad_daily_service \
+  python /app/examples/auto_put_ad_mini/run_daily_roi.py \
+  --end-date 20260725
+```
+
+确认数据库快照和本地报告后，再增加 `--send-feishu` 验证只读在线表和整批命令。
+此时必须保持 `ROI_APPLY_ENABLED=0`；回复确认只应得到“写操作关闭”的提示。
+
 ## 6. 验收与回滚
 
 验收：
@@ -158,6 +189,8 @@ docker compose --env-file /dev/null logs --since=30m ad_daily_service
 - 飞书只有一个 WebSocket 消费者。
 - 10:30 只出现一次创建任务。
 - 审核扫描每 2 小时最多一个实例。
+- 11:00 只生成一个相同指标/策略版本的 ROI 批次，同日重跑不会重复发布。
+- ROI 审批执行与实时 CPM 调控不并发写腾讯，基础出价调整后两边状态一致。
 
 回滚时将 `VERSION` 改为上一个镜像版本并重新部署。数据库新增表和列可以保留，
 旧代码不会读取这些字段，不需要执行破坏性回滚。

examples/auto_put_ad_mini/roi_control/__init__.py

@@ -0,0 +1 @@
+"""Versioned daily ROI metrics, policy, reporting, and approved execution."""

examples/auto_put_ad_mini/roi_control/config.py

@@ -0,0 +1,91 @@
+"""Configuration for daily ROI computation and approved execution."""
+
+from __future__ import annotations
+
+import os
+from dataclasses import asdict, dataclass
+from decimal import Decimal
+
+
+def env_flag(name: str, default: bool = False) -> bool:
+    raw = os.getenv(name)
+    if raw is None:
+        return default
+    return raw.strip().lower() in {"1", "true", "yes", "on"}
+
+
+@dataclass(frozen=True)
+class RoiConfig:
+    daily_enabled: bool = False
+    apply_enabled: bool = False
+    report_hour: int = 11
+    report_minute: int = 0
+    approval_ttl_minutes: int = 120
+    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_up_min_age: int = 3
+    self_min_avg_uv: float = 200
+    partner_min_avg_uv: float = 200
+    min_daily_cost: float = 100
+    stop_quantile: float = 0.10
+    up_quantile: float = 0.80
+    gzh_adjust_rank_min: float = 0.10
+    gzh_adjust_rank_max: float = 0.30
+
+    @classmethod
+    def from_env(cls) -> "RoiConfig":
+        config = cls(
+            daily_enabled=env_flag("DAILY_ROI_ENABLED"),
+            apply_enabled=env_flag("ROI_APPLY_ENABLED"),
+            report_hour=int(os.getenv("DAILY_ROI_HOUR", "11")),
+            report_minute=int(os.getenv("DAILY_ROI_MINUTE", "0")),
+            approval_ttl_minutes=int(
+                os.getenv("ROI_APPROVAL_TTL_MINUTES", "120")
+            ),
+            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_up_min_age=int(os.getenv("ROI_SELF_UP_MIN_AGE", "3")),
+            self_min_avg_uv=float(os.getenv("ROI_SELF_MIN_AVG_UV", "200")),
+            partner_min_avg_uv=float(
+                os.getenv("ROI_PARTNER_MIN_AVG_UV", "200")
+            ),
+            min_daily_cost=float(os.getenv("ROI_MIN_DAILY_COST", "100")),
+            stop_quantile=float(os.getenv("ROI_STOP_QUANTILE", "0.10")),
+            up_quantile=float(os.getenv("ROI_UP_QUANTILE", "0.80")),
+            gzh_adjust_rank_min=float(
+                os.getenv("ROI_GZH_ADJUST_RANK_MIN", "0.10")
+            ),
+            gzh_adjust_rank_max=float(
+                os.getenv("ROI_GZH_ADJUST_RANK_MAX", "0.30")
+            ),
+        )
+        config.validate()
+        return config
+
+    def validate(self) -> None:
+        if self.apply_enabled and not self.daily_enabled:
+            raise ValueError(
+                "ROI_APPLY_ENABLED=1 requires DAILY_ROI_ENABLED=1"
+            )
+        if not 0 <= self.report_hour <= 23 or not 0 <= self.report_minute <= 59:
+            raise ValueError("DAILY_ROI_HOUR/MINUTE is invalid")
+        if self.approval_ttl_minutes < 1:
+            raise ValueError("ROI_APPROVAL_TTL_MINUTES must be positive")
+        if self.scale_ratio <= 1:
+            raise ValueError("ROI_SCALE_RATIO must be greater than 1")
+        if self.max_base_ratio < self.scale_ratio:
+            raise ValueError("ROI_MAX_BASE_RATIO must cover ROI_SCALE_RATIO")
+        if self.scale_cooldown_days < 1:
+            raise ValueError("ROI_SCALE_COOLDOWN_DAYS must be positive")
+        if not 0 < self.stop_quantile < self.up_quantile < 1:
+            raise ValueError("ROI quantiles must satisfy 0 < stop < up < 1")
+
+    def snapshot(self) -> dict[str, object]:
+        values = asdict(self)
+        values["scale_ratio"] = str(self.scale_ratio)
+        values["max_base_ratio"] = str(self.max_base_ratio)
+        return values

examples/auto_put_ad_mini/roi_control/data_source.py

@@ -0,0 +1,179 @@
+"""ODPS 数据读取与日聚合 SQL。"""
+
+from __future__ import annotations
+
+from datetime import datetime, timedelta
+from typing import Tuple
+from zoneinfo import ZoneInfo
+
+import pandas as pd
+
+from .metrics import GZH_CHANNEL, QIWEI_CHANNEL, SELF_CHANNEL
+from .odps_client import ODPSClient
+
+
+TABLE_NAME = "loghubods.opengid_base_data"
+SHANGHAI = ZoneInfo("Asia/Shanghai")
+
+
+def parse_yyyymmdd(value: str) -> datetime:
+    try:
+        return datetime.strptime(value, "%Y%m%d")
+    except ValueError as exc:
+        raise ValueError(f"日期必须是 YYYYMMDD，实际为: {value}") from exc
+
+
+def date_window(end_date: str) -> Tuple[str, str]:
+    end = parse_yyyymmdd(end_date)
+    return (end - timedelta(days=2)).strftime("%Y%m%d"), end.strftime("%Y%m%d")
+
+
+def resolve_end_date(client: ODPSClient, requested: str | None = None) -> str:
+    """使用显式日期，或选择不晚于昨日的最新可用分区。"""
+
+    if requested:
+        parse_yyyymmdd(requested)
+        return requested
+
+    latest = (
+        client.odps.get_table("opengid_base_data")
+        .get_max_partition()
+        .partition_spec["dt"]
+    )
+    parse_yyyymmdd(latest)
+    yesterday = (datetime.now(SHANGHAI) - timedelta(days=1)).strftime("%Y%m%d")
+    return min(latest, yesterday)
+
+
+def build_daily_sql(start_date: str, end_date: str) -> str:
+    parse_yyyymmdd(start_date)
+    parse_yyyymmdd(end_date)
+    common_filter = f"""
+      dt BETWEEN '{start_date}' AND '{end_date}'
+      AND usersharedepth = '0'
+      AND videoid IS NOT NULL
+      AND NVL(hotsencetype, '') <> '1167'
+    """
+
+    return f"""
+SELECT
+  dt,
+  'self' AS entity_type,
+  channel,
+  MAX(NVL(代理名称, '')) AS 代理名称,
+  NVL(账号id, '') AS 账号id,
+  MAX(NVL(账号名称, '')) AS 账号名称,
+  NVL(广告id, '') AS 广告id,
+  MAX(NVL(广告名称, '')) AS 广告名称,
+  NVL(包名, '') AS 包名,
+  NVL(创意id, '') AS 创意id,
+  '' AS 合作方名,
+  '' AS 公众号名,
+  COUNT(DISTINCT mid) AS 首层UV,
+  SUM(NVL(t0裂变人数, 0)) AS T0裂变数,
+  SUM(NVL(成本, 0)) AS 成本,
+  SUM(NVL(效率收入, 0)) AS 效率收入,
+  SUM(NVL(多层t0裂变收入, 0)) AS 多层裂变收入
+FROM {TABLE_NAME}
+WHERE {common_filter}
+  AND channel = '{SELF_CHANNEL}'
+  AND 广告id IS NOT NULL
+  AND 创意id IS NOT NULL
+GROUP BY
+  dt, channel, 账号id, 广告id, 包名, 创意id
+
+UNION ALL
+
+SELECT
+  dt,
+  'gzh' AS entity_type,
+  channel,
+  '' AS 代理名称,
+  '' AS 账号id,
+  '' AS 账号名称,
+  '' AS 广告id,
+  '' AS 广告名称,
+  '' AS 包名,
+  '' AS 创意id,
+  NVL(合作方名, '') AS 合作方名,
+  NVL(公众号名, '') AS 公众号名,
+  COUNT(DISTINCT mid) AS 首层UV,
+  SUM(NVL(t0裂变人数, 0)) AS T0裂变数,
+  SUM(NVL(成本, 0)) AS 成本,
+  SUM(NVL(效率收入, 0)) AS 效率收入,
+  SUM(NVL(多层t0裂变收入, 0)) AS 多层裂变收入
+FROM {TABLE_NAME}
+WHERE {common_filter}
+  AND channel = '{GZH_CHANNEL}'
+  AND 公众号名 IS NOT NULL
+GROUP BY dt, channel, 合作方名, 公众号名
+
+UNION ALL
+
+SELECT
+  dt,
+  'qiwei' AS entity_type,
+  channel,
+  '' AS 代理名称,
+  '' AS 账号id,
+  '' AS 账号名称,
+  '' AS 广告id,
+  '' AS 广告名称,
+  '' AS 包名,
+  '' AS 创意id,
+  NVL(合作方名, '') AS 合作方名,
+  MAX(NVL(公众号名, '')) AS 公众号名,
+  COUNT(DISTINCT mid) AS 首层UV,
+  SUM(NVL(t0裂变人数, 0)) AS T0裂变数,
+  SUM(NVL(成本, 0)) AS 成本,
+  SUM(NVL(效率收入, 0)) AS 效率收入,
+  SUM(NVL(多层t0裂变收入, 0)) AS 多层裂变收入
+FROM {TABLE_NAME}
+WHERE {common_filter}
+  AND channel = '{QIWEI_CHANNEL}'
+  AND 合作方名 IS NOT NULL
+GROUP BY dt, channel, 合作方名
+""".strip()
+
+
+def build_ad_age_sql(end_date: str, lookback_days: int = 30) -> str:
+    end = parse_yyyymmdd(end_date)
+    start_date = (end - timedelta(days=lookback_days - 1)).strftime("%Y%m%d")
+    return f"""
+SELECT
+  广告id,
+  MIN(dt) AS 首次出现日期
+FROM {TABLE_NAME}
+WHERE dt BETWEEN '{start_date}' AND '{end_date}'
+  AND usersharedepth = '0'
+  AND channel = '{SELF_CHANNEL}'
+  AND videoid IS NOT NULL
+  AND NVL(hotsencetype, '') <> '1167'
+  AND 广告id IS NOT NULL
+GROUP BY 广告id
+""".strip()
+
+
+def fetch_daily_data(
+    client: ODPSClient,
+    start_date: str,
+    end_date: str,
+) -> pd.DataFrame:
+    return client.execute_sql(build_daily_sql(start_date, end_date))
+
+
+def fetch_ad_age(
+    client: ODPSClient,
+    end_date: str,
+    lookback_days: int = 30,
+) -> pd.DataFrame:
+    raw = client.execute_sql(build_ad_age_sql(end_date, lookback_days))
+    if raw.empty:
+        return pd.DataFrame(columns=["广告id", "广告age"])
+
+    end = parse_yyyymmdd(end_date)
+    result = raw.copy()
+    result["广告id"] = result["广告id"].fillna("").astype(str)
+    first_seen = pd.to_datetime(result["首次出现日期"].astype(str), format="%Y%m%d")
+    result["广告age"] = (end - first_seen).dt.days + 1
+    return result[["广告id", "广告age"]]

examples/auto_put_ad_mini/roi_control/execution.py

@@ -0,0 +1,469 @@
+"""Execute an approved ROI batch with Tencent read-back and state coordination."""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+from datetime import date, datetime, timedelta
+from decimal import Decimal, ROUND_HALF_UP
+from typing import Any
+
+from storage import (
+    advisory_lock,
+    load_ad_states,
+    load_managed_accounts,
+    sync_roi_base_bid,
+)
+from tencent_client import (
+    ACTIVE_STATUS,
+    SUSPEND_STATUS,
+    PostWriteVerificationError,
+    TencentClient,
+    TencentWriteOutcomeUnknownError,
+    current_bid_fen,
+    resolve_bid_field,
+)
+
+from .config import RoiConfig
+from .policy import ACTION_PAUSE_CREATIVE, ACTION_SCALE_BID
+from .repository import (
+    FINAL_STATUSES,
+    claim_run_for_execution,
+    finalize_run,
+    load_action_items,
+    load_run,
+    reject_run,
+    update_action_item,
+)
+
+
+logger = logging.getLogger("auto_put_ad_mini.roi_execution")
+
+
+def _round_fen(value: Decimal) -> int:
+    return int(value.quantize(Decimal("1"), rounding=ROUND_HALF_UP))
+
+
+def _managed_accounts() -> dict[int, dict[str, Any]]:
+    return {
+        int(row["account_id"]): row
+        for row in load_managed_accounts()
+    }
+
+
+def _failure_status(exc: Exception) -> str:
+    if isinstance(exc, PostWriteVerificationError):
+        return "VERIFY_FAILED"
+    if isinstance(exc, TencentWriteOutcomeUnknownError):
+        return "OUTCOME_UNKNOWN"
+    return "FAILED"
+
+
+def plan_scale_bid(
+    *,
+    current_bid_fen: int,
+    base_bid_fen: int,
+    initial_base_bid_fen: int,
+    boosted_date: date | None,
+    control_date: date,
+    intraday_ratio: Decimal,
+    scale_ratio: Decimal,
+    max_base_ratio: Decimal,
+) -> dict[str, Any]:
+    """Plan a permanent base increase without reapplying a restored boost."""
+
+    boosted_date_today = boosted_date == control_date
+    expected_boosted = _round_fen(Decimal(base_bid_fen) * intraday_ratio)
+    actively_boosted = boosted_date_today and current_bid_fen == expected_boosted
+    if current_bid_fen != base_bid_fen and not actively_boosted:
+        return {
+            "allowed": False,
+            "reason": (
+                f"current bid={current_bid_fen} does not match managed "
+                f"base={base_bid_fen} or intraday boost={expected_boosted}"
+            ),
+        }
+
+    max_base = _round_fen(Decimal(initial_base_bid_fen) * max_base_ratio)
+    new_base = min(
+        _round_fen(Decimal(base_bid_fen) * scale_ratio),
+        max_base,
+    )
+    if new_base <= base_bid_fen:
+        return {
+            "allowed": False,
+            "reason": f"permanent base reached cap={max_base}",
+            "cap_reached": True,
+            "max_base_bid_fen": max_base,
+        }
+    target = (
+        _round_fen(Decimal(new_base) * intraday_ratio)
+        if actively_boosted
+        else new_base
+    )
+    return {
+        "allowed": True,
+        "new_base_bid_fen": new_base,
+        "target_bid_fen": target,
+        "boosted_date_today": boosted_date_today,
+        "preserve_intraday_boost": actively_boosted,
+    }
+
+
+def _execute_pause(
+    item: dict[str, Any],
+    *,
+    client: TencentClient,
+    now: datetime,
+) -> None:
+    account_id = int(item["account_id"])
+    adgroup_id = int(item["adgroup_id"])
+    creative_id = int(item["dynamic_creative_id"])
+    creative = client.get_dynamic_creative(account_id, creative_id)
+    actual_adgroup_id = int(creative.get("adgroup_id") or 0)
+    before_status = str(creative.get("configured_status") or "")
+    if actual_adgroup_id != adgroup_id:
+        update_action_item(
+            int(item["id"]),
+            execution_status="SKIPPED_MAPPING_MISMATCH",
+            skip_reason=(
+                f"creative belongs to adgroup={actual_adgroup_id}, expected={adgroup_id}"
+            ),
+            pre_state_json=creative,
+            executed_at=now,
+        )
+        return
+    if item["execution_status"] == "PREPARED" and before_status == SUSPEND_STATUS:
+        update_action_item(
+            int(item["id"]),
+            before_status=item.get("before_status"),
+            target_status=SUSPEND_STATUS,
+            readback_status=before_status,
+            execution_status="SUCCESS",
+            readback_json=creative,
+            executed_at=now,
+        )
+        return
+    if before_status != ACTIVE_STATUS:
+        update_action_item(
+            int(item["id"]),
+            before_status=before_status,
+            target_status=SUSPEND_STATUS,
+            readback_status=before_status,
+            execution_status="SKIPPED_STATE_MISMATCH",
+            skip_reason="creative is no longer active",
+            pre_state_json=creative,
+            executed_at=now,
+        )
+        return
+
+    update_action_item(
+        int(item["id"]),
+        before_status=before_status,
+        target_status=SUSPEND_STATUS,
+        execution_status="PREPARED",
+        pre_state_json=creative,
+    )
+    try:
+        readback = client.update_dynamic_creative_status(
+            account_id,
+            creative_id,
+            SUSPEND_STATUS,
+        )
+        update_action_item(
+            int(item["id"]),
+            readback_status=str(readback.get("configured_status") or ""),
+            execution_status="SUCCESS",
+            readback_json=readback,
+            executed_at=now,
+        )
+    except Exception as exc:
+        update_action_item(
+            int(item["id"]),
+            execution_status=_failure_status(exc),
+            error_message=str(exc),
+            readback_json=(exc.actual if isinstance(exc, PostWriteVerificationError) else None),
+            executed_at=now,
+        )
+
+
+def _resume_prepared_scale(
+    item: dict[str, Any],
+    ad: dict[str, Any],
+    *,
+    client: TencentClient,
+    now: datetime,
+) -> None:
+    details = json.loads(item.get("pre_state_json") or "{}")
+    bid_field = str(item.get("bid_field") or "")
+    current = current_bid_fen(ad, bid_field)
+    before = int(item.get("before_bid_fen") or 0)
+    target = int(item.get("target_bid_fen") or 0)
+    if current == target:
+        sync_roi_base_bid(
+            account_id=int(item["account_id"]),
+            adgroup_id=int(item["adgroup_id"]),
+            adgroup_name=str(ad.get("adgroup_name") or ""),
+            bid_field=bid_field,
+            initial_base_bid_fen=int(item["initial_base_bid_fen"]),
+            new_base_bid_fen=int(details["new_base_bid_fen"]),
+            boosted_date=(
+                now.date() if details.get("boosted_date_today") else None
+            ),
+            action_at=now,
+        )
+        update_action_item(
+            int(item["id"]),
+            execution_status="SUCCESS",
+            readback_json=ad,
+            executed_at=now,
+        )
+        return
+    if current != before:
+        update_action_item(
+            int(item["id"]),
+            execution_status="SKIPPED_STATE_MISMATCH",
+            skip_reason=f"bid changed after preparation: before={before} current={current}",
+            readback_json=ad,
+            executed_at=now,
+        )
+        return
+    readback = client.update_ad(
+        int(item["account_id"]),
+        int(item["adgroup_id"]),
+        bid_field=bid_field,
+        target_bid_fen=target,
+    )
+    sync_roi_base_bid(
+        account_id=int(item["account_id"]),
+        adgroup_id=int(item["adgroup_id"]),
+        adgroup_name=str(ad.get("adgroup_name") or ""),
+        bid_field=bid_field,
+        initial_base_bid_fen=int(item["initial_base_bid_fen"]),
+        new_base_bid_fen=int(details["new_base_bid_fen"]),
+        boosted_date=(now.date() if details.get("boosted_date_today") else None),
+        action_at=now,
+    )
+    update_action_item(
+        int(item["id"]),
+        execution_status="SUCCESS",
+        readback_json=readback,
+        executed_at=now,
+    )
+
+
+def _execute_scale(
+    item: dict[str, Any],
+    account: dict[str, Any],
+    *,
+    client: TencentClient,
+    config: RoiConfig,
+    now: datetime,
+) -> None:
+    account_id = int(item["account_id"])
+    adgroup_id = int(item["adgroup_id"])
+    ad = client.get_ad(account_id, adgroup_id)
+    if str(ad.get("configured_status") or "") != ACTIVE_STATUS:
+        update_action_item(
+            int(item["id"]),
+            execution_status="SKIPPED_STATE_MISMATCH",
+            skip_reason="ad is no longer active",
+            pre_state_json=ad,
+            executed_at=now,
+        )
+        return
+    if item["execution_status"] == "PREPARED":
+        try:
+            _resume_prepared_scale(item, ad, client=client, now=now)
+        except Exception as exc:
+            update_action_item(
+                int(item["id"]),
+                execution_status=_failure_status(exc),
+                error_message=str(exc),
+                executed_at=now,
+            )
+        return
+
+    states = load_ad_states(account_id)
+    state = states.get(adgroup_id) or {}
+    last_scaled = state.get("last_roi_scaled_at")
+    if last_scaled and now < last_scaled + timedelta(days=config.scale_cooldown_days):
+        update_action_item(
+            int(item["id"]),
+            execution_status="SKIPPED_COOLDOWN",
+            skip_reason=f"last ROI scale at {last_scaled}",
+            pre_state_json=ad,
+            executed_at=now,
+        )
+        return
+
+    bid_field = resolve_bid_field(ad, account.get("bid_scene"))
+    current = current_bid_fen(ad, bid_field)
+    if current is None or current <= 0:
+        raise ValueError(f"ad has no valid {bid_field}")
+    base = int(state.get("base_bid_fen") or current)
+    initial = int(state.get("initial_base_bid_fen") or base)
+    intraday_ratio = Decimal(os.getenv("RTC_BID_UP_RATIO", "1.25"))
+    plan = plan_scale_bid(
+        current_bid_fen=current,
+        base_bid_fen=base,
+        initial_base_bid_fen=initial,
+        boosted_date=state.get("boosted_date"),
+        control_date=now.date(),
+        intraday_ratio=intraday_ratio,
+        scale_ratio=config.scale_ratio,
+        max_base_ratio=config.max_base_ratio,
+    )
+    if not plan["allowed"]:
+        update_action_item(
+            int(item["id"]),
+            bid_field=bid_field,
+            initial_base_bid_fen=initial,
+            base_bid_fen=base,
+            before_bid_fen=current,
+            target_bid_fen=current,
+            execution_status=(
+                "SKIPPED_CAP_REACHED"
+                if plan.get("cap_reached")
+                else "SKIPPED_STATE_MISMATCH"
+            ),
+            skip_reason=str(plan["reason"]),
+            pre_state_json=ad,
+            executed_at=now,
+        )
+        return
+    new_base = int(plan["new_base_bid_fen"])
+    target = int(plan["target_bid_fen"])
+    boosted_date_today = bool(plan["boosted_date_today"])
+    prepared_state = {
+        **ad,
+        "managed_base_bid_fen": base,
+        "new_base_bid_fen": new_base,
+        "boosted_date_today": boosted_date_today,
+        "preserve_intraday_boost": bool(plan["preserve_intraday_boost"]),
+    }
+    update_action_item(
+        int(item["id"]),
+        bid_field=bid_field,
+        initial_base_bid_fen=initial,
+        base_bid_fen=base,
+        before_bid_fen=current,
+        target_bid_fen=target,
+        execution_status="PREPARED",
+        pre_state_json=prepared_state,
+    )
+    try:
+        readback = client.update_ad(
+            account_id,
+            adgroup_id,
+            bid_field=bid_field,
+            target_bid_fen=target,
+        )
+        sync_roi_base_bid(
+            account_id=account_id,
+            adgroup_id=adgroup_id,
+            adgroup_name=str(ad.get("adgroup_name") or ""),
+            bid_field=bid_field,
+            initial_base_bid_fen=initial,
+            new_base_bid_fen=new_base,
+            boosted_date=now.date() if boosted_date_today else None,
+            action_at=now,
+        )
+        update_action_item(
+            int(item["id"]),
+            execution_status="SUCCESS",
+            readback_json=readback,
+            executed_at=now,
+        )
+    except Exception as exc:
+        update_action_item(
+            int(item["id"]),
+            execution_status=_failure_status(exc),
+            error_message=str(exc),
+            readback_json=(exc.actual if isinstance(exc, PostWriteVerificationError) else None),
+            executed_at=now,
+        )
+
+
+def execute_roi_batch(
+    run_id: str,
+    *,
+    sender_open_id: str,
+    now: datetime,
+    lock_name: str,
+    tencent: TencentClient | None = None,
+) -> dict[str, Any]:
+    config = RoiConfig.from_env()
+    if not config.apply_enabled:
+        raise RuntimeError("ROI_APPLY_ENABLED=0, ROI Tencent writes are disabled")
+    with advisory_lock(lock_name) as acquired:
+        if not acquired:
+            raise RuntimeError("实时控制正在执行，请稍后再次确认 ROI 批次")
+        run = claim_run_for_execution(
+            run_id,
+            sender_open_id=sender_open_id,
+            now=now,
+        )
+        if run.get("status") in FINAL_STATUSES:
+            return run
+        accounts = _managed_accounts()
+        client = tencent or TencentClient()
+        for item in load_action_items(run_id):
+            if item["execution_status"] not in {"PENDING", "PREPARED"}:
+                continue
+            account_id = int(item["account_id"])
+            account = accounts.get(account_id)
+            if not account:
+                update_action_item(
+                    int(item["id"]),
+                    execution_status="SKIPPED_NOT_MANAGED",
+                    skip_reason="account left automated management scope",
+                    executed_at=now,
+                )
+                continue
+            try:
+                if item["action_type"] == ACTION_PAUSE_CREATIVE:
+                    _execute_pause(item, client=client, now=now)
+                elif item["action_type"] == ACTION_SCALE_BID:
+                    _execute_scale(
+                        item,
+                        account,
+                        client=client,
+                        config=config,
+                        now=now,
+                    )
+                else:
+                    update_action_item(
+                        int(item["id"]),
+                        execution_status="SKIPPED_UNSUPPORTED",
+                        skip_reason=f"unsupported action={item['action_type']}",
+                        executed_at=now,
+                    )
+            except Exception as exc:
+                logger.exception(
+                    "ROI action failed run=%s item=%s", run_id, item["id"]
+                )
+                update_action_item(
+                    int(item["id"]),
+                    execution_status=_failure_status(exc),
+                    error_message=str(exc),
+                    executed_at=now,
+                )
+        return {"run_id": run_id, **finalize_run(run_id, now=now)}
+
+
+def reject_roi_batch(
+    run_id: str,
+    *,
+    sender_open_id: str,
+    now: datetime,
+) -> dict[str, Any]:
+    return reject_run(run_id, sender_open_id=sender_open_id, now=now)
+
+
+def roi_batch_status(run_id: str) -> dict[str, Any]:
+    run = load_run(run_id)
+    if not run:
+        raise ValueError(f"ROI批次不存在: {run_id}")
+    return run

examples/auto_put_ad_mini/roi_control/feishu.py

@@ -0,0 +1,204 @@
+"""Publish read-only ROI workbooks and whole-batch approval instructions."""
+
+from __future__ import annotations
+
+import json
+import os
+import time
+from pathlib import Path
+from typing import Any
+
+import httpx
+
+
+BASE_URL = "https://open.feishu.cn/open-apis"
+
+
+class RoiFeishuPublisher:
+    def __init__(self, timeout: float = 30.0) -> None:
+        self.app_id = os.getenv("FEISHU_APP_ID", "").strip()
+        self.app_secret = os.getenv("FEISHU_APP_SECRET", "").strip()
+        self.chat_id = (
+            os.getenv("ROI_FEISHU_CHAT_ID", "").strip()
+            or os.getenv("RTC_COMMAND_CHAT_ID", "").strip()
+            or os.getenv("FEISHU_AD_PROJECT_CHAT_ID", "").strip()
+        )
+        missing = [
+            name
+            for name, value in {
+                "FEISHU_APP_ID": self.app_id,
+                "FEISHU_APP_SECRET": self.app_secret,
+                "ROI_FEISHU_CHAT_ID/RTC_COMMAND_CHAT_ID/FEISHU_AD_PROJECT_CHAT_ID": self.chat_id,
+            }.items()
+            if not value
+        ]
+        if missing:
+            raise RuntimeError(f"Missing Feishu configuration: {', '.join(missing)}")
+        self.client = httpx.Client(timeout=timeout)
+
+    def close(self) -> None:
+        self.client.close()
+
+    @staticmethod
+    def _json(response: httpx.Response, action: str) -> dict[str, Any]:
+        response.raise_for_status()
+        payload = response.json()
+        if payload.get("code") != 0:
+            raise RuntimeError(f"{action} failed: {payload.get('msg', payload)}")
+        return payload
+
+    def _token(self) -> str:
+        response = self.client.post(
+            f"{BASE_URL}/auth/v3/tenant_access_token/internal",
+            json={"app_id": self.app_id, "app_secret": self.app_secret},
+        )
+        return self._json(response, "get tenant token")["tenant_access_token"]
+
+    @staticmethod
+    def _headers(token: str) -> dict[str, str]:
+        return {"Authorization": f"Bearer {token}"}
+
+    def _upload(self, token: str, path: Path) -> str:
+        with path.open("rb") as handle:
+            response = self.client.post(
+                f"{BASE_URL}/drive/v1/medias/upload_all",
+                headers=self._headers(token),
+                data={
+                    "file_name": path.name,
+                    "parent_type": "explorer",
+                    "parent_node": "",
+                    "size": str(path.stat().st_size),
+                },
+                files={"file": (path.name, handle, "application/octet-stream")},
+                timeout=60,
+            )
+        return self._json(response, "upload ROI workbook")["data"]["file_token"]
+
+    def _import_sheet(self, token: str, file_token: str, path: Path) -> str:
+        response = self.client.post(
+            f"{BASE_URL}/drive/v1/import_tasks",
+            headers={**self._headers(token), "Content-Type": "application/json"},
+            json={
+                "file_extension": "xlsx",
+                "file_token": file_token,
+                "type": "sheet",
+                "file_name": path.stem,
+                "point": {"mount_type": 1, "mount_key": ""},
+            },
+        )
+        return self._json(response, "create ROI sheet import")["data"]["ticket"]
+
+    def _wait_import(self, token: str, ticket: str) -> dict[str, Any]:
+        deadline = time.monotonic() + 60
+        while time.monotonic() < deadline:
+            response = self.client.get(
+                f"{BASE_URL}/drive/v1/import_tasks/{ticket}",
+                headers=self._headers(token),
+            )
+            result = (
+                self._json(response, "wait ROI sheet import")
+                .get("data", {})
+                .get("result", {})
+            )
+            if result.get("job_status") == 0:
+                return result
+            if result.get("job_status") == 3:
+                raise RuntimeError(
+                    f"ROI sheet import failed: {result.get('job_error_msg', 'unknown')}"
+                )
+            time.sleep(2)
+        raise RuntimeError("ROI sheet import timed out after 60 seconds")
+
+    def _set_read_only_link(self, token: str, sheet_token: str) -> None:
+        response = self.client.patch(
+            f"{BASE_URL}/drive/v1/permissions/{sheet_token}/public",
+            headers={**self._headers(token), "Content-Type": "application/json"},
+            params={"type": "sheet"},
+            json={
+                "external_access_entity": "open",
+                "link_share_entity": "anyone_readable",
+            },
+        )
+        self._json(response, "set ROI sheet read-only permission")
+
+    def _send_card(
+        self,
+        token: str,
+        *,
+        run_id: str,
+        url: str,
+        summary: str,
+        requires_approval: bool,
+    ) -> str:
+        instructions = (
+            f"\n\n请在有效期内回复：`确认 {run_id}` 或 `拒绝 {run_id}`"
+            if requires_approval
+            else "\n\n本批次没有可执行动作，无需审批。"
+        )
+        card = {
+            "config": {"wide_screen_mode": True},
+            "header": {
+                "template": "orange" if requires_approval else "blue",
+                "title": {"tag": "plain_text", "content": "日级 ROI 调控"},
+            },
+            "elements": [
+                {
+                    "tag": "div",
+                    "text": {
+                        "tag": "lark_md",
+                        "content": f"批次：`{run_id}`\n{summary}{instructions}",
+                    },
+                },
+                {"tag": "hr"},
+                {
+                    "tag": "action",
+                    "actions": [
+                        {
+                            "tag": "button",
+                            "type": "primary",
+                            "text": {"tag": "plain_text", "content": "查看 ROI 明细"},
+                            "url": url,
+                        }
+                    ],
+                },
+            ],
+        }
+        response = self.client.post(
+            f"{BASE_URL}/im/v1/messages",
+            headers={**self._headers(token), "Content-Type": "application/json"},
+            params={"receive_id_type": "chat_id"},
+            json={
+                "receive_id": self.chat_id,
+                "msg_type": "interactive",
+                "content": json.dumps(card, ensure_ascii=False),
+            },
+        )
+        return self._json(response, "send ROI batch card")["data"]["message_id"]
+
+    def publish(
+        self,
+        path: Path,
+        *,
+        run_id: str,
+        summary: str,
+        requires_approval: bool,
+    ) -> dict[str, str]:
+        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_read_only_link(token, sheet_token)
+        message_id = self._send_card(
+            token,
+            run_id=run_id,
+            url=url,
+            summary=summary,
+            requires_approval=requires_approval,
+        )
+        return {"url": url, "sheet_token": sheet_token, "message_id": message_id}

examples/auto_put_ad_mini/roi_control/metrics.py

@@ -0,0 +1,220 @@
+"""Versioned north-star ROI metric computation.
+
+This module only transforms pandas DataFrames. It does not know about policy,
+approval, Tencent writes, storage, scheduling, or Feishu.
+"""
+
+from __future__ import annotations
+
+from typing import Dict, Iterable, List, Mapping, Sequence
+
+import numpy as np
+import pandas as pd
+
+
+METRIC_VERSION = "north_star_roi_v1"
+
+SELF_CHANNEL = "小程序投流-稳定"
+GZH_CHANNEL = "公众号合作-即转-稳定"
+QIWEI_CHANNEL = "群/企微合作-稳定"
+
+ENTITY_SELF = "self"
+ENTITY_GZH = "gzh"
+ENTITY_QIWEI = "qiwei"
+
+DIMENSION_COLUMNS = [
+    "channel",
+    "代理名称",
+    "账号id",
+    "账号名称",
+    "广告id",
+    "广告名称",
+    "包名",
+    "创意id",
+    "合作方名",
+    "公众号名",
+]
+
+NUMERIC_COLUMNS = [
+    "首层UV",
+    "T0裂变数",
+    "成本",
+    "效率收入",
+    "多层裂变收入",
+]
+
+ENTITY_KEYS: Mapping[str, Sequence[str]] = {
+    ENTITY_SELF: (
+        "channel",
+        "账号id",
+        "广告id",
+        "包名",
+        "创意id",
+    ),
+    ENTITY_GZH: ("channel", "合作方名", "公众号名"),
+    ENTITY_QIWEI: ("channel", "合作方名"),
+}
+
+def prepare_daily_metrics(raw: pd.DataFrame) -> pd.DataFrame:
+    """清洗 ODPS 日聚合结果并计算统一 ROI。"""
+
+    raw = raw.rename(
+        columns={
+            "首层uv": "首层UV",
+            "t0裂变数": "T0裂变数",
+        }
+    )
+    required = {"dt", "entity_type", *DIMENSION_COLUMNS, *NUMERIC_COLUMNS}
+    missing = sorted(required - set(raw.columns))
+    if missing:
+        raise ValueError(f"日聚合数据缺少字段: {', '.join(missing)}")
+
+    daily = raw.copy()
+    daily["dt"] = daily["dt"].astype(str)
+    daily["entity_type"] = daily["entity_type"].astype(str)
+
+    for column in DIMENSION_COLUMNS:
+        daily[column] = daily[column].fillna("").astype(str)
+    for column in NUMERIC_COLUMNS:
+        daily[column] = pd.to_numeric(daily[column], errors="coerce").fillna(0.0)
+
+    daily["全链路效率收入"] = daily["效率收入"] + daily["多层裂变收入"]
+    daily["ROI"] = np.where(
+        daily["成本"] > 0,
+        daily["全链路效率收入"] / daily["成本"],
+        np.nan,
+    )
+    daily["T0裂变率"] = np.where(
+        daily["首层UV"] > 0,
+        daily["T0裂变数"] / daily["首层UV"],
+        np.nan,
+    )
+    return daily
+
+
+def _age_map(ad_age: pd.DataFrame | None) -> Dict[str, int]:
+    if ad_age is None or ad_age.empty:
+        return {}
+    required = {"广告id", "广告age"}
+    missing = required - set(ad_age.columns)
+    if missing:
+        raise ValueError(f"广告age数据缺少字段: {', '.join(sorted(missing))}")
+    ages = ad_age.copy()
+    ages["广告id"] = ages["广告id"].fillna("").astype(str)
+    ages["广告age"] = pd.to_numeric(ages["广告age"], errors="coerce").fillna(0).astype(int)
+    return dict(zip(ages["广告id"], ages["广告age"]))
+
+
+def _expected_three_days(expected_dates: Iterable[str]) -> List[str]:
+    dates = sorted({str(value) for value in expected_dates})
+    if len(dates) != 3:
+        raise ValueError(f"必须提供连续三个日期，实际为: {dates}")
+    parsed = pd.to_datetime(dates, format="%Y%m%d")
+    gaps = parsed.to_series().diff().dropna().dt.days.tolist()
+    if gaps != [1, 1]:
+        raise ValueError(f"日期必须连续，实际为: {dates}")
+    return dates
+
+
+def _summarize_entity(
+    entity_daily: pd.DataFrame,
+    entity_type: str,
+    expected_dates: Sequence[str],
+    ages: Mapping[str, int],
+) -> pd.DataFrame:
+    keys = list(ENTITY_KEYS[entity_type])
+    records: List[dict] = []
+
+    for key_values, group in entity_daily.groupby(keys, dropna=False, sort=False):
+        if not isinstance(key_values, tuple):
+            key_values = (key_values,)
+        by_date = group.groupby("dt", as_index=False)[
+            ["首层UV", "T0裂变数", "成本", "效率收入", "多层裂变收入"]
+        ].sum()
+        if set(by_date["dt"]) != set(expected_dates):
+            continue
+
+        by_date = by_date.set_index("dt").loc[list(expected_dates)]
+        daily_roi = np.where(
+            by_date["成本"] > 0,
+            (by_date["效率收入"] + by_date["多层裂变收入"]) / by_date["成本"],
+            np.nan,
+        )
+
+        record = dict(zip(keys, key_values))
+        for column in DIMENSION_COLUMNS:
+            if column in record:
+                continue
+            values = group.sort_values("dt")[column].dropna().astype(str)
+            record[column] = next(
+                (value for value in reversed(values.tolist()) if value),
+                "",
+            )
+        record["entity_type"] = entity_type
+        record["覆盖天数"] = 3
+        record["三日首层UV"] = float(by_date["首层UV"].sum())
+        record["日均首层UV"] = float(by_date["首层UV"].mean())
+        record["三日T0裂变数"] = float(by_date["T0裂变数"].sum())
+        record["T0裂变率"] = (
+            record["三日T0裂变数"] / record["三日首层UV"]
+            if record["三日首层UV"] > 0
+            else np.nan
+        )
+        record["成本"] = float(by_date["成本"].sum())
+        record["三日最小单日成本"] = float(by_date["成本"].min())
+        record["效率收入"] = float(by_date["效率收入"].sum())
+        record["多层裂变收入"] = float(by_date["多层裂变收入"].sum())
+        record["全链路效率收入"] = record["效率收入"] + record["多层裂变收入"]
+        record["ROI"] = (
+            record["全链路效率收入"] / record["成本"]
+            if record["成本"] > 0
+            else np.nan
+        )
+        for index, dt in enumerate(expected_dates):
+            roi_value = float(daily_roi[index]) if np.isfinite(daily_roi[index]) else np.nan
+            record[f"首层UV_{dt}"] = float(by_date.iloc[index]["首层UV"])
+            record[f"T0裂变数_{dt}"] = float(by_date.iloc[index]["T0裂变数"])
+            record[f"成本_{dt}"] = float(by_date.iloc[index]["成本"])
+            record[f"效率收入_{dt}"] = float(by_date.iloc[index]["效率收入"])
+            record[f"多层裂变收入_{dt}"] = float(
+                by_date.iloc[index]["多层裂变收入"]
+            )
+            record[f"ROI_{dt}"] = roi_value
+
+        if entity_type == ENTITY_SELF:
+            record["广告age"] = int(ages.get(str(record.get("广告id", "")), 0))
+        records.append(record)
+
+    return pd.DataFrame(records)
+
+
+def compute_roi_summary(
+    raw_daily: pd.DataFrame,
+    expected_dates: Iterable[str],
+    ad_age: pd.DataFrame | None = None,
+) -> tuple[pd.DataFrame, list[str]]:
+    """Return complete three-day entity ROI snapshots and normalized dates."""
+
+    dates = _expected_three_days(expected_dates)
+    daily = prepare_daily_metrics(raw_daily)
+    daily = daily[daily["dt"].isin(dates)].copy()
+    ages = _age_map(ad_age)
+    summaries = []
+    for entity_type in (ENTITY_SELF, ENTITY_GZH, ENTITY_QIWEI):
+        entity_daily = daily[daily["entity_type"].eq(entity_type)]
+        if entity_daily.empty:
+            continue
+        entity_summary = _summarize_entity(
+            entity_daily,
+            entity_type,
+            dates,
+            ages,
+        )
+        if not entity_summary.empty:
+            summaries.append(entity_summary)
+
+    if not summaries:
+        raise ValueError("三个目标渠道均没有可用的连续三日数据")
+
+    summary = pd.concat(summaries, ignore_index=True, sort=False)
+    return summary, dates

examples/auto_put_ad_mini/roi_control/odps_client.py

@@ -0,0 +1,53 @@
+#!/usr/bin/env python
+# coding=utf-8
+
+import os
+
+from odps import ODPS, options
+
+
+class ODPSClient(object):
+    def __init__(self, project="loghubods"):
+        self.accessId = os.environ.get("ODPS_ACCESS_ID")
+        self.accessSecret = os.environ.get("ODPS_ACCESS_SECRET")
+        if not self.accessId or not self.accessSecret:
+            raise RuntimeError(
+                "请设置 ODPS_ACCESS_ID 和 ODPS_ACCESS_SECRET 环境变量"
+            )
+
+        self.endpoint = os.getenv(
+            "ODPS_ENDPOINT", "http://service.odps.aliyun.com/api"
+        )
+
+        # 调长超时，应对复杂 SQL 长时间运行（20 分钟）
+        options.connect_timeout = 60
+        options.read_timeout = 1200
+        options.retry_times = 3
+
+        self.odps = ODPS(
+            self.accessId,
+            self.accessSecret,
+            project,
+            self.endpoint,
+        )
+
+    def execute_sql(self, sql: str):
+        hints = {
+            'odps.sql.submit.mode': 'script'
+        }
+        # 异步提交以便先打印 instanceId / logview，方便用户在 ODPS 控制台跟踪进度
+        inst = self.odps.run_sql(sql, hints=hints)
+        try:
+            logview = inst.get_logview_address()
+        except Exception:
+            logview = '(获取 logview 失败)'
+        print(f'[ODPS] InstanceId: {inst.id}', flush=True)
+        print(f'[ODPS] LogView   : {logview}', flush=True)
+        inst.wait_for_success()
+        with inst.open_reader(tunnel=True) as reader:
+            pd_df = reader.to_pandas()
+        return pd_df
+
+    def execute_sql_result_save_file(self, sql: str, output_file: str):
+        data_df = self.execute_sql(sql)
+        data_df.to_csv(output_file, index=False)

examples/auto_put_ad_mini/roi_control/policy.py

@@ -0,0 +1,162 @@
+"""Translate versioned ROI metrics into auditable recommendations and actions."""
+
+from __future__ import annotations
+
+import hashlib
+import math
+from typing import Any
+
+import pandas as pd
+
+from .metrics import ENTITY_KEYS, ENTITY_SELF
+
+
+ACTION_PAUSE_CREATIVE = "PAUSE_CREATIVE"
+ACTION_SCALE_BID = "SCALE_BID"
+MODE_ACTIONABLE = "ACTIONABLE"
+MODE_NOTIFY_ONLY = "NOTIFY_ONLY"
+
+
+def safe_int(value: Any) -> int | None:
+    if value is None or value == "":
+        return None
+    try:
+        number = int(float(value))
+    except (TypeError, ValueError):
+        return None
+    return number if number > 0 else None
+
+
+def _entity_hash(row: pd.Series) -> str:
+    keys = ENTITY_KEYS[str(row["entity_type"])]
+    raw = "\x1f".join(str(row.get(key, "")) for key in keys)
+    return hashlib.sha256(raw.encode("utf-8")).hexdigest()
+
+
+def _finite(value: Any) -> float | None:
+    try:
+        number = float(value)
+    except (TypeError, ValueError):
+        return None
+    return number if math.isfinite(number) else None
+
+
+def annotate_execution(
+    summary: pd.DataFrame,
+    managed_account_ids: set[int],
+    run_id: str,
+) -> tuple[pd.DataFrame, list[dict[str, Any]], list[dict[str, Any]]]:
+    """Keep policy recommendations separate from Tencent eligibility.
+
+    Returns annotated summary, complete metric snapshots, and deduplicated
+    executable action items.
+    """
+
+    annotated = summary.copy()
+    annotated["执行模式"] = MODE_NOTIFY_ONLY
+    annotated["执行说明"] = "无调控建议"
+    actions: list[dict[str, Any]] = []
+    snapshots: list[dict[str, Any]] = []
+    seen_actions: set[str] = set()
+
+    for index, row in annotated.iterrows():
+        entity_type = str(row.get("entity_type") or "")
+        recommendation = str(row.get("动作") or "")
+        account_id = safe_int(row.get("账号id"))
+        adgroup_id = safe_int(row.get("广告id"))
+        creative_id = safe_int(row.get("创意id"))
+        execution_mode = MODE_NOTIFY_ONLY
+        execution_reason = "无调控建议"
+        action_type: str | None = None
+
+        if recommendation:
+            if entity_type != ENTITY_SELF:
+                execution_reason = "合作渠道当前仅通知，不执行腾讯写操作"
+            elif account_id not in managed_account_ids:
+                execution_reason = "账户不在自动化管理范围，仅通知"
+            elif not adgroup_id:
+                execution_reason = "缺少有效广告ID，仅通知"
+            elif recommendation == "关停" and not creative_id:
+                execution_reason = "缺少有效动态创意ID，仅通知"
+            elif recommendation == "关停":
+                action_type = ACTION_PAUSE_CREATIVE
+            elif recommendation == "扩量":
+                action_type = ACTION_SCALE_BID
+            else:
+                execution_reason = "该建议当前没有自动执行器，仅通知"
+
+        if action_type:
+            suffix = creative_id if action_type == ACTION_PAUSE_CREATIVE else adgroup_id
+            idempotency_key = f"{run_id}:{action_type}:{account_id}:{suffix}"
+            execution_mode = MODE_ACTIONABLE
+            execution_reason = (
+                "审批后暂停该动态创意"
+                if action_type == ACTION_PAUSE_CREATIVE
+                else "审批后按ROI策略提高广告永久基础出价"
+            )
+            if idempotency_key not in seen_actions:
+                actions.append(
+                    {
+                        "idempotency_key": idempotency_key,
+                        "action_type": action_type,
+                        "account_id": account_id,
+                        "adgroup_id": adgroup_id,
+                        "dynamic_creative_id": creative_id,
+                        "execution_status": "PENDING",
+                    }
+                )
+                seen_actions.add(idempotency_key)
+
+        annotated.at[index, "执行模式"] = execution_mode
+        annotated.at[index, "执行说明"] = execution_reason
+
+        daily_metrics = {
+            column: _finite(row.get(column))
+            for column in annotated.columns
+            if column.startswith(
+                (
+                    "首层UV_20",
+                    "T0裂变数_20",
+                    "成本_20",
+                    "效率收入_20",
+                    "多层裂变收入_20",
+                    "ROI_20",
+                )
+            )
+        }
+        snapshots.append(
+            {
+                "entity_hash": _entity_hash(row),
+                "entity_type": entity_type,
+                "channel": str(row.get("channel") or ""),
+                "account_id": account_id,
+                "account_name": str(row.get("账号名称") or ""),
+                "adgroup_id": adgroup_id,
+                "adgroup_name": str(row.get("广告名称") or ""),
+                "dynamic_creative_id": creative_id,
+                "audience_name": str(row.get("包名") or ""),
+                "partner_name": str(row.get("合作方名") or ""),
+                "official_account_name": str(row.get("公众号名") or ""),
+                "ad_age": safe_int(row.get("广告age")),
+                "avg_first_uv": _finite(row.get("日均首层UV")),
+                "first_uv": _finite(row.get("三日首层UV")),
+                "t0_fission_count": _finite(row.get("三日T0裂变数")),
+                "t0_fission_rate": _finite(row.get("T0裂变率")),
+                "cost": _finite(row.get("成本")),
+                "efficiency_revenue": _finite(row.get("效率收入")),
+                "fission_revenue": _finite(row.get("多层裂变收入")),
+                "total_revenue": _finite(row.get("全链路效率收入")),
+                "roi": _finite(row.get("ROI")),
+                "stop_threshold": _finite(row.get("t_stop")),
+                "scale_threshold": _finite(row.get("t_up")),
+                "recommended_action": recommendation or None,
+                "action_reason": str(row.get("动作原因") or ""),
+                "execution_mode": execution_mode,
+                "ineligible_reason": (
+                    execution_reason if execution_mode == MODE_NOTIFY_ONLY else None
+                ),
+                "daily_metrics_json": daily_metrics,
+            }
+        )
+
+    return annotated, snapshots, actions

examples/auto_put_ad_mini/roi_control/reporting.py

@@ -0,0 +1,321 @@
+"""调控结果 Excel 报告生成。"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Dict, Iterable, Mapping, Sequence
+
+import numpy as np
+import pandas as pd
+from openpyxl import Workbook
+from openpyxl.formatting.rule import ColorScaleRule
+from openpyxl.styles import Alignment, Font, PatternFill
+from openpyxl.utils import get_column_letter
+from .metrics import ENTITY_GZH, ENTITY_QIWEI, ENTITY_SELF
+
+
+HEADER_FILL = PatternFill("solid", fgColor="1F4E78")
+HEADER_FONT = Font(color="FFFFFF", bold=True)
+STOP_FILL = PatternFill("solid", fgColor="F4CCCC")
+UP_FILL = PatternFill("solid", fgColor="D9EAD3")
+ADJUST_FILL = PatternFill("solid", fgColor="FFF2CC")
+
+
+BASE_COLUMNS: Dict[str, Sequence[str]] = {
+    "小程序投流": (
+        "渠道",
+        "代理名称",
+        "账号id",
+        "账号名称",
+        "广告id",
+        "广告名称",
+        "包名",
+        "创意id",
+        "广告age",
+        "日均首层UV",
+        "三日首层UV",
+        "三日T0裂变数",
+        "T0裂变率",
+        "效率收入",
+        "多层裂变收入",
+        "全链路效率收入",
+        "成本",
+        "三日最小单日成本",
+        "三日聚合ROI",
+        "关停线",
+        "扩量线",
+        "动作",
+        "执行模式",
+        "执行说明",
+    ),
+    "公众号即转": (
+        "渠道",
+        "合作方名",
+        "公众号名",
+        "日均首层UV",
+        "三日首层UV",
+        "三日T0裂变数",
+        "T0裂变率",
+        "效率收入",
+        "多层裂变收入",
+        "全链路效率收入",
+        "成本",
+        "三日最小单日成本",
+        "三日聚合ROI",
+        "关停线",
+        "扩量线",
+        "渠道内ROI排名百分位",
+        "动作",
+        "执行模式",
+        "执行说明",
+    ),
+    "企微群合作": (
+        "渠道",
+        "合作方名",
+        "公众号名",
+        "日均首层UV",
+        "三日首层UV",
+        "三日T0裂变数",
+        "T0裂变率",
+        "效率收入",
+        "多层裂变收入",
+        "全链路效率收入",
+        "成本",
+        "三日最小单日成本",
+        "三日聚合ROI",
+        "关停线",
+        "扩量线",
+        "动作",
+        "执行模式",
+        "执行说明",
+    ),
+}
+
+ENTITY_TO_SHEET = {
+    ENTITY_SELF: "小程序投流",
+    ENTITY_GZH: "公众号即转",
+    ENTITY_QIWEI: "企微群合作",
+}
+
+FREEZE_PANES = {
+    "小程序投流": "F2",
+    "公众号即转": "D2",
+    "企微群合作": "C2",
+}
+
+
+def _date_detail_columns(columns: Iterable[str]) -> list[str]:
+    result = [
+        column
+        for column in columns
+        if column.startswith(("首层UV_20", "成本_20", "ROI_20"))
+    ]
+    return sorted(result, key=lambda value: (value.rsplit("_", 1)[-1], value.split("_", 1)[0]))
+
+
+def _sheet_frame(candidates: pd.DataFrame, sheet_name: str) -> pd.DataFrame:
+    entity_type = next(
+        key for key, value in ENTITY_TO_SHEET.items() if value == sheet_name
+    )
+    subset = candidates[candidates["entity_type"].eq(entity_type)].copy()
+    subset = subset.rename(columns={"channel": "渠道"})
+    subset["三日聚合ROI"] = subset["ROI"]
+    subset["关停线"] = subset["t_stop"]
+    subset["扩量线"] = subset["t_up"]
+    date_columns = _date_detail_columns(subset.columns)
+    ordered = list(BASE_COLUMNS[sheet_name]) + date_columns
+    for column in ordered:
+        if column not in subset:
+            subset[column] = ""
+    subset = subset[ordered]
+    if not subset.empty:
+        subset = subset.sort_values(["动作", "三日聚合ROI"], ascending=[True, False])
+    return subset
+
+
+def _write_dataframe(ws, frame: pd.DataFrame) -> None:
+    ws.append(list(frame.columns))
+    for values in frame.itertuples(index=False, name=None):
+        ws.append(
+            [
+                None if isinstance(value, float) and np.isnan(value) else value
+                for value in values
+            ]
+        )
+
+
+def _format_sheet(ws, sheet_name: str) -> None:
+    max_column = max(ws.max_column, 1)
+    max_row = max(ws.max_row, 1)
+    ws.freeze_panes = FREEZE_PANES[sheet_name]
+    ws.auto_filter.ref = f"A1:{get_column_letter(max_column)}{max_row}"
+    ws.row_dimensions[1].height = 28
+
+    headers = {}
+    for cell in ws[1]:
+        cell.fill = HEADER_FILL
+        cell.font = HEADER_FONT
+        cell.alignment = Alignment(horizontal="center", vertical="center")
+        headers[cell.value] = cell.column
+
+    for column_index in range(1, max_column + 1):
+        header = ws.cell(1, column_index).value or ""
+        values = [str(ws.cell(row, column_index).value or "") for row in range(1, max_row + 1)]
+        width = min(max(max(map(len, values)), len(str(header))) + 2, 42)
+        ws.column_dimensions[get_column_letter(column_index)].width = max(width, 11)
+
+        if "ROI" in str(header) or "裂变率" in str(header) or "百分位" in str(header):
+            for row in range(2, max_row + 1):
+                ws.cell(row, column_index).number_format = "0.000"
+        elif "成本" in str(header) or header in {
+            "效率收入",
+            "多层裂变收入",
+            "全链路效率收入",
+        }:
+            for row in range(2, max_row + 1):
+                ws.cell(row, column_index).number_format = "#,##0.00"
+        elif "UV" in str(header) or "裂变数" in str(header):
+            for row in range(2, max_row + 1):
+                ws.cell(row, column_index).number_format = "#,##0.00"
+
+        if "ROI" in str(header) and max_row >= 2:
+            letter = get_column_letter(column_index)
+            ws.conditional_formatting.add(
+                f"{letter}2:{letter}{max_row}",
+                ColorScaleRule(
+                    start_type="min",
+                    start_color="F8696B",
+                    mid_type="percentile",
+                    mid_value=50,
+                    mid_color="FFEB84",
+                    end_type="max",
+                    end_color="63BE7B",
+                ),
+            )
+
+    action_column = headers.get("动作")
+    if action_column:
+        for row in range(2, max_row + 1):
+            action = ws.cell(row, action_column).value
+            fill = {
+                "关停": STOP_FILL,
+                "扩量": UP_FILL,
+                "调整封面&落地页视频": ADJUST_FILL,
+            }.get(action)
+            if fill:
+                ws.cell(row, action_column).fill = fill
+
+def _write_summary(
+    workbook: Workbook,
+    thresholds: pd.DataFrame,
+    candidates: pd.DataFrame,
+    expected_dates: Sequence[str],
+    qiwei_fission_is_zero: bool,
+    rule_config: Mapping[str, object] | None,
+) -> None:
+    ws = workbook.active
+    ws.title = "运行摘要"
+    rows = [
+        ("统计窗口", f"{expected_dates[0]} 至 {expected_dates[-1]}"),
+        ("首层口径", "usersharedepth='0'"),
+        ("UV口径", "COUNT(DISTINCT mid)"),
+        ("ROI口径", "三日聚合：(三日效率收入 + 三日多层t0裂变收入) / 三日成本"),
+        ("阈值口径", "三类合格实体混入同一全局池：三日聚合ROI的P10=关停线、P80=扩量线"),
+        (
+            "合格池口径",
+            "完整覆盖3天；UV、单日成本、广告age与分位门槛按本批次规则配置执行",
+        ),
+        ("动作优先级", "关停 > 扩量 > 调整封面&落地页视频"),
+        ("审批方式", "飞书消息整批确认或拒绝；表格仅用于查看，不做逐行审批"),
+    ]
+    for label, value in rows:
+        ws.append([label, value])
+
+    if rule_config:
+        ws.append([])
+        ws.append(["规则配置", "值"])
+        for key in (
+            "self_stop_min_age",
+            "self_up_min_age",
+            "self_min_avg_uv",
+            "partner_min_avg_uv",
+            "min_daily_cost",
+            "stop_quantile",
+            "up_quantile",
+            "scale_ratio",
+            "scale_cooldown_days",
+            "max_base_ratio",
+        ):
+            ws.append([key, rule_config.get(key)])
+
+    ws.append([])
+    display_thresholds = thresholds.rename(
+        columns={"t_stop": "关停线", "t_up": "扩量线"}
+    )
+    threshold_columns = [
+        "统计窗口",
+        "关停线",
+        "扩量线",
+        "阈值样本数",
+        "小程序样本数",
+        "公众号样本数",
+        "企微样本数",
+    ]
+    ws.append(threshold_columns)
+    for values in display_thresholds[threshold_columns].itertuples(
+        index=False,
+        name=None,
+    ):
+        ws.append(list(values))
+
+    ws.append([])
+    ws.append(["渠道", "建议动作数"])
+    for entity_type, sheet_name in ENTITY_TO_SHEET.items():
+        count = int(candidates["entity_type"].eq(entity_type).sum())
+        ws.append([sheet_name, count])
+
+    if qiwei_fission_is_zero:
+        ws.append([])
+        ws.append(
+            [
+                "数据质量告警",
+                "企微群合作在统计窗口内的多层t0裂变收入合计为0；本报告严格按附件公式计算，未使用ARPU估算。",
+            ]
+        )
+
+    ws.column_dimensions["A"].width = 22
+    ws.column_dimensions["B"].width = 92
+    for row in ws.iter_rows():
+        if row[0].value in {"统计窗口", "渠道"}:
+            for cell in row:
+                cell.font = Font(bold=True)
+    ws.freeze_panes = "A2"
+
+
+def write_workbook(
+    candidates: pd.DataFrame,
+    thresholds: pd.DataFrame,
+    expected_dates: Sequence[str],
+    output_path: Path,
+    qiwei_fission_is_zero: bool,
+    rule_config: Mapping[str, object] | None = None,
+) -> Path:
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+    workbook = Workbook()
+    _write_summary(
+        workbook,
+        thresholds,
+        candidates,
+        expected_dates,
+        qiwei_fission_is_zero,
+        rule_config,
+    )
+
+    for sheet_name in BASE_COLUMNS:
+        ws = workbook.create_sheet(sheet_name)
+        frame = _sheet_frame(candidates, sheet_name)
+        _write_dataframe(ws, frame)
+        _format_sheet(ws, sheet_name)
+
+    workbook.save(output_path)
+    return output_path

examples/auto_put_ad_mini/roi_control/repository.py

@@ -0,0 +1,377 @@
+"""MySQL persistence for versioned ROI runs, snapshots, and action audit."""
+
+from __future__ import annotations
+
+import json
+from datetime import date, datetime
+from decimal import Decimal
+from typing import Any, Iterable
+
+from storage import connect
+
+
+FINAL_STATUSES = {
+    "COMPLETED",
+    "PARTIAL",
+    "REJECTED",
+    "EXPIRED",
+}
+
+
+def json_dumps(value: Any) -> str:
+    def default(item: Any) -> Any:
+        if isinstance(item, (datetime, date)):
+            return item.isoformat()
+        if isinstance(item, Decimal):
+            return str(item)
+        if hasattr(item, "item"):
+            return item.item()
+        if hasattr(item, "tolist"):
+            return item.tolist()
+        raise TypeError(f"Unsupported JSON value: {type(item)!r}")
+
+    return json.dumps(value, ensure_ascii=False, default=default)
+
+
+def create_or_load_run(record: dict[str, Any]) -> dict[str, Any]:
+    connection = connect()
+    try:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                """
+                INSERT IGNORE INTO roi_metric_run
+                    (run_id, run_key, metric_version, policy_version,
+                     start_date, end_date, config_json, status)
+                VALUES (%s,%s,%s,%s,%s,%s,%s,'COMPUTING')
+                """,
+                (
+                    record["run_id"],
+                    record["run_key"],
+                    record["metric_version"],
+                    record["policy_version"],
+                    record["start_date"],
+                    record["end_date"],
+                    json_dumps(record["config"]),
+                ),
+            )
+            cursor.execute(
+                "SELECT * FROM roi_metric_run WHERE run_key=%s",
+                (record["run_key"],),
+            )
+            return cursor.fetchone() or {}
+    finally:
+        connection.close()
+
+
+def load_run(run_id: str) -> dict[str, Any] | None:
+    connection = connect()
+    try:
+        with connection.cursor() as cursor:
+            cursor.execute("SELECT * FROM roi_metric_run WHERE run_id=%s", (run_id,))
+            return cursor.fetchone()
+    finally:
+        connection.close()
+
+
+def replace_run_results(
+    run_id: str,
+    *,
+    snapshots: Iterable[dict[str, Any]],
+    actions: Iterable[dict[str, Any]],
+    thresholds: dict[str, Any],
+) -> tuple[int, int]:
+    snapshot_rows = list(snapshots)
+    action_rows = list(actions)
+    connection = connect()
+    connection.autocommit(False)
+    try:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                "SELECT status FROM roi_metric_run WHERE run_id=%s FOR UPDATE",
+                (run_id,),
+            )
+            run = cursor.fetchone()
+            if not run:
+                raise ValueError(f"ROI run not found: {run_id}")
+            if run["status"] not in {"COMPUTING", "COMPUTED", "FAILED"}:
+                raise RuntimeError(
+                    f"ROI run {run_id} cannot be replaced in status {run['status']}"
+                )
+            cursor.execute("DELETE FROM roi_action_item WHERE run_id=%s", (run_id,))
+            cursor.execute("DELETE FROM roi_entity_snapshot WHERE run_id=%s", (run_id,))
+
+            snapshot_columns = [
+                "run_id", "entity_hash", "entity_type", "channel",
+                "account_id", "account_name", "adgroup_id", "adgroup_name",
+                "dynamic_creative_id", "audience_name", "partner_name",
+                "official_account_name", "ad_age", "avg_first_uv", "first_uv",
+                "t0_fission_count", "t0_fission_rate", "cost",
+                "efficiency_revenue", "fission_revenue", "total_revenue", "roi",
+                "stop_threshold", "scale_threshold", "recommended_action",
+                "action_reason", "execution_mode", "ineligible_reason",
+                "daily_metrics_json",
+            ]
+            if snapshot_rows:
+                cursor.executemany(
+                    f"INSERT INTO roi_entity_snapshot ({', '.join(snapshot_columns)}) "
+                    f"VALUES ({', '.join(['%s'] * len(snapshot_columns))})",
+                    [
+                        [
+                            json_dumps(row[column])
+                            if column == "daily_metrics_json"
+                            else row.get(column)
+                            for column in snapshot_columns
+                        ]
+                        for row in snapshot_rows
+                    ],
+                )
+
+            action_columns = [
+                "run_id", "idempotency_key", "action_type", "account_id",
+                "adgroup_id", "dynamic_creative_id", "execution_status",
+            ]
+            if action_rows:
+                cursor.executemany(
+                    f"INSERT INTO roi_action_item ({', '.join(action_columns)}) "
+                    f"VALUES ({', '.join(['%s'] * len(action_columns))})",
+                    [
+                        [row.get(column) for column in action_columns]
+                        for row in action_rows
+                    ],
+                )
+
+            cursor.execute(
+                """
+                UPDATE roi_metric_run
+                SET thresholds_json=%s,
+                    status='COMPUTED',
+                    entity_count=%s,
+                    candidate_count=%s,
+                    actionable_count=%s,
+                    error_message=NULL
+                WHERE run_id=%s
+                """,
+                (
+                    json_dumps(thresholds),
+                    len(snapshot_rows),
+                    sum(bool(row.get("recommended_action")) for row in snapshot_rows),
+                    len(action_rows),
+                    run_id,
+                ),
+            )
+        connection.commit()
+        return len(snapshot_rows), len(action_rows)
+    except Exception:
+        connection.rollback()
+        raise
+    finally:
+        connection.close()
+
+
+def mark_published(
+    run_id: str,
+    *,
+    sheet_token: str,
+    sheet_url: str,
+    message_id: str,
+    expires_at: datetime | None,
+    requires_approval: bool,
+) -> None:
+    connection = connect()
+    try:
+        with connection.cursor() as cursor:
+            affected = cursor.execute(
+                """
+                UPDATE roi_metric_run
+                SET status=%s, sheet_token=%s, sheet_url=%s,
+                    message_id=%s, expires_at=%s, error_message=NULL
+                WHERE run_id=%s AND status IN ('COMPUTED','PENDING_APPROVAL')
+                """,
+                (
+                    "PENDING_APPROVAL" if requires_approval else "COMPLETED",
+                    sheet_token,
+                    sheet_url,
+                    message_id,
+                    expires_at,
+                    run_id,
+                ),
+            )
+            if not affected:
+                raise RuntimeError(f"ROI run cannot be published: {run_id}")
+    finally:
+        connection.close()
+
+
+def mark_failed(run_id: str, error_message: str) -> None:
+    connection = connect()
+    try:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                """
+                UPDATE roi_metric_run
+                SET status='FAILED', error_message=%s
+                WHERE run_id=%s AND status NOT IN ('COMPLETED','PARTIAL','REJECTED')
+                """,
+                (error_message[:4000], run_id),
+            )
+    finally:
+        connection.close()
+
+
+def reject_run(run_id: str, *, sender_open_id: str, now: datetime) -> dict[str, Any]:
+    connection = connect()
+    connection.autocommit(False)
+    try:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                "SELECT * FROM roi_metric_run WHERE run_id=%s FOR UPDATE",
+                (run_id,),
+            )
+            run = cursor.fetchone()
+            if not run:
+                raise ValueError(f"ROI批次不存在: {run_id}")
+            if run["status"] in FINAL_STATUSES:
+                connection.commit()
+                return run
+            if run["status"] != "PENDING_APPROVAL":
+                raise RuntimeError(f"ROI批次当前不可拒绝: {run['status']}")
+            if run.get("expires_at") and now > run["expires_at"]:
+                cursor.execute(
+                    "UPDATE roi_metric_run SET status='EXPIRED' WHERE run_id=%s",
+                    (run_id,),
+                )
+            else:
+                cursor.execute(
+                    """
+                    UPDATE roi_metric_run
+                    SET status='REJECTED', rejected_by=%s, rejected_at=%s
+                    WHERE run_id=%s
+                    """,
+                    (sender_open_id, now, run_id),
+                )
+        connection.commit()
+        return load_run(run_id) or run
+    except Exception:
+        connection.rollback()
+        raise
+    finally:
+        connection.close()
+
+
+def claim_run_for_execution(
+    run_id: str,
+    *,
+    sender_open_id: str,
+    now: datetime,
+) -> dict[str, Any]:
+    connection = connect()
+    connection.autocommit(False)
+    try:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                "SELECT * FROM roi_metric_run WHERE run_id=%s FOR UPDATE",
+                (run_id,),
+            )
+            run = cursor.fetchone()
+            if not run:
+                raise ValueError(f"ROI批次不存在: {run_id}")
+            if run["status"] in FINAL_STATUSES:
+                connection.commit()
+                return run
+            if run["status"] == "PENDING_APPROVAL":
+                if run.get("expires_at") and now > run["expires_at"]:
+                    cursor.execute(
+                        "UPDATE roi_metric_run SET status='EXPIRED' WHERE run_id=%s",
+                        (run_id,),
+                    )
+                    connection.commit()
+                    return load_run(run_id) or run
+                cursor.execute(
+                    """
+                    UPDATE roi_metric_run
+                    SET status='EXECUTING', approved_by=%s, approved_at=%s
+                    WHERE run_id=%s
+                    """,
+                    (sender_open_id, now, run_id),
+                )
+            elif run["status"] != "EXECUTING":
+                raise RuntimeError(f"ROI批次当前不可确认: {run['status']}")
+        connection.commit()
+        return load_run(run_id) or run
+    except Exception:
+        connection.rollback()
+        raise
+    finally:
+        connection.close()
+
+
+def load_action_items(run_id: str) -> list[dict[str, Any]]:
+    connection = connect()
+    try:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                "SELECT * FROM roi_action_item WHERE run_id=%s ORDER BY id",
+                (run_id,),
+            )
+            return list(cursor.fetchall())
+    finally:
+        connection.close()
+
+
+def update_action_item(item_id: int, **values: Any) -> None:
+    allowed = {
+        "bid_field", "initial_base_bid_fen", "base_bid_fen", "before_bid_fen",
+        "target_bid_fen", "before_status", "target_status", "readback_status",
+        "execution_status", "skip_reason", "error_message", "pre_state_json",
+        "readback_json", "executed_at",
+    }
+    unknown = set(values) - allowed
+    if unknown:
+        raise ValueError(f"Unsupported ROI action fields: {sorted(unknown)}")
+    assignments: list[str] = []
+    params: list[Any] = []
+    for column, value in values.items():
+        assignments.append(f"{column}=%s")
+        params.append(
+            json_dumps(value)
+            if column in {"pre_state_json", "readback_json"} and value is not None
+            else value
+        )
+    if not assignments:
+        return
+    params.append(item_id)
+    connection = connect()
+    try:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                f"UPDATE roi_action_item SET {', '.join(assignments)} WHERE id=%s",
+                params,
+            )
+    finally:
+        connection.close()
+
+
+def finalize_run(run_id: str, *, now: datetime) -> dict[str, Any]:
+    connection = connect()
+    try:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                """
+                SELECT
+                    SUM(execution_status='SUCCESS') AS successes,
+                    SUM(execution_status IN ('FAILED','OUTCOME_UNKNOWN','VERIFY_FAILED')) AS failures,
+                    SUM(execution_status LIKE 'SKIPPED%%') AS skipped
+                FROM roi_action_item WHERE run_id=%s
+                """,
+                (run_id,),
+            )
+            counts = cursor.fetchone() or {}
+            failures = int(counts.get("failures") or 0)
+            status = "PARTIAL" if failures else "COMPLETED"
+            cursor.execute(
+                "UPDATE roi_metric_run SET status=%s, executed_at=%s WHERE run_id=%s",
+                (status, now, run_id),
+            )
+        return {"status": status, **{key: int(value or 0) for key, value in counts.items()}}
+    finally:
+        connection.close()

examples/auto_put_ad_mini/roi_control/rules.py

@@ -0,0 +1,166 @@
+"""Versioned ROI thresholds and action recommendation policy."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Iterable
+
+import numpy as np
+import pandas as pd
+
+from .metrics import (
+    ENTITY_GZH,
+    ENTITY_QIWEI,
+    ENTITY_SELF,
+    compute_roi_summary,
+)
+
+
+POLICY_VERSION = "roi_policy_v1"
+
+@dataclass(frozen=True)
+class RuleConfig:
+    self_stop_min_age: int = 5
+    self_up_min_age: int = 3
+    self_min_avg_uv: float = 200
+    partner_min_avg_uv: float = 200
+    min_daily_cost: float = 100
+    stop_quantile: float = 0.10
+    up_quantile: float = 0.80
+    gzh_adjust_rank_min: float = 0.10
+    gzh_adjust_rank_max: float = 0.30
+
+
+def compute_global_thresholds(
+    summary: pd.DataFrame,
+    expected_dates: list[str],
+    config: RuleConfig,
+) -> pd.DataFrame:
+    self_eligible = (
+        summary["entity_type"].eq(ENTITY_SELF)
+        & summary["日均首层UV"].gt(config.self_min_avg_uv)
+    )
+    partner_eligible = (
+        summary["entity_type"].isin([ENTITY_GZH, ENTITY_QIWEI])
+        & summary["日均首层UV"].gt(config.partner_min_avg_uv)
+    )
+    valid = summary.loc[
+        (self_eligible | partner_eligible)
+        & summary["成本"].gt(0)
+        & summary["三日最小单日成本"].gt(config.min_daily_cost)
+        & np.isfinite(summary["ROI"])
+    ].copy()
+    if valid.empty:
+        raise ValueError("没有满足渠道UV门槛且三日聚合ROI有效的实体，无法计算全局阈值")
+
+    type_counts = valid["entity_type"].value_counts()
+    return pd.DataFrame(
+        [
+            {
+                "统计窗口": f"{expected_dates[0]} 至 {expected_dates[-1]}",
+                "t_stop": float(valid["ROI"].quantile(config.stop_quantile)),
+                "t_up": float(valid["ROI"].quantile(config.up_quantile)),
+                "阈值样本数": int(len(valid)),
+                "小程序样本数": int(type_counts.get(ENTITY_SELF, 0)),
+                "公众号样本数": int(type_counts.get(ENTITY_GZH, 0)),
+                "企微样本数": int(type_counts.get(ENTITY_QIWEI, 0)),
+            }
+        ]
+    )
+
+
+def apply_actions(
+    summary: pd.DataFrame,
+    thresholds: pd.DataFrame,
+    config: RuleConfig,
+) -> pd.DataFrame:
+    result = summary.copy()
+    result["动作"] = ""
+    result["动作原因"] = ""
+    result["渠道内ROI排名百分位"] = np.nan
+    t_stop = float(thresholds.iloc[0]["t_stop"])
+    t_up = float(thresholds.iloc[0]["t_up"])
+    result["t_stop"] = t_stop
+    result["t_up"] = t_up
+
+    gzh_mask = result["entity_type"].eq(ENTITY_GZH) & np.isfinite(result["ROI"])
+    if gzh_mask.any():
+        result.loc[gzh_mask, "渠道内ROI排名百分位"] = result.loc[gzh_mask, "ROI"].rank(
+            method="average", ascending=False, pct=True
+        )
+
+    for index, row in result.iterrows():
+        entity_type = row["entity_type"]
+        avg_uv = row["日均首层UV"]
+        roi = row["ROI"]
+        cost_confident = row["三日最小单日成本"] > config.min_daily_cost
+        stop_side = bool(np.isfinite(roi) and (roi <= t_stop or roi == 0))
+        up_side = bool(np.isfinite(roi) and roi > 0 and roi >= t_up)
+
+        if entity_type == ENTITY_SELF:
+            if (
+                row["广告age"] >= config.self_stop_min_age
+                and stop_side
+                and avg_uv > config.self_min_avg_uv
+                and cost_confident
+            ):
+                result.at[index, "动作"] = "关停"
+                result.at[index, "动作原因"] = (
+                    f"广告age≥{config.self_stop_min_age}天，三日聚合ROI不高于"
+                    f"全局t_stop（ROI=0明确关停），日均首层UV>"
+                    f"{config.self_min_avg_uv:g}，连续3天每天成本>"
+                    f"{config.min_daily_cost:g}"
+                )
+            elif (
+                row["广告age"] >= config.self_up_min_age
+                and up_side
+                and avg_uv > config.self_min_avg_uv
+                and cost_confident
+            ):
+                result.at[index, "动作"] = "扩量"
+                result.at[index, "动作原因"] = (
+                    f"广告age≥{config.self_up_min_age}天，三日聚合ROI不低于"
+                    f"全局t_up，日均首层UV>{config.self_min_avg_uv:g}，"
+                    f"连续3天每天成本>{config.min_daily_cost:g}"
+                )
+        elif entity_type in (ENTITY_GZH, ENTITY_QIWEI):
+            if stop_side and avg_uv > config.partner_min_avg_uv and cost_confident:
+                result.at[index, "动作"] = "关停"
+                result.at[index, "动作原因"] = (
+                    "三日聚合ROI不高于全局t_stop（ROI=0明确关停），"
+                    f"日均首层UV>{config.partner_min_avg_uv:g}，"
+                    f"连续3天每天成本>{config.min_daily_cost:g}"
+                )
+            elif up_side and avg_uv > config.partner_min_avg_uv and cost_confident:
+                result.at[index, "动作"] = "扩量"
+                result.at[index, "动作原因"] = (
+                    "三日聚合ROI不低于全局t_up，"
+                    f"日均首层UV>{config.partner_min_avg_uv:g}，"
+                    f"连续3天每天成本>{config.min_daily_cost:g}"
+                )
+            elif entity_type == ENTITY_GZH:
+                rank_pct = row["渠道内ROI排名百分位"]
+                if (
+                    np.isfinite(rank_pct)
+                    and config.gzh_adjust_rank_min <= rank_pct <= config.gzh_adjust_rank_max
+                ):
+                    result.at[index, "动作"] = "调整封面&落地页视频"
+                    result.at[index, "动作原因"] = (
+                        "三日综合ROI在公众号即转渠道排名"
+                        f"{config.gzh_adjust_rank_min:.0%}–"
+                        f"{config.gzh_adjust_rank_max:.0%}"
+                    )
+    return result
+
+
+def evaluate_rules(
+    raw_daily: pd.DataFrame,
+    expected_dates: Iterable[str],
+    ad_age: pd.DataFrame | None = None,
+    config: RuleConfig = RuleConfig(),
+) -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]:
+    summary, dates = compute_roi_summary(raw_daily, expected_dates, ad_age)
+    thresholds = compute_global_thresholds(summary, dates, config)
+    evaluated = apply_actions(summary, thresholds, config)
+    candidates = evaluated[evaluated["动作"].ne("")].copy()
+    return candidates, thresholds, evaluated

examples/auto_put_ad_mini/roi_control/service.py

@@ -0,0 +1,210 @@
+"""Orchestrate one idempotent daily ROI metric and approval batch."""
+
+from __future__ import annotations
+
+import logging
+import os
+from datetime import datetime, timedelta
+from pathlib import Path
+from typing import Any
+from zoneinfo import ZoneInfo
+
+from storage import initialize_schema, load_managed_accounts
+
+from .config import RoiConfig
+from .data_source import (
+    ODPSClient,
+    date_window,
+    fetch_ad_age,
+    fetch_daily_data,
+    resolve_end_date,
+)
+from .feishu import RoiFeishuPublisher
+from .metrics import ENTITY_QIWEI, METRIC_VERSION
+from .policy import annotate_execution
+from .reporting import write_workbook
+from .rules import POLICY_VERSION, RuleConfig, evaluate_rules
+from .repository import (
+    FINAL_STATUSES,
+    create_or_load_run,
+    mark_failed,
+    mark_published,
+    replace_run_results,
+)
+
+
+SHANGHAI = ZoneInfo("Asia/Shanghai")
+logger = logging.getLogger("auto_put_ad_mini.roi_control")
+
+
+def _rule_config(config: RoiConfig) -> RuleConfig:
+    return RuleConfig(
+        self_stop_min_age=config.self_stop_min_age,
+        self_up_min_age=config.self_up_min_age,
+        self_min_avg_uv=config.self_min_avg_uv,
+        partner_min_avg_uv=config.partner_min_avg_uv,
+        min_daily_cost=config.min_daily_cost,
+        stop_quantile=config.stop_quantile,
+        up_quantile=config.up_quantile,
+        gzh_adjust_rank_min=config.gzh_adjust_rank_min,
+        gzh_adjust_rank_max=config.gzh_adjust_rank_max,
+    )
+
+
+def _dates(start_date: str) -> list[str]:
+    start = datetime.strptime(start_date, "%Y%m%d")
+    return [
+        (start + timedelta(days=offset)).strftime("%Y%m%d")
+        for offset in range(3)
+    ]
+
+
+def _run_identity(end_date: str) -> tuple[str, str]:
+    run_id = f"roi_{end_date}_{METRIC_VERSION}_{POLICY_VERSION}"
+    run_key = f"{METRIC_VERSION}:{POLICY_VERSION}:{end_date}"
+    return run_id, run_key
+
+
+def run_daily_roi(
+    *,
+    requested_end_date: str | None = None,
+    output_dir: Path,
+    send_feishu: bool,
+    now: datetime | None = None,
+) -> dict[str, Any]:
+    """Compute, snapshot, report, and optionally publish one ROI batch."""
+
+    config = RoiConfig.from_env()
+    initialize_schema()
+    client = ODPSClient(project=os.getenv("ODPS_PROJECT", "loghubods"))
+    end_date = resolve_end_date(client, requested_end_date)
+    start_date, end_date = date_window(end_date)
+    expected_dates = _dates(start_date)
+    run_id, run_key = _run_identity(end_date)
+    run = create_or_load_run(
+        {
+            "run_id": run_id,
+            "run_key": run_key,
+            "metric_version": METRIC_VERSION,
+            "policy_version": POLICY_VERSION,
+            "start_date": datetime.strptime(start_date, "%Y%m%d").date(),
+            "end_date": datetime.strptime(end_date, "%Y%m%d").date(),
+            "config": config.snapshot(),
+        }
+    )
+    if run.get("status") in FINAL_STATUSES | {"PENDING_APPROVAL", "EXECUTING"}:
+        logger.info("Reuse ROI run=%s status=%s", run_id, run.get("status"))
+        return {
+            "run_id": run_id,
+            "status": run.get("status"),
+            "reused": True,
+            "sheet_url": run.get("sheet_url"),
+        }
+
+    try:
+        logger.info("ROI %s loading ODPS daily data %s..%s", run_id, start_date, end_date)
+        daily = fetch_daily_data(client, start_date, end_date)
+        ad_age = fetch_ad_age(client, end_date)
+        _, thresholds, summary = evaluate_rules(
+            daily,
+            expected_dates,
+            ad_age,
+            _rule_config(config),
+        )
+        managed_ids = {
+            int(row["account_id"])
+            for row in load_managed_accounts()
+        }
+        annotated, snapshots, actions = annotate_execution(
+            summary,
+            managed_ids,
+            run_id,
+        )
+        actionable_candidates = annotated[annotated["动作"].ne("")].copy()
+        for row in snapshots:
+            row["run_id"] = run_id
+        for row in actions:
+            row["run_id"] = run_id
+        threshold_record = thresholds.iloc[0].to_dict()
+        replace_run_results(
+            run_id,
+            snapshots=snapshots,
+            actions=actions,
+            thresholds=threshold_record,
+        )
+
+        qiwei_rows = daily[daily["entity_type"].eq(ENTITY_QIWEI)]
+        qiwei_fission_is_zero = bool(
+            not qiwei_rows.empty
+            and qiwei_rows["多层裂变收入"].fillna(0).sum() == 0
+        )
+        output_dir.mkdir(parents=True, exist_ok=True)
+        output_path = output_dir / f"{run_id}.xlsx"
+        write_workbook(
+            actionable_candidates,
+            thresholds,
+            expected_dates,
+            output_path,
+            qiwei_fission_is_zero,
+            config.snapshot(),
+        )
+
+        result: dict[str, Any] = {
+            "run_id": run_id,
+            "status": "COMPUTED",
+            "reused": False,
+            "start_date": start_date,
+            "end_date": end_date,
+            "entity_count": len(snapshots),
+            "candidate_count": len(actionable_candidates),
+            "actionable_count": len(actions),
+            "thresholds": threshold_record,
+            "report": str(output_path),
+        }
+        if not send_feishu:
+            return result
+
+        counts = actionable_candidates.groupby("动作").size().to_dict()
+        summary_text = (
+            f"统计窗口：{start_date} - {end_date}\n"
+            f"关停建议：{counts.get('关停', 0)}\n"
+            f"扩量建议：{counts.get('扩量', 0)}\n"
+            f"素材调整建议：{counts.get('调整封面&落地页视频', 0)}\n"
+            f"可执行动作：{len(actions)}\n"
+            f"审批有效期：发送后 {config.approval_ttl_minutes} 分钟"
+        )
+        publisher = RoiFeishuPublisher()
+        try:
+            published = publisher.publish(
+                output_path,
+                run_id=run_id,
+                summary=summary_text,
+                requires_approval=bool(actions),
+            )
+        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",
+                "sheet_url": published["url"],
+                "expires_at": expires_at.isoformat() if expires_at else None,
+            }
+        )
+        return result
+    except Exception as exc:
+        mark_failed(run_id, str(exc))
+        raise

examples/auto_put_ad_mini/run_daily_roi.py

@@ -0,0 +1,56 @@
+#!/usr/bin/env python
+"""Run one versioned daily ROI computation and optional Feishu approval batch."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import logging
+import os
+import sys
+from pathlib import Path
+
+from dotenv import load_dotenv
+
+
+HERE = Path(__file__).resolve().parent
+ROOT = HERE.parents[1]
+RTC_DIR = ROOT / "examples" / "tencent_realtime_control"
+for path in (ROOT, HERE, RTC_DIR):
+    if str(path) not in sys.path:
+        sys.path.insert(0, str(path))
+
+load_dotenv(HERE / ".env", override=False)
+load_dotenv(Path.cwd() / ".env", override=False)
+
+from roi_control.service import run_daily_roi  # noqa: E402
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--end-date", help="ODPS end partition in YYYYMMDD")
+    parser.add_argument(
+        "--output-dir",
+        type=Path,
+        default=HERE / "outputs" / "roi_control",
+    )
+    parser.add_argument("--send-feishu", action="store_true")
+    return parser.parse_args()
+
+
+def main() -> None:
+    args = parse_args()
+    result = run_daily_roi(
+        requested_end_date=args.end_date,
+        output_dir=args.output_dir,
+        send_feishu=args.send_feishu,
+    )
+    print(json.dumps(result, ensure_ascii=False, default=str))
+
+
+if __name__ == "__main__":
+    logging.basicConfig(
+        level=os.getenv("LOG_LEVEL", "INFO"),
+        format="%(asctime)s %(levelname)s %(name)s %(message)s",
+    )
+    main()

examples/auto_put_ad_mini/run_daily_service.py

@@ -28,7 +28,8 @@ load_dotenv(Path.cwd() / ".env", override=False)
 
 from config import TIME_SERIES_DEFAULT  # noqa: E402
 from db.connection import get_connection  # noqa: E402
-from storage import advisory_lock  # noqa: E402
+from storage import advisory_lock, initialize_schema  # noqa: E402
+from roi_control.config import RoiConfig  # noqa: E402
 
 
 logger = logging.getLogger("auto_put_ad_mini.daily_service")
@@ -61,14 +62,18 @@ def sync_enabled_delivery_templates() -> int:
         connection.close()
 
 
-def _run_script(script_name: str, lock_name: str) -> None:
+def _run_script(
+    script_name: str,
+    lock_name: str,
+    extra_args: list[str] | None = None,
+) -> None:
     with advisory_lock(lock_name) as acquired:
         if not acquired:
             logger.warning("Skip %s: another instance holds %s", script_name, lock_name)
             return
         logger.info("Starting %s", script_name)
         completed = subprocess.run(
-            [sys.executable, str(HERE / script_name)],
+            [sys.executable, str(HERE / script_name), *(extra_args or [])],
             cwd=HERE,
             check=False,
         )
@@ -93,7 +98,17 @@ def run_creative_review() -> None:
     )
 
 
+def run_daily_roi() -> None:
+    _run_script(
+        "run_daily_roi.py",
+        os.getenv("DAILY_ROI_LOCK_NAME", "ad_daily_roi"),
+        ["--send-feishu"],
+    )
+
+
 def main() -> None:
+    roi_config = RoiConfig.from_env()
+    initialize_schema()
     if _env_flag("DAILY_SYNC_DELIVERY_TEMPLATE", False):
         changed = sync_enabled_delivery_templates()
         logger.info("Synchronized %d enabled delivery template(s)", changed)
@@ -130,6 +145,22 @@ def main() -> None:
             coalesce=True,
             misfire_grace_time=1800,
         )
+    if roi_config.daily_enabled:
+        scheduler.add_job(
+            run_daily_roi,
+            CronTrigger(
+                hour=roi_config.report_hour,
+                minute=roi_config.report_minute,
+                timezone="Asia/Shanghai",
+            ),
+            id="daily_roi",
+            name="日级ROI计算与整批审批",
+            max_instances=1,
+            coalesce=True,
+            misfire_grace_time=int(
+                os.getenv("DAILY_ROI_MISFIRE_GRACE_SECONDS", "3600")
+            ),
+        )
     if _env_flag("DAILY_RUN_ON_STARTUP", False):
         scheduler.add_job(
             run_creation,
@@ -142,6 +173,12 @@ def main() -> None:
             id="creative_review_startup",
             name="启动时审核扫描",
         )
+    if roi_config.daily_enabled and _env_flag("DAILY_ROI_RUN_ON_STARTUP", False):
+        scheduler.add_job(
+            run_daily_roi,
+            id="daily_roi_startup",
+            name="启动时ROI计算",
+        )
     logger.info(
         "Daily service started jobs=%s",
         [job.id for job in scheduler.get_jobs()],

examples/auto_put_ad_mini/test_roi_control_metrics.py

@@ -0,0 +1,242 @@
+import unittest
+
+import pandas as pd
+
+from roi_control.metrics import (
+    ENTITY_GZH,
+    ENTITY_QIWEI,
+    ENTITY_SELF,
+    GZH_CHANNEL,
+    QIWEI_CHANNEL,
+    SELF_CHANNEL,
+)
+from roi_control.rules import evaluate_rules
+
+
+DATES = ["20260720", "20260721", "20260722"]
+
+
+def row(entity_type, channel, entity_id, dt, roi, uv=600, age=10, cost=200.0):
+    common = {
+        "dt": dt,
+        "entity_type": entity_type,
+        "channel": channel,
+        "代理名称": "",
+        "账号id": "",
+        "账号名称": "",
+        "广告id": "",
+        "广告名称": "",
+        "包名": "",
+        "创意id": "",
+        "合作方名": "",
+        "公众号名": "",
+        "首层UV": uv,
+        "T0裂变数": uv * 0.2,
+        "成本": cost,
+        "效率收入": roi * cost,
+        "多层裂变收入": 0,
+    }
+    if entity_type == ENTITY_SELF:
+        common.update(
+            {
+                "代理名称": "代理",
+                "账号id": "account",
+                "账号名称": "账户",
+                "广告id": entity_id,
+                "广告名称": f"广告{entity_id}",
+                "包名": "人群包",
+                "创意id": f"creative-{entity_id}",
+            }
+        )
+    elif entity_type == ENTITY_GZH:
+        common.update({"合作方名": "合作方", "公众号名": entity_id})
+    else:
+        common.update({"合作方名": entity_id})
+    return common
+
+
+class RoiRulesTest(unittest.TestCase):
+    def build_daily(self):
+        rows = []
+        # 每日形成稳定分布，低值实体连续低于后10%，高值实体连续高于前20%。
+        for dt in DATES:
+            for index, roi in enumerate([0.1, 0.5, 0.8, 1.0, 1.2, 1.5, 2.0, 3.0, 4.0, 5.0]):
+                rows.append(row(ENTITY_SELF, SELF_CHANNEL, f"ad-{index}", dt, roi))
+            rows.append(row(ENTITY_GZH, GZH_CHANNEL, "低ROI公众号", dt, 0.05, uv=300))
+            rows.append(row(ENTITY_GZH, GZH_CHANNEL, "高ROI公众号", dt, 6.0, uv=300))
+            rows.append(row(ENTITY_QIWEI, QIWEI_CHANNEL, "低ROI企微", dt, 0.02, uv=300))
+            rows.append(row(ENTITY_QIWEI, QIWEI_CHANNEL, "高ROI企微", dt, 7.0, uv=300))
+        return pd.DataFrame(rows)
+
+    def test_stop_and_up_actions(self):
+        daily = self.build_daily()
+        ages = pd.DataFrame(
+            {
+                "广告id": [f"ad-{index}" for index in range(10)],
+                "广告age": [10] * 10,
+            }
+        )
+        candidates, thresholds, summary = evaluate_rules(daily, DATES, ages)
+
+        actions = set(candidates["动作"])
+        self.assertIn("关停", actions)
+        self.assertIn("扩量", actions)
+        self.assertEqual(len(thresholds), 1)
+        self.assertEqual(thresholds.iloc[0]["阈值样本数"], len(summary))
+        self.assertTrue((summary["覆盖天数"] == 3).all())
+        self.assertTrue((summary["t_stop"] == thresholds.iloc[0]["t_stop"]).all())
+        self.assertTrue((summary["t_up"] == thresholds.iloc[0]["t_up"]).all())
+
+    def test_missing_day_is_not_candidate(self):
+        daily = self.build_daily()
+        missing = daily[
+            ~(
+                daily["entity_type"].eq(ENTITY_QIWEI)
+                & daily["合作方名"].eq("低ROI企微")
+                & daily["dt"].eq(DATES[-1])
+            )
+        ]
+        candidates, _, _ = evaluate_rules(missing, DATES)
+        self.assertFalse(candidates["合作方名"].eq("低ROI企微").any())
+
+    def test_no_complete_three_day_entity_fails_clearly(self):
+        daily = self.build_daily()
+        incomplete = daily[daily["dt"].ne(DATES[-1])]
+        with self.assertRaisesRegex(ValueError, "连续三日数据"):
+            evaluate_rules(incomplete, DATES)
+
+    def test_multi_layer_income_is_added_once(self):
+        daily = self.build_daily()
+        mask = (
+            daily["entity_type"].eq(ENTITY_GZH)
+            & daily["公众号名"].eq("高ROI公众号")
+        )
+        daily.loc[mask, "效率收入"] = 100
+        daily.loc[mask, "多层裂变收入"] = 200
+        daily.loc[mask, "成本"] = 100
+
+        _, _, summary = evaluate_rules(daily, DATES)
+        target = summary[
+            summary["entity_type"].eq(ENTITY_GZH)
+            & summary["公众号名"].eq("高ROI公众号")
+        ].iloc[0]
+        self.assertAlmostEqual(target["ROI"], 3.0)
+
+    def test_self_age_blocks_actions(self):
+        daily = self.build_daily()
+        ages = pd.DataFrame({"广告id": ["ad-0", "ad-9"], "广告age": [2, 2]})
+        candidates, _, _ = evaluate_rules(daily, DATES, ages)
+        blocked_ids = set(candidates.loc[candidates["entity_type"].eq(ENTITY_SELF), "广告id"])
+        self.assertNotIn("ad-0", blocked_ids)
+        self.assertNotIn("ad-9", blocked_ids)
+
+    def test_pyodps_lowercase_aliases_are_normalized(self):
+        daily = self.build_daily().rename(
+            columns={"首层UV": "首层uv", "T0裂变数": "t0裂变数"}
+        )
+        candidates, thresholds, _ = evaluate_rules(daily, DATES)
+        self.assertFalse(candidates.empty)
+        self.assertEqual(len(thresholds), 1)
+
+    def test_zero_roi_at_stop_boundary_is_stopped(self):
+        rows = []
+        for dt in DATES:
+            for index, roi in enumerate([0, 0, 0, 0, 0, 1, 2, 3, 4, 5]):
+                rows.append(row(ENTITY_SELF, SELF_CHANNEL, f"boundary-{index}", dt, roi))
+        ages = pd.DataFrame(
+            {
+                "广告id": [f"boundary-{index}" for index in range(10)],
+                "广告age": [10] * 10,
+            }
+        )
+        candidates, thresholds, _ = evaluate_rules(pd.DataFrame(rows), DATES, ages)
+        self.assertEqual(thresholds.iloc[0]["t_stop"], 0)
+        stopped = candidates[candidates["动作"].eq("关停")]
+        self.assertTrue(stopped["广告id"].eq("boundary-0").any())
+
+    def test_display_name_change_does_not_break_three_day_identity(self):
+        daily = self.build_daily()
+        mask = daily["广告id"].eq("ad-9")
+        daily.loc[mask & daily["dt"].eq(DATES[1]), "广告名称"] = "改名后的广告"
+        ages = pd.DataFrame({"广告id": ["ad-9"], "广告age": [10]})
+        candidates, _, summary = evaluate_rules(daily, DATES, ages)
+        target = summary[
+            summary["entity_type"].eq(ENTITY_SELF)
+            & summary["广告id"].eq("ad-9")
+        ]
+        self.assertEqual(len(target), 1)
+        self.assertTrue(candidates["广告id"].eq("ad-9").any())
+
+    def test_global_threshold_pool_mixes_all_entity_types_after_uv_gate(self):
+        rows = []
+        specifications = [
+            (ENTITY_SELF, SELF_CHANNEL, "self-qualified", 1.0, 300),
+            (ENTITY_GZH, GZH_CHANNEL, "gzh-qualified", 2.0, 300),
+            (ENTITY_QIWEI, QIWEI_CHANNEL, "qiwei-qualified", 3.0, 300),
+            (ENTITY_GZH, GZH_CHANNEL, "gzh-low-uv-excluded", 100.0, 100),
+        ]
+        for entity_type, channel, entity_id, roi, uv in specifications:
+            for dt in DATES:
+                rows.append(row(entity_type, channel, entity_id, dt, roi, uv=uv))
+
+        _, thresholds, summary = evaluate_rules(pd.DataFrame(rows), DATES)
+        threshold = thresholds.iloc[0]
+        expected = pd.Series([1.0, 2.0, 3.0])
+        self.assertAlmostEqual(threshold["t_stop"], expected.quantile(0.10))
+        self.assertAlmostEqual(threshold["t_up"], expected.quantile(0.80))
+        self.assertEqual(threshold["阈值样本数"], 3)
+        self.assertEqual(threshold["小程序样本数"], 1)
+        self.assertEqual(threshold["公众号样本数"], 1)
+        self.assertEqual(threshold["企微样本数"], 1)
+        self.assertEqual(len(summary), 4)
+
+    def test_actions_use_three_day_aggregate_not_each_daily_roi(self):
+        rows = []
+        daily_rois = [0.0, 0.0, 6.0]
+        for dt, roi in zip(DATES, daily_rois):
+            rows.append(row(ENTITY_GZH, GZH_CHANNEL, "波动公众号", dt, roi, uv=300))
+        for dt in DATES:
+            rows.append(row(ENTITY_SELF, SELF_CHANNEL, "基准创意", dt, 1.0, uv=600))
+            rows.append(row(ENTITY_QIWEI, QIWEI_CHANNEL, "基准企微", dt, 3.0, uv=300))
+
+        candidates, thresholds, summary = evaluate_rules(pd.DataFrame(rows), DATES)
+        target = summary[summary["公众号名"].eq("波动公众号")].iloc[0]
+        self.assertAlmostEqual(target["ROI"], 2.0)
+        self.assertAlmostEqual(target["t_stop"], thresholds.iloc[0]["t_stop"])
+        self.assertAlmostEqual(target["t_up"], thresholds.iloc[0]["t_up"])
+        self.assertNotEqual(
+            candidates[candidates["公众号名"].eq("波动公众号")]["动作"].tolist(),
+            ["关停"],
+        )
+
+    def test_uv_gate_uses_three_day_average_not_each_day(self):
+        daily = self.build_daily()
+        target_mask = daily["广告id"].eq("ad-9")
+        daily.loc[target_mask, "首层UV"] = [100, 250, 400]
+        ages = pd.DataFrame({"广告id": ["ad-9"], "广告age": [10]})
+
+        candidates, thresholds, summary = evaluate_rules(daily, DATES, ages)
+        target = summary[summary["广告id"].eq("ad-9")].iloc[0]
+        self.assertEqual(target["日均首层UV"], 250)
+        self.assertEqual(
+            candidates[candidates["广告id"].eq("ad-9")]["动作"].tolist(),
+            ["扩量"],
+        )
+        self.assertEqual(thresholds.iloc[0]["小程序样本数"], 10)
+
+    def test_every_day_cost_must_be_above_100(self):
+        daily = self.build_daily()
+        target_mask = daily["广告id"].eq("ad-9")
+        daily.loc[target_mask, "成本"] = [100, 200, 200]
+        daily.loc[target_mask, "效率收入"] = [500, 1000, 1000]
+        ages = pd.DataFrame({"广告id": ["ad-9"], "广告age": [10]})
+
+        candidates, thresholds, summary = evaluate_rules(daily, DATES, ages)
+        target = summary[summary["广告id"].eq("ad-9")].iloc[0]
+        self.assertEqual(target["三日最小单日成本"], 100)
+        self.assertFalse(candidates["广告id"].eq("ad-9").any())
+        self.assertEqual(thresholds.iloc[0]["小程序样本数"], 9)
+
+
+if __name__ == "__main__":
+    unittest.main()

examples/auto_put_ad_mini/test_roi_control_policy.py

@@ -0,0 +1,153 @@
+import os
+import unittest
+from datetime import date
+from decimal import Decimal
+from unittest.mock import patch
+
+import pandas as pd
+
+from operator_commands import ACTION_CONFIRM, ACTION_REJECT, parse_command
+from roi_control.config import RoiConfig
+from roi_control.execution import plan_scale_bid
+from roi_control.metrics import ENTITY_GZH, ENTITY_SELF
+from roi_control.policy import (
+    ACTION_PAUSE_CREATIVE,
+    ACTION_SCALE_BID,
+    MODE_ACTIONABLE,
+    MODE_NOTIFY_ONLY,
+    annotate_execution,
+)
+
+
+class RoiControlPolicyTest(unittest.TestCase):
+    def test_actionability_is_separate_from_metric_recommendation(self):
+        summary = pd.DataFrame(
+            [
+                {
+                    "entity_type": ENTITY_SELF,
+                    "channel": "小程序投流-稳定",
+                    "账号id": "84502354",
+                    "广告id": "1001",
+                    "创意id": "2001",
+                    "动作": "关停",
+                    "动作原因": "低ROI",
+                },
+                {
+                    "entity_type": ENTITY_SELF,
+                    "channel": "小程序投流-稳定",
+                    "账号id": "84502354",
+                    "广告id": "1001",
+                    "创意id": "2002",
+                    "动作": "扩量",
+                    "动作原因": "高ROI",
+                },
+                {
+                    "entity_type": ENTITY_SELF,
+                    "channel": "小程序投流-稳定",
+                    "账号id": "99999999",
+                    "广告id": "3001",
+                    "创意id": "4001",
+                    "动作": "关停",
+                    "动作原因": "低ROI",
+                },
+                {
+                    "entity_type": ENTITY_GZH,
+                    "channel": "公众号合作-即转-稳定",
+                    "合作方名": "合作方",
+                    "公众号名": "公众号",
+                    "动作": "扩量",
+                    "动作原因": "高ROI",
+                },
+            ]
+        )
+        annotated, snapshots, actions = annotate_execution(
+            summary,
+            {84502354},
+            "roi_20260725_north_star_roi_v1_roi_policy_v1",
+        )
+
+        self.assertEqual(len(snapshots), 4)
+        self.assertEqual(
+            {action["action_type"] for action in actions},
+            {ACTION_PAUSE_CREATIVE, ACTION_SCALE_BID},
+        )
+        self.assertEqual(
+            annotated.iloc[0]["执行模式"],
+            MODE_ACTIONABLE,
+        )
+        self.assertEqual(annotated.iloc[2]["执行模式"], MODE_NOTIFY_ONLY)
+        self.assertEqual(annotated.iloc[3]["执行模式"], MODE_NOTIFY_ONLY)
+
+    def test_scale_actions_are_deduplicated_by_ad(self):
+        rows = [
+            {
+                "entity_type": ENTITY_SELF,
+                "channel": "小程序投流-稳定",
+                "账号id": "84502354",
+                "广告id": "1001",
+                "创意id": str(2000 + index),
+                "动作": "扩量",
+                "动作原因": "高ROI",
+            }
+            for index in range(3)
+        ]
+        _, _, actions = annotate_execution(
+            pd.DataFrame(rows),
+            {84502354},
+            "roi_20260725_north_star_roi_v1_roi_policy_v1",
+        )
+        self.assertEqual(len(actions), 1)
+        self.assertEqual(actions[0]["action_type"], ACTION_SCALE_BID)
+
+    def test_invalid_apply_without_daily_job_is_rejected(self):
+        environment = {
+            "DAILY_ROI_ENABLED": "0",
+            "ROI_APPLY_ENABLED": "1",
+        }
+        with patch.dict(os.environ, environment, clear=False):
+            with self.assertRaisesRegex(ValueError, "requires DAILY_ROI_ENABLED"):
+                RoiConfig.from_env()
+
+    def test_roi_batch_commands_are_deterministic(self):
+        confirm = parse_command("确认 roi_20260725_north_star_roi_v1")
+        reject = parse_command("拒绝 roi_20260725_north_star_roi_v1")
+        old_confirm = parse_command("确认 cmd_20260725120000_abcd1234")
+        self.assertEqual(confirm.action, ACTION_CONFIRM)
+        self.assertEqual(reject.action, ACTION_REJECT)
+        self.assertEqual(confirm.command_id, "roi_20260725_north_star_roi_v1")
+        self.assertEqual(old_confirm.action, ACTION_CONFIRM)
+
+    def test_restored_intraday_boost_is_not_reapplied_by_roi(self):
+        plan = plan_scale_bid(
+            current_bid_fen=100,
+            base_bid_fen=100,
+            initial_base_bid_fen=100,
+            boosted_date=date(2026, 7, 26),
+            control_date=date(2026, 7, 26),
+            intraday_ratio=Decimal("1.25"),
+            scale_ratio=Decimal("1.10"),
+            max_base_ratio=Decimal("2.00"),
+        )
+        self.assertTrue(plan["allowed"])
+        self.assertEqual(plan["new_base_bid_fen"], 110)
+        self.assertEqual(plan["target_bid_fen"], 110)
+        self.assertTrue(plan["boosted_date_today"])
+        self.assertFalse(plan["preserve_intraday_boost"])
+
+    def test_active_intraday_boost_is_preserved_relative_to_new_base(self):
+        plan = plan_scale_bid(
+            current_bid_fen=125,
+            base_bid_fen=100,
+            initial_base_bid_fen=100,
+            boosted_date=date(2026, 7, 26),
+            control_date=date(2026, 7, 26),
+            intraday_ratio=Decimal("1.25"),
+            scale_ratio=Decimal("1.10"),
+            max_base_ratio=Decimal("2.00"),
+        )
+        self.assertEqual(plan["target_bid_fen"], 138)
+        self.assertTrue(plan["preserve_intraday_boost"])
+
+
+if __name__ == "__main__":
+    unittest.main()

examples/tencent_realtime_control/.env.example

@@ -19,6 +19,17 @@ RTC_COMMAND_ALLOWED_OPEN_IDS=
 RTC_COMMAND_CONFIRM_TTL_MINUTES=10
 RTC_COMMAND_WORKERS=2
 
+DAILY_ROI_ENABLED=0
+DAILY_ROI_HOUR=11
+DAILY_ROI_MINUTE=0
+DAILY_ROI_RUN_ON_STARTUP=0
+ROI_APPLY_ENABLED=0
+ROI_APPROVAL_TTL_MINUTES=120
+ROI_SCALE_RATIO=1.10
+ROI_SCALE_COOLDOWN_DAYS=3
+ROI_MAX_BASE_RATIO=2.00
+ROI_FEISHU_CHAT_ID=
+
 ODPS_ACCESS_ID=
 ODPS_ACCESS_SECRET=
 ODPS_PROJECT=loghubods

examples/tencent_realtime_control/README.md

@@ -155,6 +155,21 @@ dry-run 切换步骤见
 `examples/auto_put_ad_mini/docs/unified_services_deployment.md`。
 生产 Compose 文件兼容当前 ECS 的 Docker Compose `2.27.0`。
 
+## 日级 ROI 整批审批
+
+`ad-daily-service` 可在每天 11:00 计算版本化的三日 ROI 指标并发布只读飞书表；
+`ad-control-service` 复用本服务的唯一飞书 WebSocket 处理整批命令：
+
+```text
+确认 roi_<run_id>
+拒绝 roi_<run_id>
+```
+
+`DAILY_ROI_ENABLED=1` 只开启计算和通知；`ROI_APPLY_ENABLED=1` 才允许审批后
+暂停低 ROI 动态创意或提高高 ROI 广告的永久基础出价。ROI 腾讯写操作与实时
+CPM 调控共用 `RTC_DB_LOCK_NAME`，并在写入后回读验证。详细开关和分阶段上线
+步骤见统一部署文档。
+
 ## ECS 首次迁移 ODPS 配置
 
 仅用于旧 `auto_put_ad_mini` 容器仍在运行、但 ECS 的

examples/tencent_realtime_control/feishu_command_service.py

@@ -20,6 +20,7 @@ from operator_commands import (
     ACTION_CONFIRM,
     ACTION_DAY_PAUSE,
     ACTION_RESUME,
+    ACTION_REJECT,
     ACTION_STATUS,
     ACTION_STOP,
     parse_command,
@@ -31,6 +32,8 @@ from operator_control import (
     preview_write_command,
 )
 from realtime_config import RealtimeControlConfig
+from roi_control.config import RoiConfig
+from roi_control.execution import execute_roi_batch, reject_roi_batch
 
 
 SHANGHAI = ZoneInfo("Asia/Shanghai")
@@ -72,6 +75,7 @@ class FeishuCommandService:
             )
         self.apply = apply
         self.config = RealtimeControlConfig.from_env()
+        self.roi_config = RoiConfig.from_env()
         self.confirmation_ttl_minutes = int(
             os.getenv("RTC_COMMAND_CONFIRM_TTL_MINUTES", "10")
         )
@@ -132,6 +136,8 @@ class FeishuCommandService:
                 )
                 return
             if parsed.action == ACTION_CANCEL:
+                if (parsed.command_id or "").startswith("roi_"):
+                    raise ValueError("ROI批次请使用“拒绝 roi_<run_id>”")
                 command = cancel_command(
                     parsed.command_id or "",
                     event.sender_open_id,
@@ -150,7 +156,40 @@ class FeishuCommandService:
                     message,
                 )
                 return
+            if parsed.action == ACTION_REJECT:
+                run = reject_roi_batch(
+                    parsed.command_id or "",
+                    sender_open_id=event.sender_open_id,
+                    now=now,
+                )
+                self._reply(
+                    event,
+                    f"ROI批次 {run['run_id']} 已处理\n- 状态：{run['status']}",
+                )
+                return
             if parsed.action == ACTION_CONFIRM:
+                if (parsed.command_id or "").startswith("roi_"):
+                    if not self.roi_config.apply_enabled:
+                        self._reply(
+                            event,
+                            "ROI_APPLY_ENABLED=0，当前只生成报告，不执行腾讯写操作。",
+                        )
+                        return
+                    result = execute_roi_batch(
+                        parsed.command_id or "",
+                        sender_open_id=event.sender_open_id,
+                        now=now,
+                        lock_name=self.config.lock_name,
+                    )
+                    self._reply(
+                        event,
+                        f"ROI批次 {result['run_id']} 执行完成\n"
+                        f"- 状态：{result['status']}\n"
+                        f"- 成功：{result.get('successes', 0)} 条\n"
+                        f"- 跳过：{result.get('skipped', 0)} 条\n"
+                        f"- 失败：{result.get('failures', 0)} 条",
+                    )
+                    return
                 if not self.apply:
                     self._reply(event, "当前服务为 dry-run，禁止执行腾讯写操作。")
                     return

examples/tencent_realtime_control/operator_commands.py

@@ -12,10 +12,11 @@ ACTION_RESUME = "RESUME"
 ACTION_STATUS = "STATUS"
 ACTION_CONFIRM = "CONFIRM"
 ACTION_CANCEL = "CANCEL"
+ACTION_REJECT = "REJECT"
 
 WRITE_ACTIONS = {ACTION_DAY_PAUSE, ACTION_STOP, ACTION_RESUME}
 
-_COMMAND_ID_RE = re.compile(r"\bcmd_[0-9A-Za-z_-]+\b", re.IGNORECASE)
+_COMMAND_ID_RE = re.compile(r"\b(?:cmd|roi)_[0-9A-Za-z_-]+\b", re.IGNORECASE)
 _ACCOUNT_ID_RE = re.compile(r"(?<!\d)(\d{7,12})(?!\d)")
 
 
@@ -66,6 +67,13 @@ def parse_command(raw: str) -> ParsedCommand | None:
             action=ACTION_CANCEL,
             command_id=command_match.group(0).lower(),
         )
+    if lowered.startswith("拒绝"):
+        if not command_match or not command_match.group(0).lower().startswith("roi_"):
+            raise ValueError("拒绝命令缺少 roi_run_id")
+        return ParsedCommand(
+            action=ACTION_REJECT,
+            command_id=command_match.group(0).lower(),
+        )
 
     if "查看暂停状态" in text or "查询暂停状态" in text:
         return ParsedCommand(action=ACTION_STATUS, scope_type="ALL")

examples/tencent_realtime_control/run_control_service.py

@@ -11,12 +11,15 @@ from pathlib import Path
 
 
 ROOT = Path(__file__).resolve().parents[2]
-if str(ROOT) not in sys.path:
-    sys.path.insert(0, str(ROOT))
+AUTO_DIR = ROOT / "examples" / "auto_put_ad_mini"
+for path in (ROOT, AUTO_DIR):
+    if str(path) not in sys.path:
+        sys.path.insert(0, str(path))
 
 from run_once import load_environment  # noqa: E402
 from run_scheduler import run_forever  # noqa: E402
 from storage import initialize_schema  # noqa: E402
+from roi_control.config import RoiConfig  # noqa: E402
 
 
 logger = logging.getLogger("tencent_realtime_control.service")
@@ -43,6 +46,7 @@ def main() -> None:
     load_environment()
     args = parse_args()
     apply = args.apply or _env_flag("RTC_APPLY_ENABLED")
+    RoiConfig.from_env()
     initialize_schema()
     if os.getenv("RTC_COMMAND_ENABLED", "1").strip().lower() in {
         "1",

examples/tencent_realtime_control/schema.sql

@@ -17,7 +17,9 @@ CREATE TABLE IF NOT EXISTS realtime_control_ad_state (
     adgroup_name VARCHAR(255) DEFAULT NULL,
     bid_field VARCHAR(32) NOT NULL,
     base_bid_fen INT NOT NULL,
+    initial_base_bid_fen INT DEFAULT NULL,
     boosted_date DATE DEFAULT NULL,
+    last_roi_scaled_at DATETIME DEFAULT NULL,
     paused_by_strategy BOOLEAN NOT NULL DEFAULT FALSE,
     pause_reason VARCHAR(32) DEFAULT NULL,
     operator_pause_mode VARCHAR(32) DEFAULT NULL,
@@ -37,6 +39,104 @@ CREATE TABLE IF NOT EXISTS realtime_control_ad_state (
     KEY idx_last_seen (last_seen_at)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='实时CPM调控广告基准与状态';
 
+CREATE TABLE IF NOT EXISTS roi_metric_run (
+    run_id VARCHAR(64) NOT NULL PRIMARY KEY,
+    run_key VARCHAR(128) NOT NULL,
+    metric_version VARCHAR(64) NOT NULL,
+    policy_version VARCHAR(64) NOT NULL,
+    start_date DATE NOT NULL,
+    end_date DATE NOT NULL,
+    config_json LONGTEXT NOT NULL,
+    thresholds_json LONGTEXT DEFAULT NULL,
+    status VARCHAR(32) NOT NULL,
+    entity_count INT NOT NULL DEFAULT 0,
+    candidate_count INT NOT NULL DEFAULT 0,
+    actionable_count INT NOT NULL DEFAULT 0,
+    sheet_token VARCHAR(128) DEFAULT NULL,
+    sheet_url TEXT DEFAULT NULL,
+    message_id VARCHAR(128) DEFAULT NULL,
+    expires_at DATETIME DEFAULT NULL,
+    approved_by VARCHAR(128) DEFAULT NULL,
+    approved_at DATETIME DEFAULT NULL,
+    rejected_by VARCHAR(128) DEFAULT NULL,
+    rejected_at DATETIME DEFAULT NULL,
+    executed_at DATETIME DEFAULT NULL,
+    error_message TEXT DEFAULT NULL,
+    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+    UNIQUE KEY uk_roi_run_key (run_key),
+    KEY idx_roi_run_status_expiry (status, expires_at),
+    KEY idx_roi_run_end_date (end_date)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='版本化ROI计算与整批审批运行';
+
+CREATE TABLE IF NOT EXISTS roi_entity_snapshot (
+    id BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY,
+    run_id VARCHAR(64) NOT NULL,
+    entity_hash CHAR(64) NOT NULL,
+    entity_type VARCHAR(32) NOT NULL,
+    channel VARCHAR(128) DEFAULT NULL,
+    account_id BIGINT DEFAULT NULL,
+    account_name VARCHAR(255) DEFAULT NULL,
+    adgroup_id BIGINT DEFAULT NULL,
+    adgroup_name VARCHAR(255) DEFAULT NULL,
+    dynamic_creative_id BIGINT DEFAULT NULL,
+    audience_name VARCHAR(255) DEFAULT NULL,
+    partner_name VARCHAR(255) DEFAULT NULL,
+    official_account_name VARCHAR(255) DEFAULT NULL,
+    ad_age INT DEFAULT NULL,
+    avg_first_uv DECIMAL(18,4) DEFAULT NULL,
+    first_uv DECIMAL(18,4) DEFAULT NULL,
+    t0_fission_count DECIMAL(18,4) DEFAULT NULL,
+    t0_fission_rate DECIMAL(18,8) DEFAULT NULL,
+    cost DECIMAL(20,4) DEFAULT NULL,
+    efficiency_revenue DECIMAL(20,4) DEFAULT NULL,
+    fission_revenue DECIMAL(20,4) DEFAULT NULL,
+    total_revenue DECIMAL(20,4) DEFAULT NULL,
+    roi DECIMAL(18,8) DEFAULT NULL,
+    stop_threshold DECIMAL(18,8) DEFAULT NULL,
+    scale_threshold DECIMAL(18,8) DEFAULT NULL,
+    recommended_action VARCHAR(64) DEFAULT NULL,
+    action_reason TEXT DEFAULT NULL,
+    execution_mode VARCHAR(32) NOT NULL,
+    ineligible_reason TEXT DEFAULT NULL,
+    daily_metrics_json LONGTEXT NOT NULL,
+    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    UNIQUE KEY uk_roi_entity (run_id, entity_hash),
+    KEY idx_roi_entity_action (run_id, recommended_action),
+    KEY idx_roi_entity_ad (account_id, adgroup_id),
+    KEY idx_roi_entity_creative (account_id, dynamic_creative_id)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='ROI实体指标快照';
+
+CREATE TABLE IF NOT EXISTS roi_action_item (
+    id BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY,
+    run_id VARCHAR(64) NOT NULL,
+    idempotency_key VARCHAR(160) NOT NULL,
+    action_type VARCHAR(32) NOT NULL,
+    account_id BIGINT NOT NULL,
+    adgroup_id BIGINT NOT NULL,
+    dynamic_creative_id BIGINT DEFAULT NULL,
+    bid_field VARCHAR(32) DEFAULT NULL,
+    initial_base_bid_fen INT DEFAULT NULL,
+    base_bid_fen INT DEFAULT NULL,
+    before_bid_fen INT DEFAULT NULL,
+    target_bid_fen INT DEFAULT NULL,
+    before_status VARCHAR(50) DEFAULT NULL,
+    target_status VARCHAR(50) DEFAULT NULL,
+    readback_status VARCHAR(50) DEFAULT NULL,
+    execution_status VARCHAR(32) NOT NULL DEFAULT 'PENDING',
+    skip_reason TEXT DEFAULT NULL,
+    error_message TEXT DEFAULT NULL,
+    pre_state_json LONGTEXT DEFAULT NULL,
+    readback_json LONGTEXT DEFAULT NULL,
+    executed_at DATETIME DEFAULT NULL,
+    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+    UNIQUE KEY uk_roi_action_idempotency (idempotency_key),
+    KEY idx_roi_action_run_status (run_id, execution_status),
+    KEY idx_roi_action_ad (account_id, adgroup_id),
+    KEY idx_roi_action_creative (account_id, dynamic_creative_id)
+) 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

@@ -58,6 +58,8 @@ def initialize_schema() -> None:
             )
             existing_columns = {row["COLUMN_NAME"] for row in cursor.fetchall()}
             migrations = {
+                "initial_base_bid_fen": "INT DEFAULT NULL",
+                "last_roi_scaled_at": "DATETIME DEFAULT NULL",
                 "operator_pause_mode": "VARCHAR(32) DEFAULT NULL",
                 "operator_resume_at": "DATETIME DEFAULT NULL",
                 "operator_command_id": "VARCHAR(64) DEFAULT NULL",
@@ -70,6 +72,13 @@ def initialize_schema() -> None:
                         f"ALTER TABLE realtime_control_ad_state "
                         f"ADD COLUMN {column} {definition}"
                     )
+            cursor.execute(
+                """
+                UPDATE realtime_control_ad_state
+                SET initial_base_bid_fen=base_bid_fen
+                WHERE initial_base_bid_fen IS NULL
+                """
+            )
             cursor.execute(
                 """
                 SELECT INDEX_NAME
@@ -243,12 +252,17 @@ def upsert_ad_state(
                 """
                 INSERT INTO realtime_control_ad_state
                     (account_id, adgroup_id, adgroup_name, bid_field, base_bid_fen,
+                     initial_base_bid_fen,
                      boosted_date, paused_by_strategy, pause_reason, last_action,
                      last_action_at, last_seen_at)
-                VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
+                VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
                 ON DUPLICATE KEY UPDATE
                     adgroup_name=VALUES(adgroup_name),
                     bid_field=VALUES(bid_field),
+                    initial_base_bid_fen=COALESCE(
+                        initial_base_bid_fen,
+                        VALUES(initial_base_bid_fen)
+                    ),
                     boosted_date=VALUES(boosted_date),
                     paused_by_strategy=VALUES(paused_by_strategy),
                     pause_reason=VALUES(pause_reason),
@@ -262,6 +276,7 @@ def upsert_ad_state(
                     adgroup_name,
                     bid_field,
                     base_bid_fen,
+                    base_bid_fen,
                     boosted_date,
                     paused_by_strategy,
                     pause_reason,
@@ -273,6 +288,61 @@ def upsert_ad_state(
     finally:
         connection.close()
 
+
+def sync_roi_base_bid(
+    *,
+    account_id: int,
+    adgroup_id: int,
+    adgroup_name: str,
+    bid_field: str,
+    initial_base_bid_fen: int,
+    new_base_bid_fen: int,
+    boosted_date: date | None,
+    action_at: datetime,
+) -> None:
+    """Persist a permanent ROI base bid without losing intraday boost state."""
+
+    connection = connect()
+    try:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                """
+                INSERT INTO realtime_control_ad_state
+                    (account_id, adgroup_id, adgroup_name, bid_field,
+                     base_bid_fen, initial_base_bid_fen, boosted_date,
+                     paused_by_strategy, pause_reason, last_action,
+                     last_action_at, last_seen_at, last_roi_scaled_at)
+                VALUES (%s,%s,%s,%s,%s,%s,%s,FALSE,NULL,'ROI_SCALE',%s,%s,%s)
+                ON DUPLICATE KEY UPDATE
+                    adgroup_name=VALUES(adgroup_name),
+                    bid_field=VALUES(bid_field),
+                    base_bid_fen=VALUES(base_bid_fen),
+                    initial_base_bid_fen=COALESCE(
+                        initial_base_bid_fen,
+                        VALUES(initial_base_bid_fen)
+                    ),
+                    boosted_date=VALUES(boosted_date),
+                    last_action='ROI_SCALE',
+                    last_action_at=VALUES(last_action_at),
+                    last_seen_at=VALUES(last_seen_at),
+                    last_roi_scaled_at=VALUES(last_roi_scaled_at)
+                """,
+                (
+                    account_id,
+                    adgroup_id,
+                    adgroup_name,
+                    bid_field,
+                    new_base_bid_fen,
+                    initial_base_bid_fen,
+                    boosted_date,
+                    action_at,
+                    action_at,
+                    action_at,
+                ),
+            )
+    finally:
+        connection.close()
+
 def insert_action_log(record: dict[str, Any]) -> None:
     columns = [
         "run_id",

examples/tencent_realtime_control/tencent_client.py

@@ -29,6 +29,12 @@ AD_FIELDS = [
     "end_date",
     "time_series",
 ]
+DYNAMIC_CREATIVE_FIELDS = [
+    "dynamic_creative_id",
+    "adgroup_id",
+    "configured_status",
+    "system_status",
+]
 
 
 class PostWriteVerificationError(RuntimeError):
@@ -303,6 +309,108 @@ class TencentClient:
             actual=last_actual,
         )
 
+    def get_dynamic_creative(
+        self,
+        account_id: int,
+        dynamic_creative_id: int,
+    ) -> dict[str, Any]:
+        params = {
+            **self._common_params(account_id),
+            "account_id": account_id,
+            "fields": json.dumps(DYNAMIC_CREATIVE_FIELDS, ensure_ascii=False),
+            "filtering": json.dumps(
+                [
+                    {
+                        "field": "dynamic_creative_id",
+                        "operator": "IN",
+                        "values": [str(dynamic_creative_id)],
+                    }
+                ]
+            ),
+            "page": 1,
+            "page_size": 10,
+        }
+        response = self.session.get(
+            f"{self.base_url}/dynamic_creatives/get",
+            params=params,
+            timeout=self.timeout,
+        )
+        response.raise_for_status()
+        data = self._check(response.json(), "get_dynamic_creative")
+        for row in data.get("list") or []:
+            if int(row.get("dynamic_creative_id") or 0) == dynamic_creative_id:
+                return row
+        raise RuntimeError(
+            "Dynamic creative not found: "
+            f"account={account_id} creative={dynamic_creative_id}"
+        )
+
+    def update_dynamic_creative_status(
+        self,
+        account_id: int,
+        dynamic_creative_id: int,
+        target_status: str,
+    ) -> dict[str, Any]:
+        try:
+            params = {
+                **self._common_params(account_id),
+                "user_token": self._user_token(account_id),
+            }
+        except Exception as exc:
+            raise TencentWriteNotSentError(str(exc)) from exc
+        try:
+            response = self.session.post(
+                f"{self.base_url}/dynamic_creatives/update",
+                params=params,
+                json={
+                    "account_id": account_id,
+                    "dynamic_creative_id": dynamic_creative_id,
+                    "configured_status": target_status,
+                },
+                timeout=self.timeout,
+            )
+        except requests.RequestException as exc:
+            raise TencentWriteOutcomeUnknownError(str(exc)) from exc
+        if response.status_code == 408 or response.status_code >= 500:
+            raise TencentWriteOutcomeUnknownError(
+                f"Tencent HTTP {response.status_code}: {response.text[:500]}"
+            )
+        try:
+            response.raise_for_status()
+            payload = response.json()
+            self._check(payload, "update_dynamic_creative")
+        except requests.HTTPError as exc:
+            raise TencentWriteRejectedError(str(exc)) from exc
+        except (ValueError, TypeError) as exc:
+            raise TencentWriteOutcomeUnknownError(
+                f"Tencent returned invalid JSON: {response.text[:500]}"
+            ) from exc
+        except RuntimeError as exc:
+            raise TencentWriteRejectedError(str(exc)) from exc
+
+        last_actual: dict[str, Any] = {}
+        for attempt in range(1, self.verify_attempts + 1):
+            try:
+                creative = self.get_dynamic_creative(
+                    account_id,
+                    dynamic_creative_id,
+                )
+                last_actual = {
+                    "configured_status": creative.get("configured_status")
+                }
+                if creative.get("configured_status") == target_status:
+                    return creative
+            except Exception as exc:
+                last_actual = {"verification_error": str(exc)}
+            if attempt < self.verify_attempts:
+                time.sleep(self.verify_delay_seconds)
+        raise PostWriteVerificationError(
+            account_id=account_id,
+            adgroup_id=dynamic_creative_id,
+            expected={"configured_status": target_status},
+            actual=last_actual,
+        )
+
     def update_ad_begin_dates(
         self,
         account_id: int,