import yaml from pathlib import Path from typing import Any, Optional class Config: """配置管理类""" def __init__(self, config_path: str = "config/config.yaml"): self.config_path = Path(config_path) self._config_data = self._load_config() def _load_config(self) -> dict: """加载配置文件""" if not self.config_path.exists(): raise FileNotFoundError(f"配置文件不存在: {self.config_path}") with open(self.config_path, 'r', encoding='utf-8') as f: config = yaml.safe_load(f) return config def get(self, key: str, default: Any = None) -> Any: """获取配置值,支持点号分隔的嵌套键""" keys = key.split('.') value = self._config_data for k in keys: if isinstance(value, dict) and k in value: value = value[k] else: return default return value def set(self, key: str, value: Any): """设置配置值""" keys = key.split('.') config = self._config_data for k in keys[:-1]: if k not in config: config[k] = {} config = config[k] config[keys[-1]] = value def save(self): """保存配置到文件""" with open(self.config_path, 'w', encoding='utf-8') as f: yaml.dump(self._config_data, f, allow_unicode=True, default_flow_style=False) def reload(self): """重新加载配置文件""" self._config_data = self._load_config() @property def database(self) -> dict: """获取数据库配置""" return self.get('database', {}) @property def redis(self) -> dict: """获取Redis配置""" return self.get('redis', {}) @property def broker(self) -> dict: """获取券商配置""" return self.get('broker', {}) @property def risk_control(self) -> dict: """获取风控配置""" return self.get('risk_control', {}) @property def strategy(self) -> dict: """获取策略配置""" return self.get('strategy', {}) @property def logging(self) -> dict: """获取日志配置""" return self.get('logging', {}) @property def monitoring(self) -> dict: """获取监控配置""" return self.get('monitoring', {})