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 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 plan_scale_bid
from roi_control.metrics import ENTITY_GZH, ENTITY_SELF
from roi_control.policy import (
    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 parse_approval_rows


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",
                },
            ]
        )
        annotated, snapshots, actions = annotate_execution(
            summary,
            {84502354},
            "roi_20260725_north_star_roi_v1_roi_policy_v1",
        )

        self.assertEqual(len(snapshots), 4)
        self.assertEqual(
            {action["action_type"] for action in actions},
            {ACTION_PAUSE_CREATIVE, 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]["审批选择"], "不可执行")

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