Agent/examples/auto_put_ad_mini/test_roi_control_metrics.py

import tempfile
import unittest
from datetime import datetime
from pathlib import Path
from types import SimpleNamespace

import pandas as pd
from openpyxl import load_workbook

from roi_control.data_source import (
    SourceDataNotReadyError,
    build_daily_sql,
    build_source_readiness_sql,
    date_window,
    resolve_end_date,
)
from roi_control.fission_multiplier import (
    DISPLAY_MULTIPLIER_COLUMN,
    load_fission_multiplier_parameters,
)
from roi_control.metrics import (
    ENTITY_GZH,
    ENTITY_SELF,
    ENTITY_SELF_AD,
    GZH_CHANNEL,
    SELF_CHANNEL,
    prepare_daily_metrics,
)
from roi_control.reporting import (
    AGENCY_SUMMARY_SHEETS,
    DAILY_SHEETS,
    SUMMARY_SHEETS,
    _agency_summary_frame,
    _daily_frame,
    _summary_frame,
    _visible_columns,
    write_agency_workbooks,
    write_workbook,
)
from roi_control.rules import evaluate_rules as _evaluate_rules
from roi_control.service import _annotate_current_creative_status, _run_identity
from tencent_client import ACTIVE_STATUS, SUSPEND_STATUS


DATES = ["20260720", "20260721", "20260722"]
FISSION_PARAMETERS = load_fission_multiplier_parameters()


def evaluate_rules(*args, **kwargs):
    kwargs["fission_parameters"] = FISSION_PARAMETERS
    return _evaluate_rules(*args, **kwargs)


def row(entity_type, entity_id, dt, roi, *, uv=600, cost=200.0):
    common = {
        "dt": dt,
        "entity_type": entity_type,
        "channel": SELF_CHANNEL if entity_type != ENTITY_GZH else GZH_CHANNEL,
        "代理名称": "",
        "账号id": "",
        "账号名称": "",
        "广告id": "",
        "广告名称": "",
        "包名": "",
        "广告优化目标": "",
        "创意id": "",
        "合作方名": "",
        "公众号名": "",
        "首层UV": uv,
        "T0裂变数": uv * 0.2,
        "成本": cost,
        "效率收入": roi * cost,
        "裂变效率收入": 0,
    }
    if entity_type in (ENTITY_SELF, ENTITY_SELF_AD):
        common.update(
            {
                "代理名称": "代理",
                "账号id": "84502354",
                "账号名称": "账户",
                "广告id": entity_id,
                "广告名称": f"广告{entity_id}",
                "包名": "泛人群",
                "广告优化目标": "关键页面访问次数",
                "创意id": f"creative-{entity_id}" if entity_type == ENTITY_SELF else "",
            }
        )
    else:
        common.update({"合作方名": "合作方", "公众号名": entity_id})
    return common


