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()