Agent/examples/auto_put_ad_mini/roi_control/policy.py

"""Translate versioned ROI metrics into auditable recommendations and actions."""

from __future__ import annotations

import hashlib
import math
from typing import Any

import pandas as pd

from .fission_multiplier import (
    DISPLAY_MULTIPLIER_COLUMN,
    DISPLAY_TOTAL_TO_FIRST_COLUMN,
)
from .metrics import ENTITY_KEYS, ENTITY_SELF, ENTITY_SELF_AD


ACTION_PAUSE_CREATIVE = "PAUSE_CREATIVE"
ACTION_PAUSE_AD = "PAUSE_AD"
ACTION_SCALE_BID = "SCALE_BID"
MODE_ACTIONABLE = "ACTIONABLE"
MODE_NOTIFY_ONLY = "NOTIFY_ONLY"

POLICY_AUDIT_TEXT_FIELDS = (
    "阈值样本状态",
    "关停规则",
    "关停预算选择状态",
)
POLICY_AUDIT_NUMBER_FIELDS = (
    "昨日成本",
    "小程序昨日总成本",
    "小程序实际关停成本占比",
    "关停线分位点",
    "整体三日ROI排名百分位",
    "创意三日ROI排名百分位",
    "单日ROI排名百分位",
    "单日P10线",
    "单日合并资格线",
)


def safe_int(value: Any) -> int | None:
    if value is None or value == "":
        return None
    try:
        number = int(float(value))
    except (TypeError, ValueError):
        return None
    return number if number > 0 else None


def _entity_hash(row: pd.Series) -> str:
    keys = ENTITY_KEYS[str(row["entity_type"])]
    raw = "\x1f".join(str(row.get(key, "")) for key in keys)
    return hashlib.sha256(raw.encode("utf-8")).hexdigest()


def _finite(value: Any) -> float | None:
    try:
        number = float(value)
    except (TypeError, ValueError):
        return None
    return number if math.isfinite(number) else None


