"""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