Agent/examples/auto_put_ad_mini/roi_control/rules.py

"""Versioned three-day policy with a latest-day creative stop supplement."""

from __future__ import annotations

from dataclasses import dataclass
from typing import Iterable

import numpy as np
import pandas as pd

from .fission_multiplier import FissionMultiplierParameters
from .metrics import (
    ENTITY_GZH,
    ENTITY_SELF,
    ENTITY_SELF_AD,
    GZH_CHANNEL,
    SELF_CHANNEL,
    compute_roi_summary,
)


POLICY_VERSION = "roi_policy_v15"
POLICY_RUN_SUFFIX = "p15"

STOP_RULE_THREE_DAY = "三日持续低ROI"
STOP_RULE_ONE_DAY = "单日绝对低ROI且P10"


@dataclass(frozen=True)
class RuleConfig:
    self_stop_min_age: int = 4
    self_up_min_age: int = 3
    self_min_daily_uv: float = 200
    partner_min_daily_uv: float = 200
    observe_min_latest_uv: float = 200
    one_day_min_uv: float = 200
    one_day_hard_stop_roi: float = 0.20
    one_day_stop_quantile: float = 0.10
    self_stop_cost_soft_lower_ratio: float = 0.045
    self_stop_cost_target_ratio: float = 0.05
    self_stop_cost_soft_upper_ratio: float = 0.055
    self_stop_cost_hard_cap_ratio: float = 0.06
    self_stop_three_day_share: float = 0.60
    self_stop_one_day_share: float = 0.40
    stop_quantile: float = 0.20
    up_quantile: float = 0.80


def _daily_sample_mask(
    summary: pd.DataFrame,
    entity_type: str,
    dt: str,
    config: RuleConfig,
) -> pd.Series:
    min_uv = (
        config.partner_min_daily_uv
        if entity_type == ENTITY_GZH
        else config.self_min_daily_uv
    )
    return (
        summary["entity_type"].eq(entity_type)
        & pd.to_numeric(summary[f"首层UV_{dt}"], errors="coerce").gt(min_uv)
        & pd.to_numeric(summary[f"成本_{dt}"], errors="coerce").gt(0)
        & np.isfinite(pd.to_numeric(summary[f"ROI_{dt}"], errors="coerce"))
    )


def entity_eligibility_mask(
    summary: pd.DataFrame,
    entity_type: str,
    config: RuleConfig,
    expected_dates: list[str],
) -> pd.Series:
    eligible = summary["entity_type"].eq(entity_type)
    for dt in expected_dates:
        eligible &= _daily_sample_mask(summary, entity_type, dt, config)
    return eligible


def threshold_eligibility_mask(
    summary: pd.DataFrame,
    config: RuleConfig,
    expected_dates: list[str],
) -> pd.Series:
    """Return formal creative-level samples for channel-independent lines."""

    return entity_eligibility_mask(
        summary, ENTITY_SELF, config, expected_dates
    ) | entity_eligibility_mask(summary, ENTITY_GZH, config, expected_dates)


def report_formal_mask(
    summary: pd.DataFrame,
    config: RuleConfig,
    expected_dates: list[str],
) -> pd.Series:
    return threshold_eligibility_mask(
        summary, config, expected_dates
    ) | entity_eligibility_mask(summary, ENTITY_SELF_AD, config, expected_dates)


def observation_mask(
    summary: pd.DataFrame,
    config: RuleConfig,
    expected_dates: list[str],
) -> pd.Series:
    latest = expected_dates[-1]
    formal = report_formal_mask(summary, config, expected_dates)
    return (
        summary["entity_type"].isin([ENTITY_SELF, ENTITY_SELF_AD, ENTITY_GZH])
        & ~formal
        & pd.to_numeric(summary[f"首层UV_{latest}"], errors="coerce").gt(
            config.observe_min_latest_uv
        )
    )


def one_day_supplement_mask(
    summary: pd.DataFrame,
    config: RuleConfig,
    expected_dates: list[str],
) -> pd.Series:
    latest = expected_dates[-1]
    latest_roi = pd.to_numeric(summary[f"ROI_{latest}"], errors="coerce")
    formal_creative = entity_eligibility_mask(
        summary, ENTITY_SELF, config, expected_dates
    )
    return (
        summary["entity_type"].eq(ENTITY_SELF)
        & ~formal_creative
        & pd.to_numeric(summary[f"首层UV_{latest}"], errors="coerce").gt(
            config.one_day_min_uv
        )
        & pd.to_numeric(summary[f"成本_{latest}"], errors="coerce").gt(0)
        & np.isfinite(latest_roi)
    )


