""" 配置层抽象接口(通用) 只提供配置加载和验证机制 """ import yaml from typing import Dict, Any, Optional from pathlib import Path class ConfigLoader: """ 配置加载器(通用) 支持从YAML文件加载配置 """ def __init__(self, config_path: Optional[str] = None): """初始化配置加载器""" self._config: Dict[str, Any] = {} if config_path: self.load(config_path) def load(self, config_path: str) -> Dict[str, Any]: """从YAML文件加载配置""" path = Path(config_path) if not path.exists(): raise FileNotFoundError(f"Config file not found: {config_path}") with open(path, 'r', encoding='utf-8') as f: self._config = yaml.safe_load(f) or {} return self._config def get(self, key: str, default: Any = None) -> Any: """获取配置项""" return self._config.get(key, default) def get_section(self, section: str) -> Dict[str, Any]: """获取配置区块""" return self._config.get(section, {}) def validate(self, required_keys: list) -> bool: """验证必填配置项""" missing = [key for key in required_keys if key not in self._config] if missing: raise ValueError(f"Missing required config keys: {missing}") return True def __repr__(self) -> str: return f"ConfigLoader(keys={list(self._config.keys())})" class StrategyConfig: """ 策略配置类(通用) 用于封装策略配置 """ def __init__(self, config: Dict[str, Any]): """初始化策略配置""" self._config = config @property def name(self) -> str: """策略名称""" return self._config.get('strategy', {}).get('name', 'unknown') @property def select_num(self) -> int: """选中数量""" return self._config.get('signal', {}).get('select_num', 3) @property def stoploss(self) -> float: """止损阈值""" return self._config.get('risk', {}).get('stop_loss', -0.05) def get_factor_config(self) -> list: """获取因子配置""" return self._config.get('factors', []) def get_signal_config(self) -> dict: """获取信号配置""" return self._config.get('signal', {}) def get_risk_config(self) -> list: """获取风控配置""" return self._config.get('risk', []) def to_dict(self) -> Dict[str, Any]: """转换为字典""" return self._config.copy() # 导出抽象接口 __all__ = ['ConfigLoader', 'StrategyConfig']