Agent/examples/tencent_realtime_control/realtime_config.py

"""Runtime configuration for Tencent real-time control."""

from __future__ import annotations

import os
from dataclasses import dataclass
from decimal import Decimal


@dataclass(frozen=True)
class RealtimeControlConfig:
    start_hour: int = 12
    stop_hour: int = 21
    poll_seconds: int = 600
    high_cpm: Decimal = Decimal("250")
    low_cpm: Decimal = Decimal("190")
    bid_up_ratio: Decimal = Decimal("1.05")
    inventory_refresh_minutes: int = 60
    max_partition_lag_hours: int = 2
    lock_name: str = "tencent_realtime_control"
    next_delivery_hour: int = 6

    @classmethod
    def from_env(cls) -> "RealtimeControlConfig":
        config = cls(
            start_hour=int(os.getenv("RTC_START_HOUR", "12")),
            stop_hour=int(os.getenv("RTC_STOP_HOUR", "21")),
            poll_seconds=int(os.getenv("RTC_POLL_SECONDS", "600")),
            high_cpm=Decimal(os.getenv("RTC_HIGH_CPM", "250")),
            low_cpm=Decimal(os.getenv("RTC_LOW_CPM", "190")),
            bid_up_ratio=Decimal(os.getenv("RTC_BID_UP_RATIO", "1.05")),
            inventory_refresh_minutes=int(
                os.getenv("RTC_INVENTORY_REFRESH_MINUTES", "60")
            ),
            max_partition_lag_hours=int(
                os.getenv("RTC_MAX_PARTITION_LAG_HOURS", "2")
            ),
            lock_name=os.getenv(
                "RTC_DB_LOCK_NAME", "tencent_realtime_control"
            ).strip(),
            next_delivery_hour=int(
                os.getenv("RTC_COMMAND_NEXT_DELIVERY_HOUR", "6")
            ),
        )
        if not 0 <= config.start_hour < config.stop_hour <= 23:
            raise ValueError("RTC hours must satisfy 0 <= start < stop <= 23")
        if config.poll_seconds < 60:
            raise ValueError("RTC_POLL_SECONDS must be at least 60")
        if config.low_cpm >= config.high_cpm:
            raise ValueError("RTC_LOW_CPM must be lower than RTC_HIGH_CPM")
        if config.bid_up_ratio <= 1:
            raise ValueError("RTC_BID_UP_RATIO must be greater than 1")
        if config.inventory_refresh_minutes < 5:
            raise ValueError("RTC_INVENTORY_REFRESH_MINUTES must be at least 5")
        if config.max_partition_lag_hours < 1:
            raise ValueError("RTC_MAX_PARTITION_LAG_HOURS must be at least 1")
        if not config.lock_name:
            raise ValueError("RTC_DB_LOCK_NAME must not be empty")
        if not 0 <= config.next_delivery_hour <= 23:
            raise ValueError("RTC_COMMAND_NEXT_DELIVERY_HOUR must be in [0, 23]")
        return config