config — 配置读取

所有可调参数集中在 config.json,通过点号路径访问:

from StockDataMaster.config import get_config
cfg = get_config()
max_days = cfg.get('cache.max_days_per_stock', 120)

config

配置管理模块

负责加载和管理StockDataMaster的所有配置项

Config

Config(config_path: str = None)

配置管理类

初始化配置管理器

参数：

名称	类型	描述	默认
config_path	str	配置文件路径，默认为StockDataMaster/config.json	None

源代码位于： config.py

def __init__(self, config_path: str = None):
    """
    初始化配置管理器

    Args:
        config_path: 配置文件路径，默认为StockDataMaster/config.json
    """
    if config_path is None:
        # 获取当前文件所在目录
        current_dir = os.path.dirname(os.path.abspath(__file__))
        config_path = os.path.join(current_dir, "config.json")

    self.config_path = config_path
    self.config: Dict[str, Any] = {}
    self.load()

load

load()

加载配置文件

源代码位于： config.py

def load(self):
    """加载配置文件"""
    try:
        with open(self.config_path, 'r', encoding='utf-8') as f:
            self.config = json.load(f)
        self._load_dotenv()
        self._apply_env_overrides()
    except FileNotFoundError:
        raise FileNotFoundError(f"配置文件不存在: {self.config_path}")
    except json.JSONDecodeError as e:
        raise ValueError(f"配置文件JSON格式错误: {e}")

reload

reload()

热重载配置文件

源代码位于： config.py

def reload(self):
    """热重载配置文件"""
    self.load()

get

get(key: str, default: Any = None) -> Any

获取配置项

参数：

名称	类型	描述	默认
key	str	配置键,支持点号分隔的嵌套键,如 'cache.max_days_per_stock'	必需
default	Any	默认值	None

返回：

类型	描述
Any	配置值

源代码位于： config.py

def get(self, key: str, default: Any = None) -> Any:
    """
    获取配置项

    Args:
        key: 配置键,支持点号分隔的嵌套键,如 'cache.max_days_per_stock'
        default: 默认值

    Returns:
        配置值
    """
    keys = key.split('.')
    value = self.config

    for k in keys:
        if isinstance(value, dict) and k in value:
            value = value[k]
        else:
            return default

    return value

get_data_source_config

get_data_source_config(source_name: str) -> Dict[str, Any]

获取指定数据源的配置

参数：

名称	类型	描述	默认
source_name	str	数据源名称 (mootdx/baostock/tushare/xtquant)	必需

返回：

类型	描述
Dict[str, Any]	数据源配置字典

源代码位于： config.py

def get_data_source_config(self, source_name: str) -> Dict[str, Any]:
    """
    获取指定数据源的配置

    Args:
        source_name: 数据源名称 (mootdx/baostock/tushare/xtquant)

    Returns:
        数据源配置字典
    """
    return self.config.get('data_sources', {}).get(source_name, {})

get_enabled_sources

get_enabled_sources() -> list

获取所有启用的数据源名称列表,按优先级排序

返回：

类型	描述
list	数据源名称列表

源代码位于： config.py

def get_enabled_sources(self) -> list:
    """
    获取所有启用的数据源名称列表,按优先级排序

    Returns:
        数据源名称列表
    """
    sources = self.config.get('data_sources', {})
    enabled = [(name, cfg) for name, cfg in sources.items() if cfg.get('enabled', False)]
    # 按优先级排序
    enabled.sort(key=lambda x: x[1].get('priority', 999))
    return [name for name, _ in enabled]

get_sources_by_usage

get_sources_by_usage(usage: str) -> list

获取支持特定用途的数据源列表,按优先级排序

参数：

名称	类型	描述	默认
usage	str	用途类型 (kline/valuation/tick)	必需

返回：

类型	描述
list	数据源名称列表

源代码位于： config.py

def get_sources_by_usage(self, usage: str) -> list:
    """
    获取支持特定用途的数据源列表,按优先级排序

    Args:
        usage: 用途类型 (kline/valuation/tick)

    Returns:
        数据源名称列表
    """
    sources = self.config.get('data_sources', {})
    matched = [
        (name, cfg) for name, cfg in sources.items()
        if cfg.get('enabled', False) and usage in cfg.get('use_for', [])
    ]
    # 按优先级排序
    matched.sort(key=lambda x: x[1].get('priority', 999))
    return [name for name, _ in matched]

