Agent/examples/auto_put_ad_mini/roi_control/config.py

"""Configuration for daily ROI computation and approved execution."""

from __future__ import annotations

import hashlib
import json
import math
import os
from dataclasses import asdict, dataclass
from decimal import Decimal
from typing import Mapping


def env_flag(name: str, default: bool = False) -> bool:
    raw = os.getenv(name)
    if raw is None:
        return default
    return raw.strip().lower() in {"1", "true", "yes", "on"}


AGENCY_WEBHOOK_PREFIX = "https://open.feishu.cn/open-apis/bot/v2/hook/"


@dataclass(frozen=True)
class AgencyWebhookConfig:
    """Secret-backed routing for per-agency ROI report notifications."""

    enabled: bool = False
    webhooks: Mapping[str, str] | None = None

    @classmethod
    def from_env(cls) -> "AgencyWebhookConfig":
        enabled = env_flag("ROI_AGENCY_WEBHOOK_ENABLED")
        if not enabled:
            return cls(enabled=False, webhooks={})
        raw = os.getenv("ROI_AGENCY_WEBHOOKS_JSON", "").strip()
        if not raw:
            raise ValueError(
                "ROI_AGENCY_WEBHOOK_ENABLED=1 requires ROI_AGENCY_WEBHOOKS_JSON"
            )
        try:
            values = json.loads(raw)
        except json.JSONDecodeError as exc:
            raise ValueError("ROI_AGENCY_WEBHOOKS_JSON must be valid JSON") from exc
        if not isinstance(values, dict):
            raise ValueError("ROI_AGENCY_WEBHOOKS_JSON must be a JSON object")
        webhooks: dict[str, str] = {}
        for raw_name, raw_url in values.items():
            name = str(raw_name).strip()
            url = str(raw_url).strip()
            if not name:
                raise ValueError("ROI agency webhook name cannot be empty")
            if not url.startswith(AGENCY_WEBHOOK_PREFIX):
                raise ValueError(f"Invalid Feishu webhook for agency: {name}")
            if name in webhooks:
                raise ValueError(f"Duplicate ROI agency webhook: {name}")
            webhooks[name] = url
        return cls(enabled=True, webhooks=webhooks)

    def snapshot(self) -> dict[str, object]:
        routes = self.webhooks or {}
        return {
            "enabled": self.enabled,
            "agencies": sorted(routes),
            "route_fingerprints": {
                name: hashlib.sha256(url.encode("utf-8")).hexdigest()[:12]
                for name, url in sorted(routes.items())
            },
        }


