refactor(framework): 框架只保留抽象接口,具体实现移至strategies/shared

- FactorBase/FactorRegistry/FactorCombiner: 因子抽象接口
- SignalGenerator: 信号生成抽象接口
- RiskControl/Position/CallbackHook: 风控抽象接口
- StrategyBase: 策略抽象基类
- Executor/Portfolio: 执行器抽象接口
- ConfigLoader: 配置加载器
- 删除framework/factors/momentum.py(具体实现)
This commit is contained in:
2026-05-11 23:09:01 +08:00
parent 9a8a0d7c72
commit 30ea2970bd
8 changed files with 503 additions and 1516 deletions

View File

@@ -1,83 +1,103 @@
"""
配置层抽象设计
配置层抽象接口(通用)
核心组件:
- ConfigLoader: 配置加载器
只提供配置加载和验证机制
"""
import yaml
from typing import Dict, Any, Optional
from pathlib import Path
from dataclasses import dataclass
@dataclass
class StrategyConfig:
"""策略配置"""
name: str
version: int
factors: list
signal: dict
callbacks: dict
params: dict
class ConfigLoader:
"""
配置加载器
配置加载器(通用)
支持YAML配置文件加载和验证
支持YAML文件加载配置
"""
def __init__(self, config_path: str):
"""
初始化配置加载器
def __init__(self, config_path: Optional[str] = None):
"""初始化配置加载器"""
self._config: Dict[str, Any] = {}
Args:
config_path: 配置文件路径
"""
self._config_path = Path(config_path)
self._config = None
if config_path:
self.load(config_path)
def load(self) -> Dict:
"""加载配置"""
if not self._config_path.exists():
raise FileNotFoundError(f"Config file not found: {self._config_path}")
def load(self, config_path: str) -> Dict[str, Any]:
"""从YAML文件加载配置"""
path = Path(config_path)
with open(self._config_path, 'r', encoding='utf-8') as f:
self._config = yaml.safe_load(f)
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 validate(self) -> bool:
"""验证配置"""
if self._config is None:
self.load()
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]
# 必须字段
required_fields = ['strategy', 'factors', 'signal']
for field in required_fields:
if field not in self._config:
raise ValueError(f"Missing required field: {field}")
if missing:
raise ValueError(f"Missing required config keys: {missing}")
return True
def get_strategy_config(self) -> StrategyConfig:
"""获取策略配置"""
if self._config is None:
self.load()
return StrategyConfig(
name=self._config['strategy']['name'],
version=self._config['strategy'].get('version', 1),
factors=self._config['factors'],
signal=self._config['signal'],
callbacks=self._config.get('callbacks', {}),
params=self._config.get('params', {})
)
def __repr__(self) -> str:
return f"ConfigLoader(keys={list(self._config.keys())})"
class StrategyConfig:
"""
策略配置类(通用)
@staticmethod
def from_yaml(yaml_str: str) -> Dict:
"""从YAML字符串加载"""
return yaml.safe_load(yaml_str)
用于封装策略配置
"""
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']