Agent/examples/auto_put_ad_mini/roi_control/config.py

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

from __future__ import annotations

import os
from dataclasses import asdict, dataclass
from decimal import Decimal


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"}


@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 = 11
    report_minute: int = 0
    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 = 5
    self_up_min_age: int = 3
    self_min_avg_uv: float = 200
    partner_min_avg_uv: float = 200
    min_daily_cost: float = 100
    stop_quantile: float = 0.20
    up_quantile: float = 0.80
    gzh_adjust_rank_min: float = 0.10
    gzh_adjust_rank_max: float = 0.30

    @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", "11")),
            report_minute=int(os.getenv("DAILY_ROI_MINUTE", "0")),
            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", "5")),
            self_up_min_age=int(os.getenv("ROI_SELF_UP_MIN_AGE", "3")),
            self_min_avg_uv=float(os.getenv("ROI_SELF_MIN_AVG_UV", "200")),
            partner_min_avg_uv=float(
                os.getenv("ROI_PARTNER_MIN_AVG_UV", "200")
            ),
            min_daily_cost=float(os.getenv("ROI_MIN_DAILY_COST", "100")),
            stop_quantile=float(os.getenv("ROI_STOP_QUANTILE", "0.20")),
            up_quantile=float(os.getenv("ROI_UP_QUANTILE", "0.80")),
            gzh_adjust_rank_min=float(
                os.getenv("ROI_GZH_ADJUST_RANK_MIN", "0.10")
            ),
            gzh_adjust_rank_max=float(
                os.getenv("ROI_GZH_ADJUST_RANK_MAX", "0.30")
            ),
        )
        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 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 not 0 < self.stop_quantile < self.up_quantile < 1:
            raise ValueError("ROI quantiles must satisfy 0 < stop < up < 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