Agent/examples/auto_put_ad_mini/roi_control/rules.py

"""Versioned ROI thresholds and action recommendation policy."""

from __future__ import annotations

from dataclasses import dataclass
from typing import Iterable

import numpy as np
import pandas as pd

from .metrics import (
    ENTITY_GZH,
    ENTITY_QIWEI,
    ENTITY_SELF,
    compute_roi_summary,
)
from .fission_multiplier import FissionMultiplierParameters


POLICY_VERSION = "roi_policy_v1"
POLICY_RUN_SUFFIX = "p1"

@dataclass(frozen=True)
class RuleConfig:
    self_stop_min_age: int = 5
    self_up_min_age: int = 3
    self_min_avg_uv: float = 200
    partner_min_avg_uv: float = 200
    min_daily_cost: float = 100
    stop_quantile: float = 0.10
    up_quantile: float = 0.80
    gzh_adjust_rank_min: float = 0.10
    gzh_adjust_rank_max: float = 0.30


def threshold_eligibility_mask(
    summary: pd.DataFrame,
    config: RuleConfig,
) -> pd.Series:
    self_eligible = (
        summary["entity_type"].eq(ENTITY_SELF)
        & summary["日均首层UV"].gt(config.self_min_avg_uv)
    )
    gzh_eligible = (
        summary["entity_type"].eq(ENTITY_GZH)
        & summary["日均首层UV"].gt(config.partner_min_avg_uv)
    )
    return (
        (self_eligible | gzh_eligible)
        & summary["成本"].gt(0)
        & summary["三日最小单日成本"].gt(config.min_daily_cost)
        & np.isfinite(summary["ROI"])
    )


def compute_global_thresholds(
    summary: pd.DataFrame,
    expected_dates: list[str],
    config: RuleConfig,
) -> pd.DataFrame:
    valid = summary.loc[threshold_eligibility_mask(summary, config)].copy()
    if valid.empty:
        raise ValueError("没有满足渠道UV门槛且三日聚合ROI有效的实体，无法计算全局阈值")

    type_counts = valid["entity_type"].value_counts()
    return pd.DataFrame(
        [
            {
                "统计窗口": f"{expected_dates[0]} 至 {expected_dates[-1]}",
                "t_stop": float(valid["ROI"].quantile(config.stop_quantile)),
                "t_up": float(valid["ROI"].quantile(config.up_quantile)),
                "阈值样本数": int(len(valid)),
                "小程序样本数": int(type_counts.get(ENTITY_SELF, 0)),
                "公众号样本数": int(type_counts.get(ENTITY_GZH, 0)),
                "企微参考实体数": int(
                    summary["entity_type"].eq(ENTITY_QIWEI).sum()
                ),
            }
        ]
    )


def apply_actions(
    summary: pd.DataFrame,
    thresholds: pd.DataFrame,
    config: RuleConfig,
) -> pd.DataFrame:
    result = summary.copy()
    result["动作"] = ""
    result["动作原因"] = ""
    result["渠道内ROI排名百分位"] = np.nan
    t_stop = float(thresholds.iloc[0]["t_stop"])
    t_up = float(thresholds.iloc[0]["t_up"])
    result["t_stop"] = t_stop
    result["t_up"] = t_up

    gzh_mask = result["entity_type"].eq(ENTITY_GZH) & np.isfinite(result["ROI"])
    if gzh_mask.any():
        result.loc[gzh_mask, "渠道内ROI排名百分位"] = result.loc[gzh_mask, "ROI"].rank(
            method="average", ascending=False, pct=True
        )

    for index, row in result.iterrows():
        entity_type = row["entity_type"]
        avg_uv = row["日均首层UV"]
        roi = row["ROI"]
        cost_confident = row["三日最小单日成本"] > config.min_daily_cost
        stop_side = bool(np.isfinite(roi) and (roi <= t_stop or roi == 0))
        up_side = bool(np.isfinite(roi) and roi > 0 and roi >= t_up)

        if entity_type == ENTITY_SELF:
            if (
                row["广告age"] >= config.self_stop_min_age
                and stop_side
                and avg_uv > config.self_min_avg_uv
                and cost_confident
            ):
                result.at[index, "动作"] = "关停"
                result.at[index, "动作原因"] = (
                    f"广告age≥{config.self_stop_min_age}天，三日聚合ROI不高于"
                    f"全局t_stop（ROI=0明确关停），日均首层UV>"
                    f"{config.self_min_avg_uv:g}，连续3天每天成本>"
                    f"{config.min_daily_cost:g}"
                )
            elif (
                row["广告age"] >= config.self_up_min_age
                and up_side
                and avg_uv > config.self_min_avg_uv
                and cost_confident
            ):
                result.at[index, "动作"] = "扩量"
                result.at[index, "动作原因"] = (
                    f"广告age≥{config.self_up_min_age}天，三日聚合ROI不低于"
                    f"全局t_up，日均首层UV>{config.self_min_avg_uv:g}，"
                    f"连续3天每天成本>{config.min_daily_cost:g}"
                )
        elif entity_type == ENTITY_GZH:
            if stop_side and avg_uv > config.partner_min_avg_uv and cost_confident:
                result.at[index, "动作"] = "关停"
                result.at[index, "动作原因"] = (
                    "三日聚合ROI不高于全局t_stop（ROI=0明确关停），"
                    f"日均首层UV>{config.partner_min_avg_uv:g}，"
                    f"连续3天每天成本>{config.min_daily_cost:g}"
                )
            elif up_side and avg_uv > config.partner_min_avg_uv and cost_confident:
                result.at[index, "动作"] = "扩量"
                result.at[index, "动作原因"] = (
                    "三日聚合ROI不低于全局t_up，"
                    f"日均首层UV>{config.partner_min_avg_uv:g}，"
                    f"连续3天每天成本>{config.min_daily_cost:g}"
                )
            else:
                rank_pct = row["渠道内ROI排名百分位"]
                if (
                    np.isfinite(rank_pct)
                    and config.gzh_adjust_rank_min <= rank_pct <= config.gzh_adjust_rank_max
                ):
                    result.at[index, "动作"] = "调整封面&落地页视频"
                    result.at[index, "动作原因"] = (
                        "三日综合ROI在公众号即转渠道排名"
                        f"{config.gzh_adjust_rank_min:.0%}–"
                        f"{config.gzh_adjust_rank_max:.0%}"
                    )
    return result


def evaluate_rules(
    raw_daily: pd.DataFrame,
    expected_dates: Iterable[str],
    ad_age: pd.DataFrame | None = None,
    config: RuleConfig = RuleConfig(),
    *,
    fission_parameters: FissionMultiplierParameters,
) -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]:
    summary, dates = compute_roi_summary(
        raw_daily,
        expected_dates,
        ad_age,
        fission_parameters=fission_parameters,
    )
    thresholds = compute_global_thresholds(summary, dates, config)
    evaluated = apply_actions(summary, thresholds, config)
    evaluated["阈值样本状态"] = "未达到阈值样本门槛"
    evaluated.loc[
        threshold_eligibility_mask(evaluated, config),
        "阈值样本状态",
    ] = "进入阈值样本池"
    evaluated.loc[
        evaluated["entity_type"].eq(ENTITY_QIWEI),
        "阈值样本状态",
    ] = "企微仅展示_不进入阈值样本池"
    candidates = evaluated[evaluated["动作"].ne("")].copy()
    return candidates, thresholds, evaluated