feat(realtime-control): add pause-only account scope

AGENTS.md

@@ -19,6 +19,7 @@
 - 生产使用同一个完整 `ad-put-agent` 镜像启动两个容器:`ad-control-service` 和 `ad-daily-service`。
 - `ad-control-service` 负责唯一飞书 WebSocket、运营暂停/停止/恢复、ROI 表格逐行审批执行和每 10 分钟实时 CPM 调控。
 - 实时调控账户范围是历史 `ad_creation_account_config` 与启用白名单的交集;飞书关闭创建配置只停止新建/补创意,不能让既有广告退出实时管理。
+- `realtime_control_account_scope` 可独立纳管非自动创建账户;显式配置覆盖自动化账户默认模式。`PAUSE_ONLY` 账户只执行 CPM 关停,不能执行 BOOST 或 RESTORE 调价。
 - `ad-daily-service` 每天 10:30 调用现有广告/创意创建流程,每 2 小时扫描腾讯创意正式审核结果,并可在每天 11:00 计算和发布日级 ROI 批次。
 - 旧 `server.py`、`execute_once.py` 和 `run.py` 的模型调控链路不再作为生产入口,但暂不删除历史代码。
 - 飞书生产控制命令使用确定性解析,不能让模型直接决定账户范围或执行腾讯写操作。
@@ -27,6 +28,7 @@
 - ROI 表格逐行审批是独立入口:黄色【审批选择】列填写“批准”即为最终确认,不再经过群聊二次确认。审批表使用获得链接者可编辑权限,但执行目标必须只按数据库中的隐藏幂等键回读,不能信任表格内可编辑的账户、广告、创意、成本或 ROI 字段。
 - 运营暂停状态和 CPM 暂停状态必须分开持久化。原本人工暂停的广告不能被系统认领或自动开启;腾讯后台人工重新开启时以人工操作为准。
 - 实时控制和飞书写操作必须共用 MySQL advisory lock,并执行腾讯写后回读校验。
+- 广告设置 `bid_hold` 时,实时 CPM 调控不得修改其出价;解除 `bid_hold` 只能恢复实时调价资格,不能自动恢复历史价格。
 - `RTC_APPLY_ENABLED`、`DAILY_ROI_ENABLED`、`ROI_APPLY_ENABLED` 默认关闭;生产切换前必须先完成 dry-run 和只通知验证。
 
 ## ROI 北极星指标与日级调控

examples/tencent_realtime_control/adjust_bid_experiment_20260729.py

