aszerW
30ea2970bd
- FactorBase/FactorRegistry/FactorCombiner: 因子抽象接口 - SignalGenerator: 信号生成抽象接口 - RiskControl/Position/CallbackHook: 风控抽象接口 - StrategyBase: 策略抽象基类 - Executor/Portfolio: 执行器抽象接口 - ConfigLoader: 配置加载器 - 删除framework/factors/momentum.py（具体实现）
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'] |