refactor(archive): move unused modules to archive/

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/
This commit is contained in:
2026-06-03 23:41:46 +08:00
parent d700bc1dfd
commit c905230a40
98 changed files with 0 additions and 714 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)