class RoiThreeDayRulesTest(unittest.TestCase):
    @staticmethod
    def source_client(latest_dt, *, row_count=100, self_rows=60, gzh_rows=40):
        partition = SimpleNamespace(partition_spec={"dt": latest_dt})
        table = SimpleNamespace(get_max_partition=lambda: partition)
        odps = SimpleNamespace(get_table=lambda _name: table)
        return SimpleNamespace(
            odps=odps,
            execute_sql=lambda _sql: pd.DataFrame(
                [
                    {
                        "row_count": row_count,
                        "self_rows": self_rows,
                        "gzh_rows": gzh_rows,
                    }
                ]
            ),
        )

    def test_source_readiness_requires_exact_t_minus_one(self):
        client = self.source_client("20260803")
        end_date = resolve_end_date(
            client,
            now=datetime(2026, 8, 4, 9, 0),
        )
        self.assertEqual(end_date, "20260803")
        sql = build_source_readiness_sql(end_date)
        self.assertIn("dt = '20260803'", sql)
        self.assertIn(SELF_CHANNEL, sql)
        self.assertIn(GZH_CHANNEL, sql)

    def test_source_readiness_never_falls_back_to_older_partition(self):
        client = self.source_client("20260802")
        with self.assertRaisesRegex(SourceDataNotReadyError, "required_dt=20260803"):
            resolve_end_date(client, now=datetime(2026, 8, 4, 9, 0))

    def test_source_readiness_rejects_missing_report_channel(self):
        client = self.source_client("20260803", gzh_rows=0)
        with self.assertRaisesRegex(SourceDataNotReadyError, "gzh_rows"):
            resolve_end_date(client, now=datetime(2026, 8, 4, 9, 0))

    def build_daily(self):
        rows = []
        for dt in DATES:
            for index, roi in enumerate([0.1, 1.0, 3.0, 4.0]):
                rows.append(row(ENTITY_SELF, f"creative-ad-{index}", dt, roi))
            for index, roi in enumerate([0.5, 2.0, 5.0, 6.0]):
                rows.append(row(ENTITY_GZH, f"公众号-{index}", dt, roi, uv=300))
            for index, roi in enumerate([0.2, 3.5]):
                rows.append(row(ENTITY_SELF_AD, f"ad-{index}", dt, roi))
        return pd.DataFrame(rows)

    def ages(self):
        return pd.DataFrame(
            {
                "广告id": [f"creative-ad-{i}" for i in range(4)]
                + [f"ad-{i}" for i in range(2)],
                "广告age": [10] * 6,
            }
        )

    def test_versions_and_three_day_window(self):
        self.assertEqual(date_window("20260722"), ("20260720", "20260722"))
        run_id, _ = _run_identity("20260722", FISSION_PARAMETERS)
        self.assertIn("m8", run_id)
        self.assertIn("p15", run_id)
        self.assertIn("r41", run_id)

    def test_current_creative_status_only_reads_stop_decisions(self):
        rows = pd.DataFrame(
            [
                {"entity_type": ENTITY_SELF, "动作": "关停", "账号id": "1", "创意id": "11"},
                {"entity_type": ENTITY_SELF, "动作": "扩量", "账号id": "1", "创意id": "12"},
                {"entity_type": ENTITY_SELF, "动作": "关停", "账号id": "1", "创意id": "13"},
                {"entity_type": ENTITY_GZH, "动作": "关停", "账号id": "", "创意id": ""},
            ]
        )

        class FakeTencent:
            def __init__(self):
                self.calls = []

            def get_dynamic_creative(self, account_id, creative_id):
                self.calls.append((account_id, creative_id))
                return {
                    "configured_status": (
                        ACTIVE_STATUS if creative_id == 11 else SUSPEND_STATUS
                    )
                }

        client = FakeTencent()
        result = _annotate_current_creative_status(rows, client)
        self.assertEqual(client.calls, [(1, 11), (1, 13)])
        self.assertEqual(
            result["当前创意状态"].tolist(),
            ["正常", "", "已停止", ""],
        )

    def test_daily_sql_has_direct_ad_grain_and_excludes_qiwei(self):
        sql = build_daily_sql(DATES[0], DATES[-1])
        self.assertEqual(sql.count("SUM(NVL(t0_fission_uv_root, 0))"), 3)
        self.assertIn("'self_ad' AS entity_type", sql)
        self.assertNotIn("'qiwei' AS entity_type", sql)
        self.assertIn("COUNT(DISTINCT mid) AS 首层UV", sql)
        self.assertIn("usersharedepth <= 1", sql)
        self.assertNotIn("usersharedepth = '0'", sql)

    def test_channel_independent_stop_samples_and_creative_only_p80_scale_actions(self):
        candidates, thresholds, summary = evaluate_rules(
            self.build_daily(), DATES, self.ages()
        )
        self.assertEqual(len(thresholds), 2)
        by_type = thresholds.set_index("entity_type")
        self.assertTrue(pd.isna(by_type.loc[ENTITY_SELF, "t_stop"]))
        self.assertAlmostEqual(float(by_type.loc[ENTITY_GZH, "t_stop"]), 1.4)
        self.assertAlmostEqual(float(by_type.loc[ENTITY_SELF, "t_up"]), 3.4)
        self.assertEqual(int(by_type.loc[ENTITY_SELF, "阈值样本数"]), 4)
        self.assertEqual(int(by_type.loc[ENTITY_GZH, "阈值样本数"]), 4)
        self.assertEqual(int(by_type.loc[ENTITY_SELF, "扩量样本数"]), 4)
        self.assertEqual(
            by_type.loc[ENTITY_SELF, "关停线口径"],
            "T-1实际成本5%软预算动态线",
        )
        self.assertEqual(
            by_type.loc[ENTITY_SELF, "扩量线口径"],
            "合格小程序创意实体等权P80",
        )
        scale_rows = candidates[candidates["动作"].eq("扩量")]
        self.assertEqual(len(scale_rows), 1)
        self.assertEqual(scale_rows.iloc[0]["entity_type"], ENTITY_SELF)
        self.assertEqual(scale_rows.iloc[0]["广告id"], "creative-ad-3")
        self.assertTrue(scale_rows.iloc[0]["是否位于创意三日ROI前20%"])
        self.assertTrue((summary["覆盖天数"] == 3).all())
        pool = summary[summary["entity_type"].isin([ENTITY_SELF, ENTITY_GZH])]
        self.assertTrue(
            pool["阈值样本状态"].eq("进入三日渠道独立阈值样本池").all()
        )

    def test_self_stop_budget_uses_latest_cost_with_six_four_soft_allocation(self):
        rows = []
        formal = [
            ("formal-1", 0.10, 30.0),
            ("formal-2", 0.20, 30.0),
            ("formal-3", 1.00, 500.0),
            ("formal-4", 2.00, 500.0),
        ]
        for entity_id, roi, latest_cost in formal:
            for dt in DATES:
                rows.append(
                    row(
                        ENTITY_SELF,
                        entity_id,
                        dt,
                        roi,
                        cost=latest_cost if dt == DATES[-1] else 100.0,
                    )
                )
        rows.extend(
            [
                row(ENTITY_SELF, "hard-1", DATES[-1], 0.05, uv=300, cost=20.0),
                row(ENTITY_SELF, "hard-2", DATES[-1], 0.10, uv=300, cost=200.0),
                row(ENTITY_SELF, "rank-1", DATES[-1], 0.05, uv=600, cost=20.0),
                row(ENTITY_SELF, "rank-2", DATES[-1], 0.40, uv=600, cost=700.0),
            ]
        )
        ages = pd.DataFrame(
            {
                "广告id": [item[0] for item in formal]
                + ["hard-1", "hard-2", "rank-1", "rank-2"],
                "广告age": [10] * 8,
            }
        )

        _, thresholds, summary = evaluate_rules(pd.DataFrame(rows), DATES, ages)
        self_threshold = thresholds.set_index("entity_type").loc[ENTITY_SELF]
        selected = summary[
            summary["entity_type"].eq(ENTITY_SELF) & summary["动作"].eq("关停")
        ]
        selected_ids = set(selected["广告id"])
        latest_cost = selected[f"成本_{DATES[-1]}"].sum()

        self.assertEqual(
            selected_ids,
            {"formal-1", "formal-2", "hard-1", "rank-1"},
        )
        self.assertAlmostEqual(float(latest_cost), 100.0)
        self.assertAlmostEqual(float(self_threshold["小程序昨日总成本"]), 2000.0)
        self.assertAlmostEqual(float(self_threshold["目标关停成本"]), 100.0)
        self.assertAlmostEqual(float(self_threshold["实际关停成本"]), 100.0)
        self.assertAlmostEqual(float(self_threshold["实际关停成本占比"]), 0.05)
        self.assertEqual(self_threshold["关停成本预算状态"], "正常范围")
        self.assertAlmostEqual(float(self_threshold["三日基础预算成本"]), 60.0)
        self.assertAlmostEqual(float(self_threshold["单日基础预算成本"]), 40.0)
        self.assertAlmostEqual(float(self_threshold["三日关停线"]), 0.20)
        self.assertAlmostEqual(float(self_threshold["三日关停线分位点"]), 0.50)
        self.assertAlmostEqual(float(self_threshold["单日P10线"]), 0.05)
        self.assertAlmostEqual(float(self_threshold["单日合并资格线"]), 0.05)
        self.assertAlmostEqual(float(self_threshold["单日实际关停线"]), 0.05)
        self.assertEqual(
            selected.groupby("关停规则")[f"成本_{DATES[-1]}"].sum().to_dict(),
            {
                "三日持续低ROI": 60.0,
                "单日绝对低ROI且P10": 40.0,
            },
        )

    def test_self_and_official_account_stop_lines_are_independent(self):
        daily = self.build_daily()
        latest_costs = {
            "creative-ad-0": 10.0,
            "creative-ad-1": 330.0,
            "creative-ad-2": 330.0,
            "creative-ad-3": 330.0,
        }
        for entity_id, cost in latest_costs.items():
            mask = (
                daily["entity_type"].eq(ENTITY_SELF)
                & daily["广告id"].eq(entity_id)
                & daily["dt"].eq(DATES[-1])
            )
            roi = float(daily.loc[mask, "效率收入"].iloc[0]) / float(
                daily.loc[mask, "成本"].iloc[0]
            )
            daily.loc[mask, "成本"] = cost
            daily.loc[mask, "效率收入"] = roi * cost

        _, thresholds, summary = evaluate_rules(daily, DATES, self.ages())
        by_type = thresholds.set_index("entity_type")

        self.assertEqual(set(by_type.index), {ENTITY_SELF, ENTITY_GZH})
        self.assertAlmostEqual(float(by_type.loc[ENTITY_SELF, "t_stop"]), 0.10)
        self.assertAlmostEqual(
            float(by_type.loc[ENTITY_SELF, "关停线分位点"]),
            0.25,
        )
        self.assertAlmostEqual(float(by_type.loc[ENTITY_GZH, "t_stop"]), 1.40)
        self.assertAlmostEqual(
            float(by_type.loc[ENTITY_GZH, "关停线分位点"]),
            0.20,
        )
        self_rows = summary[summary["entity_type"].eq(ENTITY_SELF)]
        gzh_rows = summary[summary["entity_type"].eq(ENTITY_GZH)]
        self.assertTrue(self_rows["t_stop"].eq(0.10).all())
        self.assertTrue(
            gzh_rows["t_stop"].apply(
                lambda value: abs(float(value) - 1.40) < 1e-9
            ).all()
        )

    def test_unused_one_day_budget_rolls_to_three_day_candidates(self):
        rows = []
        formal = [
            ("formal-roll-1", 0.10, 60.0),
            ("formal-roll-2", 0.20, 20.0),
            ("formal-roll-3", 1.00, 900.0),
        ]
        for entity_id, roi, latest_cost in formal:
            for dt in DATES:
                rows.append(
                    row(
                        ENTITY_SELF,
                        entity_id,
                        dt,
                        roi,
                        cost=latest_cost if dt == DATES[-1] else 100.0,
                    )
                )
        rows.extend(
            [
                row(ENTITY_SELF, "rank-roll-1", DATES[-1], 0.05, uv=600, cost=20),
                row(ENTITY_SELF, "rank-roll-2", DATES[-1], 0.40, uv=600, cost=1000),
            ]
        )
        ages = pd.DataFrame(
            {
                "广告id": [item[0] for item in formal]
                + ["rank-roll-1", "rank-roll-2"],
                "广告age": [10] * 5,
            }
        )

        _, thresholds, summary = evaluate_rules(pd.DataFrame(rows), DATES, ages)
        selected = summary[
            summary["entity_type"].eq(ENTITY_SELF) & summary["动作"].eq("关停")
        ]
        self_threshold = thresholds.set_index("entity_type").loc[ENTITY_SELF]

        self.assertEqual(
            set(selected["广告id"]),
            {"formal-roll-1", "formal-roll-2", "rank-roll-1"},
        )
        self.assertAlmostEqual(float(self_threshold["实际关停成本"]), 100.0)
        self.assertAlmostEqual(float(self_threshold["三日基础预算成本"]), 60.0)
        self.assertAlmostEqual(float(self_threshold["三日实际关停成本"]), 80.0)
        self.assertAlmostEqual(float(self_threshold["单日实际关停成本"]), 20.0)
        self.assertEqual(self_threshold["关停成本预算状态"], "正常范围")

    def test_creative_p80_scale_requires_ad_age_at_least_three_days(self):
        ages = self.ages()
        ages.loc[ages["广告id"].eq("creative-ad-3"), "广告age"] = 2
        candidates, _, _ = evaluate_rules(self.build_daily(), DATES, ages)
        target = candidates[candidates["广告id"].eq("creative-ad-3")].iloc[0]
        self.assertEqual(target["动作"], "观察")
        self.assertIn("广告age<3天", target["动作原因"])

    def test_ad_level_reuses_threshold_without_entering_sample_and_can_stop(self):
        daily = self.build_daily()
        low_creative = daily["entity_type"].eq(ENTITY_SELF) & daily["广告id"].eq(
            "creative-ad-0"
        )
        latest_low = low_creative & daily["dt"].eq(DATES[-1])
        daily.loc[latest_low, ["成本", "效率收入"]] = [10.0, 1.0]
        low_ad = daily["entity_type"].eq(ENTITY_SELF_AD) & daily["广告id"].eq("ad-0")
        daily.loc[low_ad, "效率收入"] = daily.loc[low_ad, "成本"] * 0.05
        _, thresholds, summary = evaluate_rules(daily, DATES, self.ages())
        ad_rows = summary[summary["entity_type"].eq(ENTITY_SELF_AD)]
        self.assertTrue(
            ad_rows["阈值样本状态"].eq("广告级三日合格_不进入阈值样本池").all()
        )
        actions = dict(zip(ad_rows["广告id"], ad_rows["动作"]))
        self.assertEqual(actions["ad-0"], "关停")
        self.assertEqual(actions["ad-1"], "")
        self_line = float(
            thresholds.set_index("entity_type").loc[ENTITY_SELF, "t_stop"]
        )
        self.assertTrue(ad_rows["t_stop"].eq(self_line).all())
        self.assertTrue(ad_rows["调控参与状态"].str.contains("审批后可暂停广告").all())

    def test_latest_day_uv_over_200_is_appended_with_fixed_three_day_average(self):
        daily = self.build_daily()
        latest_only = row(ENTITY_SELF, "latest-only", DATES[-1], 1.5, uv=300)
        daily = pd.concat([daily, pd.DataFrame([latest_only])], ignore_index=True)
        candidates, _, summary = evaluate_rules(daily, DATES, self.ages())
        target = summary[summary["广告id"].eq("latest-only")].iloc[0]
        self.assertEqual(target["覆盖天数"], 1)
        self.assertEqual(target["阈值样本状态"], "单日补充决策_昨日UV>200")
        self.assertEqual(target["动作"], "观察")
        self.assertEqual(target["日均首层UV"], 100)
        self.assertTrue(candidates["广告id"].eq("latest-only").any())

    def test_one_day_rule_requires_absolute_low_roi_and_p10_and_age(self):
        daily = self.build_daily()
        supplemental = [
            row(ENTITY_SELF, "one-day-both", DATES[-1], 0.05, uv=300, cost=10),
            row(ENTITY_SELF, "one-day-not-p10", DATES[-1], 0.15, uv=300, cost=10),
            row(ENTITY_SELF, "one-day-young", DATES[-1], 0.01, uv=300, cost=10),
            row(ENTITY_SELF, "one-day-mid", DATES[-1], 1.0, uv=300, cost=10),
            row(ENTITY_SELF, "one-day-high", DATES[-1], 2.0, uv=300, cost=900),
        ]
        daily = pd.concat([daily, pd.DataFrame(supplemental)], ignore_index=True)
        ages = pd.concat(
            [
                self.ages(),
                pd.DataFrame(
                    {
                        "广告id": [
                            "one-day-both",
                            "one-day-not-p10",
                            "one-day-young",
                            "one-day-mid",
                            "one-day-high",
                        ],
                        "广告age": [4, 4, 3, 4, 4],
                    }
                ),
            ],
            ignore_index=True,
        )
        _, thresholds, summary = evaluate_rules(daily, DATES, ages)
        self_threshold = thresholds.set_index("entity_type").loc[ENTITY_SELF]
        self.assertEqual(int(self_threshold["单日候选池样本数"]), 4)
        self.assertAlmostEqual(float(self_threshold["单日P10线"]), 0.08)
        self.assertAlmostEqual(float(self_threshold["单日合并资格线"]), 0.08)
        targets = summary.set_index("广告id")
        self.assertEqual(targets.loc["one-day-both", "动作"], "关停")
        self.assertEqual(
            targets.loc["one-day-both", "关停规则"],
            "单日绝对低ROI且P10",
        )
        self.assertIn("同时满足", targets.loc["one-day-both", "动作原因"])
        self.assertIn("ROI≤0.20", targets.loc["one-day-both", "动作原因"])
        self.assertIn("P10", targets.loc["one-day-both", "动作原因"])
        self.assertEqual(targets.loc["one-day-not-p10", "动作"], "观察")
        self.assertIn("未进入后10%", targets.loc["one-day-not-p10", "动作原因"])
        self.assertEqual(targets.loc["one-day-young", "动作"], "观察")
        self.assertIn("广告age=3≤3天", targets.loc["one-day-young", "动作原因"])
        creative_frame = _summary_frame(summary, ENTITY_SELF)
        creative_stops = creative_frame[
            creative_frame["建议动作"].eq("关停创意")
        ]
        self.assertIn("one-day-both", creative_stops["广告id"].tolist())
        self.assertNotIn("one-day-not-p10", creative_stops["广告id"].tolist())
        self.assertTrue(creative_stops["动作"].eq("关停").all())
        self.assertTrue(
            summary[
                summary["entity_type"].isin([ENTITY_SELF, ENTITY_SELF_AD])
                & summary["动作"].eq("关停")
            ]["广告age"].ge(4).all()
        )

    def test_one_day_p10_without_absolute_low_roi_is_observe(self):
        daily = self.build_daily()
        supplemental = [
            row(ENTITY_SELF, f"one-day-high-{index}", DATES[-1], roi, uv=300, cost=10)
            for index, roi in enumerate([0.30, 0.40, 0.50, 0.60, 0.70])
        ]
        daily = pd.concat([daily, pd.DataFrame(supplemental)], ignore_index=True)
        ages = pd.concat(
            [
                self.ages(),
                pd.DataFrame(
                    {
                        "广告id": [f"one-day-high-{index}" for index in range(5)],
                        "广告age": [10] * 5,
                    }
                ),
            ],
            ignore_index=True,
        )

        _, thresholds, summary = evaluate_rules(daily, DATES, ages)
        self_threshold = thresholds.set_index("entity_type").loc[ENTITY_SELF]
        target = summary[summary["广告id"].eq("one-day-high-0")].iloc[0]
        self.assertGreater(float(self_threshold["单日P10线"]), 0.20)
        self.assertAlmostEqual(float(self_threshold["单日合并资格线"]), 0.20)
        self.assertEqual(target["动作"], "观察")
        self.assertIn("高于绝对线0.20", target["动作原因"])

    def test_three_day_formal_entity_never_reenters_one_day_rule(self):
        daily = self.build_daily()
        ages = self.ages()
        _, _, summary = evaluate_rules(daily, DATES, ages)
        target = summary[summary["广告id"].eq("creative-ad-0")].iloc[0]
        self.assertEqual(target["阈值样本状态"], "进入三日渠道独立阈值样本池")
        self.assertNotEqual(target["关停规则"], "单日绝对低ROI且P10")
        self.assertNotIn("单日", target["动作原因"])

    def test_daily_prediction_uses_same_day_t0_fission_revenue_once(self):
        raw = pd.DataFrame([row(ENTITY_SELF, "formula", DATES[-1], 0)])
        raw.loc[0, ["成本", "效率收入", "裂变效率收入"]] = [100, 50, 30]
        result = prepare_daily_metrics(raw, FISSION_PARAMETERS).iloc[0]
        multiplier = result[DISPLAY_MULTIPLIER_COLUMN]
        self.assertEqual(result["T0实际裂变收入"], 30)
        self.assertAlmostEqual(result["预测全链路效率收入"], 50 + 30 * multiplier)
        self.assertAlmostEqual(result["ROI"], (50 + 30 * multiplier) / 100)

    def test_agency_workbooks_filter_and_physically_remove_sensitive_columns(self):
        _, _, summary = evaluate_rules(self.build_daily(), DATES, self.ages())
        agency_rows = summary.copy()
        creative_indexes = agency_rows[
            agency_rows["entity_type"].eq(ENTITY_SELF)
        ].index.tolist()
        ad_indexes = agency_rows[
            agency_rows["entity_type"].eq(ENTITY_SELF_AD)
        ].index.tolist()
        agency_rows.loc[creative_indexes, "代理名称"] = "代理B"
        agency_rows.loc[ad_indexes, "代理名称"] = "代理B"
        agency_rows.loc[creative_indexes[0], "代理名称"] = "小程序-代投-贝湉"
        agency_rows.loc[creative_indexes[1], "代理名称"] = "小程序 -代投-贝湉"
        agency_rows.loc[ad_indexes[0], "代理名称"] = "小程序-代投-贝湉"
        original = agency_rows.copy(deep=True)

        with tempfile.TemporaryDirectory() as directory:
            outputs = write_agency_workbooks(
                agency_rows,
                Path(directory),
                "20260803",
            )
            self.assertEqual(
                [row["agency_name"] for row in outputs],
                ["代理B", "小程序-代投-贝湉"],
            )
            pd.testing.assert_frame_equal(agency_rows, original)
            bay = next(
                row for row in outputs if row["agency_name"] == "小程序-代投-贝湉"
            )
            self.assertEqual(bay["creative_rows"], 2)
            self.assertEqual(bay["ad_rows"], 1)
            self.assertEqual(
                Path(bay["report"]).name,
                "20260803_小程序-代投-贝湉_调控建议.xlsx",
            )
            filtered = write_agency_workbooks(
                agency_rows,
                Path(directory) / "filtered",
                "20260803",
                agency_names={"代理B"},
            )
            self.assertEqual(
                [row["agency_name"] for row in filtered],
                ["代理B"],
            )

            forbidden_fragments = (
                "ROI",
                "收入",
                "关停线",
                "扩量线",
                "排名",
                "是否位于",
                "t_stop",
                "t_up",
                "审批",
                "执行",
                "幂等键",
            )
            for output in outputs:
                workbook = load_workbook(output["report"], read_only=False)
                expected_sheets = {
                    AGENCY_SUMMARY_SHEETS[ENTITY_SELF],
                    AGENCY_SUMMARY_SHEETS[ENTITY_SELF_AD],
                }
                self.assertEqual(set(workbook.sheetnames), expected_sheets)
                self.assertEqual(
                    workbook[AGENCY_SUMMARY_SHEETS[ENTITY_SELF_AD]].sheet_state,
                    "hidden",
                )
                for sheet_name in expected_sheets:
                    sheet = workbook[sheet_name]
                    headers = [cell.value for cell in sheet[1]]
                    for removed in (
                        "包名",
                        "广告age",
                        "日均首层UV",
                        "建议说明",
                        "裂变系数-总裂变UV/T0裂变UV",
                        "裂变系数-总裂变UV/首层UV",
                        "日均T0裂变人数",
                        "日均T0裂变率",
                    ):
                        self.assertNotIn(removed, headers)
                    self.assertFalse(
                        any(
                            fragment in str(header)
                            for header in headers
                            for fragment in forbidden_fragments
                        )
                    )
                    self.assertTrue(
                        all(
                            not sheet.column_dimensions[cell.column_letter].hidden
                            for cell in sheet[1]
                        )
                    )
                creative_headers = [
                    cell.value
                    for cell in workbook[AGENCY_SUMMARY_SHEETS[ENTITY_SELF]][1]
                ]
                self.assertEqual(
                    creative_headers,
                    [
                        "渠道",
                        "代理名称",
                        "账号id",
                        "账号名称",
                        "广告id",
                        "广告名称",
                        "广告优化目标",
                        "创意id",
                        "日均成本",
                        "评分",
                        "建议动作",
                        "当前创意状态",
                    ],
                )
                creative = workbook[AGENCY_SUMMARY_SHEETS[ENTITY_SELF]]
                score_column = creative_headers.index("评分") + 1
                self.assertEqual(
                    creative.cell(2, score_column).number_format,
                    "0.00",
                )
                self.assertNotIn("当日效率ROI", creative_headers)

    def test_summary_and_daily_report_frames(self):
        daily = self.build_daily()
        daily = pd.concat(
            [
                daily,
                pd.DataFrame(
                    [
                        row(ENTITY_SELF, "latest-only-low", DATES[-1], 0.1, uv=220),
                        row(ENTITY_SELF, "latest-only-high", DATES[-1], 9.0, uv=280),
                    ]
                ),
            ],
            ignore_index=True,
        )
        _, thresholds, summary = evaluate_rules(daily, DATES, self.ages())
        frame = _summary_frame(summary, ENTITY_SELF)
        detail = _daily_frame(summary, ENTITY_SELF, DATES)
        self.assertEqual(frame.iloc[-1]["阈值样本状态"], "单日补充决策_昨日UV>200")
        observations = frame[
            frame["阈值样本状态"].eq("单日补充决策_昨日UV>200")
        ]
        self.assertEqual(observations.iloc[0]["广告id"], "latest-only-low")
        formal_actions = frame[
            ~frame["阈值样本状态"].eq("单日补充决策_昨日UV>200")
        ]["建议动作"].tolist()
        action_rank = {"关停创意": 0, "扩量": 1, "观察": 2}
        self.assertEqual(
            [action_rank[action] for action in formal_actions],
            sorted(action_rank[action] for action in formal_actions),
        )
        self.assertIn("日均总预估效率收入", frame.columns)
        neutral = frame[frame["动作"].eq("")]
        self.assertTrue(neutral["建议动作"].eq("观察").all())
        self.assertTrue(
            neutral["建议说明"].str.contains("当前无需关停或扩量").all()
        )
        self.assertIn("日均T0裂变率", frame.columns)
        self.assertNotIn("三日加权平均T0裂变率", _visible_columns(SUMMARY_SHEETS[ENTITY_SELF]))
        self.assertIn("三日均值ROI", frame.columns)
        self.assertIn("当日效率ROI", frame.columns)
        self.assertIn("预测总效率ROI", frame.columns)
        visible = _visible_columns(SUMMARY_SHEETS[ENTITY_SELF])
        self.assertNotIn("整体三日ROI排名百分位", visible)
        self.assertNotIn("是否位于三日ROI后20%", visible)
        self.assertNotIn("最新日首层UV", visible)
        self.assertNotIn("覆盖天数", visible)
        self.assertNotIn("审批选择", visible)
        self.assertEqual(visible[-1], "当前创意状态")
        for removed in ("动作", "动作原因", "阈值样本状态", "执行状态", "执行结果"):
            self.assertNotIn(removed, visible)
        self.assertEqual(
            visible[visible.index("三日均值ROI") : visible.index("建议说明") + 1],
            [
                "三日均值ROI",
                "当日效率ROI",
                "预测总效率ROI",
                "关停线",
                "扩量线（P80）",
                "建议动作",
                "建议说明",
            ],
        )
        self.assertIn("关停线", frame.columns)
        self.assertNotIn("关停线分位点", frame.columns)
        self.assertNotIn("关停线分位点", visible)
        self_threshold = thresholds.set_index("entity_type").loc[ENTITY_SELF]
        self.assertTrue(
            frame["扩量线（P80）"].eq(float(self_threshold["t_up"])).all()
        )
        gzh_frame = _summary_frame(summary, ENTITY_GZH)
        self.assertTrue(gzh_frame["扩量线（P80）"].isna().all())
        self.assertIn("是否低于三日关停线", frame.columns)
        self.assertNotIn("关停线（P25）", frame.columns)

        self.assertNotIn("整体实体等权P25关停线", frame.columns)
        self.assertEqual(set(detail["dt"]), set(DATES))
        self.assertTrue((detail.groupby("广告id").size() == 3).all())
        self.assertEqual(detail.columns[0], "dt")
        self.assertTrue(detail["dt"].astype(str).is_monotonic_decreasing)
        self.assertIn("当日效率ROI", detail.columns)
        self.assertIn("预测总效率ROI", detail.columns)

        with tempfile.TemporaryDirectory() as directory:
            output = Path(directory) / "roi.xlsx"
            write_workbook(
                summary,
                thresholds,
                DATES,
                output,
                {},
                pd.DataFrame(
                    [
                        {
                            "实体类型": "self",
                            "匹配层级": "miniapp_package_goal_exact",
                            "实体数": 1,
                            "渠道实体数": 1,
                            "匹配率": 1.0,
                            "参数版本": "test",
                        }
                    ]
                ),
            )
            workbook = load_workbook(output, read_only=False)
            expected = set(SUMMARY_SHEETS.values()) | set(DAILY_SHEETS.values())
            self.assertTrue(expected.issubset(workbook.sheetnames))
            self.assertNotIn("企微群合作", workbook.sheetnames)
            self.assertIn("传播裂变系数匹配", workbook.sheetnames)
            self.assertEqual(
                workbook["传播裂变系数匹配"].sheet_state,
                "hidden",
            )
            for sheet_name in expected:
                self.assertEqual(workbook[sheet_name].freeze_panes, "H2")
                expected_rules = 3 if sheet_name in SUMMARY_SHEETS.values() else 2
                self.assertEqual(
                    len(workbook[sheet_name].conditional_formatting),
                    expected_rules,
                )
            self.assertEqual(
                workbook[SUMMARY_SHEETS[ENTITY_SELF_AD]].sheet_state, "hidden"
            )
            self.assertEqual(
                workbook[DAILY_SHEETS[ENTITY_SELF_AD]].sheet_state, "hidden"
            )
            creative_sheet = workbook[SUMMARY_SHEETS[ENTITY_SELF]]
            color_rules = [
                rule
                for rules in creative_sheet.conditional_formatting._cf_rules.values()
                for rule in rules
            ]
            self.assertEqual(len(color_rules), 3)
            for rule in color_rules:
                self.assertEqual(rule.type, "colorScale")
                colors = [color.rgb[-6:] for color in rule.colorScale.color]
                self.assertEqual(colors, ["C00000", "FFEB84", "00B050"])
            creative_headers = {
                cell.value: cell.column for cell in creative_sheet[1]
            }
            first_scale_row = next(
                row_number
                for row_number in range(2, creative_sheet.max_row + 1)
                if creative_sheet.cell(
                    row_number, creative_headers["建议动作"]
                ).value
                == "扩量"
            )
            self.assertEqual(
                creative_sheet.cell(first_scale_row, 1).border.top.style,
                "medium",
            )
            self.assertEqual(
                creative_sheet.cell(
                    first_scale_row, creative_headers["建议动作"]
                ).border.top.style,
                "medium",
            )
            first_observe_row = next(
                row_number
                for row_number in range(first_scale_row + 1, creative_sheet.max_row + 1)
                if creative_sheet.cell(
                    row_number, creative_headers["建议动作"]
                ).value
                == "观察"
            )
            self.assertEqual(
                creative_sheet.cell(first_observe_row, 1).border.top.style,
                "medium",
            )
            self.assertEqual(
                creative_sheet.cell(
                    first_observe_row, creative_headers["建议动作"]
                ).border.top.style,
                "medium",
            )
            self.assertEqual(
                creative_sheet.cell(2, creative_headers["日均首层UV"]).number_format,
                "0",
            )
            self.assertEqual(
                creative_sheet.cell(
                    2, creative_headers["日均T0裂变人数"]
                ).number_format,
                "0",
            )
            self.assertEqual(
                creative_sheet.cell(2, creative_headers["广告age"]).number_format,
                "0",
            )
            self.assertEqual(
                creative_sheet.cell(
                    2, creative_headers["三日均值ROI"]
                ).number_format,
                "0.00",
            )
            self.assertEqual(
                creative_sheet.cell(
                    2, creative_headers["当日效率ROI"]
                ).number_format,
                "0.00",
            )
            self.assertEqual(
                creative_sheet.cell(
                    2, creative_headers["预测总效率ROI"]
                ).number_format,
                "0.00",
            )
            for header in (
                "裂变系数-总裂变UV/T0裂变UV",
                "裂变系数-总裂变UV/首层UV",
            ):
                self.assertEqual(
                    creative_sheet.cell(
                        2, creative_headers[header]
                    ).number_format,
                    "0.00",
                )
            self.assertEqual(
                creative_sheet.cell(
                    2, creative_headers["扩量线（P80）"]
                ).number_format,
                "0.00",
            )
            run_summary = workbook["运行摘要"]
            summary_rows = {
                run_summary.cell(row_number, 1).value: row_number
                for row_number in range(1, run_summary.max_row + 1)
            }
            self.assertEqual(
                run_summary.cell(summary_rows["创意扩量线（P80）"], 2).number_format,
                "0.00",
            )
            self.assertEqual(
                run_summary.cell(summary_rows["扩量样本数"], 2).number_format,
                "0",
            )
            self.assertEqual(
                run_summary.cell(
                    summary_rows["小程序单日实际关停线"], 2
                ).number_format,
                "0.00",
            )
            self.assertEqual(
                run_summary.cell(
                    summary_rows["小程序实际关停成本占比"], 2
                ).number_format,
                "0.00%",
            )
            ad_sheet = workbook[SUMMARY_SHEETS[ENTITY_SELF_AD]]
            ad_headers = [cell.value for cell in ad_sheet[1]]
            self.assertIn("审批选择", ad_headers)
            approval_column = ad_headers.index("审批选择") + 1
            self.assertTrue(
                ad_sheet.column_dimensions[
                    ad_sheet.cell(1, approval_column).column_letter
                ].hidden
            )
            self.assertGreater(len(ad_sheet.data_validations.dataValidation), 0)

    def test_summary_latest_actual_roi_and_agency_score_use_latest_day(self):
        rows = pd.DataFrame(
            [
                row(ENTITY_SELF, "score-window", dt, roi)
                for dt, roi in zip(DATES, [1.0, 2.0, 3.0])
            ]
        )
        rows.loc[rows["dt"].eq(DATES[-1]), "裂变效率收入"] = 100.0
        ages = pd.DataFrame({"广告id": ["score-window"], "广告age": [10]})
        _, _, summary = evaluate_rules(rows, DATES, ages)

        internal = _summary_frame(summary, ENTITY_SELF).iloc[0]
        agency = _agency_summary_frame(summary, ENTITY_SELF).iloc[0]
        snapshot = summary.iloc[0]
        latest_actual = float(snapshot[f"实际ROI_{DATES[-1]}"])
        latest_predicted = float(snapshot[f"ROI_{DATES[-1]}"])

        self.assertNotAlmostEqual(latest_actual, latest_predicted)
        self.assertAlmostEqual(float(internal["三日均值ROI"]), float(snapshot["实际ROI"]))
        self.assertAlmostEqual(float(internal["当日效率ROI"]), latest_actual)
        self.assertAlmostEqual(float(internal["预测总效率ROI"]), float(snapshot["ROI"]))
        self.assertAlmostEqual(float(agency["评分"]), latest_actual)


if __name__ == "__main__":
    unittest.main()