def compute_channel_thresholds(
    summary: pd.DataFrame,
    expected_dates: list[str],
    config: RuleConfig,
) -> pd.DataFrame:
    self_eligible = entity_eligibility_mask(
        summary, ENTITY_SELF, config, expected_dates
    )
    gzh_eligible = entity_eligibility_mask(
        summary, ENTITY_GZH, config, expected_dates
    )
    self_sample = pd.to_numeric(
        summary.loc[self_eligible, "ROI"], errors="coerce"
    )
    self_sample = self_sample[np.isfinite(self_sample)]
    gzh_sample = pd.to_numeric(
        summary.loc[gzh_eligible, "ROI"], errors="coerce"
    )
    gzh_sample = gzh_sample[np.isfinite(gzh_sample)]
    if self_sample.empty and gzh_sample.empty:
        raise ValueError("最近三日没有满足每日UV和成本门槛的渠道独立阈值样本")

    t_up = (
        float(self_sample.quantile(config.up_quantile))
        if not self_sample.empty
        else np.nan
    )
    gzh_stop = (
        float(gzh_sample.quantile(config.stop_quantile))
        if not gzh_sample.empty
        else np.nan
    )
    latest = expected_dates[-1]
    return pd.DataFrame(
        [
            {
                "统计窗口": f"{expected_dates[0]} 至 {latest}",
                "entity_type": ENTITY_SELF,
                "渠道": SELF_CHANNEL,
                "t_stop": np.nan,
                "t_up": t_up,
                "t_one_day_stop": np.nan,
                "关停线分位点": np.nan,
                "单日实际关停线分位点": np.nan,
                "阈值样本数": int(len(self_sample)),
                "扩量样本数": int(len(self_sample)),
                "单日候选池样本数": 0,
                "单日硬关停ROI": config.one_day_hard_stop_roi,
                "单日P10线": np.nan,
                "单日合并资格线": np.nan,
                "单日最低UV": config.one_day_min_uv,
                "关停线口径": "T-1实际成本5%软预算动态线",
                "扩量线口径": (
                    f"合格小程序创意实体等权P{int(config.up_quantile * 100)}"
                ),
                "广告级是否入池": "否_仅复用小程序三日动态关停线",
            },
            {
                "统计窗口": f"{expected_dates[0]} 至 {latest}",
                "entity_type": ENTITY_GZH,
                "渠道": GZH_CHANNEL,
                "t_stop": gzh_stop,
                "t_up": np.nan,
                "t_one_day_stop": np.nan,
                "关停线分位点": (
                    config.stop_quantile if np.isfinite(gzh_stop) else np.nan
                ),
                "单日实际关停线分位点": np.nan,
                "阈值样本数": int(len(gzh_sample)),
                "扩量样本数": 0,
                "单日候选池样本数": 0,
                "单日硬关停ROI": np.nan,
                "单日P10线": np.nan,
                "单日合并资格线": np.nan,
                "单日最低UV": np.nan,
                "关停线口径": (
                    f"公众号合格实体等权P{int(config.stop_quantile * 100)}"
                ),
                "扩量线口径": "不适用",
                "广告级是否入池": "不适用",
            },
        ]
    )


def _sample_percentile(values: pd.Series, target: float) -> float:
    clean = pd.to_numeric(values, errors="coerce")
    clean = clean[np.isfinite(clean)]
    if clean.empty or not np.isfinite(target):
        return np.nan
    return float((clean <= target).mean())


def _ordered_candidate_indices(
    result: pd.DataFrame,
    mask: pd.Series,
    roi_column: str,
    cost_column: str,
) -> list[object]:
    roi = pd.to_numeric(result[roi_column], errors="coerce")
    cost = pd.to_numeric(result[cost_column], errors="coerce")
    eligible = mask & np.isfinite(roi) & cost.gt(0)
    return result.loc[eligible].assign(_decision_roi=roi.loc[eligible]).sort_values(
        "_decision_roi",
        ascending=True,
        kind="stable",
    ).index.tolist()