@@ -0,0 +1,284 @@
+#!/usr/bin/env python
+"""一次性执行 2026-07-29 八广告分组调价实验。默认只预览。"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+import uuid
+from dataclasses import asdict, dataclass
+from datetime import datetime
+from decimal import Decimal, ROUND_HALF_UP
+from pathlib import Path
+from typing import Any
+
+from dotenv import load_dotenv
+
+
+ROOT = Path(__file__).resolve().parent
+REPO_ROOT = ROOT.parents[1]
+MINIAPP_ROOT = ROOT.parent / "auto_put_ad_mini"
+for path in (ROOT, MINIAPP_ROOT, REPO_ROOT):
+    if str(path) not in sys.path:
+        sys.path.insert(0, str(path))
+load_dotenv(MINIAPP_ROOT / ".env", override=False)
+load_dotenv(REPO_ROOT / ".env", override=False)
+
+from run_once import SHANGHAI  # noqa: E402
+from storage import (  # noqa: E402
+    advisory_lock,
+    clear_bid_experiment_hold,
+    initialize_schema,
+    insert_action_log,
+    set_bid_experiment_hold,
+    upsert_realtime_account_scope,
+)
+from tencent_client import (  # noqa: E402
+    ACTIVE_STATUS,
+    TencentClient,
+    current_bid_fen,
+    resolve_bid_field,
+)
+
+
+@dataclass(frozen=True)
+class ExperimentAd:
+    group: str
+    factor: Decimal
+    agency: str
+    audience_name: str
+    account_id: int
+    adgroup_id: int
+    adgroup_name: str
+    hold_realtime_bid: bool = False
+
+
+EXPERIMENT_ADS = (
+    ExperimentAd("下调10%", Decimal("0.90"), "小程序-代投-像素", "R50*已转化", 84207102, 107115658070, "R50*已转化-含小程序"),
+    ExperimentAd("下调10%", Decimal("0.90"), "小程序-自动化", "泛人群", 86748335, 117650209221, "泛人群-20260716-关键页面-targeted", True),
+    ExperimentAd("下调5%", Decimal("0.95"), "小程序-代投-像素", "", 82037573, 116560437525, "泛人群-公小朋-45+-0713"),
+    ExperimentAd("下调5%", Decimal("0.95"), "小程序-代投-翱鲨", "回流330以上人群", 86532580, 118337357855, "票圈-图片-朋友圈+公小-0719-zc-R330人群-关键页"),
+    ExperimentAd("上调5%", Decimal("1.05"), "小程序-代投-棱镜", "R50*泛知识*时政历史", 79911661, 108000527445, "0612-朋友圈+公众号小程序-关键页r50历史-玉"),
+    ExperimentAd("上调5%", Decimal("1.05"), "小程序-自动化", "回流330以上人群", 86197371, 115272976405, "回流330以上人群-20260708-关键页面-targeted", True),
+    ExperimentAd("上调10%", Decimal("1.10"), "小程序-代投-翱鲨", "R50*泛知识*生活科普", 83290935, 108653638689, "票圈-图片-朋友圈+公小-0615-zc-R50*泛知识*生活科普-关键页-2"),
+    ExperimentAd("上调10%", Decimal("1.10"), "小程序-代投-棱镜", "wx*商业", 85504853, 119352748609, "0722朋友圈+公众号小程序-wx*商业*玉-点击"),
+)
+
+
+def target_bid_fen(current_bid: int, factor: Decimal) -> int:
+    return int(
+        (Decimal(current_bid) * factor).quantize(
+            Decimal("1"), rounding=ROUND_HALF_UP
+        )
+    )
+
+
+def prepare(client: TencentClient) -> list[dict[str, Any]]:
+    rows: list[dict[str, Any]] = []
+    for spec in EXPERIMENT_ADS:
+        ad = client.get_ad(spec.account_id, spec.adgroup_id)
+        actual_name = str(ad.get("adgroup_name") or "").strip()
+        if actual_name != spec.adgroup_name:
+            raise ValueError(
+                f"广告名称不一致: account={spec.account_id} "
+                f"adgroup={spec.adgroup_id} actual={actual_name!r}"
+            )
+        status = str(ad.get("configured_status") or "")
+        if status != ACTIVE_STATUS:
+            raise ValueError(
+                f"广告当前不是正常状态: account={spec.account_id} "
+                f"adgroup={spec.adgroup_id} status={status}"
+            )
+        bid_field = resolve_bid_field(ad, None)
+        current_bid = current_bid_fen(ad, bid_field)
+        if current_bid is None or current_bid <= 0:
+            raise ValueError(
+                f"广告当前出价无效: account={spec.account_id} "
+                f"adgroup={spec.adgroup_id} field={bid_field}"
+            )
+        rows.append(
+            {
+                **asdict(spec),
+                "factor": str(spec.factor),
+                "configured_status": status,
+                "bid_field": bid_field,
+                "current_bid_fen": current_bid,
+                "target_bid_fen": target_bid_fen(current_bid, spec.factor),
+                "status": "planned",
+            }
+        )
+    return rows
+
+
+def write_report(rows: list[dict[str, Any]], mode: str) -> Path:
+    output_dir = ROOT / "outputs"
+    output_dir.mkdir(parents=True, exist_ok=True)
+    timestamp = datetime.now(SHANGHAI).strftime("%Y%m%d_%H%M%S")
+    path = output_dir / f"bid_experiment_20260729_{mode}_{timestamp}.json"
+    path.write_text(json.dumps(rows, ensure_ascii=False, indent=2), encoding="utf-8")
+    return path
+
+
+def release_holds(apply: bool) -> int:
+    now = datetime.now(SHANGHAI)
+    rows = [
+        {
+            "account_id": spec.account_id,
+            "adgroup_id": spec.adgroup_id,
+            "adgroup_name": spec.adgroup_name,
+            "status": "planned",
+        }
+        for spec in EXPERIMENT_ADS
+        if spec.hold_realtime_bid
+    ]
+    if apply:
+        with advisory_lock("tencent_realtime_control") as acquired:
+            if not acquired:
+                raise RuntimeError("实时调控正在执行，未获得数据库锁")
+            for row in rows:
+                clear_bid_experiment_hold(
+                    int(row["account_id"]),
+                    int(row["adgroup_id"]),
+                    action_at=now,
+                )
+                row["status"] = "success"
+    report = write_report(rows, "release_apply" if apply else "release_dry_run")
+    print(json.dumps({"rows": rows, "report": str(report)}, ensure_ascii=False))
+    return 0
+
+
+def register_pause_only_scope(apply: bool) -> int:
+    rows = [
+        {
+            "account_id": spec.account_id,
+            "audience_name": spec.audience_name,
+            "control_mode": "PAUSE_ONLY",
+            "status": "planned",
+        }
+        for spec in EXPERIMENT_ADS
+    ]
+    if apply:
+        for row in rows:
+            upsert_realtime_account_scope(
+                account_id=int(row["account_id"]),
+                control_mode="PAUSE_ONLY",
+                audience_name=str(row["audience_name"]),
+                bid_scene=None,
+                source="bid_experiment_20260729",
+                note="八账户只参与实时CPM关停",
+            )
+            row["status"] = "success"
+    report = write_report(
+        rows,
+        "scope_apply" if apply else "scope_dry_run",
+    )
+    print(json.dumps({"rows": rows, "report": str(report)}, ensure_ascii=False))
+    return 0
+
+
+def apply_experiment(client: TencentClient, rows: list[dict[str, Any]]) -> int:
+    run_id = f"bid-exp-20260729-{uuid.uuid4().hex[:12]}"
+    now = datetime.now(SHANGHAI)
+    failures = 0
+    with advisory_lock("tencent_realtime_control") as acquired:
+        if not acquired:
+            raise RuntimeError("实时调控正在执行，未获得数据库锁")
+        for row in rows:
+            before = int(row["current_bid_fen"])
+            target = int(row["target_bid_fen"])
+            try:
+                readback = client.update_ad(
+                    int(row["account_id"]),
+                    int(row["adgroup_id"]),
+                    bid_field=str(row["bid_field"]),
+                    target_bid_fen=target,
+                )
+                row["readback_bid_fen"] = int(readback[row["bid_field"]])
+                if row["hold_realtime_bid"]:
+                    try:
+                        set_bid_experiment_hold(
+                            account_id=int(row["account_id"]),
+                            adgroup_id=int(row["adgroup_id"]),
+                            adgroup_name=str(row["adgroup_name"]),
+                            bid_field=str(row["bid_field"]),
+                            base_bid_fen=target,
+                            action_at=now,
+                            reason="20260729分组调价实验_手动解除",
+                        )
+                    except Exception:
+                        client.update_ad(
+                            int(row["account_id"]),
+                            int(row["adgroup_id"]),
+                            bid_field=str(row["bid_field"]),
+                            target_bid_fen=before,
+                        )
+                        raise
+                row["status"] = "success"
+            except Exception as exc:
+                failures += 1
+                row["status"] = "failed"
+                row["error"] = str(exc)
+            try:
+                insert_action_log(
+                    {
+                        "run_id": run_id,
+                        "control_date": now.date(),
+                        "decision": "BID_EXPERIMENT",
+                        "account_id": row["account_id"],
+                        "adgroup_id": row["adgroup_id"],
+                        "adgroup_name": row["adgroup_name"],
+                        "bid_field": row["bid_field"],
+                        "base_bid_fen": row["target_bid_fen"],
+                        "before_bid_fen": row["current_bid_fen"],
+                        "target_bid_fen": row["target_bid_fen"],
+                        "before_status": row["configured_status"],
+                        "apply_mode": True,
+                        "execution_status": row["status"],
+                        "error_message": row.get("error"),
+                    }
+                )
+            except Exception as exc:
+                row["audit_error"] = str(exc)
+    report = write_report(rows, "apply")
+    print(
+        json.dumps(
+            {"run_id": run_id, "failures": failures, "report": str(report)},
+            ensure_ascii=False,
+        )
+    )
+    return 1 if failures else 0
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="执行20260729八广告分组调价实验")
+    parser.add_argument("--apply", action="store_true", help="真实修改腾讯出价")
+    parser.add_argument(
+        "--release-hold",
+        action="store_true",
+        help="只解除两条自动化广告的实时出价冻结，不修改腾讯价格",
+    )
+    parser.add_argument(
+        "--register-pause-only",
+        action="store_true",
+        help="只把八个账户登记为实时CPM关停范围，不再次调价",
+    )
+    args = parser.parse_args()
+
+    initialize_schema()
+    if args.release_hold:
+        return release_holds(args.apply)
+    if args.register_pause_only:
+        return register_pause_only_scope(args.apply)
+
+    client = TencentClient()
+    rows = prepare(client)
+    if args.apply:
+        return apply_experiment(client, rows)
+    report = write_report(rows, "dry_run")
+    print(json.dumps({"rows": rows, "report": str(report)}, ensure_ascii=False))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

examples/tencent_realtime_control/run_once.py

@@ -198,6 +198,12 @@ def execute_inventory_action(
 
     for account in accounts:
         account_id = int(account["account_id"])
+        control_mode = str(account.get("control_mode") or "FULL").upper()
+        if control_mode == "PAUSE_ONLY" and decision not in (
+            DECISION_PAUSE,
+            DECISION_CUTOFF,
+        ):
+            continue
         states = load_ad_states(account_id)
         try:
             ads = tencent.get_ads(account_id)
@@ -309,6 +315,11 @@ def execute_inventory_action(
             if plan is None:
                 continue
 
+            bid_hold = bool((state or {}).get("bid_hold"))
+            if bid_hold:
+                plan["target_bid_fen"] = current_bid
+                plan["bid_change_needed"] = False
+
             before_begin_date = str(ad.get("begin_date") or "")
             target_begin_date = None
             begin_date_change_needed = False
@@ -401,6 +412,7 @@ def execute_inventory_action(
                 "before_status": current_status,
                 "target_status": plan["target_status"],
                 "boost_limit_reached": plan["boost_limit_reached"],
+                "bid_hold": bid_hold,
                 "bid_change_needed": plan["bid_change_needed"],
                 "status_change_needed": plan["status_change_needed"],
                 "before_begin_date": before_begin_date,

examples/tencent_realtime_control/schema.sql

@@ -19,6 +19,8 @@ CREATE TABLE IF NOT EXISTS realtime_control_ad_state (
     base_bid_fen INT NOT NULL,
     initial_base_bid_fen INT DEFAULT NULL,
     boosted_date DATE DEFAULT NULL,
+    bid_hold BOOLEAN NOT NULL DEFAULT FALSE,
+    bid_hold_reason VARCHAR(255) DEFAULT NULL,
     last_roi_scaled_at DATETIME DEFAULT NULL,
     paused_by_strategy BOOLEAN NOT NULL DEFAULT FALSE,
     pause_reason VARCHAR(32) DEFAULT NULL,
@@ -39,6 +41,19 @@ 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 realtime_control_account_scope (
+    account_id BIGINT NOT NULL PRIMARY KEY,
+    control_mode VARCHAR(32) NOT NULL,
+    audience_name VARCHAR(255) DEFAULT NULL,
+    bid_scene VARCHAR(64) DEFAULT NULL,
+    enabled BOOLEAN NOT NULL DEFAULT TRUE,
+    source VARCHAR(64) DEFAULT NULL,
+    note VARCHAR(512) DEFAULT NULL,
+    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+    KEY idx_realtime_scope_enabled_mode (enabled, control_mode)
+) 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,

examples/tencent_realtime_control/storage.py

@@ -60,6 +60,8 @@ def initialize_schema() -> None:
             migrations = {
                 "initial_base_bid_fen": "INT DEFAULT NULL",
                 "last_roi_scaled_at": "DATETIME DEFAULT NULL",
+                "bid_hold": "BOOLEAN NOT NULL DEFAULT FALSE",
+                "bid_hold_reason": "VARCHAR(255) DEFAULT NULL",
                 "operator_pause_mode": "VARCHAR(32) DEFAULT NULL",
                 "operator_resume_at": "DATETIME DEFAULT NULL",
                 "operator_command_id": "VARCHAR(64) DEFAULT NULL",
@@ -207,20 +209,75 @@ def load_enabled_accounts() -> list[dict[str, Any]]:
 
 
 def load_realtime_accounts() -> list[dict[str, Any]]:
-    """All historical automation accounts still enabled by the whitelist."""
+    """Load automation accounts plus explicit real-time scope overrides."""
     connection = connect()
     try:
         with connection.cursor() as cursor:
             cursor.execute(
                 """
