Agent/examples/tencent_realtime_control/adjust_bid_experiment_20260729.py

#!/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())