def _select_prefix_within_budget(
    result: pd.DataFrame,
    ordered: list[object],
    cost_column: str,
    budget: float,
) -> list[object]:
    selected: list[object] = []
    cost = 0.0
    for index in ordered:
        candidate_cost = float(result.at[index, cost_column])
        if cost + candidate_cost > budget + 1e-9:
            break
        selected.append(index)
        cost += candidate_cost
    return selected


def _selection_line(
    result: pd.DataFrame,
    ordered: list[object],
    selected: set[object],
    roi_column: str,
) -> tuple[float, float]:
    selected_indices = [index for index in ordered if index in selected]
    if not selected_indices:
        return np.nan, np.nan
    line = float(
        pd.to_numeric(
            result.loc[selected_indices, roi_column], errors="coerce"
        ).max()
    )
    sample = pd.to_numeric(result.loc[ordered, roi_column], errors="coerce")
    return line, _sample_percentile(sample, line)


def _allocate_self_stop_budget(
    result: pd.DataFrame,
    config: RuleConfig,
    expected_dates: list[str],
    formal_creative: pd.Series,
    one_day_supplement: pd.Series,
) -> tuple[pd.Series, pd.Series, dict[str, float | str]]:
    latest = expected_dates[-1]
    cost_column = f"成本_{latest}"
    latest_roi_column = f"ROI_{latest}"
    latest_cost = pd.to_numeric(result[cost_column], errors="coerce").fillna(0.0)
    latest_roi = pd.to_numeric(result[latest_roi_column], errors="coerce")
    age = pd.to_numeric(result.get("广告age"), errors="coerce").fillna(0)
    actionable_age = age.ge(config.self_stop_min_age)

    three_day_mask = formal_creative & actionable_age
    one_day_pool = one_day_supplement & actionable_age
    one_day_sample = latest_roi.loc[one_day_pool]
    one_day_sample = one_day_sample[np.isfinite(one_day_sample)]
    one_day_p10_line = (
        float(one_day_sample.quantile(config.one_day_stop_quantile))
        if not one_day_sample.empty
        else np.nan
    )
    one_day_eligibility_line = (
        min(config.one_day_hard_stop_roi, one_day_p10_line)
        if np.isfinite(one_day_p10_line)
        else np.nan
    )
    one_day_candidate = (
        one_day_pool
        & latest_roi.le(config.one_day_hard_stop_roi)
        & latest_roi.le(one_day_p10_line)
    )
    result["单日ROI排名百分位"] = np.nan
    if not one_day_sample.empty:
        result.loc[one_day_pool, "单日ROI排名百分位"] = latest_roi.loc[
            one_day_pool
        ].apply(lambda value: _sample_percentile(one_day_sample, float(value)))
    result["单日P10线"] = one_day_p10_line
    result["单日合并资格线"] = one_day_eligibility_line
    masks = {
        STOP_RULE_THREE_DAY: three_day_mask,
        STOP_RULE_ONE_DAY: one_day_candidate,
    }
    roi_columns = {
        STOP_RULE_THREE_DAY: "ROI",
        STOP_RULE_ONE_DAY: latest_roi_column,
    }
    shares = {
        STOP_RULE_THREE_DAY: config.self_stop_three_day_share,
        STOP_RULE_ONE_DAY: config.self_stop_one_day_share,
    }
    all_self = result["entity_type"].eq(ENTITY_SELF)
    total_cost = float(latest_cost.loc[all_self & latest_cost.gt(0)].sum())
    target_cost = total_cost * config.self_stop_cost_target_ratio
    soft_lower_cost = total_cost * config.self_stop_cost_soft_lower_ratio
    soft_upper_cost = total_cost * config.self_stop_cost_soft_upper_ratio
    hard_cap_cost = total_cost * config.self_stop_cost_hard_cap_ratio

    ordered = {
        rule: _ordered_candidate_indices(
            result,
            mask,
            roi_columns[rule],
            cost_column,
        )
        for rule, mask in masks.items()
    }
    selected: set[object] = set()
    for rule in (STOP_RULE_THREE_DAY, STOP_RULE_ONE_DAY):
        selected.update(
            _select_prefix_within_budget(
                result,
                ordered[rule],
                cost_column,
                target_cost * shares[rule],
            )
        )

    selected_cost = float(latest_cost.loc[list(selected)].sum()) if selected else 0.0
    for rule in (STOP_RULE_ONE_DAY, STOP_RULE_THREE_DAY):
        for index in ordered[rule]:
            if index in selected:
                continue
            candidate_cost = float(latest_cost.loc[index])
            proposed = selected_cost + candidate_cost
            if proposed <= target_cost + 1e-9:
                selected.add(index)
                selected_cost = proposed
                continue
            if (
                proposed <= soft_upper_cost + 1e-9
                and abs(proposed - target_cost) < abs(selected_cost - target_cost)
            ):
                selected.add(index)
                selected_cost = proposed
            break

    # The normal selection stays within the 5.5% soft upper bound. If discrete
    # creative costs still leave the result below the 4.5% soft lower bound,
    # allow the next uninterrupted ROI prefix to approach 5%, but never exceed
    # the 6% absolute cap.
    if selected_cost < soft_lower_cost:
        for rule in (STOP_RULE_ONE_DAY, STOP_RULE_THREE_DAY):
            for index in ordered[rule]:
                if index in selected:
                    continue
                candidate_cost = float(latest_cost.loc[index])
                proposed = selected_cost + candidate_cost
                if (
                    proposed <= hard_cap_cost + 1e-9
                    and abs(proposed - target_cost)
                    < abs(selected_cost - target_cost)
                ):
                    selected.add(index)
                    selected_cost = proposed
                break

    if selected_cost > hard_cap_cost + 1e-9:
        raise ValueError("小程序关停候选成本超过绝对上限")

    selected_mask = pd.Series(result.index.isin(selected), index=result.index)
    candidate_rule = pd.Series("", index=result.index, dtype="object")
    for rule, mask in masks.items():
        candidate_rule.loc[mask] = rule

    lines: dict[str, float] = {}
    for rule in ordered:
        line, percentile = _selection_line(
            result,
            ordered[rule],
            selected,
            roi_columns[rule],
        )
        lines[f"{rule}_line"] = line
        lines[f"{rule}_percentile"] = percentile
        selected_indices = [index for index in ordered[rule] if index in selected]
        lines[f"{rule}_cost"] = (
            float(latest_cost.loc[selected_indices].sum())
            if selected_indices
            else 0.0
        )

    stats = {
        "小程序昨日总成本": total_cost,
        "软下限关停成本": soft_lower_cost,
        "目标关停成本": target_cost,
        "软上限关停成本": soft_upper_cost,
        "绝对上限关停成本": hard_cap_cost,
        "实际关停成本": selected_cost,
        "实际关停成本占比": selected_cost / total_cost if total_cost > 0 else 0.0,
        "关停成本预算状态": (
            "正常范围"
            if soft_lower_cost <= selected_cost <= soft_upper_cost
            else "低于软下限"
            if selected_cost < soft_lower_cost
            else "高于软上限"
        ),
        "三日基础预算成本": target_cost * config.self_stop_three_day_share,
        "单日基础预算成本": target_cost * config.self_stop_one_day_share,
        "三日候选数": len(ordered[STOP_RULE_THREE_DAY]),
        "单日候选池样本数": int(one_day_pool.sum()),
        "单日合并候选数": len(ordered[STOP_RULE_ONE_DAY]),
        "单日P10线": one_day_p10_line,
        "单日合并资格线": one_day_eligibility_line,
        "三日实际关停成本": lines[f"{STOP_RULE_THREE_DAY}_cost"],
        "单日实际关停成本": lines[f"{STOP_RULE_ONE_DAY}_cost"],
        "三日关停线": lines[f"{STOP_RULE_THREE_DAY}_line"],
        "三日关停线分位点": lines[f"{STOP_RULE_THREE_DAY}_percentile"],
        "单日实际关停线": lines[f"{STOP_RULE_ONE_DAY}_line"],
        "单日实际关停线分位点": lines[f"{STOP_RULE_ONE_DAY}_percentile"],
    }
    return selected_mask, candidate_rule, stats