-                SELECT c.account_id, c.audience_name, c.bid_scene
+                SELECT c.account_id, c.audience_name, c.bid_scene,
+                       'FULL' AS control_mode
                 FROM ad_creation_account_config c
                 JOIN account_whitelist w ON w.account_id = c.account_id
                 WHERE w.enabled = TRUE
                 ORDER BY c.account_id
                 """
             )
-            return list(cursor.fetchall())
+            accounts = {
+                int(row["account_id"]): row for row in cursor.fetchall()
+            }
+            cursor.execute(
+                """
+                SELECT account_id, audience_name, bid_scene, control_mode
+                FROM realtime_control_account_scope
+                WHERE enabled=TRUE
+                ORDER BY account_id
+                """
+            )
+            for row in cursor.fetchall():
+                accounts[int(row["account_id"])] = row
+            return [accounts[key] for key in sorted(accounts)]
+    finally:
+        connection.close()
+
+
+def upsert_realtime_account_scope(
+    *,
+    account_id: int,
+    control_mode: str,
+    audience_name: str,
+    bid_scene: str | None,
+    source: str,
+    note: str,
+) -> None:
+    if control_mode not in {"FULL", "PAUSE_ONLY"}:
+        raise ValueError(f"Unsupported real-time control mode: {control_mode}")
+    connection = connect()
+    try:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                """
+                INSERT INTO realtime_control_account_scope
+                    (account_id, control_mode, audience_name, bid_scene,
+                     enabled, source, note)
+                VALUES (%s,%s,%s,%s,TRUE,%s,%s)
+                ON DUPLICATE KEY UPDATE
+                    control_mode=VALUES(control_mode),
+                    audience_name=VALUES(audience_name),
+                    bid_scene=VALUES(bid_scene),
+                    enabled=TRUE,
+                    source=VALUES(source),
+                    note=VALUES(note)
+                """,
+                (
+                    account_id,
+                    control_mode,
+                    audience_name,
+                    bid_scene,
+                    source,
+                    note,
+                ),
+            )
     finally:
         connection.close()
 
@@ -410,6 +467,92 @@ def sync_roi_base_bid(
     finally:
         connection.close()
 
+
+def set_bid_experiment_hold(
+    *,
+    account_id: int,
+    adgroup_id: int,
+    adgroup_name: str,
+    bid_field: str,
+    base_bid_fen: int,
+    action_at: datetime,
+    reason: str,
+) -> None:
+    """Persist an experiment base bid and prevent real-time bid overrides."""
+
+    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,
+                     bid_hold, bid_hold_reason, paused_by_strategy,
+                     last_action, last_action_at, last_seen_at)
+                VALUES (%s,%s,%s,%s,%s,%s,%s,TRUE,%s,FALSE,
+                        'BID_EXPERIMENT_HOLD',%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),
+                    bid_hold=TRUE,
+                    bid_hold_reason=VALUES(bid_hold_reason),
+                    last_action='BID_EXPERIMENT_HOLD',
+                    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,
+                    base_bid_fen,
+                    action_at.date(),
+                    reason,
+                    action_at,
+                    action_at,
+                ),
+            )
+    finally:
+        connection.close()
+
+
+def clear_bid_experiment_hold(
+    account_id: int,
+    adgroup_id: int,
+    *,
+    action_at: datetime,
+) -> None:
+    connection = connect()
+    try:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                """
+                UPDATE realtime_control_ad_state
+                SET bid_hold=FALSE,
+                    bid_hold_reason=NULL,
+                    last_action='BID_EXPERIMENT_RELEASE',
+                    last_action_at=%s,
+                    last_seen_at=%s
+                WHERE account_id=%s AND adgroup_id=%s
+                """,
+                (action_at, action_at, account_id, adgroup_id),
+            )
+            if cursor.rowcount != 1:
+                raise ValueError(
+                    "未找到实验出价状态: "
+                    f"account={account_id} adgroup={adgroup_id}"
+                )
+    finally:
+        connection.close()
+
 def insert_action_log(record: dict[str, Any]) -> None:
     columns = [
         "run_id",