aszerW
c905230a40
Archive legacy framework and utility modules that are no longer referenced by the active core (datasource/ and rotation/): - framework/ -> archive/framework/ - framework_v2/ -> archive/framework_v2/ - strategies/ -> archive/strategies/ - config/ -> archive/config/ - visualization/ -> archive/visualization/ - scripts/ -> archive/scripts/ - tests/ -> archive/tests/ - run_rotation.py, run_us_rotation.py -> archive/single_files/ - compare_*.py, test_api_dates.py -> archive/single_files/
103 lines
2.7 KiB
Python
103 lines
2.7 KiB
Python
"""
|
|
配置层抽象接口（通用）
|
|
|
|
只提供配置加载和验证机制
|
|
"""
|
|
|
|
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'] |