90 lines
2.5 KiB
Python
90 lines
2.5 KiB
Python
|
import yaml
|
||
|
|
from pathlib import Path
|
||
|
|
from typing import Any, Optional
|
||
|
|
|
||
|
|
|
||
|
|
class Config:
|
||
|
|
"""配置管理类"""
|
||
|
|
|
||
|
|
def __init__(self, config_path: str = "config/config.yaml"):
|
||
|
|
self.config_path = Path(config_path)
|
||
|
|
self._config_data = self._load_config()
|
||
|
|
|
||
|
|
def _load_config(self) -> dict:
|
||
|
|
"""加载配置文件"""
|
||
|
|
if not self.config_path.exists():
|
||
|
|
raise FileNotFoundError(f"配置文件不存在: {self.config_path}")
|
||
|
|
|
||
|
|
with open(self.config_path, 'r', encoding='utf-8') as f:
|
||
|
|
config = yaml.safe_load(f)
|
||
|
|
|
||
|
|
return config
|
||
|
|
|
||
|
|
def get(self, key: str, default: Any = None) -> Any:
|
||
|
|
"""获取配置值，支持点号分隔的嵌套键"""
|
||
|
|
keys = key.split('.')
|
||
|
|
value = self._config_data
|
||
|
|
|
||
|
|
for k in keys:
|
||
|
|
if isinstance(value, dict) and k in value:
|
||
|
|
value = value[k]
|
||
|
|
else:
|
||
|
|
return default
|
||
|
|
|
||
|
|
return value
|
||
|
|
|
||
|
|
def set(self, key: str, value: Any):
|
||
|
|
"""设置配置值"""
|
||
|
|
keys = key.split('.')
|
||
|
|
config = self._config_data
|
||
|
|
|
||
|
|
for k in keys[:-1]:
|
||
|
|
if k not in config:
|
||
|
|
config[k] = {}
|
||
|
|
config = config[k]
|
||
|
|
|
||
|
|
config[keys[-1]] = value
|
||
|
|
|
||
|
|
def save(self):
|
||
|
|
"""保存配置到文件"""
|
||
|
|
with open(self.config_path, 'w', encoding='utf-8') as f:
|
||
|
|
yaml.dump(self._config_data, f, allow_unicode=True, default_flow_style=False)
|
||
|
|
|
||
|
|
def reload(self):
|
||
|
|
"""重新加载配置文件"""
|
||
|
|
self._config_data = self._load_config()
|
||
|
|
|
||
|
|
@property
|
||
|
|
def database(self) -> dict:
|
||
|
|
"""获取数据库配置"""
|
||
|
|
return self.get('database', {})
|
||
|
|
|
||
|
|
@property
|
||
|
|
def redis(self) -> dict:
|
||
|
|
"""获取Redis配置"""
|
||
|
|
return self.get('redis', {})
|
||
|
|
|
||
|
|
@property
|
||
|
|
def broker(self) -> dict:
|
||
|
|
"""获取券商配置"""
|
||
|
|
return self.get('broker', {})
|
||
|
|
|
||
|
|
@property
|
||
|
|
def risk_control(self) -> dict:
|
||
|
|
"""获取风控配置"""
|
||
|
|
return self.get('risk_control', {})
|
||
|
|
|
||
|
|
@property
|
||
|
|
def strategy(self) -> dict:
|
||
|
|
"""获取策略配置"""
|
||
|
|
return self.get('strategy', {})
|
||
|
|
|
||
|
|
@property
|
||
|
|
def logging(self) -> dict:
|
||
|
|
"""获取日志配置"""
|
||
|
|
return self.get('logging', {})
|
||
|
|
|
||
|
|
@property
|
||
|
|
def monitoring(self) -> dict:
|
||
|
|
"""获取监控配置"""
|
||
|
|
return self.get('monitoring', {})
|