@dataclass(frozen=True)
class RoiConfig:
    daily_enabled: bool = False
    apply_enabled: bool = False
    sheet_approval_enabled: bool = False
    sheet_approval_poll_seconds: int = 60
    report_hour: int = 9
    report_minute: int = 0
    internal_test_enabled: bool = False
    internal_test_hour: int = 15
    internal_test_minute: int = 30
    internal_test_dedup_enabled: bool = True
    approval_ttl_minutes: int = 120
    scale_ratio: Decimal = Decimal("1.10")
    scale_cooldown_days: int = 3
    max_base_ratio: Decimal = Decimal("2.00")
    self_stop_min_age: int = 4
    self_up_min_age: int = 3
    self_min_daily_uv: float = 200
    partner_min_daily_uv: float = 200
    observe_min_latest_uv: float = 200
    one_day_min_uv: float = 200
    one_day_p30_min_uv: float = 500
    one_day_hard_stop_roi: float = 0.20
    one_day_stop_quantile: float = 0.30
    stop_quantile: float = 0.20
    up_quantile: float = 0.80
    stop_weight_cap_quantile: float = 0.95

    @classmethod
    def from_env(cls) -> "RoiConfig":
        config = cls(
            daily_enabled=env_flag("DAILY_ROI_ENABLED"),
            apply_enabled=env_flag("ROI_APPLY_ENABLED"),
            sheet_approval_enabled=env_flag("ROI_SHEET_APPROVAL_ENABLED"),
            sheet_approval_poll_seconds=int(
                os.getenv("ROI_SHEET_APPROVAL_POLL_SECONDS", "60")
            ),
            report_hour=int(os.getenv("DAILY_ROI_HOUR", "9")),
            report_minute=int(os.getenv("DAILY_ROI_MINUTE", "0")),
            internal_test_enabled=env_flag("ROI_INTERNAL_TEST_ENABLED"),
            internal_test_hour=int(os.getenv("ROI_INTERNAL_TEST_HOUR", "15")),
            internal_test_minute=int(os.getenv("ROI_INTERNAL_TEST_MINUTE", "30")),
            internal_test_dedup_enabled=env_flag(
                "ROI_INTERNAL_TEST_DEDUP_ENABLED",
                True,
            ),
            approval_ttl_minutes=int(
                os.getenv("ROI_APPROVAL_TTL_MINUTES", "120")
            ),
            scale_ratio=Decimal(os.getenv("ROI_SCALE_RATIO", "1.10")),
            scale_cooldown_days=int(os.getenv("ROI_SCALE_COOLDOWN_DAYS", "3")),
            max_base_ratio=Decimal(os.getenv("ROI_MAX_BASE_RATIO", "2.00")),
            self_stop_min_age=int(os.getenv("ROI_SELF_STOP_MIN_AGE", "4")),
            self_up_min_age=int(os.getenv("ROI_SELF_UP_MIN_AGE", "3")),
            self_min_daily_uv=float(
                os.getenv(
                    "ROI_SELF_MIN_DAILY_UV",
                    os.getenv("ROI_SELF_MIN_AVG_UV", "200"),
                )
            ),
            partner_min_daily_uv=float(
                os.getenv(
                    "ROI_PARTNER_MIN_DAILY_UV",
                    os.getenv("ROI_PARTNER_MIN_AVG_UV", "200"),
                )
            ),
            observe_min_latest_uv=float(
                os.getenv("ROI_OBSERVE_MIN_LATEST_UV", "200")
            ),
            one_day_min_uv=float(os.getenv("ROI_ONE_DAY_MIN_UV", "200")),
            one_day_p30_min_uv=float(
                os.getenv("ROI_ONE_DAY_P30_MIN_UV", "500")
            ),
            one_day_hard_stop_roi=float(
                os.getenv("ROI_ONE_DAY_HARD_STOP_ROI", "0.20")
            ),
            one_day_stop_quantile=float(
                os.getenv("ROI_ONE_DAY_STOP_QUANTILE", "0.30")
            ),
            stop_quantile=float(os.getenv("ROI_STOP_QUANTILE", "0.20")),
            up_quantile=float(os.getenv("ROI_UP_QUANTILE", "0.80")),
            stop_weight_cap_quantile=float(
                os.getenv("ROI_STOP_WEIGHT_CAP_QUANTILE", "0.95")
            ),
        )
        config.validate()
        return config

    def validate(self) -> None:
        if self.apply_enabled and not self.daily_enabled:
            raise ValueError(
                "ROI_APPLY_ENABLED=1 requires DAILY_ROI_ENABLED=1"
            )
        if self.sheet_approval_enabled and not self.apply_enabled:
            raise ValueError(
                "ROI_SHEET_APPROVAL_ENABLED=1 requires ROI_APPLY_ENABLED=1"
            )
        if self.sheet_approval_poll_seconds < 30:
            raise ValueError("ROI_SHEET_APPROVAL_POLL_SECONDS must be at least 30")
        if not 0 <= self.report_hour <= 23 or not 0 <= self.report_minute <= 59:
            raise ValueError("DAILY_ROI_HOUR/MINUTE is invalid")
        if not 0 <= self.internal_test_hour <= 23 or not 0 <= self.internal_test_minute <= 59:
            raise ValueError("ROI_INTERNAL_TEST_HOUR/MINUTE is invalid")
        if self.approval_ttl_minutes < 1:
            raise ValueError("ROI_APPROVAL_TTL_MINUTES must be positive")
        if self.scale_ratio <= 1:
            raise ValueError("ROI_SCALE_RATIO must be greater than 1")
        if self.max_base_ratio < self.scale_ratio:
            raise ValueError("ROI_MAX_BASE_RATIO must cover ROI_SCALE_RATIO")
        if self.scale_cooldown_days < 1:
            raise ValueError("ROI_SCALE_COOLDOWN_DAYS must be positive")
        if min(
            self.self_min_daily_uv,
            self.partner_min_daily_uv,
            self.observe_min_latest_uv,
            self.one_day_min_uv,
            self.one_day_p30_min_uv,
        ) <= 0:
            raise ValueError("ROI UV thresholds must be positive")
        if self.one_day_p30_min_uv <= self.one_day_min_uv:
            raise ValueError(
                "ROI_ONE_DAY_P30_MIN_UV must be greater than ROI_ONE_DAY_MIN_UV"
            )
        if not math.isfinite(self.one_day_hard_stop_roi):
            raise ValueError("ROI_ONE_DAY_HARD_STOP_ROI must be finite")
        if not 0 < self.one_day_stop_quantile < 1:
            raise ValueError(
                "ROI_ONE_DAY_STOP_QUANTILE must satisfy 0 < value < 1"
            )
        if not 0 < self.stop_quantile < 1:
            raise ValueError("ROI_STOP_QUANTILE must satisfy 0 < value < 1")
        if not self.stop_quantile < self.up_quantile < 1:
            raise ValueError(
                "ROI_UP_QUANTILE must satisfy ROI_STOP_QUANTILE < value < 1"
            )
        if not 0 < self.stop_weight_cap_quantile <= 1:
            raise ValueError(
                "ROI_STOP_WEIGHT_CAP_QUANTILE must satisfy 0 < value <= 1"
            )

    def snapshot(self) -> dict[str, object]:
        values = asdict(self)
        values["scale_ratio"] = str(self.scale_ratio)
        values["max_base_ratio"] = str(self.max_base_ratio)
        return values