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,
    _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("p12", run_id)
        self.assertIn("r36", 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_global_p20_and_creative_only_p80_scale_actions(self):
        candidates, thresholds, summary = evaluate_rules(
            self.build_daily(), DATES, self.ages()
        )
        self.assertEqual(len(thresholds), 1)
        self.assertAlmostEqual(float(thresholds.iloc[0]["t_stop"]), 0.7)
        self.assertAlmostEqual(float(thresholds.iloc[0]["t_up"]), 3.4)
        self.assertEqual(int(thresholds.iloc[0]["阈值样本数"]), 8)
        self.assertEqual(int(thresholds.iloc[0]["扩量样本数"]), 4)
        self.assertEqual(thresholds.iloc[0]["关停线口径"], "合格实体等权P20")
        self.assertEqual(thresholds.iloc[0]["扩量线口径"], "合格小程序创意实体等权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_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):
        _, thresholds, summary = evaluate_rules(self.build_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.assertTrue(ad_rows["t_stop"].eq(float(thresholds.iloc[0]["t_stop"])).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_supplement_requires_ad_age_over_three_days(self):
        daily = self.build_daily()
        supplemental = [
            row(ENTITY_SELF, "one-day-hard", DATES[-1], 0.2, uv=300),
            row(ENTITY_SELF, "one-day-p30", DATES[-1], 0.5, uv=600),
            row(ENTITY_SELF, "one-day-young", DATES[-1], 0.6, uv=600),
            row(ENTITY_SELF, "one-day-mid", DATES[-1], 8.0, uv=600),
            row(ENTITY_SELF, "one-day-high", DATES[-1], 9.0, uv=600),
        ]
        daily = pd.concat([daily, pd.DataFrame(supplemental)], ignore_index=True)
        ages = pd.concat(
            [
                self.ages(),
                pd.DataFrame(
                    {
                        "广告id": [
                            "one-day-hard",
                            "one-day-p30",
                            "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.assertEqual(int(thresholds.iloc[0]["单日P30样本数"]), 8)
        self.assertAlmostEqual(float(thresholds.iloc[0]["t_one_day_stop"]), 0.64)
        targets = summary.set_index("广告id")
        self.assertEqual(targets.loc["one-day-hard", "动作"], "关停")
        self.assertIn("单日硬关停线", targets.loc["one-day-hard", "动作原因"])
        self.assertIn("启用单日补充规则", targets.loc["one-day-hard", "动作原因"])
        self.assertIn("最新日预测总效率ROI", targets.loc["one-day-hard", "动作原因"])
        self.assertIn("三日预测总效率ROI", targets.loc["one-day-hard", "动作原因"])
        self.assertIn("不参与本次单日判断", targets.loc["one-day-hard", "动作原因"])
        self.assertIn("广告age=4>3天", targets.loc["one-day-hard", "动作原因"])
        self.assertEqual(targets.loc["one-day-p30", "动作"], "关停")
        self.assertIn("单日实体等权P30", targets.loc["one-day-p30", "动作原因"])
        self.assertEqual(targets.loc["one-day-young", "动作"], "观察")
        self.assertIn("广告age=3≤3天", targets.loc["one-day-young", "动作原因"])
        self.assertEqual(targets.loc["one-day-mid", "动作"], "观察")
        self.assertIn("未命中单日关停规则", targets.loc["one-day-mid", "动作原因"])
        creative_frame = _summary_frame(summary, ENTITY_SELF)
        creative_stops = creative_frame[
            creative_frame["建议动作"].eq("关停创意")
        ]
        self.assertEqual(
            creative_stops["广告id"].tolist(),
            ["creative-ad-0", "one-day-hard", "one-day-p30"],
        )
        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_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)
        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",
                "关停线（P20）",
                "扩量线（P80）",
                "建议动作",
                "建议说明",
            ],
        )
        self.assertIn("关停线（P20）", frame.columns)
        self.assertTrue(frame["扩量线（P80）"].eq(float(thresholds.iloc[0]["t_up"])).all())
        gzh_frame = _summary_frame(summary, ENTITY_GZH)
        self.assertTrue(gzh_frame["扩量线（P80）"].isna().all())
        self.assertIn("是否位于三日ROI后20%", 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")
                self.assertEqual(len(workbook[sheet_name].conditional_formatting), 2)
            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), 2)
            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",
            )
            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["单日关停线（P30）"], 2).number_format,
                "0.00",
            )
            self.assertEqual(
                run_summary.cell(summary_rows["单日P30样本数"], 2).number_format,
                "0",
            )
            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)


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