get_sources_by_role

get_sources_by_role(role: str, time_slot: str = None) -> list

按 roles 获取支持特定角色的数据源列表,按时段过滤并按优先级排序

参数：

名称	类型	描述	默认
role	str	角色 (kline_day/kline_minute/tick/validation)	必需
time_slot	str	时段 ('trading'/'after_hours'), None=不过滤	None

返回：

类型	描述
list	数据源名称列表,按优先级排序

源代码位于： config.py

def get_sources_by_role(self, role: str, time_slot: str = None) -> list:
    """
    按 roles 获取支持特定角色的数据源列表,按时段过滤并按优先级排序

    Args:
        role: 角色 (kline_day/kline_minute/tick/validation)
        time_slot: 时段 ('trading'/'after_hours'), None=不过滤

    Returns:
        数据源名称列表,按优先级排序
    """
    self._migrate_legacy_config()

    sources = self.config.get('data_sources', {})
    matched = []

    for name, cfg in sources.items():
        if not cfg.get('enabled', False):
            continue

        roles = cfg.get('roles', {})
        if role not in roles:
            continue

        role_cfg = roles[role]

        # 时段过滤: 如果角色定义了 time_slot,只在匹配时段时包含
        role_time_slot = role_cfg.get('time_slot')
        if time_slot and role_time_slot and role_time_slot != time_slot:
            continue

        priority = role_cfg.get('priority', 999)
        matched.append((name, priority))

    matched.sort(key=lambda x: x[1])
    return [name for name, _ in matched]

get_validation_config

get_validation_config() -> dict

获取投票校验配置

源代码位于： config.py

def get_validation_config(self) -> dict:
    """获取投票校验配置"""
    defaults = {
        'mode': 'voting',
        'quorum': 2,
        'strategy': 'first_to_quorum',
        'sources': ['xtquant', 'baostock', 'mootdx'],
        'price_tolerance_abs': 0.01,
        'price_tolerance_pct': 0.005,
        'volume_tolerance_pct': 0.05,
        'min_pass_rate': 0.8,
        'skip_today_in_trading_hours': True
    }
    validation_cfg = self.config.get('validation', {})
    defaults.update(validation_cfg)
    return defaults

get_stock_name_config

get_stock_name_config() -> dict

获取股票名称配置

源代码位于： config.py

def get_stock_name_config(self) -> dict:
    """获取股票名称配置"""
    defaults = {
        'cache_enabled': True,
        'cleanup_day': 5,
        'baostock_max_consecutive_failures': 3,
        'baostock_retry_cooldown': 300
    }
    sn_cfg = self.config.get('stock_name', {})
    defaults.update(sn_cfg)
    return defaults

is_cache_enabled

is_cache_enabled() -> bool

缓存是否启用

源代码位于： config.py

def is_cache_enabled(self) -> bool:
    """缓存是否启用"""
    return self.get('cache.enabled', False)

is_health_check_enabled

is_health_check_enabled() -> bool

健康检查是否启用

源代码位于： config.py

def is_health_check_enabled(self) -> bool:
    """健康检查是否启用"""
    return self.get('health_check.enabled', False)

is_hot_switch_enabled

is_hot_switch_enabled() -> bool

热切换是否启用

源代码位于： config.py

def is_hot_switch_enabled(self) -> bool:
    """热切换是否启用"""
    return self.get('hot_switch.enabled', False)

get_cache_max_days

get_cache_max_days() -> int

获取缓存最大天数

源代码位于： config.py

def get_cache_max_days(self) -> int:
    """获取缓存最大天数"""
    return self.get('cache.max_days_per_stock', 120)

get_health_check_interval

get_health_check_interval() -> int

获取健康检查间隔(秒)

源代码位于： config.py

def get_health_check_interval(self) -> int:
    """获取健康检查间隔(秒)"""
    return self.get('health_check.interval_seconds', 60)

get_config

get_config(config_path: str = None) -> Config

获取全局配置实例(单例模式)

参数：

名称	类型	描述	默认
config_path	str	配置文件路径	None

返回：

类型	描述
Config	Config实例

源代码位于： config.py

def get_config(config_path: str = None) -> Config:
    """
    获取全局配置实例(单例模式)

    Args:
        config_path: 配置文件路径

    Returns:
        Config实例
    """
    global _global_config
    if _global_config is None:
        _global_config = Config(config_path)
    return _global_config