feat(framework_v2): 实现通用配置系统,支持扁平化资产池和策略分组

- 使用 Pydantic Schema 验证配置类型安全
- 实现扁平化 AssetPool,移除预设分类(equity/commodity/fixed_income)
- 移除 MarketType 枚举,改用 group 字符串字段实现策略分组
- AssetConfig 引入 signal_source/trade_source 分离,支持跨市场场景
- ConfigLoader 支持通用 StrategyConfig,向后兼容 RotationStrategyConfig
- 新增 GroupConfig 替代 MarketGroupConfig,支持分散化选股

重构核心:
- market → group(策略分组语义,组内竞争强制分散)
- by_market → by_group
- MarketGroupConfig → GroupConfig
This commit is contained in:
2026-05-24 14:25:25 +08:00
parent 99d3584d05
commit 341611c32b
3 changed files with 763 additions and 0 deletions

View File

@@ -0,0 +1,247 @@
"""
配置加载器
支持:
1. YAML 配置文件加载
2. Pydantic Schema 验证
3. 环境变量替换
4. 配置合并(默认值 + 用户配置)
"""
import os
import re
import yaml
from pathlib import Path
from typing import Optional, Dict, Any
from framework_v2.config.schemas import StrategyConfig, RotationStrategyConfig
class ConfigLoader:
"""
配置加载器
用法:
loader = ConfigLoader()
config = loader.load('config/rotation.yaml')
"""
def __init__(self, config_dir: str = None):
"""
初始化
Args:
config_dir: 配置目录(默认 framework_v2/config
"""
if config_dir is None:
# 默认配置目录
config_dir = Path(__file__).parent
self.config_dir = Path(config_dir)
def load(self, config_file: str, config_type: str = 'strategy') -> StrategyConfig:
"""
加载配置文件
Args:
config_file: 配置文件路径(相对路径或绝对路径)
config_type: 配置类型('strategy''rotation'
Returns:
验证后的配置对象
示例:
>>> loader = ConfigLoader()
>>> config = loader.load('rotation_example.yaml')
>>> print(config.factor.n_days)
25
"""
# 1. 解析文件路径
config_path = self._resolve_path(config_file)
# 2. 读取 YAML
with open(config_path, 'r', encoding='utf-8') as f:
config_dict = yaml.safe_load(f)
# 3. 环境变量替换
config_dict = self._substitute_env_vars(config_dict)
# 4. Pydantic 验证(根据类型选择 Schema
if config_type == 'rotation':
# 兼容旧版本:轮动策略专用配置
config = RotationStrategyConfig(**config_dict)
else:
# 通用策略配置(推荐)
config = StrategyConfig(**config_dict)
return config
def load_dict(self, config_dict: Dict[str, Any], config_type: str = 'strategy') -> StrategyConfig:
"""
从字典加载配置
Args:
config_dict: 配置字典
config_type: 配置类型('strategy''rotation'
Returns:
验证后的配置对象
"""
# 环境变量替换
config_dict = self._substitute_env_vars(config_dict)
# Pydantic 验证(根据类型选择 Schema
if config_type == 'rotation':
config = RotationStrategyConfig(**config_dict)
else:
config = StrategyConfig(**config_dict)
return config
def _resolve_path(self, config_file: str) -> Path:
"""
解析配置文件路径
Args:
config_file: 配置文件路径
Returns:
绝对路径
"""
path = Path(config_file)
# 如果是绝对路径,直接返回
if path.is_absolute():
if not path.exists():
raise FileNotFoundError(f"配置文件不存在: {path}")
return path
# 相对路径:先在配置目录查找
config_path = self.config_dir / path
if config_path.exists():
return config_path
# 然后在当前工作目录查找
cwd_path = Path.cwd() / path
if cwd_path.exists():
return cwd_path
raise FileNotFoundError(
f"配置文件未找到: {path}\n"
f"搜索路径:\n"
f" - {config_path}\n"
f" - {cwd_path}"
)
def _substitute_env_vars(self, config: Any) -> Any:
"""
替换配置中的环境变量
支持格式:
- ${VAR_NAME}
- ${VAR_NAME:default_value}
Args:
config: 配置对象dict/list/str
Returns:
替换后的配置
"""
if isinstance(config, dict):
return {
key: self._substitute_env_vars(value)
for key, value in config.items()
}
elif isinstance(config, list):
return [
self._substitute_env_vars(item)
for item in config
]
elif isinstance(config, str):
# 匹配 ${VAR_NAME} 或 ${VAR_NAME:default}
pattern = r'\$\{([^}]+)\}'
def replace_match(match):
var_expr = match.group(1)
# 检查是否有默认值
if ':' in var_expr:
var_name, default = var_expr.split(':', 1)
else:
var_name = var_expr
default = None
# 从环境变量读取
value = os.getenv(var_name)
if value is None:
if default is not None:
return default
else:
raise ValueError(
f"环境变量未设置: {var_name}\n"
f"请设置环境变量或在配置中使用默认值: ${{{var_name}:default_value}}"
)
return value
return re.sub(pattern, replace_match, config)
else:
return config
def get_available_configs(self) -> list:
"""
获取可用的配置文件列表
Returns:
配置文件名列表
"""
if not self.config_dir.exists():
return []
return [
f.name
for f in self.config_dir.glob('*.yaml')
if f.is_file()
]
# 全局实例
_config_loader: Optional[ConfigLoader] = None
def get_config_loader(config_dir: str = None) -> ConfigLoader:
"""
获取配置加载器单例
Args:
config_dir: 配置目录
Returns:
ConfigLoader 实例
"""
global _config_loader
if _config_loader is None:
_config_loader = ConfigLoader(config_dir)
return _config_loader
def load_config(config_file: str, config_type: str = 'strategy') -> StrategyConfig:
"""
快捷函数:加载配置文件
Args:
config_file: 配置文件路径
config_type: 配置类型('strategy''rotation'
Returns:
验证后的配置对象
示例:
>>> from framework_v2.config import load_config
>>> config = load_config('rotation_example.yaml')
"""
loader = get_config_loader()
return loader.load(config_file, config_type=config_type)