feat(realtime-control): add CPM-driven Tencent ad control

examples/tencent_realtime_control/.env.example

@@ -0,0 +1,32 @@
+RTC_START_HOUR=7
+RTC_STOP_HOUR=21
+RTC_POLL_SECONDS=300
+RTC_HIGH_CPM=250
+RTC_LOW_CPM=190
+RTC_BID_UP_RATIO=1.25
+RTC_INVENTORY_REFRESH_MINUTES=60
+RTC_MAX_PARTITION_LAG_HOURS=2
+RTC_DB_LOCK_NAME=tencent_realtime_control
+RTC_FEISHU_NOTIFY_ENABLED=1
+RTC_FEISHU_CHAT_ID=
+RTC_FEISHU_TIMEOUT_SECONDS=30
+RTC_VERIFY_ATTEMPTS=3
+RTC_VERIFY_DELAY_SECONDS=1
+
+ODPS_ACCESS_ID=
+ODPS_ACCESS_SECRET=
+ODPS_PROJECT=loghubods
+ODPS_ENDPOINT=http://service.odps.aliyun.com/api
+DB_HOST=
+DB_PORT=3306
+DB_USER=
+DB_PASSWORD=
+DB_NAME=
+
+TENCENT_AD_BASE_URL=https://api.e.qq.com/v3.0
+TENCENT_AD_TOKEN_API=https://api.piaoquantv.com/ad/put/tencent/getAccessToken
+TENCENT_AD_USER_TOKEN_API=https://api.piaoquantv.com/ad/put/tencent/getUserToken
+
+FEISHU_APP_ID=
+FEISHU_APP_SECRET=
+FEISHU_AD_PROJECT_CHAT_ID=

examples/tencent_realtime_control/.gitignore

