config_loader.py - master - keorlov/collsim

config_loader.py
165 строк5 KB
Konstantin Orlov
wip
14 июл 2026, 13:09
14 июл 2026, 13:09688f811
"""Загрузчик конфигурации из YAML с использованием dataclasses."""

from dataclasses import dataclass, asdict
from typing import Optional, Any
import yaml


@dataclass(slots=True)
class SimulationConfig:
    """Параметры симуляции."""
    days: int
    optimization_mode: str
    gini: float
    cure_recovery_rate: float
    hard_cure_rate: float = 0.002


@dataclass(slots=True)
class PortfolioConfig:
    """Параметры портфеля."""
    initial_clients: int
    daily_growth_rate: float


@dataclass(slots=True)
class ClientTypeConfig:
    """Параметры типа клиента."""
    share: float
    hazard_multiplier: float


@dataclass(slots=True)
class ProductConfig:
    """Параметры продукта."""
    share: float
    avg_balance: float
    term_days: Optional[int]
    hazard_mult: float
    daily_interest_rate: float = 0.0004
    nim_annual_rate: float = 0.10
    daily_close_prob: float = 0.0


@dataclass(slots=True)
class HazardConfig:
    """Параметры хазарда."""
    base_daily_rate: float


@dataclass(slots=True)
class ChannelConfig:
    """Параметры канала связи."""
    cost: int
    p_base: float


@dataclass(slots=True)
class BudgetsConfig:
    """Бюджеты."""
    daily_sms: int
    daily_call: int


@dataclass(slots=True)
class LegalLimitsConfig:
    """Лимиты 230-ФЗ."""
    max_calls_per_week: int
    max_sms_per_week: int


@dataclass(slots=True)
class HardCollectionConfig:
    """Параметры жесткого взыскания."""
    cession_dpd: int
    cession_rate: float


@dataclass(slots=True)
class InitialStateConfig:
    """Параметры начального среза портфеля (тёплый старт)."""
    soft_share: float = 0.038
    hard_share: float = 0.048
    hard_dpd_max: int = 350


@dataclass(slots=True)
class Config:
    """Полная конфигурация симуляции."""
    simulation: SimulationConfig
    portfolio: PortfolioConfig
    client_types: dict[str, ClientTypeConfig]
    products: dict[str, ProductConfig]
    hazard: HazardConfig
    channels: dict[str, ChannelConfig]
    budgets: BudgetsConfig
    legal_limits: LegalLimitsConfig
    hard_collection: HardCollectionConfig
    initial_state: InitialStateConfig


def load_config(path: str = "config.yaml") -> Config:
    """
    Загружает конфигурацию из YAML файла в dataclasses.
    
    Args:
        path: Путь к YAML файлу конфигурации.
        
    Returns:
        Config: Заполненный объект конфигурации.
    """
    with open(path, "r", encoding="utf-8") as f:
        data = yaml.safe_load(f)
    
    return Config(
        simulation=SimulationConfig(**data["simulation"]),
        portfolio=PortfolioConfig(**data["portfolio"]),
        client_types={k: ClientTypeConfig(**v) for k, v in data["client_types"].items()},
        products={k: ProductConfig(**v) for k, v in data["products"].items()},
        hazard=HazardConfig(**data["hazard"]),
        channels={k: ChannelConfig(**v) for k, v in data["channels"].items()},
        budgets=BudgetsConfig(**data["budgets"]),
        legal_limits=LegalLimitsConfig(**data["legal_limits"]),
        hard_collection=HardCollectionConfig(**data["hard_collection"]),
        initial_state=InitialStateConfig(**data.get("initial_state", {})),
    )


def merge_config(base: Config, override: dict[str, Any]) -> Config:
    """
    Создаёт новый Config с наложением переопределений поверх базового.
    
    Args:
        base: Базовая конфигурация.
        override: Словарь с переопределениями (той же структуры что YAML).
        
    Returns:
        Config: Новый конфиг с объединёнными параметрами.
    """
    def merge_dict(base_dict: dict, override_dict: dict) -> dict:
        result = base_dict.copy()
        for k, v in override_dict.items():
            if k in result and isinstance(result[k], dict) and isinstance(v, dict):
                result[k] = merge_dict(result[k], v)
            else:
                result[k] = v
        return result
    
    base_dict = asdict(base)
    merged = merge_dict(base_dict, override)
    
    return Config(
        simulation=SimulationConfig(**merged["simulation"]),
        portfolio=PortfolioConfig(**merged["portfolio"]),
        client_types={k: ClientTypeConfig(**v) for k, v in merged["client_types"].items()},
        products={k: ProductConfig(**v) for k, v in merged["products"].items()},
        hazard=HazardConfig(**merged["hazard"]),
        channels={k: ChannelConfig(**v) for k, v in merged["channels"].items()},
        budgets=BudgetsConfig(**merged["budgets"]),
        legal_limits=LegalLimitsConfig(**merged["legal_limits"]),
        hard_collection=HardCollectionConfig(**merged["hard_collection"]),
        initial_state=InitialStateConfig(**merged["initial_state"]),
    )


config = load_config()