def annotate_execution(
    summary: pd.DataFrame,
    managed_account_ids: set[int],
    run_id: str,
) -> tuple[pd.DataFrame, list[dict[str, Any]], list[dict[str, Any]]]:
    """Keep policy recommendations separate from Tencent eligibility.

    Returns annotated summary, complete metric snapshots, and deduplicated
    executable action items.
    """

    annotated = summary.copy()
    annotated["执行模式"] = MODE_NOTIFY_ONLY
    annotated["执行说明"] = "无调控建议"
    annotated["审批选择"] = "不可执行"
    annotated["执行状态"] = ""
    annotated["执行结果"] = ""
    annotated["动作幂等键"] = ""
    actions: list[dict[str, Any]] = []
    snapshots: list[dict[str, Any]] = []
    seen_actions: set[str] = set()

    for index, row in annotated.iterrows():
        entity_type = str(row.get("entity_type") or "")
        recommendation = str(row.get("动作") or "")
        account_id = safe_int(row.get("账号id"))
        adgroup_id = safe_int(row.get("广告id"))
        creative_id = safe_int(row.get("创意id"))
        execution_mode = MODE_NOTIFY_ONLY
        execution_reason = "无调控建议"
        action_type: str | None = None

        if recommendation:
            if recommendation == "观察":
                execution_reason = "观察行仅展示，不执行腾讯写操作"
            elif entity_type not in (ENTITY_SELF, ENTITY_SELF_AD):
                execution_reason = "合作渠道当前仅通知，不执行腾讯写操作"
            elif account_id not in managed_account_ids:
                execution_reason = "账户不在自动化管理范围，仅通知"
            elif not adgroup_id:
                execution_reason = "缺少有效广告ID，仅通知"
            elif (
                recommendation == "关停"
                and entity_type == ENTITY_SELF
                and not creative_id
            ):
                execution_reason = "缺少有效动态创意ID，仅通知"
            elif recommendation == "关停" and entity_type == ENTITY_SELF_AD:
                action_type = ACTION_PAUSE_AD
            elif recommendation == "关停" and entity_type == ENTITY_SELF:
                action_type = ACTION_PAUSE_CREATIVE
            elif recommendation == "扩量" and entity_type == ENTITY_SELF:
                action_type = ACTION_SCALE_BID
            else:
                execution_reason = "该建议当前没有自动执行器，仅通知"

        if action_type:
            suffix = creative_id if action_type == ACTION_PAUSE_CREATIVE else adgroup_id
            idempotency_key = f"{run_id}:{action_type}:{account_id}:{suffix}"
            execution_mode = MODE_ACTIONABLE
            if action_type == ACTION_PAUSE_CREATIVE:
                execution_reason = "审批后暂停该动态创意"
            elif action_type == ACTION_PAUSE_AD:
                execution_reason = "审批后暂停整个广告"
            else:
                execution_reason = "审批后按ROI策略提高广告永久基础出价"
            if idempotency_key not in seen_actions:
                actions.append(
                    {
                        "idempotency_key": idempotency_key,
                        "action_type": action_type,
                        "account_id": account_id,
                        "adgroup_id": adgroup_id,
                        "dynamic_creative_id": creative_id,
                        "execution_status": "PENDING",
                    }
                )
                seen_actions.add(idempotency_key)

        annotated.at[index, "执行模式"] = execution_mode
        annotated.at[index, "执行说明"] = execution_reason
        if execution_mode == MODE_ACTIONABLE:
            annotated.at[index, "审批选择"] = ""
            annotated.at[index, "执行状态"] = "待审批"
            annotated.at[index, "动作幂等键"] = idempotency_key

        daily_metrics = {
            column: _finite(row.get(column))
            for column in annotated.columns
            if column.startswith(
                (
                    "首层UV_20",
                    "T0裂变数_20",
                    "成本_20",
                    "效率收入_20",
                    "裂变效率收入_20",
                    "实际ROI_20",
                    "ROI_20",
                )
            )
        }
        daily_metrics.update(
            {
                "fission_parameter_version": str(
                    row.get("传播裂变参数版本") or ""
                ),
                "fission_cohort_date": str(
                    row.get("传播裂变参数cohort日期") or ""
                ),
                "fission_match_level": str(
                    row.get("传播裂变系数匹配层级") or ""
                ),
                "fission_source": str(row.get("传播裂变系数来源") or ""),
                "fission_parameter_status": str(
                    row.get("传播裂变参数状态") or ""
                ),
                "control_participation_status": str(
                    row.get("调控参与状态") or ""
                ),
                "roi_method": str(row.get("ROI计算口径") or ""),
                **{
                    column: str(row.get(column) or "")
                    for column in POLICY_AUDIT_TEXT_FIELDS
                },
                **{
                    column: _finite(row.get(column))
                    for column in POLICY_AUDIT_NUMBER_FIELDS
                },
            }
        )
        snapshots.append(
            {
                "entity_hash": _entity_hash(row),
                "entity_type": entity_type,
                "channel": str(row.get("channel") or ""),
                "account_id": account_id,
                "account_name": str(row.get("账号名称") or ""),
                "adgroup_id": adgroup_id,
                "adgroup_name": str(row.get("广告名称") or ""),
                "dynamic_creative_id": creative_id,
                "audience_name": str(row.get("包名") or ""),
                "conversion_goal": str(row.get("广告优化目标") or ""),
                "partner_name": str(row.get("合作方名") or ""),
                "official_account_name": str(row.get("公众号名") or ""),
                "fission_parameter_version": str(
                    row.get("传播裂变参数版本") or ""
                ),
                "fission_cohort_date": str(
                    row.get("传播裂变参数cohort日期") or ""
                ),
                "fission_multiplier_vs_t0": _finite(
                    row.get(DISPLAY_MULTIPLIER_COLUMN)
                ),
                "fission_multiplier_vs_first": _finite(
                    row.get(DISPLAY_TOTAL_TO_FIRST_COLUMN)
                ),
                "fission_match_level": str(
                    row.get("传播裂变系数匹配层级") or ""
                ),
                "fission_source": str(row.get("传播裂变系数来源") or ""),
                "ad_age": safe_int(row.get("广告age")),
                "avg_first_uv": _finite(row.get("日均首层UV")),
                "first_uv": _finite(row.get("窗口首层UV")),
                "t0_fission_count": _finite(row.get("窗口T0裂变数")),
                "t0_fission_rate": _finite(row.get("T0裂变率")),
                "cost": _finite(row.get("成本")),
                "efficiency_revenue": _finite(row.get("效率收入")),
                "fission_revenue": _finite(row.get("T0实际裂变收入")),
                "actual_total_revenue": _finite(
                    row.get("实际全链路效率收入")
                ),
                "actual_roi": _finite(row.get("实际ROI")),
                "predicted_tail_revenue": _finite(
                    row.get("预测T1-T15裂变收入")
                ),
                "predicted_fission_revenue": _finite(
                    row.get("预测T0-T15裂变收入")
                ),
                "total_revenue": _finite(row.get("全链路效率收入")),
                "roi": _finite(row.get("ROI")),
                "stop_threshold": _finite(row.get("t_stop")),
                "scale_threshold": _finite(row.get("t_up")),
                "recommended_action": recommendation or None,
                "action_reason": str(row.get("动作原因") or ""),
                "execution_mode": execution_mode,
                "ineligible_reason": (
                    execution_reason if execution_mode == MODE_NOTIFY_ONLY else None
                ),
                "daily_metrics_json": daily_metrics,
            }
        )

    return annotated, snapshots, actions