@@ -0,0 +1,4 @@
+__pycache__/
+outputs/*.csv
+outputs/*.json
+outputs/*.xlsx

examples/tencent_realtime_control/README.md

@@ -0,0 +1,119 @@
+# 腾讯广告实时调控
+
+独立管理实时投放指标采集、状态判断和腾讯广告调控，不依赖广告创建或历史调价分析流程。
+
+## 数据和范围
+
+- 从 `loghubods.advertiser_data_da_hour` 读取当天整体小时数据。
+- CPM 直接使用 `真实cpm_总`。
+- 固定过滤名称、广告主、公司、行业、客户和落地页类型均为 `SUM`。
+- 账户只取 `ad_creation_account_config.enabled=1` 与
+  `account_whitelist.enabled=1` 的交集。
+- 普通出价使用 `bid_amount`，最大转化量使用 `custom_cost_cap`。
+- 每条广告首次纳管时将腾讯当前实际出价持久化为基础出价。
+
+## 调控规则
+
+北京时间每天 `07:00-21:00` 每 5 分钟检查一次：
+
+- `CPM > 250`：基础出价上调 25%。
+- `190 <= CPM <= 250`：恢复基础出价并恢复策略暂停的广告。
+- `CPM < 190`：先恢复基础出价，再暂停广告。
+- 同一广告每天最多上调一次；恢复基础价后，当天也不会再次上调。
+- `21:00`：先恢复基础出价，再暂停广告。
+- 次日 `07:00`：只恢复由本策略暂停的广告，不打开人工暂停广告。
+
+每个真实发生调价或关停动作的窗口会生成一张飞书在线表格并发送到调控群。
+群消息包含触发时间、数据分区、当前 CPM、决策、影响账户/广告数、动作分布和
+受影响广告当日累计消耗。表格包含日期、数据分区、当前 CPM、账户、广告 ID、
+广告名称、人群包、广告当日累计消耗/曝光/点击、基础出价、调整前出价、执行后
+当前出价和动作。
+同一窗口的所有成功动作合并在一张表中；无动作、dry-run 和已达到当日上调
+上限的广告不发送通知。飞书发送失败不会回滚已经成功的腾讯操作，本地 xlsx
+仍保存在 `outputs/`，错误会写入本轮 JSON 报告和进程日志。
+
+腾讯写接口返回 `code=0` 后不会立即判定成功。流程会按广告 ID 重新读取腾讯
+广告并核对本次修改的出价和状态字段，默认最多校验 3 次、间隔 1 秒。只有读回
+值与目标一致才更新成功状态并进入成功通知；持续不一致按失败写入动作审计，
+飞书卡片标红并在表格中展示错误。
+
+ODPS 小时分区可能延迟。当天尚未出现 `06` 点及之后的分区时只等待，不使用
+夜间低流量分区做关停判断。最新分区相对当前时间最多允许延迟 2 小时，超过后
+本轮不执行基于 CPM 的调价、恢复或关停；每天 07:00 的定时恢复不依赖 CPM。
+
+状态保存在 MySQL，使用数据库锁避免多 Pod 同时执行。相同分区和相同决策默认
+不重复扫描腾讯；每 60 分钟至少刷新一次广告清单以纳管新广告。
+
+## 初始化
+
+```bash
+.venv/bin/python examples/tencent_realtime_control/init_db.py
+```
+
+## 数据检查
+
+读取上海时区当天分时 CPM：
+
+```bash
+.venv/bin/python examples/tencent_realtime_control/fetch_daily_hourly_cpm.py
+```
+
+指定日期：
+
+```bash
+.venv/bin/python examples/tencent_realtime_control/fetch_daily_hourly_cpm.py \
+  --date 20260724
+```
+
+## 单次执行
+
+默认 dry-run，不修改腾讯：
+
+```bash
+.venv/bin/python examples/tencent_realtime_control/run_once.py
+```
+
+验证 CPM 分支：
+
+```bash
+.venv/bin/python examples/tencent_realtime_control/run_once.py \
+  --at 2026-07-24T15:00:00 \
+  --cpm-override 260
+```
+
+使用真实账户、广告、出价和当日指标模拟完整通知链路，不修改腾讯广告：
+
+```bash
+.venv/bin/python examples/tencent_realtime_control/run_once.py \
+  --at 2026-07-24T15:00:00 \
+  --cpm-override 260 \
+  --test-notification
+```
+
+模拟卡片、动作列和表格标题都会标记“模拟”。`--test-notification` 不能与
+`--apply` 同时使用。
+
+真实执行：
+
+```bash
+.venv/bin/python examples/tencent_realtime_control/run_once.py --apply
+```
+
+## Docker 调度
+
+长驻进程会在投放时段按 5 分钟边界执行，21 点收尾后休眠到次日 7 点：
+
+```bash
+.venv/bin/python examples/tencent_realtime_control/run_scheduler.py --apply
+```
+
+默认所有执行入口均为 dry-run，必须显式传 `--apply` 才会调用腾讯写接口。
+`--at` 和 `--cpm-override` 只允许 dry-run。
+
+## 环境变量
+
+参考 `.env.example`。部署必须提供 ODPS、MySQL、腾讯 token 服务和飞书应用
+配置，不能在代码中写入凭证。通知群优先读取 `RTC_FEISHU_CHAT_ID`，为空时
+复用广告调控流程的 `FEISHU_AD_PROJECT_CHAT_ID`；不会回退发送到个人或运营
+审批群。`RTC_FEISHU_NOTIFY_ENABLED=0` 可临时关闭
+通知。

examples/tencent_realtime_control/__init__.py

@@ -0,0 +1 @@
+"""Tencent advertising real-time control module."""

examples/tencent_realtime_control/feishu_notifier.py

@@ -0,0 +1,552 @@
+"""Send successful real-time control actions as a Feishu spreadsheet."""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+import time
+from datetime import datetime
+from decimal import Decimal
+from pathlib import Path
+from typing import Any
+
+import requests
+from openpyxl import Workbook
+from openpyxl.styles import Alignment, Font, PatternFill
+
+
+ROOT = Path(__file__).resolve().parent
+FEISHU_BASE_URL = "https://open.feishu.cn/open-apis"
+HEADERS = [
+    "日期",
+    "数据分区",
+    "当前CPM",
+    "账户ID",
+    "广告ID",
+    "广告名称",
+    "人群包",
+    "今日消耗(元)",
+    "今日曝光",
+    "今日点击",
+    "基础出价(元)",
+    "调整前出价(元)",
+    "当前出价(元)",
+    "动作",
+    "执行结果",
+    "错误信息",
+]
+
+logger = logging.getLogger("tencent_realtime_control.feishu")
+
+
+def _yuan(fen: int) -> float:
+    return float((Decimal(fen) / Decimal("100")).quantize(Decimal("0.00")))
+
+
+def _action_text(result: dict[str, Any], bid_up_ratio: Decimal) -> str:
+    decision = result["decision"]
+    bid_changed = bool(result.get("bid_change_needed"))
+    status_changed = bool(result.get("status_change_needed"))
+
+    if decision == "BOOST":
+        percent = int((bid_up_ratio - Decimal("1")) * Decimal("100"))
+        return f"出价上调{percent}%"
+    if decision == "RESTORE":
+        if bid_changed and status_changed:
+            return "恢复基础出价并开启投放"
+        if bid_changed:
+            return "恢复基础出价"
+        return "恢复投放"
+    if decision == "PAUSE":
+        return "恢复基础出价并关停（CPM过低）" if bid_changed else "关停（CPM过低）"
+    if decision == "CUTOFF":
+        return "恢复基础出价并关停（投放截止）" if bid_changed else "关停（投放截止）"
+    return decision
+
+
+def notification_rows(
+    *,
+    now: datetime,
+    results: list[dict[str, Any]],
+    bid_up_ratio: Decimal,
+    observed_partition: str | None,
+    observed_cpm: Decimal | None,
+    test_mode: bool = False,
+) -> list[list[Any]]:
+    rows: list[list[Any]] = []
+    for result in results:
+        expected_statuses = {"planned"} if test_mode else {"success", "failed"}
+        if result.get("status") not in expected_statuses:
+            continue
+
+        is_bid_change = bool(result.get("bid_change_needed"))
+        is_pause = (
+            result.get("decision") in {"PAUSE", "CUTOFF"}
+            and bool(result.get("status_change_needed"))
+        )
+        if not is_bid_change and not is_pause:
+            continue
+
+        rows.append(
+            [
+                now.strftime("%Y-%m-%d %H:%M:%S"),
+                observed_partition or "",
+                float(observed_cpm) if observed_cpm is not None else "",
+                str(result["account_id"]),
+                str(result["adgroup_id"]),
+                str(result.get("adgroup_name") or ""),
+                str(result.get("audience_name") or ""),
+                _yuan(int(result.get("today_cost_fen") or 0)),
+                int(result.get("today_impressions") or 0),
+                int(result.get("today_clicks") or 0),
+                _yuan(int(result["base_bid_fen"])),
+                _yuan(int(result["before_bid_fen"])),
+                (
+                    _yuan(int(result["target_bid_fen"]))
+                    if result.get("status") != "failed"
+                    else ""
+                ),
+                (
+                    f"【模拟】{_action_text(result, bid_up_ratio)}"
+                    if test_mode
+                    else _action_text(result, bid_up_ratio)
+                ),
+                (
+                    "模拟"
+                    if test_mode
+                    else "成功"
+                    if result.get("status") == "success"
+                    else "失败"
+                ),
+                str(result.get("error") or ""),
+            ]
+        )
+    return rows
+
+
+def create_notification_xlsx(
+    *,
+    run_id: str,
+    now: datetime,
+    rows: list[list[Any]],
+    test_mode: bool = False,
+) -> Path:
+    output_dir = ROOT / "outputs"
+    output_dir.mkdir(parents=True, exist_ok=True)
+    mode = "test" if test_mode else "apply"
+    path = output_dir / (
+        f"realtime_actions_{mode}_{now.strftime('%Y%m%d_%H%M%S')}_"
+        f"{run_id[:8]}.xlsx"
+    )
+
+    workbook = Workbook()
+    sheet = workbook.active
+    sheet.title = "实时调控动作_模拟" if test_mode else "实时调控动作"
+    sheet.append(HEADERS)
+    for row in rows:
+        sheet.append(row)
+
+    header_fill = PatternFill("solid", fgColor="1F4E78")
+    for cell in sheet[1]:
+        cell.fill = header_fill
+        cell.font = Font(color="FFFFFF", bold=True)
+        cell.alignment = Alignment(horizontal="center", vertical="center")
+
+    widths = [
+        21, 18, 12, 14, 18, 30, 22, 16, 14, 14, 16, 16, 16, 30, 12, 55
+    ]
+    for index, width in enumerate(widths, start=1):
+        sheet.column_dimensions[chr(64 + index)].width = width
+    for row in sheet.iter_rows(min_row=2):
+        for cell in row:
+            cell.alignment = Alignment(horizontal="center", vertical="center")
+    for column in ("C", "H", "K", "L", "M"):
+        for cell in sheet[column][1:]:
+            cell.number_format = "0.00"
+    sheet.freeze_panes = "A2"
+    sheet.auto_filter.ref = sheet.dimensions
+    workbook.save(path)
+    return path
+
+
+class FeishuNotifier:
+    def __init__(self) -> None:
+        self.app_id = os.getenv("FEISHU_APP_ID", "").strip()
+        self.app_secret = os.getenv("FEISHU_APP_SECRET", "").strip()
+        self.chat_id = (
+            os.getenv("RTC_FEISHU_CHAT_ID", "").strip()
+            or os.getenv("FEISHU_AD_PROJECT_CHAT_ID", "").strip()
+        )
+        self.timeout = int(os.getenv("RTC_FEISHU_TIMEOUT_SECONDS", "30"))
+        self.session = requests.Session()
+
+    def _require_config(self) -> None:
+        missing = [
+            name
+            for name, value in (
+                ("FEISHU_APP_ID", self.app_id),
+                ("FEISHU_APP_SECRET", self.app_secret),
+                ("RTC_FEISHU_CHAT_ID/FEISHU_AD_PROJECT_CHAT_ID", self.chat_id),
+            )
+            if not value
+        ]
+        if missing:
+            raise RuntimeError(
+                f"Missing Feishu notification configuration: {', '.join(missing)}"
+            )
+
+    def _tenant_token(self) -> str:
+        response = self.session.post(
+            f"{FEISHU_BASE_URL}/auth/v3/tenant_access_token/internal",
+            json={"app_id": self.app_id, "app_secret": self.app_secret},
+            timeout=self.timeout,
+        )
+        response.raise_for_status()
+        payload = response.json()
+        if payload.get("code") != 0:
+            raise RuntimeError(f"Feishu token failed: {payload}")
+        return str(payload["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 file:
+            response = self.session.post(
+                f"{FEISHU_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,
+                        file,
+                        "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
+                    )
+                },
+                timeout=max(self.timeout, 60),
+            )
+        response.raise_for_status()
+        payload = response.json()
+        if payload.get("code") != 0:
+            raise RuntimeError(f"Feishu upload failed: {payload}")
+        return str(payload["data"]["file_token"])
+
+    def _import_sheet(self, token: str, file_token: str, path: Path) -> dict[str, Any]:
+        response = self.session.post(
+            f"{FEISHU_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": ""},
+            },
+            timeout=self.timeout,
+        )
+        response.raise_for_status()
+        payload = response.json()
+        if payload.get("code") != 0:
+            raise RuntimeError(f"Feishu import task failed: {payload}")
+        ticket = str(payload["data"]["ticket"])
+
+        deadline = time.monotonic() + 60
+        while time.monotonic() < deadline:
+            response = self.session.get(
+                f"{FEISHU_BASE_URL}/drive/v1/import_tasks/{ticket}",
+                headers=self._headers(token),
+                timeout=self.timeout,
+            )
+            response.raise_for_status()
+            payload = response.json()
+            if payload.get("code") != 0:
+                raise RuntimeError(f"Feishu import query failed: {payload}")
+            result = (payload.get("data") or {}).get("result") or {}
+            if result.get("job_status") == 0:
+                return result
+            if result.get("job_status") == 3:
+                raise RuntimeError(
+                    f"Feishu import failed: {result.get('job_error_msg')}"
+                )
+            time.sleep(2)
+        raise RuntimeError("Feishu spreadsheet import timed out after 60 seconds")
+
+    def _set_read_permission(
+        self, token: str, sheet_token: str, file_type: str
+    ) -> None:
+        response = self.session.patch(
+            f"{FEISHU_BASE_URL}/drive/v1/permissions/{sheet_token}/public",
+            headers={**self._headers(token), "Content-Type": "application/json"},
+            params={"type": file_type},
+            json={
+                "external_access_entity": "open",
+                "link_share_entity": "anyone_readable",
+            },
+            timeout=self.timeout,
+        )
+        response.raise_for_status()
+        payload = response.json()
+        if payload.get("code") != 0:
+            logger.warning("Feishu permission update failed: %s", payload)
+
+    def _send_card(
+        self,
+        token: str,
+        *,
+        url: str,
+        now: datetime,
+        summary: dict[str, Any],
+        test_mode: bool,
+    ) -> None:
+        action_counts = summary["action_counts"]
+        action_text = "，".join(
+            f"{name} {count} 条"
+            for name, count in action_counts.items()
+            if count
+        )
+        cpm_text = (
+            f"{summary['observed_cpm']:.2f}"
+            if summary.get("observed_cpm") is not None
+            else "无（定时收尾）"
+        )
+        summary_lines = [
+            f"**触发时间：** {now.strftime('%Y-%m-%d %H:%M:%S')}",
+        ]
+        if test_mode:
+            summary_lines.append("**执行模式：** 仅模拟，未修改腾讯广告")
+        summary_lines.extend(
+            [
+                f"**数据分区：** {summary.get('observed_partition') or '无'}",
+                f"**当前 CPM：** {cpm_text}",
+                f"**本轮决策：** {summary['decision']}",
+                (
+                    f"**影响范围：** {summary['account_count']} 个账户，"
+                    f"{summary['row_count']} 条广告"
+                ),
+                f"**动作概要：** {action_text}",
+                (
+                    "**受影响广告今日消耗：** "
+                    f"{summary['today_cost_yuan']:.2f} 元"
+                ),
+                f"**执行结果：** 成功 {summary['success_count']} 条，失败 "
+                f"{summary['failure_count']} 条",
+            ]
+        )
+        card = {
+            "config": {"wide_screen_mode": True},
+            "header": {
+                "title": {
+                    "tag": "plain_text",
+                    "content": (
+                        "腾讯广告实时调控通知（模拟）"
+                        if test_mode
+                        else "腾讯广告实时调控通知"
+                    ),
+                },
+                "template": (
+                    "orange"
+                    if test_mode
+                    else "red"
+                    if summary["failure_count"]
+                    else "blue"
+                ),
+            },
+            "elements": [
+                {
+                    "tag": "div",
+                    "text": {
+                        "tag": "lark_md",
+                        "content": "\n".join(summary_lines),
+                    },
+                },
+                {
+                    "tag": "action",
+                    "actions": [
+                        {
+                            "tag": "button",
+                            "text": {"tag": "plain_text", "content": "打开动作明细"},
+                            "type": "primary",
+                            "url": url,
+                        }
+                    ],
+                },
+            ],
+        }
+        response = self.session.post(
+            f"{FEISHU_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),
+            },
+            timeout=self.timeout,
+        )
+        response.raise_for_status()
+        payload = response.json()
+        if payload.get("code") != 0:
+            raise RuntimeError(f"Feishu message failed: {payload}")
+
+    def send(
+        self,
+        *,
+        path: Path,
+        now: datetime,
+        summary: dict[str, Any],
+        test_mode: bool = False,
+    ) -> str:
+        self._require_config()
+        token = self._tenant_token()
+        file_token = self._upload(token, path)
+        result = self._import_sheet(token, file_token, path)
+        url = str(result.get("url") or "")
+        if not url:
+            raise RuntimeError("Feishu import succeeded without spreadsheet URL")
+        sheet_token = str(result.get("token") or "")
+        if sheet_token:
+            self._set_read_permission(
+                token,
+                sheet_token,
+                str(result.get("type") or "sheet"),
+            )
+        self._send_card(
+            token,
+            url=url,
+            now=now,
+            summary=summary,
+            test_mode=test_mode,
+        )
+        return url
+
+
+def build_notification_summary(
+    *,
+    rows: list[list[Any]],
+    results: list[dict[str, Any]],
+    decision: str,
+    observed_partition: str | None,
+    observed_cpm: Decimal | None,
+    test_mode: bool = False,
+) -> dict[str, Any]:
+    expected_statuses = {"planned"} if test_mode else {"success", "failed"}
+    notified = [
+        result
+        for result in results
+        if result.get("status") in expected_statuses
+        and (
+            result.get("bid_change_needed")
+            or (
+                result.get("decision") in {"PAUSE", "CUTOFF"}
+                and result.get("status_change_needed")
+            )
+        )
+    ]
+    return {
+        "decision": decision,
+        "observed_partition": observed_partition,
+        "observed_cpm": float(observed_cpm) if observed_cpm is not None else None,
+        "row_count": len(rows),
+        "account_count": len({int(row["account_id"]) for row in notified}),
+        "today_cost_yuan": sum(
+            int(row.get("today_cost_fen") or 0) for row in notified
+        )
+        / 100,
+        "success_count": sum(
+            row.get("status") in {"success", "planned"} for row in notified
+        ),
+        "failure_count": sum(row.get("status") == "failed" for row in notified),
+        "action_counts": {
+            "上调": sum(row.get("decision") == "BOOST" for row in notified),
+            "恢复基础价": sum(
+                row.get("decision") == "RESTORE"
+                and row.get("bid_change_needed")
+                for row in notified
+            ),
+            "关停": sum(
+                row.get("decision") in {"PAUSE", "CUTOFF"}
+                and row.get("status_change_needed")
+                for row in notified
+            ),
+        },
+    }
+
+
+def send_action_notification(
+    *,
+    run_id: str,
+    now: datetime,
+    results: list[dict[str, Any]],
+    bid_up_ratio: Decimal,
+    decision: str,
+    observed_partition: str | None,
+    observed_cpm: Decimal | None,
+    test_mode: bool = False,
+) -> dict[str, Any]:
+    rows = notification_rows(
+        now=now,
+        results=results,
+        bid_up_ratio=bid_up_ratio,
+        observed_partition=observed_partition,
+        observed_cpm=observed_cpm,
+        test_mode=test_mode,
+    )
+    if not rows:
+        return {"status": "no_actions", "row_count": 0}
+
+    path = create_notification_xlsx(
+        run_id=run_id,
+        now=now,
+        rows=rows,
+        test_mode=test_mode,
+    )
+    summary = build_notification_summary(
+        rows=rows,
+        results=results,
+        decision=decision,
+        observed_partition=observed_partition,
+        observed_cpm=observed_cpm,
+        test_mode=test_mode,
+    )
+    enabled = os.getenv("RTC_FEISHU_NOTIFY_ENABLED", "1").strip().lower() in {
+        "1",
+        "true",
+        "yes",
+        "y",
+        "on",
+    }
+    if not enabled:
+        return {
+            "status": "disabled",
+            "row_count": len(rows),
+            "xlsx_path": str(path),
+        }
+
+    try:
+        url = FeishuNotifier().send(
+            path=path,
+            now=now,
+            summary=summary,
+            test_mode=test_mode,
+        )
+        return {
+            "status": "sent",
+            "row_count": len(rows),
+            "xlsx_path": str(path),
+            "url": url,
+        }
+    except Exception as exc:
+        logger.exception("Feishu action notification failed")
+        return {
+            "status": "failed",
+            "row_count": len(rows),
+            "xlsx_path": str(path),
+            "error": str(exc),
+        }

examples/tencent_realtime_control/fetch_daily_hourly_cpm.py

@@ -0,0 +1,60 @@
+#!/usr/bin/env python
+"""Fetch one day's overall hourly CPM for real-time ad control."""
+
+from __future__ import annotations
+
+import argparse
+import re
+from datetime import datetime
+from pathlib import Path
+from zoneinfo import ZoneInfo
+
+from dotenv import load_dotenv
+
+from odps_source import build_odps_client, fetch_hourly_cpm
+
+SHANGHAI = ZoneInfo("Asia/Shanghai")
+ROOT = Path(__file__).resolve().parent
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--date",
+        default=datetime.now(SHANGHAI).strftime("%Y%m%d"),
+        help="Data date in YYYYMMDD format; defaults to today in Asia/Shanghai.",
+    )
+    parser.add_argument(
+        "--output",
+        type=Path,
+        help="CSV output path; defaults to outputs/hourly_cpm_<date>.csv.",
+    )
+    return parser.parse_args()
+
+
+def main() -> None:
+    args = parse_args()
+    if not re.fullmatch(r"\d{8}", args.date):
+        raise ValueError("--date must use YYYYMMDD format")
+
+    load_dotenv(ROOT.parent / "auto_put_ad_mini" / ".env", override=False)
+    load_dotenv(Path.cwd() / ".env", override=False)
+    frame = fetch_hourly_cpm(build_odps_client(), args.date)
+    if frame.empty:
+        print(f"No hourly CPM data found for {args.date}.")
+        return
+
+    output_path = args.output or ROOT / "outputs" / f"hourly_cpm_{args.date}.csv"
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+    frame.to_csv(output_path, index=False)
+
+    display = frame.copy()
+    display["hour"] = display["hour_of_day"].map(lambda value: f"{value:02d}:00")
+    display["cpm"] = display["cpm"].map(lambda value: f"{value:.2f}")
+    print(display[["hour", "impressions", "cpm"]].to_string(index=False))
+    print(f"hours={len(frame)} latest_partition={frame.iloc[-1]['dt']}")
+    print(f"output={output_path}")
+
+
+if __name__ == "__main__":
+    main()

examples/tencent_realtime_control/init_db.py

@@ -0,0 +1,18 @@
+#!/usr/bin/env python
+"""Initialize real-time control state tables."""
+
+from pathlib import Path
+
+from dotenv import load_dotenv
+
+from storage import initialize_schema
+
+
+ROOT = Path(__file__).resolve().parent
+
+
+if __name__ == "__main__":
+    load_dotenv(ROOT.parent / "auto_put_ad_mini" / ".env", override=False)
+    load_dotenv(Path.cwd() / ".env", override=False)
+    initialize_schema()
+    print("Real-time control schema initialized.")

examples/tencent_realtime_control/odps_source.py

@@ -0,0 +1,74 @@
+"""ODPS source for the overall hourly CPM signal."""
+
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+from datetime import date
+
+import pandas as pd
+from odps import ODPS
+
+
+@dataclass(frozen=True)
+class HourlyCpm:
+    partition: str
+    hour_of_day: int
+    impressions: int
+    cpm: float
+
+
+def build_odps_client() -> ODPS:
+    missing = [
+        key
+        for key in ("ODPS_ACCESS_ID", "ODPS_ACCESS_SECRET")
+        if not os.getenv(key)
+    ]
+    if missing:
+        raise RuntimeError(f"Missing ODPS environment variables: {', '.join(missing)}")
+    return ODPS(
+        os.environ["ODPS_ACCESS_ID"],
+        os.environ["ODPS_ACCESS_SECRET"],
+        os.getenv("ODPS_PROJECT", "loghubods"),
+        endpoint=os.getenv("ODPS_ENDPOINT", "http://service.odps.aliyun.com/api"),
+    )
+
+
+def fetch_hourly_cpm(client: ODPS, data_date: str) -> pd.DataFrame:
+    sql = f"""
+SELECT dt,
+       CAST(SUBSTR(dt, 9, 2) AS BIGINT) AS hour_of_day,
+       `曝光次数_总` AS impressions,
+       `真实cpm_总` AS cpm
+FROM loghubods.advertiser_data_da_hour
+WHERE dt LIKE '{data_date}%'
+  AND `名称` = 'SUM'
+  AND advertisercode = 'SUM'
+  AND company = 'SUM'
+  AND `行业` = 'SUM'
+  AND `客户` = 'SUM'
+  AND `落地页类型` = 'SUM'
+ORDER BY dt
+"""
+    instance = client.execute_sql(sql, hints={"odps.sql.submit.mode": "script"})
+    with instance.open_reader(tunnel=True) as reader:
+        frame = reader.to_pandas()
+    if frame.empty:
+        return frame
+    frame["hour_of_day"] = pd.to_numeric(frame["hour_of_day"]).astype("int64")
+    frame["impressions"] = pd.to_numeric(frame["impressions"]).astype("int64")
+    frame["cpm"] = pd.to_numeric(frame["cpm"]).astype(float)
+    return frame
+
+
+def fetch_latest_cpm(client: ODPS, data_date: date) -> HourlyCpm | None:
+    frame = fetch_hourly_cpm(client, data_date.strftime("%Y%m%d"))
+    if frame.empty:
+        return None
+    row = frame.iloc[-1]
+    return HourlyCpm(
+        partition=str(row["dt"]),
+        hour_of_day=int(row["hour_of_day"]),
+        impressions=int(row["impressions"]),
+        cpm=float(row["cpm"]),
+    )

examples/tencent_realtime_control/realtime_config.py

@@ -0,0 +1,55 @@
+"""Runtime configuration for Tencent real-time control."""
+
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+from decimal import Decimal
+
+
+@dataclass(frozen=True)
+class RealtimeControlConfig:
+    start_hour: int = 7
+    stop_hour: int = 21
+    poll_seconds: int = 300
+    high_cpm: Decimal = Decimal("250")
+    low_cpm: Decimal = Decimal("190")
+    bid_up_ratio: Decimal = Decimal("1.25")
+    inventory_refresh_minutes: int = 60
+    max_partition_lag_hours: int = 2
+    lock_name: str = "tencent_realtime_control"
+
+    @classmethod
+    def from_env(cls) -> "RealtimeControlConfig":
+        config = cls(
+            start_hour=int(os.getenv("RTC_START_HOUR", "7")),
+            stop_hour=int(os.getenv("RTC_STOP_HOUR", "21")),
+            poll_seconds=int(os.getenv("RTC_POLL_SECONDS", "300")),
+            high_cpm=Decimal(os.getenv("RTC_HIGH_CPM", "250")),
+            low_cpm=Decimal(os.getenv("RTC_LOW_CPM", "190")),
+            bid_up_ratio=Decimal(os.getenv("RTC_BID_UP_RATIO", "1.25")),
+            inventory_refresh_minutes=int(
+                os.getenv("RTC_INVENTORY_REFRESH_MINUTES", "60")
+            ),
+            max_partition_lag_hours=int(
+                os.getenv("RTC_MAX_PARTITION_LAG_HOURS", "2")
+            ),
+            lock_name=os.getenv(
+                "RTC_DB_LOCK_NAME", "tencent_realtime_control"
+            ).strip(),
+        )
+        if not 0 <= config.start_hour < config.stop_hour <= 23:
+            raise ValueError("RTC hours must satisfy 0 <= start < stop <= 23")
+        if config.poll_seconds < 60:
+            raise ValueError("RTC_POLL_SECONDS must be at least 60")
+        if config.low_cpm >= config.high_cpm:
+            raise ValueError("RTC_LOW_CPM must be lower than RTC_HIGH_CPM")
+        if config.bid_up_ratio <= 1:
+            raise ValueError("RTC_BID_UP_RATIO must be greater than 1")
+        if config.inventory_refresh_minutes < 5:
+            raise ValueError("RTC_INVENTORY_REFRESH_MINUTES must be at least 5")
+        if config.max_partition_lag_hours < 1:
+            raise ValueError("RTC_MAX_PARTITION_LAG_HOURS must be at least 1")
+        if not config.lock_name:
+            raise ValueError("RTC_DB_LOCK_NAME must not be empty")
+        return config

examples/tencent_realtime_control/run_once.py

@@ -0,0 +1,678 @@
+#!/usr/bin/env python
+"""Run one idempotent real-time CPM control cycle."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import logging
+import time
+import uuid
+from datetime import datetime, timedelta
+from decimal import Decimal, ROUND_HALF_UP
+from pathlib import Path
+from typing import Any
+from zoneinfo import ZoneInfo
+
+from dotenv import load_dotenv
+
+
+ROOT = Path(__file__).resolve().parent
+SHANGHAI = ZoneInfo("Asia/Shanghai")
+ACTIVE_STATUS = "AD_STATUS_NORMAL"
+SUSPEND_STATUS = "AD_STATUS_SUSPEND"
+
+DECISION_BOOST = "BOOST"
+DECISION_RESTORE = "RESTORE"
+DECISION_PAUSE = "PAUSE"
+DECISION_CUTOFF = "CUTOFF"
+DECISION_OFF_HOURS = "OFF_HOURS"
+DECISION_WAIT_DATA = "WAIT_DATA"
+
+logger = logging.getLogger("tencent_realtime_control")
+
+
+def load_environment() -> None:
+    load_dotenv(ROOT.parent / "auto_put_ad_mini" / ".env", override=False)
+    load_dotenv(Path.cwd() / ".env", override=False)
+
+
+def decide(cpm: Decimal, low_cpm: Decimal, high_cpm: Decimal) -> str:
+    if cpm > high_cpm:
+        return DECISION_BOOST
+    if cpm < low_cpm:
+        return DECISION_PAUSE
+    return DECISION_RESTORE
+
+
+def boosted_bid(base_bid_fen: int, ratio: Decimal) -> int:
+    return int(
+        (Decimal(base_bid_fen) * ratio).quantize(
+            Decimal("1"), rounding=ROUND_HALF_UP
+        )
+    )
+
+
+def partition_lag_hours(partition: str, now: datetime) -> float:
+    try:
+        partition_time = datetime.strptime(partition[:10], "%Y%m%d%H").replace(
+            tzinfo=SHANGHAI
+        )
+    except (TypeError, ValueError) as exc:
+        raise ValueError(f"Invalid CPM partition: {partition!r}") from exc
+    return (now - partition_time).total_seconds() / 3600
+
+
+def should_refresh_inventory(
+    daily_state: dict[str, Any],
+    *,
+    observed_partition: str,
+    decision: str,
+    now: datetime,
+    refresh_minutes: int,
+) -> bool:
+    if (
+        daily_state.get("last_observed_partition") != observed_partition
+        or daily_state.get("last_decision") != decision
+    ):
+        return True
+    refreshed_at = daily_state.get("last_inventory_refresh_at")
+    if not refreshed_at:
+        return True
+    if refreshed_at.tzinfo is None:
+        refreshed_at = refreshed_at.replace(tzinfo=SHANGHAI)
+    return now - refreshed_at >= timedelta(minutes=refresh_minutes)
+
+
+def target_for_ad(
+    *,
+    decision: str,
+    control_date: Any,
+    current_status: str,
+    current_bid: int,
+    base_bid: int,
+    boosted_date: Any,
+    paused_by_strategy: bool,
+    bid_up_ratio: Decimal,
+) -> dict[str, Any] | None:
+    manual_suspend = current_status != ACTIVE_STATUS and not paused_by_strategy
+
+    target_bid = base_bid
+    target_status: str | None = None
+    next_boosted_date = boosted_date
+    next_paused = False
+    pause_reason = None
+    limit_reached = False
+
+    if manual_suspend:
+        if decision not in (DECISION_PAUSE, DECISION_CUTOFF):
+            return None
+        return {
+            "target_bid_fen": base_bid,
+            "target_status": None,
+            "boosted_date": boosted_date,
+            "paused_by_strategy": False,
+            "pause_reason": None,
+            "boost_limit_reached": False,
+            "bid_change_needed": current_bid != base_bid,
+            "status_change_needed": False,
+        }
+
+    if decision == DECISION_BOOST:
+        if boosted_date == control_date:
+            limit_reached = True
+            raised_bid = boosted_bid(base_bid, bid_up_ratio)
+            if current_bid == raised_bid:
+                target_bid = raised_bid
+        else:
+            target_bid = boosted_bid(base_bid, bid_up_ratio)
+            next_boosted_date = control_date
+        if paused_by_strategy:
+            target_status = ACTIVE_STATUS
+    elif decision == DECISION_RESTORE:
+        if paused_by_strategy:
+            target_status = ACTIVE_STATUS
+    elif decision in (DECISION_PAUSE, DECISION_CUTOFF):
+        target_status = SUSPEND_STATUS if current_status != SUSPEND_STATUS else None
+        next_paused = True
+        pause_reason = "low_cpm" if decision == DECISION_PAUSE else "cutoff"
+    else:
+        raise ValueError(f"Unsupported decision: {decision}")
+
+    return {
+        "target_bid_fen": target_bid,
+        "target_status": target_status,
+        "boosted_date": next_boosted_date,
+        "paused_by_strategy": next_paused,
+        "pause_reason": pause_reason,
+        "boost_limit_reached": limit_reached,
+        "bid_change_needed": current_bid != target_bid,
+        "status_change_needed": target_status is not None
+        and target_status != current_status,
+    }
+
+
+def write_run_report(payload: dict[str, Any]) -> Path:
+    output_dir = ROOT / "outputs"
+    output_dir.mkdir(parents=True, exist_ok=True)
+    timestamp = datetime.now(SHANGHAI).strftime("%Y%m%d_%H%M%S")
+    mode = "apply" if payload["apply"] else "dry_run"
+    path = output_dir / f"control_{mode}_{timestamp}.json"
+    path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
+    return path
+
+
+def execute_inventory_action(
+    *,
+    run_id: str,
+    now: datetime,
+    decision: str,
+    observed_partition: str | None,
+    observed_cpm: Decimal | None,
+    apply: bool,
+    accounts: list[dict[str, Any]],
+    config: Any,
+    tencent: Any,
+    test_notification: bool = False,
+) -> dict[str, Any]:
+    from storage import insert_action_log, load_ad_states, upsert_ad_state
+    from tencent_client import current_bid_fen, resolve_bid_field
+
+    results: list[dict[str, Any]] = []
+    failures = 0
+    api_updates = 0
+
+    for account in accounts:
+        account_id = int(account["account_id"])
+        states = load_ad_states(account_id)
+        try:
+            ads = tencent.get_ads(account_id)
+        except Exception as exc:
+            failures += 1
+            results.append(
+                {
+                    "account_id": account_id,
+                    "status": "account_fetch_failed",
+                    "error": str(exc),
+                }
+            )
+            continue
+
+        for ad in ads:
+            adgroup_id = int(ad.get("adgroup_id") or 0)
+            current_status = str(ad.get("configured_status") or "")
+            state = states.get(adgroup_id)
+            paused_by_strategy = bool(
+                (state or {}).get("paused_by_strategy")
+            )
+            if (
+                current_status != ACTIVE_STATUS
+                and not paused_by_strategy
+                and (
+                    state is None
+                    or decision not in (DECISION_PAUSE, DECISION_CUTOFF)
+                )
+            ):
+                continue
+
+            bid_field = (
+                str(state["bid_field"])
+                if state
+                else resolve_bid_field(ad, account.get("bid_scene"))
+            )
+            current_bid = current_bid_fen(ad, bid_field)
+            if current_bid is None or current_bid <= 0:
+                failures += 1
+                results.append(
+                    {
+                        "account_id": account_id,
+                        "adgroup_id": adgroup_id,
+                        "status": "invalid_current_bid",
+                        "bid_field": bid_field,
+                    }
+                )
+                continue
+
+            base_bid = int(state["base_bid_fen"]) if state else current_bid
+            plan = target_for_ad(
+                decision=decision,
+                control_date=now.date(),
+                current_status=current_status,
+                current_bid=current_bid,
+                base_bid=base_bid,
+                boosted_date=(state or {}).get("boosted_date"),
+                paused_by_strategy=paused_by_strategy,
+                bid_up_ratio=config.bid_up_ratio,
+            )
+            if plan is None:
+                continue
+
+            needs_api = plan["bid_change_needed"] or plan["status_change_needed"]
+            execution_status = "planned" if needs_api else "noop"
+            error = None
+            if apply and state is None:
+                upsert_ad_state(
+                    account_id=account_id,
+                    adgroup_id=adgroup_id,
+                    adgroup_name=str(ad.get("adgroup_name") or ""),
+                    bid_field=bid_field,
+                    base_bid_fen=base_bid,
+                    boosted_date=None,
+                    paused_by_strategy=decision
+                    in (DECISION_PAUSE, DECISION_CUTOFF),
+                    pause_reason=plan["pause_reason"],
+                    last_action="REGISTER_BASE",
+                    action_at=now,
+                )
+            elif (
+                apply
+                and decision in (DECISION_PAUSE, DECISION_CUTOFF)
+                and plan["paused_by_strategy"]
+            ):
+                upsert_ad_state(
+                    account_id=account_id,
+                    adgroup_id=adgroup_id,
+                    adgroup_name=str(ad.get("adgroup_name") or ""),
+                    bid_field=bid_field,
+                    base_bid_fen=base_bid,
+                    boosted_date=(state or {}).get("boosted_date"),
+                    paused_by_strategy=True,
+                    pause_reason=plan["pause_reason"],
+                    last_action=f"{decision}_PENDING",
+                    action_at=now,
+                )
+            if apply and needs_api:
+                try:
+                    tencent.update_ad(
+                        account_id,
+                        adgroup_id,
+                        bid_field=bid_field
+                        if plan["bid_change_needed"]
+                        else None,
+                        target_bid_fen=plan["target_bid_fen"]
+                        if plan["bid_change_needed"]
+                        else None,
+                        target_status=plan["target_status"]
+                        if plan["status_change_needed"]
+                        else None,
+                    )
+                    execution_status = "success"
+                    api_updates += 1
+                    time.sleep(0.15)
+                except Exception as exc:
+                    execution_status = "failed"
+                    error = str(exc)
+                    failures += 1
+
+            result = {
+                "account_id": account_id,
+                "audience_name": account.get("audience_name"),
+                "adgroup_id": adgroup_id,
+                "adgroup_name": ad.get("adgroup_name"),
+                "decision": decision,
+                "bid_field": bid_field,
+                "base_bid_fen": base_bid,
+                "before_bid_fen": current_bid,
+                "target_bid_fen": plan["target_bid_fen"],
+                "before_status": current_status,
+                "target_status": plan["target_status"],
+                "boost_limit_reached": plan["boost_limit_reached"],
+                "bid_change_needed": plan["bid_change_needed"],
+                "status_change_needed": plan["status_change_needed"],
+                "status": execution_status,
+                "error": error,
+            }
+            results.append(result)
+
+            if apply and execution_status != "failed":
+                upsert_ad_state(
+                    account_id=account_id,
+                    adgroup_id=adgroup_id,
+                    adgroup_name=str(ad.get("adgroup_name") or ""),
+                    bid_field=bid_field,
+                    base_bid_fen=base_bid,
+                    boosted_date=plan["boosted_date"],
+                    paused_by_strategy=plan["paused_by_strategy"],
+                    pause_reason=plan["pause_reason"],
+                    last_action=decision,
+                    action_at=now,
+                )
+            if apply and (needs_api or state is None or execution_status == "failed"):
+                insert_action_log(
+                    {
+                        "run_id": run_id,
+                        "control_date": now.date(),
+                        "observed_partition": observed_partition,
+                        "observed_cpm": observed_cpm,
+                        "decision": decision,
+                        "account_id": account_id,
+                        "adgroup_id": adgroup_id,
+                        "adgroup_name": ad.get("adgroup_name"),
+                        "bid_field": bid_field,
+                        "base_bid_fen": base_bid,
+                        "before_bid_fen": current_bid,
+                        "target_bid_fen": plan["target_bid_fen"],
+                        "before_status": current_status,
+                        "target_status": plan["target_status"],
+                        "apply_mode": True,
+                        "execution_status": execution_status,
+                        "error_message": error,
+                    }
+                )
+
+    notification = {"status": "dry_run" if not apply else "no_actions"}
+    if apply or test_notification:
+        from feishu_notifier import send_action_notification
+
+        successful_actions: dict[int, list[dict[str, Any]]] = {}
+        expected_status = "success" if apply else "planned"
+        for result in results:
+            if result.get("status") != expected_status:
+                continue
+            if not (
+                result.get("bid_change_needed")
+                or (
+                    result.get("decision") in (DECISION_PAUSE, DECISION_CUTOFF)
+                    and result.get("status_change_needed")
+                )
+            ):
+                continue
+            successful_actions.setdefault(int(result["account_id"]), []).append(
+                result
+            )
+
+        for account_id, account_results in successful_actions.items():
+            try:
+                metrics = tencent.get_today_ad_metrics(
+                    account_id,
+                    [int(row["adgroup_id"]) for row in account_results],
+                    now.date(),
+                )
+                for result in account_results:
+                    ad_metrics = metrics.get(int(result["adgroup_id"]), {})
+                    result["today_cost_fen"] = int(
+                        ad_metrics.get("cost_fen") or 0
+                    )
+                    result["today_impressions"] = int(
+                        ad_metrics.get("impressions") or 0
+                    )
+                    result["today_clicks"] = int(
+                        ad_metrics.get("clicks") or 0
+                    )
+            except Exception as exc:
+                logger.warning(
+                    "Failed to fetch today's metrics account=%s: %s",
+                    account_id,
+                    exc,
+                )
+
+        notification = send_action_notification(
+            run_id=run_id,
+            now=now,
+            results=results,
+            bid_up_ratio=config.bid_up_ratio,
+            decision=decision,
+            observed_partition=observed_partition,
+            observed_cpm=observed_cpm,
+            test_mode=test_notification,
+        )
+
+    return {
+        "failures": failures,
+        "api_updates": api_updates,
+        "results": results,
+        "notification": notification,
+    }
+
+
+def run_cycle(
+    *,
+    now: datetime,
+    apply: bool,
+    cpm_override: Decimal | None = None,
+    force_refresh: bool = False,
+    test_notification: bool = False,
+) -> dict[str, Any]:
+    from odps_source import HourlyCpm, build_odps_client, fetch_latest_cpm
+    from realtime_config import RealtimeControlConfig
+    from storage import (
+        advisory_lock,
+        load_daily_state,
+        load_enabled_accounts,
+        save_daily_state,
+    )
+    from tencent_client import TencentClient
+
+    config = RealtimeControlConfig.from_env()
+    run_id = str(uuid.uuid4())
+    payload: dict[str, Any] = {
+        "run_id": run_id,
+        "now": now.isoformat(),
+        "apply": apply,
+        "decision": None,
+        "observed_partition": None,
+        "observed_cpm": None,
+        "summary": {},
+    }
+
+    with advisory_lock(config.lock_name) as acquired:
+        if not acquired:
+            payload["decision"] = "LOCKED_BY_OTHER_WORKER"
+            return payload
+
+        daily_state = load_daily_state(now.date())
+        accounts = load_enabled_accounts()
+        if now.hour < config.start_hour:
+            payload["decision"] = DECISION_OFF_HOURS
+            return payload
+
+        if now.hour >= config.stop_hour:
+            payload["decision"] = DECISION_CUTOFF
+            if apply and bool(daily_state.get("cutoff_done")) and not force_refresh:
+                payload["summary"] = {"status": "cutoff_already_done"}
+                return payload
+            execution = execute_inventory_action(
+                run_id=run_id,
+                now=now,
+                decision=DECISION_CUTOFF,
+                observed_partition=None,
+                observed_cpm=None,
+                apply=apply,
+                accounts=accounts,
+                config=config,
+                tencent=TencentClient(),
+                test_notification=test_notification,
+            )
+            payload["summary"] = execution
+            if apply and not execution["failures"]:
+                save_daily_state(
+                    now.date(),
+                    cutoff_done=True,
+                    last_decision=DECISION_CUTOFF,
+                    last_inventory_refresh_at=now,
+                    last_evaluated_at=now,
+                )
+            return payload
+
+        if apply and not bool(daily_state.get("morning_recovery_done")):
+            execution = execute_inventory_action(
+                run_id=run_id,
+                now=now,
+                decision=DECISION_RESTORE,
+                observed_partition=None,
+                observed_cpm=None,
+                apply=True,
+                accounts=accounts,
+                config=config,
+                tencent=TencentClient(),
+                test_notification=test_notification,
+            )
+            payload["decision"] = "MORNING_RECOVERY"
+            payload["summary"] = execution
+            if not execution["failures"]:
+                save_daily_state(
+                    now.date(),
+                    morning_recovery_done=True,
+                    last_inventory_refresh_at=now,
+                    last_evaluated_at=now,
+                )
+            return payload
+
+        if cpm_override is not None:
+            signal = HourlyCpm(
+                partition=f"{now.strftime('%Y%m%d%H')}_override",
+                hour_of_day=now.hour,
+                impressions=0,
+                cpm=float(cpm_override),
+            )
+        else:
+            signal = fetch_latest_cpm(build_odps_client(), now.date())
+        earliest_signal_hour = max(0, config.start_hour - 1)
+        if signal is None or signal.hour_of_day < earliest_signal_hour:
+            payload["decision"] = DECISION_WAIT_DATA
+            payload["summary"] = {
+                "reason": (
+                    "No same-day CPM partition at or after "
+                    f"{earliest_signal_hour:02d}:00"
+                )
+            }
+            return payload
+
+        observed_cpm = Decimal(str(signal.cpm))
+        lag_hours = partition_lag_hours(signal.partition, now)
+        payload.update(
+            {
+                "observed_partition": signal.partition,
+                "observed_cpm": float(observed_cpm),
+                "impressions": signal.impressions,
+                "partition_lag_hours": round(lag_hours, 4),
+            }
+        )
+        if lag_hours < 0 or lag_hours > config.max_partition_lag_hours:
+            payload["decision"] = DECISION_WAIT_DATA
+            payload["summary"] = {
+                "reason": (
+                    f"CPM partition lag {lag_hours:.2f}h exceeds allowed "
+                    f"{config.max_partition_lag_hours}h"
+                )
+            }
+            return payload
+
+        decision = decide(observed_cpm, config.low_cpm, config.high_cpm)
+        payload.update(
+            {
+                "decision": decision,
+            }
+        )
+
+        refresh = force_refresh or should_refresh_inventory(
+            daily_state,
+            observed_partition=signal.partition,
+            decision=decision,
+            now=now,
+            refresh_minutes=config.inventory_refresh_minutes,
+        )
+        if apply and not refresh:
+            save_daily_state(now.date(), last_evaluated_at=now)
+            payload["summary"] = {"status": "same_signal_noop"}
+            return payload
+
+        execution = execute_inventory_action(
+            run_id=run_id,
+            now=now,
+            decision=decision,
+            observed_partition=signal.partition,
+            observed_cpm=observed_cpm,
+            apply=apply,
+            accounts=accounts,
+            config=config,
+            tencent=TencentClient(),
+            test_notification=test_notification,
+        )
+        payload["summary"] = execution
+        if apply and not execution["failures"]:
+            save_daily_state(
+                now.date(),
+                last_observed_partition=signal.partition,
+                last_observed_cpm=observed_cpm,
+                last_decision=decision,
+                last_inventory_refresh_at=now,
+                last_evaluated_at=now,
+            )
+        return payload
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--apply",
+        action="store_true",
+        help="Execute Tencent updates. Default is dry-run.",
+    )
+    parser.add_argument(
+        "--at",
+        help="Dry-run only: simulate Shanghai local time, ISO format.",
+    )
+    parser.add_argument(
+        "--cpm-override",
+        type=Decimal,
+        help="Dry-run only: override latest CPM to validate decision branches.",
+    )
+    parser.add_argument("--force-refresh", action="store_true")
+    parser.add_argument(
+        "--test-notification",
+        action="store_true",
+        help=(
+            "Dry-run only: use real inventory and metrics, then send a clearly "
+            "marked simulated Feishu action sheet."
+        ),
+    )
+    return parser.parse_args()
+
+
+def main() -> int:
+    load_environment()
+    args = parse_args()
+    if args.apply and (args.at or args.cpm_override is not None):
+        raise ValueError("--at and --cpm-override are forbidden with --apply")
+    if args.apply and args.test_notification:
+        raise ValueError("--test-notification is forbidden with --apply")
+    now = (
+        datetime.fromisoformat(args.at).replace(tzinfo=SHANGHAI)
+        if args.at
+        else datetime.now(SHANGHAI)
+    )
+    payload = run_cycle(
+        now=now,
+        apply=args.apply,
+        cpm_override=args.cpm_override,
+        force_refresh=args.force_refresh,
+        test_notification=args.test_notification,
+    )
+    report = write_run_report(payload)
+    compact = {
+        "decision": payload["decision"],
+        "observed_partition": payload.get("observed_partition"),
+        "observed_cpm": payload.get("observed_cpm"),
+        "apply": payload["apply"],
+        "api_updates": payload.get("summary", {}).get("api_updates", 0),
+        "failures": payload.get("summary", {}).get("failures", 0),
+        "notification": payload.get("summary", {}).get("notification", {}).get(
+            "status"
+        ),
+        "report": str(report),
+    }
+    print(json.dumps(compact, ensure_ascii=False))
+    return 1 if compact["failures"] else 0
+
+
+if __name__ == "__main__":
+    logging.basicConfig(
+        level=logging.INFO,
+        format="%(asctime)s %(levelname)s %(name)s %(message)s",
+    )
+    try:
+        raise SystemExit(main())
+    except Exception as exc:
+        logger.exception("Real-time control failed: %s", exc)
+        raise SystemExit(1)

examples/tencent_realtime_control/run_scheduler.py

@@ -0,0 +1,107 @@
+#!/usr/bin/env python
+"""Run real-time control every five minutes during delivery hours."""
+
+from __future__ import annotations
+
+import argparse
+import logging
+import signal
+import time
+from datetime import datetime, timedelta
+from zoneinfo import ZoneInfo
+
+from realtime_config import RealtimeControlConfig
+from run_once import load_environment, run_cycle, write_run_report
+
+
+SHANGHAI = ZoneInfo("Asia/Shanghai")
+logger = logging.getLogger("tencent_realtime_control.scheduler")
+stopping = False
+
+
+def request_stop(signum: int, _frame: object) -> None:
+    global stopping
+    logger.info("Received signal %s; scheduler will stop.", signum)
+    stopping = True
+
+
+def next_wake(now: datetime, config: RealtimeControlConfig) -> datetime:
+    if now.hour >= config.stop_hour:
+        tomorrow = now.date() + timedelta(days=1)
+        return datetime.combine(
+            tomorrow,
+            datetime.min.time().replace(hour=config.start_hour),
+            SHANGHAI,
+        )
+    if now.hour < config.start_hour:
+        return datetime.combine(
+            now.date(),
+            datetime.min.time().replace(hour=config.start_hour),
+            SHANGHAI,
+        )
+    seconds = int(now.timestamp())
+    next_boundary = (
+        seconds // config.poll_seconds + 1
+    ) * config.poll_seconds
+    return datetime.fromtimestamp(next_boundary, SHANGHAI)
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--apply",
+        action="store_true",
+        help="Execute Tencent updates. Default scheduler is dry-run.",
+    )
+    return parser.parse_args()
+
+
+def main() -> None:
+    load_environment()
+    args = parse_args()
+    config = RealtimeControlConfig.from_env()
+    signal.signal(signal.SIGINT, request_stop)
+    signal.signal(signal.SIGTERM, request_stop)
+
+    logger.info(
+        "Scheduler started apply=%s hours=%02d:00-%02d:00 interval=%ss",
+        args.apply,
+        config.start_hour,
+        config.stop_hour,
+        config.poll_seconds,
+    )
+    while not stopping:
+        try:
+            payload = run_cycle(
+                now=datetime.now(SHANGHAI),
+                apply=args.apply,
+            )
+            report = write_run_report(payload)
+            logger.info(
+                "Cycle decision=%s cpm=%s updates=%s failures=%s notification=%s report=%s",
+                payload.get("decision"),
+                payload.get("observed_cpm"),
+                (payload.get("summary") or {}).get("api_updates", 0),
+                (payload.get("summary") or {}).get("failures", 0),
+                ((payload.get("summary") or {}).get("notification") or {}).get(
+                    "status"
+                ),
+                report,
+            )
+        except Exception:
+            logger.exception("Real-time control cycle failed")
+
+        wake_at = next_wake(datetime.now(SHANGHAI), config)
+        while not stopping:
+            remaining = (wake_at - datetime.now(SHANGHAI)).total_seconds()
+            if remaining <= 0:
+                break
+            time.sleep(min(remaining, 5))
+
+
+if __name__ == "__main__":
+    logging.basicConfig(
+        level=logging.INFO,
+        format="%(asctime)s %(levelname)s %(name)s %(message)s",
+    )
+    main()

examples/tencent_realtime_control/schema.sql

@@ -0,0 +1,58 @@
+CREATE TABLE IF NOT EXISTS realtime_control_daily_state (
+    control_date DATE NOT NULL PRIMARY KEY,
+    morning_recovery_done BOOLEAN NOT NULL DEFAULT FALSE,
+    cutoff_done BOOLEAN NOT NULL DEFAULT FALSE,
+    last_observed_partition VARCHAR(20) DEFAULT NULL,
+    last_observed_cpm DECIMAL(12, 4) DEFAULT NULL,
+    last_decision VARCHAR(32) DEFAULT NULL,
+    last_inventory_refresh_at DATETIME DEFAULT NULL,
+    last_evaluated_at DATETIME DEFAULT NULL,
+    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='实时CPM调控每日状态';
+
+CREATE TABLE IF NOT EXISTS realtime_control_ad_state (
+    account_id BIGINT NOT NULL,
+    adgroup_id BIGINT NOT NULL,
+    adgroup_name VARCHAR(255) DEFAULT NULL,
+    bid_field VARCHAR(32) NOT NULL,
+    base_bid_fen INT NOT NULL,
+    boosted_date DATE DEFAULT NULL,
+    paused_by_strategy BOOLEAN NOT NULL DEFAULT FALSE,
+    pause_reason VARCHAR(32) DEFAULT NULL,
+    last_action VARCHAR(32) DEFAULT NULL,
+    last_action_at DATETIME DEFAULT NULL,
+    last_seen_at DATETIME DEFAULT NULL,
+    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+    PRIMARY KEY (account_id, adgroup_id),
+    KEY idx_strategy_pause (paused_by_strategy),
+    KEY idx_boosted_date (boosted_date),
+    KEY idx_last_seen (last_seen_at)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='实时CPM调控广告基准与状态';
+
+CREATE TABLE IF NOT EXISTS realtime_control_action_log (
+    id BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY,
+    run_id VARCHAR(36) NOT NULL,
+    control_date DATE NOT NULL,
+    observed_partition VARCHAR(20) DEFAULT NULL,
+    observed_cpm DECIMAL(12, 4) DEFAULT NULL,
+    decision VARCHAR(32) NOT NULL,
+    account_id BIGINT NOT NULL,
+    adgroup_id BIGINT NOT NULL,
+    adgroup_name VARCHAR(255) DEFAULT NULL,
+    bid_field VARCHAR(32) 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,
+    apply_mode BOOLEAN NOT NULL DEFAULT FALSE,
+    execution_status VARCHAR(32) NOT NULL,
+    error_message TEXT DEFAULT NULL,
+    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    KEY idx_run_id (run_id),
+    KEY idx_control_date (control_date),
+    KEY idx_adgroup (account_id, adgroup_id),
+    KEY idx_execution_status (execution_status)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='实时CPM调控动作审计';

examples/tencent_realtime_control/storage.py

@@ -0,0 +1,225 @@
+"""MySQL state and audit storage for real-time control."""
+
+from __future__ import annotations
+
+import os
+from contextlib import contextmanager
+from datetime import date, datetime
+from pathlib import Path
+from typing import Any, Iterator
+
+import pymysql
+import pymysql.cursors
+
+
+ROOT = Path(__file__).resolve().parent
+
+
+def connect() -> pymysql.Connection:
+    required = ["DB_HOST", "DB_USER", "DB_NAME"]
+    missing = [key for key in required if not os.getenv(key)]
+    if missing:
+        raise RuntimeError(f"Missing database environment variables: {', '.join(missing)}")
+    return pymysql.connect(
+        host=os.environ["DB_HOST"],
+        port=int(os.getenv("DB_PORT", "3306")),
+        user=os.environ["DB_USER"],
+        password=os.getenv("DB_PASSWORD", ""),
+        database=os.environ["DB_NAME"],
+        charset="utf8mb4",
+        cursorclass=pymysql.cursors.DictCursor,
+        autocommit=True,
+        connect_timeout=int(os.getenv("DB_CONNECT_TIMEOUT", "10")),
+        read_timeout=int(os.getenv("DB_READ_TIMEOUT", "60")),
+        write_timeout=int(os.getenv("DB_WRITE_TIMEOUT", "60")),
+    )
+
+
+def initialize_schema() -> None:
+    statements = [
+        statement.strip()
+        for statement in (ROOT / "schema.sql").read_text(encoding="utf-8").split(";")
+        if statement.strip()
+    ]
+    connection = connect()
+    try:
+        with connection.cursor() as cursor:
+            for statement in statements:
+                cursor.execute(statement)
+    finally:
+        connection.close()
+
+
+@contextmanager
+def advisory_lock(lock_name: str) -> Iterator[bool]:
+    connection = connect()
+    acquired = False
+    try:
+        with connection.cursor() as cursor:
+            cursor.execute("SELECT GET_LOCK(%s, 0) AS acquired", (lock_name,))
+            acquired = bool((cursor.fetchone() or {}).get("acquired"))
+        yield acquired
+    finally:
+        if acquired:
+            try:
+                with connection.cursor() as cursor:
+                    cursor.execute("SELECT RELEASE_LOCK(%s)", (lock_name,))
+            except Exception:
+                pass
+        connection.close()
+
+
+def load_enabled_accounts() -> list[dict[str, Any]]:
+    connection = connect()
+    try:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                """
+                SELECT c.account_id, c.audience_name, c.bid_scene
+                FROM ad_creation_account_config c
+                JOIN account_whitelist w ON w.account_id = c.account_id
+                WHERE c.enabled = TRUE
+                  AND w.enabled = TRUE
+                ORDER BY c.account_id
+                """
+            )
+            return list(cursor.fetchall())
+    finally:
+        connection.close()
+
+
+def load_daily_state(control_date: date) -> dict[str, Any]:
+    connection = connect()
+    try:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                "SELECT * FROM realtime_control_daily_state WHERE control_date=%s",
+                (control_date,),
+            )
+            return cursor.fetchone() or {}
+    finally:
+        connection.close()
+
+
+def save_daily_state(control_date: date, **values: Any) -> None:
+    allowed = {
+        "morning_recovery_done",
+        "cutoff_done",
+        "last_observed_partition",
+        "last_observed_cpm",
+        "last_decision",
+        "last_inventory_refresh_at",
+        "last_evaluated_at",
+    }
+    unknown = set(values) - allowed
+    if unknown:
+        raise ValueError(f"Unsupported daily-state fields: {sorted(unknown)}")
+    columns = ["control_date", *values]
+    params = [control_date, *values.values()]
+    updates = ", ".join(f"{column}=VALUES({column})" for column in values)
+    placeholders = ", ".join(["%s"] * len(columns))
+    sql = (
+        f"INSERT INTO realtime_control_daily_state ({', '.join(columns)}) "
+        f"VALUES ({placeholders}) ON DUPLICATE KEY UPDATE {updates}"
+    )
+    connection = connect()
+    try:
+        with connection.cursor() as cursor:
+            cursor.execute(sql, params)
+    finally:
+        connection.close()
+
+
+def load_ad_states(account_id: int) -> dict[int, dict[str, Any]]:
+    connection = connect()
+    try:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                "SELECT * FROM realtime_control_ad_state WHERE account_id=%s",
+                (account_id,),
+            )
+            return {int(row["adgroup_id"]): row for row in cursor.fetchall()}
+    finally:
+        connection.close()
+
+
+def upsert_ad_state(
+    *,
+    account_id: int,
+    adgroup_id: int,
+    adgroup_name: str,
+    bid_field: str,
+    base_bid_fen: int,
+    boosted_date: date | None,
+    paused_by_strategy: bool,
+    pause_reason: str | None,
+    last_action: str,
+    action_at: datetime,
+) -> None:
+    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,
+                     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)
+                ON DUPLICATE KEY UPDATE
+                    adgroup_name=VALUES(adgroup_name),
+                    bid_field=VALUES(bid_field),
+                    boosted_date=VALUES(boosted_date),
+                    paused_by_strategy=VALUES(paused_by_strategy),
+                    pause_reason=VALUES(pause_reason),
+                    last_action=VALUES(last_action),
+                    last_action_at=VALUES(last_action_at),
+                    last_seen_at=VALUES(last_seen_at)
+                """,
+                (
+                    account_id,
+                    adgroup_id,
+                    adgroup_name,
+                    bid_field,
+                    base_bid_fen,
+                    boosted_date,
+                    paused_by_strategy,
+                    pause_reason,
+                    last_action,
+                    action_at,
+                    action_at,
+                ),
+            )
+    finally:
+        connection.close()
+
+def insert_action_log(record: dict[str, Any]) -> None:
+    columns = [
+        "run_id",
+        "control_date",
+        "observed_partition",
+        "observed_cpm",
+        "decision",
+        "account_id",
+        "adgroup_id",
+        "adgroup_name",
+        "bid_field",
+        "base_bid_fen",
+        "before_bid_fen",
+        "target_bid_fen",
+        "before_status",
+        "target_status",
+        "apply_mode",
+        "execution_status",
+        "error_message",
+    ]
+    connection = connect()
+    try:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                f"INSERT INTO realtime_control_action_log ({', '.join(columns)}) "
+                f"VALUES ({', '.join(['%s'] * len(columns))})",
+                [record.get(column) for column in columns],
+            )
+    finally:
+        connection.close()

examples/tencent_realtime_control/tencent_client.py

@@ -0,0 +1,333 @@
+"""Minimal Tencent Ads read/write client for real-time control."""
+
+from __future__ import annotations
+
+import json
+import os
+import time
+import uuid
+from datetime import date
+from typing import Any
+
+import requests
+
+from storage import connect
+
+
+ACTIVE_STATUS = "AD_STATUS_NORMAL"
+SUSPEND_STATUS = "AD_STATUS_SUSPEND"
+AD_FIELDS = [
+    "adgroup_id",
+    "adgroup_name",
+    "configured_status",
+    "system_status",
+    "bid_amount",
+    "custom_cost_cap",
+    "smart_bid_type",
+    "cost_constraint_scene",
+]
+
+
+class TencentClient:
+    def __init__(self) -> None:
+        self.base_url = os.getenv(
+            "TENCENT_AD_BASE_URL", "https://api.e.qq.com/v3.0"
+        ).rstrip("/")
+        self.access_token_api = os.getenv(
+            "TENCENT_AD_TOKEN_API",
+            "https://api.piaoquantv.com/ad/put/tencent/getAccessToken",
+        )
+        self.user_token_api = os.getenv(
+            "TENCENT_AD_USER_TOKEN_API",
+            "https://api.piaoquantv.com/ad/put/tencent/getUserToken",
+        )
+        self.timeout = int(os.getenv("TENCENT_AD_TIMEOUT_SECONDS", "30"))
+        self.verify_attempts = int(os.getenv("RTC_VERIFY_ATTEMPTS", "3"))
+        self.verify_delay_seconds = float(
+            os.getenv("RTC_VERIFY_DELAY_SECONDS", "1")
+        )
+        if self.verify_attempts < 1:
+            raise ValueError("RTC_VERIFY_ATTEMPTS must be at least 1")
+        if self.verify_delay_seconds < 0:
+            raise ValueError("RTC_VERIFY_DELAY_SECONDS must not be negative")
+        self._access_tokens: dict[int, str] = {}
+        self._user_tokens: dict[int, str] = {}
+        self.session = requests.Session()
+
+    def _access_token(self, account_id: int) -> str:
+        if account_id not in self._access_tokens:
+            response = self.session.get(
+                self.access_token_api,
+                params={"accountId": account_id},
+                timeout=15,
+            )
+            response.raise_for_status()
+            token = response.text.strip()
+            if len(token) <= 10:
+                raise RuntimeError(f"Invalid access token for account={account_id}")
+            self._access_tokens[account_id] = token
+        return self._access_tokens[account_id]
+
+    def _user_token(self, account_id: int) -> str:
+        if account_id in self._user_tokens:
+            return self._user_tokens[account_id]
+        token = ""
+        try:
+            response = self.session.get(
+                self.user_token_api,
+                params={"accountId": account_id},
+                timeout=15,
+            )
+            response.raise_for_status()
+            candidate = response.text.strip()
+            if len(candidate) > 10:
+                token = candidate
+        except Exception:
+            pass
+        if not token:
+            connection = connect()
+            try:
+                with connection.cursor() as cursor:
+                    cursor.execute(
+                        "SELECT user_token FROM account_whitelist WHERE account_id=%s",
+                        (account_id,),
+                    )
+                    row = cursor.fetchone()
+                    token = str((row or {}).get("user_token") or "")
+            finally:
+                connection.close()
+        if not token:
+            token = os.getenv("TENCENT_AD_USER_TOKEN", "").strip()
+        if not token:
+            raise RuntimeError(f"No user_token available for account={account_id}")
+        self._user_tokens[account_id] = token
+        return token
+
+    def _common_params(self, account_id: int) -> dict[str, Any]:
+        return {
+            "access_token": self._access_token(account_id),
+            "timestamp": int(time.time()),
+            "nonce": uuid.uuid4().hex,
+        }
+
+    @staticmethod
+    def _check(payload: dict[str, Any], operation: str) -> dict[str, Any]:
+        if payload.get("code") != 0:
+            message = payload.get("message_cn") or payload.get("message") or "unknown"
+            raise RuntimeError(
+                f"{operation} failed: code={payload.get('code')} message={message}"
+            )
+        return payload.get("data") or {}
+
+    def get_ads(self, account_id: int) -> list[dict[str, Any]]:
+        ads: list[dict[str, Any]] = []
+        page = 1
+        while True:
+            params = {
+                **self._common_params(account_id),
+                "account_id": account_id,
+                "fields": json.dumps(AD_FIELDS, ensure_ascii=False),
+                "page": page,
+                "page_size": 100,
+            }
+            response = self.session.get(
+                f"{self.base_url}/adgroups/get",
+                params=params,
+                timeout=self.timeout,
+            )
+            response.raise_for_status()
+            data = self._check(response.json(), "get_ads")
+            rows = data.get("list") or []
+            ads.extend(rows)
+            page_info = data.get("page_info") or {}
+            if page >= int(page_info.get("total_page") or 1):
+                break
+            page += 1
+        return ads
+
+    def get_ad(self, account_id: int, adgroup_id: int) -> dict[str, Any]:
+        params = {
+            **self._common_params(account_id),
+            "account_id": account_id,
+            "fields": json.dumps(AD_FIELDS, ensure_ascii=False),
+            "filtering": json.dumps(
+                [
+                    {
+                        "field": "adgroup_id",
+                        "operator": "IN",
+                        "values": [str(adgroup_id)],
+                    }
+                ]
+            ),
+            "page": 1,
+            "page_size": 10,
+        }
+        response = self.session.get(
+            f"{self.base_url}/adgroups/get",
+            params=params,
+            timeout=self.timeout,
+        )
+        response.raise_for_status()
+        data = self._check(response.json(), "get_ad")
+        rows = data.get("list") or []
+        for row in rows:
+            if int(row.get("adgroup_id") or 0) == adgroup_id:
+                return row
+        raise RuntimeError(
+            f"Ad not found after update: account={account_id} adgroup={adgroup_id}"
+        )
+
+    def update_ad(
+        self,
+        account_id: int,
+        adgroup_id: int,
+        *,
+        bid_field: str | None = None,
+        target_bid_fen: int | None = None,
+        target_status: str | None = None,
+    ) -> dict[str, Any]:
+        body: dict[str, Any] = {
+            "account_id": account_id,
+            "adgroup_id": adgroup_id,
+        }
+        if bid_field and target_bid_fen is not None:
+            body[bid_field] = target_bid_fen
+        if target_status:
+            body["configured_status"] = target_status
+        params = {
+            **self._common_params(account_id),
+            "user_token": self._user_token(account_id),
+        }
+        response = self.session.post(
+            f"{self.base_url}/adgroups/update",
+            params=params,
+            json=body,
+            timeout=self.timeout,
+        )
+        response.raise_for_status()
+        self._check(response.json(), "update_ad")
+
+        expected: dict[str, Any] = {}
+        if bid_field and target_bid_fen is not None:
+            expected[bid_field] = target_bid_fen
+        if target_status:
+            expected["configured_status"] = target_status
+
+        last_actual: dict[str, Any] = {}
+        for attempt in range(1, self.verify_attempts + 1):
+            ad = self.get_ad(account_id, adgroup_id)
+            last_actual = {field: ad.get(field) for field in expected}
+            matches = True
+            for field, target in expected.items():
+                actual = ad.get(field)
+                if field in {"bid_amount", "custom_cost_cap"}:
+                    try:
+                        actual = int(actual)
+                    except (TypeError, ValueError):
+                        matches = False
+                        break
+                if actual != target:
+                    matches = False
+                    break
+            if matches:
+                return ad
+            if attempt < self.verify_attempts:
+                time.sleep(self.verify_delay_seconds)
+
+        raise RuntimeError(
+            "Post-write verification failed: "
+            f"account={account_id} adgroup={adgroup_id} "
+            f"expected={expected} actual={last_actual}"
+        )
+
+    def get_today_ad_metrics(
+        self,
+        account_id: int,
+        adgroup_ids: list[int],
+        data_date: date,
+    ) -> dict[int, dict[str, int]]:
+        if not adgroup_ids:
+            return {}
+
+        metrics: dict[int, dict[str, int]] = {}
+        page = 1
+        while True:
+            params = {
+                **self._common_params(account_id),
+                "account_id": account_id,
+                "level": "REPORT_LEVEL_ADGROUP",
+                "date_range": json.dumps(
+                    {
+                        "start_date": data_date.isoformat(),
+                        "end_date": data_date.isoformat(),
+                    }
+                ),
+                "group_by": json.dumps(["adgroup_id", "date"]),
+                "fields": json.dumps(
+                    [
+                        "account_id",
+                        "adgroup_id",
+                        "date",
+                        "view_count",
+                        "valid_click_count",
+                        "cost",
+                        "conversions_count",
+                    ]
+                ),
+                "filtering": json.dumps(
+                    [
+                        {
+                            "field": "adgroup_id",
+                            "operator": "IN",
+                            "values": [str(value) for value in adgroup_ids],
+                        }
+                    ]
+                ),
+                "page": page,
+                "page_size": 100,
+                "time_line": "REQUEST_TIME",
+            }
+            response = self.session.get(
+                f"{self.base_url}/daily_reports/get",
+                params=params,
+                timeout=self.timeout,
+            )
+            response.raise_for_status()
+            data = self._check(response.json(), "get_today_ad_metrics")
+            for row in data.get("list") or []:
+                adgroup_id = int(row.get("adgroup_id") or 0)
+                if adgroup_id <= 0:
+                    continue
+                metrics[adgroup_id] = {
+                    "cost_fen": int(row.get("cost") or 0),
+                    "impressions": int(row.get("view_count") or 0),
+                    "clicks": int(row.get("valid_click_count") or 0),
+                    "conversions": int(row.get("conversions_count") or 0),
+                }
+            page_info = data.get("page_info") or {}
+            if page >= int(page_info.get("total_page") or 1):
+                break
+            page += 1
+        return metrics
+
+
+def resolve_bid_field(ad: dict[str, Any], configured_bid_scene: str | None) -> str:
+    smart_bid_type = str(ad.get("smart_bid_type") or "").upper()
+    try:
+        has_custom_cost_cap = int(ad.get("custom_cost_cap") or 0) > 0
+    except (TypeError, ValueError):
+        has_custom_cost_cap = False
+    if (
+        has_custom_cost_cap
+        or smart_bid_type == "SMART_BID_TYPE_SYSTEMATIC"
+        or configured_bid_scene == "max_conversion"
+    ):
+        return "custom_cost_cap"
+    return "bid_amount"
+
+
+def current_bid_fen(ad: dict[str, Any], bid_field: str) -> int | None:
+    try:
+        return int(ad.get(bid_field))
+    except (TypeError, ValueError):
+        return None