def apply_actions(
    summary: pd.DataFrame,
    thresholds: pd.DataFrame,
    config: RuleConfig,
    expected_dates: list[str],
) -> tuple[pd.DataFrame, pd.DataFrame]:
    result = summary.copy()
    result["动作"] = ""
    result["动作原因"] = ""
    thresholds = thresholds.copy()
    threshold_by_type = thresholds.set_index("entity_type")
    self_threshold = threshold_by_type.loc[ENTITY_SELF]
    gzh_threshold = threshold_by_type.loc[ENTITY_GZH]
    result["t_stop"] = np.nan
    result.loc[result["entity_type"].eq(ENTITY_GZH), "t_stop"] = float(
        gzh_threshold["t_stop"]
    )
    result["t_up"] = np.nan
    result.loc[
        result["entity_type"].isin([ENTITY_SELF, ENTITY_SELF_AD]), "t_up"
    ] = float(self_threshold["t_up"])
    result["t_one_day_stop"] = np.nan
    result["关停线分位点"] = np.nan

    threshold_eligible = threshold_eligibility_mask(result, config, expected_dates)
    ad_eligible = entity_eligibility_mask(
        result, ENTITY_SELF_AD, config, expected_dates
    )
    observe_only = observation_mask(result, config, expected_dates)
    one_day_supplement = one_day_supplement_mask(
        result, config, expected_dates
    )
    creative_eligible = entity_eligibility_mask(
        result, ENTITY_SELF, config, expected_dates
    )
    gzh_eligible = entity_eligibility_mask(
        result, ENTITY_GZH, config, expected_dates
    )
    self_sample_roi = result.loc[creative_eligible, "ROI"]
    gzh_sample_roi = result.loc[gzh_eligible, "ROI"]
    creative_sample_roi = result.loc[creative_eligible, "ROI"]
    result["整体三日ROI排名百分位"] = np.nan
    for mask, sample in (
        (result["entity_type"].isin([ENTITY_SELF, ENTITY_SELF_AD]), self_sample_roi),
        (result["entity_type"].eq(ENTITY_GZH), gzh_sample_roi),
    ):
        result.loc[mask, "整体三日ROI排名百分位"] = result.loc[mask, "ROI"].apply(
            lambda value: _sample_percentile(sample, float(value))
            if pd.notna(value)
            else np.nan
        )
    result["是否低于三日关停线"] = False
    result["创意三日ROI排名百分位"] = result["ROI"].apply(
        lambda value: _sample_percentile(creative_sample_roi, float(value))
        if pd.notna(value)
        else np.nan
    )
    result["是否位于创意三日ROI前20%"] = False
    if np.isfinite(float(self_threshold["t_up"])):
        result.loc[creative_eligible, "是否位于创意三日ROI前20%"] = (
            pd.to_numeric(result.loc[creative_eligible, "ROI"], errors="coerce")
            >= result.loc[creative_eligible, "t_up"]
        )

    (
        selected_stop,
        candidate_rule,
        budget_stats,
    ) = _allocate_self_stop_budget(
        result,
        config,
        expected_dates,
        creative_eligible,
        one_day_supplement,
    )
    self_row = thresholds["entity_type"].eq(ENTITY_SELF)
    threshold_updates = {
        **budget_stats,
        "t_stop": budget_stats["三日关停线"],
        "t_one_day_stop": budget_stats["单日实际关停线"],
        "关停线分位点": budget_stats["三日关停线分位点"],
        "单日实际关停线分位点": budget_stats["单日实际关停线分位点"],
    }
    for column, value in threshold_updates.items():
        thresholds.loc[self_row, column] = value

    latest = expected_dates[-1]
    result["昨日成本"] = np.where(
        result["entity_type"].eq(ENTITY_SELF),
        pd.to_numeric(result[f"成本_{latest}"], errors="coerce"),
        np.nan,
    )
    result["关停规则"] = candidate_rule
    result["关停预算选择状态"] = ""
    result.loc[candidate_rule.ne(""), "关停预算选择状态"] = "预算未选中"
    result.loc[selected_stop, "关停预算选择状态"] = "预算已选中"
    result["小程序昨日总成本"] = budget_stats["小程序昨日总成本"]
    result["小程序实际关停成本占比"] = budget_stats["实际关停成本占比"]
    formal_line = budget_stats["三日关停线"]
    formal_percentile = budget_stats["三日关停线分位点"]
    self_or_ad = result["entity_type"].isin([ENTITY_SELF, ENTITY_SELF_AD])
    result.loc[self_or_ad, "t_stop"] = formal_line
    result.loc[self_or_ad, "关停线分位点"] = formal_percentile
    one_day_rows = candidate_rule.eq(STOP_RULE_ONE_DAY)
    result.loc[one_day_rows, "t_stop"] = budget_stats["单日实际关停线"]
    result.loc[one_day_rows, "t_one_day_stop"] = budget_stats["单日实际关停线"]
    result.loc[one_day_rows, "关停线分位点"] = budget_stats[
        "单日实际关停线分位点"
    ]
    result.loc[selected_stop & creative_eligible, "是否低于三日关停线"] = True
    result.loc[gzh_eligible, "关停线分位点"] = float(
        gzh_threshold["关停线分位点"]
    )
    result.loc[gzh_eligible, "是否低于三日关停线"] = (
        pd.to_numeric(result.loc[gzh_eligible, "ROI"], errors="coerce")
        <= float(gzh_threshold["t_stop"])
    )

    for index, row in result.iterrows():
        entity_type = str(row["entity_type"])
        if bool(one_day_supplement.loc[index]):
            latest_uv = float(row[f"首层UV_{latest}"])
            latest_roi = float(row[f"ROI_{latest}"])
            three_day_roi = float(row["ROI"])
            raw_age = row.get("广告age")
            age = int(raw_age) if pd.notna(raw_age) else 0
            decision_context = (
                "未满足连续三天每天首层UV>200、成本>0且ROI有效，"
                "启用单日补充规则；"
            )
            three_day_context = (
                f"；三日预测总效率ROI={three_day_roi:.2f}仅作参考，"
                "不参与本次单日判断"
            )
            one_day_p10_line = float(budget_stats["单日P10线"])
            meets_absolute = latest_roi <= config.one_day_hard_stop_roi
            meets_p10 = (
                np.isfinite(one_day_p10_line)
                and latest_roi <= one_day_p10_line
            )
            qualifies = meets_absolute and meets_p10
            stop_reason = (
                f"最新日首层UV={latest_uv:.0f}>{config.one_day_min_uv:g}，"
                f"最新日预测总效率ROI={latest_roi:.2f}同时满足ROI≤"
                f"{config.one_day_hard_stop_roi:.2f}和单日候选池P10线"
                f"{one_day_p10_line:.2f}"
            )

            if qualifies and age < config.self_stop_min_age:
                result.at[index, "动作"] = "观察"
                result.at[index, "动作原因"] = (
                    f"{decision_context}{stop_reason}{three_day_context}；"
                    f"广告age={age}≤{config.self_stop_min_age - 1}天，暂不关停"
                )
            elif bool(selected_stop.loc[index]):
                result.at[index, "动作"] = "关停"
                result.at[index, "动作原因"] = (
                    f"{decision_context}{stop_reason}{three_day_context}；"
                    f"广告age={age}>{config.self_stop_min_age - 1}天，"
                    f"昨日成本={float(row[f'成本_{latest}']):.2f}元，"
                    f"按5%成本软预算选中（{candidate_rule.loc[index]}）；"
                    "建议审批后暂停动态创意"
                )
            elif qualifies:
                result.at[index, "动作"] = "观察"
                result.at[index, "动作原因"] = (
                    f"{decision_context}{stop_reason}{three_day_context}；"
                    f"昨日成本={float(row[f'成本_{latest}']):.2f}元，"
                    "满足低质候选条件但未被5%成本软预算选中；建议观察"
                )
            elif not meets_absolute:
                result.at[index, "动作"] = "观察"
                result.at[index, "动作原因"] = (
                    f"{decision_context}最新日首层UV={latest_uv:.0f}>"
                    f"{config.one_day_min_uv:g}，最新日预测总效率ROI="
                    f"{latest_roi:.2f}高于绝对线{config.one_day_hard_stop_roi:.2f}，"
                    f"不满足单日两个条件{three_day_context}；建议继续观察"
                )
            elif not meets_p10:
                result.at[index, "动作"] = "观察"
                result.at[index, "动作原因"] = (
                    f"{decision_context}最新日预测总效率ROI={latest_roi:.2f}≤"
                    f"{config.one_day_hard_stop_roi:.2f}，但高于单日候选池P10线"
                    f"{one_day_p10_line:.2f}、未进入后10%{three_day_context}；"
                    "建议继续观察"
                )
            else:
                result.at[index, "动作"] = "观察"
                result.at[index, "动作原因"] = (
                    f"{decision_context}单日候选池不足，无法计算P10线"
                    f"{three_day_context}；建议继续观察"
                )
            continue
        if bool(observe_only.loc[index]):
            result.at[index, "动作"] = "观察"
            result.at[index, "动作原因"] = (
                f"最新日首层UV>{config.observe_min_latest_uv:g}，但未满足连续三天"
                "每天首层UV>200、成本>0且ROI有效；仅置底展示，不进入正式阈值和自动执行"
            )
            continue
        if bool(ad_eligible.loc[index]):
            if np.isfinite(float(row["t_stop"])) and float(row["ROI"]) <= float(
                row["t_stop"]
            ):
                raw_age = row.get("广告age")
                age = int(raw_age) if pd.notna(raw_age) else 0
                if age >= config.self_stop_min_age:
                    result.at[index, "动作"] = "关停"
                    result.at[index, "动作原因"] = (
                        "广告级三日加权平均效率ROI≤小程序三日动态关停线，"
                        f"广告age>{config.self_stop_min_age - 1}天；审批后暂停整个广告"
                    )
                else:
                    result.at[index, "动作"] = "观察"
                    result.at[index, "动作原因"] = (
                        "广告级三日加权平均效率ROI≤小程序三日动态关停线，但广告age≤"
                        f"{config.self_stop_min_age - 1}天"
                    )
            continue
        if not bool(threshold_eligible.loc[index]):
            continue
        if entity_type == ENTITY_SELF:
            raw_age = row.get("广告age")
            age = int(raw_age) if pd.notna(raw_age) else 0
            if bool(selected_stop.loc[index]):
                result.at[index, "动作"] = "关停"
                result.at[index, "动作原因"] = (
                    "三日加权平均预测总效率ROI按从低到高排序，"
                    f"昨日成本={float(row[f'成本_{latest}']):.2f}元，"
                    "按5%成本软预算的60%基础额度或结余额度选中；"
                    f"广告age>{config.self_stop_min_age - 1}天；审批后仅暂停动态创意"
                )
            elif (
                np.isfinite(float(row["t_stop"]))
                and float(row["ROI"]) <= float(row["t_stop"])
                and age < config.self_stop_min_age
            ):
                result.at[index, "动作"] = "观察"
                result.at[index, "动作原因"] = (
                    "三日加权平均预测总效率ROI低于小程序动态关停线，但广告age≤"
                    f"{config.self_stop_min_age - 1}天"
                )
            elif bool(row["是否位于创意三日ROI前20%"]):
                if age >= config.self_up_min_age:
                    result.at[index, "动作"] = "扩量"
                    result.at[index, "动作原因"] = (
                        f"三日加权平均效率ROI≥合格创意实体等权P80，"
                        f"广告age≥{config.self_up_min_age}天；审批后提高广告永久基础出价"
                    )
                else:
                    result.at[index, "动作"] = "观察"
                    result.at[index, "动作原因"] = (
                        f"三日加权平均效率ROI≥合格创意实体等权P80，但广告age<"
                        f"{config.self_up_min_age}天"
                    )
        elif (
            entity_type == ENTITY_GZH
            and np.isfinite(float(row["t_stop"]))
            and float(row["ROI"]) <= float(row["t_stop"])
        ):
            result.at[index, "动作"] = "关停"
            result.at[index, "动作原因"] = (
                "三日加权平均效率ROI≤公众号独立实体等权关停线；"
                "公众号当前仅通知参考"
            )
    return result, thresholds


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_channel_thresholds(summary, dates, config)
    evaluated, thresholds = apply_actions(summary, thresholds, config, dates)
    formal_threshold = threshold_eligibility_mask(evaluated, config, dates)
    formal_ad = entity_eligibility_mask(evaluated, ENTITY_SELF_AD, config, dates)
    one_day_supplement = one_day_supplement_mask(evaluated, config, dates)
    observe_only = observation_mask(evaluated, config, dates) & ~one_day_supplement
    evaluated["阈值样本状态"] = "未达到三日正式样本门槛"
    evaluated.loc[formal_threshold, "阈值样本状态"] = (
        "进入三日渠道独立阈值样本池"
    )
    evaluated.loc[formal_ad, "阈值样本状态"] = "广告级三日合格_不进入阈值样本池"
    evaluated.loc[one_day_supplement, "阈值样本状态"] = (
        "单日补充决策_昨日UV>200"
    )
    evaluated.loc[observe_only, "阈值样本状态"] = "补充观察_昨日UV>200"
    candidates = evaluated[evaluated["动作"].ne("")].copy()
    return candidates, thresholds, evaluated