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