Agent/examples/auto_put_ad_mini/test_roi_control_policy.py

import os
import unittest
from datetime import date, datetime
from decimal import Decimal
from unittest.mock import Mock, patch

import pandas as pd

from operator_commands import ACTION_CONFIRM, ACTION_REJECT, parse_command
from roi_control.config import RoiConfig
from roi_control.execution import _execute_pause_ad, plan_scale_bid
from roi_control.metrics import ENTITY_GZH, ENTITY_SELF, ENTITY_SELF_AD
from roi_control.policy import (
    ACTION_PAUSE_AD,
    ACTION_PAUSE_CREATIVE,
    ACTION_SCALE_BID,
    MODE_ACTIONABLE,
    MODE_NOTIFY_ONLY,
    annotate_execution,
)
from roi_control.repository import _is_expired
from roi_control.sheet_approval import RoiSheetClient, parse_approval_rows
from tencent_client import ACTIVE_STATUS, SUSPEND_STATUS


class RoiControlPolicyTest(unittest.TestCase):
    def test_mysql_naive_expiry_compares_with_aware_service_time(self):
        expires_at = datetime(2026, 7, 28, 12, 0)
        now = datetime.fromisoformat("2026-07-28T12:01:00+08:00")
        self.assertTrue(_is_expired(expires_at, now))

    def test_actionability_is_separate_from_metric_recommendation(self):
        summary = pd.DataFrame(
            [
                {
                    "entity_type": ENTITY_SELF,
                    "channel": "小程序投流-稳定",
                    "账号id": "84502354",
                    "广告id": "1001",
                    "创意id": "2001",
                    "动作": "关停",
                    "动作原因": "低ROI",
                },
                {
                    "entity_type": ENTITY_SELF,
                    "channel": "小程序投流-稳定",
                    "账号id": "84502354",
                    "广告id": "1001",
                    "创意id": "2002",
                    "动作": "扩量",
                    "动作原因": "高ROI",
                },
                {
                    "entity_type": ENTITY_SELF,
                    "channel": "小程序投流-稳定",
                    "账号id": "99999999",
                    "广告id": "3001",
                    "创意id": "4001",
                    "动作": "关停",
                    "动作原因": "低ROI",
                },
                {
                    "entity_type": ENTITY_GZH,
                    "channel": "公众号合作-即转-稳定",
                    "合作方名": "合作方",
                    "公众号名": "公众号",
                    "动作": "扩量",
                    "动作原因": "高ROI",
                },
                {
                    "entity_type": ENTITY_SELF_AD,
                    "channel": "小程序投流-稳定",
                    "账号id": "84502354",
                    "广告id": "5001",
                    "创意id": "",
                    "动作": "关停",
                    "动作原因": "广告级低ROI",
                },
            ]
        )
        annotated, snapshots, actions = annotate_execution(
            summary,
            {84502354},
            "roi_20260725_north_star_roi_v1_roi_policy_v1",
        )

        self.assertEqual(len(snapshots), 5)
        self.assertEqual(
            {action["action_type"] for action in actions},
            {ACTION_PAUSE_CREATIVE, ACTION_PAUSE_AD, ACTION_SCALE_BID},
        )
        self.assertEqual(
            annotated.iloc[0]["执行模式"],
            MODE_ACTIONABLE,
        )
        self.assertEqual(annotated.iloc[2]["执行模式"], MODE_NOTIFY_ONLY)
        self.assertEqual(annotated.iloc[3]["执行模式"], MODE_NOTIFY_ONLY)
        self.assertEqual(annotated.iloc[0]["审批选择"], "")
        self.assertEqual(annotated.iloc[0]["执行状态"], "待审批")
        self.assertEqual(
            annotated.iloc[0]["动作幂等键"], actions[0]["idempotency_key"]
        )
        self.assertEqual(annotated.iloc[2]["审批选择"], "不可执行")
        self.assertEqual(annotated.iloc[4]["执行模式"], MODE_ACTIONABLE)
        self.assertEqual(annotated.iloc[4]["审批选择"], "")
        ad_action = next(
            action for action in actions if action["action_type"] == ACTION_PAUSE_AD
        )
        self.assertIsNone(ad_action["dynamic_creative_id"])
        self.assertTrue(ad_action["idempotency_key"].endswith(":5001"))

    def test_sheet_approval_parser_uses_hidden_idempotency_key(self):
        rows = [
            ["账号id", "审批选择", "动作幂等键", "执行状态", "执行结果"],
            ["99999999", "批准", "roi_x:PAUSE_CREATIVE:1:2", "待审批", ""],
            ["1", "拒绝", "roi_x:PAUSE_CREATIVE:1:3", "待审批", ""],
            ["1", "随便填", "roi_x:PAUSE_CREATIVE:1:4", "待审批", ""],
        ]
        decisions = parse_approval_rows(rows)
        self.assertEqual(
            [(row["decision"], row["idempotency_key"]) for row in decisions],
            [
                ("APPROVED", "roi_x:PAUSE_CREATIVE:1:2"),
                ("REJECTED", "roi_x:PAUSE_CREATIVE:1:3"),
            ],
        )

    def test_sheet_client_reads_both_current_approval_sheets(self):
        client = RoiSheetClient.__new__(RoiSheetClient)
        client._token = Mock(return_value="token")
        client._sheet_ids = Mock(
            return_value=[
                ("小程序创意级三日汇总", "creative-sheet"),
                ("小程序广告级三日汇总", "ad-sheet"),
            ]
        )
        client._read_sheet_approvals = Mock(
            side_effect=[
                [{"sheet_id": "creative-sheet", "idempotency_key": "creative"}],
                [{"sheet_id": "ad-sheet", "idempotency_key": "ad"}],
            ]
        )
        decisions = client.read_approvals("spreadsheet")
        self.assertEqual(
            [decision["sheet_id"] for decision in decisions],
            ["creative-sheet", "ad-sheet"],
        )

    def test_pause_ad_marks_prepared_then_success_after_readback(self):
        now = datetime.fromisoformat("2026-08-03T12:00:00+08:00")

        class FakeTencent:
            def get_ad(self, account_id, adgroup_id):
                self.get_args = (account_id, adgroup_id)
                return {
                    "adgroup_id": adgroup_id,
                    "configured_status": ACTIVE_STATUS,
                }

            def update_ad(self, account_id, adgroup_id, **kwargs):
                self.update_args = (account_id, adgroup_id, kwargs)
                return {
                    "adgroup_id": adgroup_id,
                    "configured_status": SUSPEND_STATUS,
                }

        client = FakeTencent()
        item = {
            "id": 99,
            "account_id": 84502354,
            "adgroup_id": 5001,
            "execution_status": "PENDING",
        }
        with patch("roi_control.execution.update_action_item") as update_item:
            _execute_pause_ad(item, client=client, now=now)
        self.assertEqual(client.get_args, (84502354, 5001))
        self.assertEqual(
            client.update_args,
            (84502354, 5001, {"target_status": SUSPEND_STATUS}),
        )
        self.assertEqual(update_item.call_count, 2)
        self.assertEqual(
            update_item.call_args_list[0].kwargs["execution_status"], "PREPARED"
        )
        self.assertEqual(
            update_item.call_args_list[1].kwargs["execution_status"], "SUCCESS"
        )
        self.assertEqual(
            update_item.call_args_list[1].kwargs["readback_status"], SUSPEND_STATUS
        )

    def test_scale_actions_are_deduplicated_by_ad(self):
        rows = [
            {
                "entity_type": ENTITY_SELF,
                "channel": "小程序投流-稳定",
                "账号id": "84502354",
                "广告id": "1001",
                "创意id": str(2000 + index),
                "动作": "扩量",
                "动作原因": "高ROI",
            }
            for index in range(3)
        ]
        _, _, actions = annotate_execution(
            pd.DataFrame(rows),
            {84502354},
            "roi_20260725_north_star_roi_v1_roi_policy_v1",
        )
        self.assertEqual(len(actions), 1)
        self.assertEqual(actions[0]["action_type"], ACTION_SCALE_BID)

    def test_observe_recommendation_never_creates_tencent_action(self):
        summary = pd.DataFrame(
            [
                {
                    "entity_type": ENTITY_SELF,
                    "channel": "小程序投流-稳定",
                    "账号id": "84502354",
                    "广告id": "1001",
                    "创意id": "2001",
                    "动作": "观察",
                    "动作原因": "连续两天表现不一致",
                }
            ]
        )
        annotated, _, actions = annotate_execution(
            summary,
            {84502354},
            "roi_20260730_m6_p3",
        )
        self.assertEqual(actions, [])
        self.assertEqual(annotated.iloc[0]["执行模式"], MODE_NOTIFY_ONLY)
        self.assertEqual(
            annotated.iloc[0]["执行说明"],
            "观察行仅展示，不执行腾讯写操作",
        )

    def test_invalid_apply_without_daily_job_is_rejected(self):
        environment = {
            "DAILY_ROI_ENABLED": "0",
            "ROI_APPLY_ENABLED": "1",
        }
        with patch.dict(os.environ, environment, clear=False):
            with self.assertRaisesRegex(ValueError, "requires DAILY_ROI_ENABLED"):
                RoiConfig.from_env()

    def test_sheet_approval_requires_roi_writes(self):
        environment = {
            "DAILY_ROI_ENABLED": "1",
            "ROI_APPLY_ENABLED": "0",
            "ROI_SHEET_APPROVAL_ENABLED": "1",
        }
        with patch.dict(os.environ, environment, clear=False):
            with self.assertRaisesRegex(ValueError, "requires ROI_APPLY_ENABLED"):
                RoiConfig.from_env()

    def test_roi_batch_commands_are_deterministic(self):
        confirm = parse_command("确认 roi_20260725_north_star_roi_v1")
        reject = parse_command("拒绝 roi_20260725_north_star_roi_v1")
        old_confirm = parse_command("确认 cmd_20260725120000_abcd1234")
        self.assertEqual(confirm.action, ACTION_CONFIRM)
        self.assertEqual(reject.action, ACTION_REJECT)
        self.assertEqual(confirm.command_id, "roi_20260725_north_star_roi_v1")
        self.assertEqual(old_confirm.action, ACTION_CONFIRM)

    def test_restored_intraday_boost_is_not_reapplied_by_roi(self):
        plan = plan_scale_bid(
            current_bid_fen=100,
            base_bid_fen=100,
            initial_base_bid_fen=100,
            boosted_date=date(2026, 7, 26),
            control_date=date(2026, 7, 26),
            intraday_ratio=Decimal("1.25"),
            scale_ratio=Decimal("1.10"),
            max_base_ratio=Decimal("2.00"),
        )
        self.assertTrue(plan["allowed"])
        self.assertEqual(plan["new_base_bid_fen"], 110)
        self.assertEqual(plan["target_bid_fen"], 110)
        self.assertTrue(plan["boosted_date_today"])
        self.assertFalse(plan["preserve_intraday_boost"])

    def test_active_intraday_boost_is_preserved_relative_to_new_base(self):
        plan = plan_scale_bid(
            current_bid_fen=125,
            base_bid_fen=100,
            initial_base_bid_fen=100,
            boosted_date=date(2026, 7, 26),
            control_date=date(2026, 7, 26),
            intraday_ratio=Decimal("1.25"),
            scale_ratio=Decimal("1.10"),
            max_base_ratio=Decimal("2.00"),
        )
        self.assertEqual(plan["target_bid_fen"], 138)
        self.assertTrue(plan["preserve_intraday_boost"])


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