health — 健康检测

后台线程对数据源做存活检查,异常时降级 / 切换。

health.health_manager

health_manager

健康检测和热切换管理器

负责监控数据源健康状态,实现故障自动切换

HealthManager

HealthManager(config, adapters: Dict)

健康检测和热切换管理器

初始化健康管理器

参数：

名称	类型	描述	默认
config		配置对象	必需
adapters	Dict	适配器字典 {name: adapter}	必需

源代码位于： health/health_manager.py

def __init__(self, config, adapters: Dict):
    """
    初始化健康管理器

    Args:
        config: 配置对象
        adapters: 适配器字典 {name: adapter}
    """
    self.config = config
    self.adapters = adapters
    self.logger = logging.getLogger("DataMaster.HealthManager")

    # 健康检查配置
    self.check_interval = config.get('health_check.interval_seconds', 60)
    self.response_threshold = config.get('health_check.response_time_threshold', 5.0)
    self.failure_threshold = config.get('health_check.consecutive_failures_threshold', 3)

    # 健康状态记录
    self.health_status = {}  # {adapter_name: {'status': 'ok', 'last_check': datetime, ...}}
    self.failure_counts = {}  # {adapter_name: count}

    # 当前活跃数据源(按用途分类)
    self.active_sources = {
        'kline_day': None,      # 日K线活跃源
        'kline_minute': None,   # 分钟K线活跃源
        'valuation': None,
        'tick': None
    }

    # 切换历史记录(使用deque限制大小)
    self.switch_history = deque(maxlen=100)

    # 监控线程
    self.monitor_thread = None
    self.is_running = False
    self.lock = threading.Lock()

start_monitoring

start_monitoring()

启动健康监控线程

源代码位于： health/health_manager.py

def start_monitoring(self):
    """启动健康监控线程"""
    if self.is_running:
        self.logger.warning("健康监控已在运行中")
        return

    self.is_running = True
    self.monitor_thread = threading.Thread(target=self._monitor_loop, daemon=True)
    self.monitor_thread.start()
    self.logger.info(f"健康监控线程已启动,检查间隔: {self.check_interval}秒")

stop_monitoring

stop_monitoring()

停止健康监控线程

源代码位于： health/health_manager.py

def stop_monitoring(self):
    """停止健康监控线程"""
    self.is_running = False
    if self.monitor_thread:
        self.monitor_thread.join(timeout=5)
    self.logger.info("健康监控线程已停止")

check_all_sources

check_all_sources()

检查所有数据源的健康状态

源代码位于： health/health_manager.py

def check_all_sources(self):
    """检查所有数据源的健康状态"""
    with self.lock:
        for name, adapter in self.adapters.items():
            try:
                # 执行健康检查
                result = adapter.health_check()

                # 获取上一次的状态
                prev_status = self.health_status.get(name, {}).get('status', 'unknown')

                # 更新健康状态
                self.health_status[name] = {
                    'status': result['status'],
                    'last_check': datetime.now(),
                    'response_time': result['response_time'],
                    'data_freshness': result['data_freshness'],
                    'error_message': result['error_message']
                }

                # 更新失败计数
                if result['status'] == 'error':
                    self.failure_counts[name] = self.failure_counts.get(name, 0) + 1

                    # 只在首次失败或状态变化时记录WARNING
                    if prev_status != 'error' or self.failure_counts[name] == 1:
                        self.logger.warning(
                            f"{name} 健康检查失败: {result['error_message']}"
                        )
                    # 持续失败时只记录DEBUG级别
                    else:
                        self.logger.debug(
                            f"{name} 健康检查仍失败({self.failure_counts[name]}次): {result['error_message']}"
                        )

                    # 检查是否需要切换：同一轮连续故障只在跨过阈值时触发一次。
                    # 否则后台线程每隔 interval_seconds 会重复记录"触发数据源切换"。
                    if self.failure_counts[name] == self.failure_threshold:
                        self._trigger_switch(name, result['error_message'])
                else:
                    # 成功则重置失败计数
                    # 如果从失败恢复,记录INFO
                    if prev_status == 'error':
                        self.logger.info(f"{name} 健康检查已恢复")
                    self.failure_counts[name] = 0

            except Exception as e:
                self.logger.error(f"{name} 健康检查异常: {e}")
                self.failure_counts[name] = self.failure_counts.get(name, 0) + 1

get_active_source

get_active_source(usage: str) -> Optional[str]

获取当前活跃的数据源（兼容抽象类型和细分类型）

参数：

名称	类型	描述	默认
usage	str	用途类型，支持抽象类型和细分类型 - 抽象类型: 'kline', 'valuation', 'tick' - 细分类型: 'kline_day', 'kline_minute', 'valuation', 'tick'	必需

返回：

类型	描述
Optional[str]	数据源名称，未找到返回 None

源代码位于： health/health_manager.py

def get_active_source(self, usage: str) -> Optional[str]:
    """
    获取当前活跃的数据源（兼容抽象类型和细分类型）

    Args:
        usage: 用途类型，支持抽象类型和细分类型
            - 抽象类型: 'kline', 'valuation', 'tick'
            - 细分类型: 'kline_day', 'kline_minute', 'valuation', 'tick'

    Returns:
        数据源名称，未找到返回 None
    """
    # 抽象类型到细分类型的映射
    type_mapping = {
        'kline': 'kline_day',  # 默认映射到日K线
    }

    # 如果是抽象类型，转换为细分类型
    actual_type = type_mapping.get(usage, usage)

    with self.lock:
        # 如果还没有活跃数据源,选择一个
        # 修复: 使用 actual_type (细分类型) 而非 usage (抽象类型) 查找备用源
        # 原因: 适配器 use_for 配置中使用细分类型 ('kline_day'),
        #       若传入抽象类型 ('kline') 将无法匹配任何数据源
        if self.active_sources.get(actual_type) is None:
            self.active_sources[actual_type] = self._find_backup_source(actual_type)

        return self.active_sources[actual_type]

get_health_report

get_health_report() -> Dict[str, Any]

获取健康状态报告

返回：

类型	描述
Dict[str, Any]	健康状态字典

源代码位于： health/health_manager.py

def get_health_report(self) -> Dict[str, Any]:
    """
    获取健康状态报告

    Returns:
        健康状态字典
    """
    with self.lock:
        report = {
            'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
            'sources': {},
            'active_sources': self.active_sources.copy(),
            'recent_switches': list(self.switch_history)[-10:]  # 最近10次切换
        }

        for name, adapter in self.adapters.items():
            status = self.health_status.get(name, {})
            source_report = {
                'enabled': adapter.config.get('enabled', False),
                'connected': adapter.is_connected,
                'status': status.get('status', 'unknown'),
                'last_check': status.get('last_check', '').strftime('%H:%M:%S') if status.get('last_check') else 'N/A',
                'response_time': f"{status.get('response_time', 0):.2f}s",
                'failure_count': self.failure_counts.get(name, 0),
                'error': status.get('error_message')
            }

            # 增强: 获取xtquant连接统计
            if name == 'xtquant' and hasattr(adapter, 'get_connection_stats'):
                try:
                    conn_stats = adapter.get_connection_stats()
                    source_report['connection_stats'] = {
                        'connect_count': conn_stats.get('connect_count', 0),
                        'disconnect_count': conn_stats.get('disconnect_count', 0),
                        'reconnect_count': conn_stats.get('reconnect_count', 0),
                        'heartbeat_failures': conn_stats.get('heartbeat_failures', 0),
                        'last_heartbeat': conn_stats.get('last_heartbeat').strftime('%H:%M:%S') if conn_stats.get('last_heartbeat') else 'N/A',
                        'heartbeat_age_seconds': conn_stats.get('heartbeat_age_seconds'),
                        'connection_health': conn_stats.get('connection_health', 'unknown'),
                        'retry_stats': conn_stats.get('retry_stats', {})
                    }
                except Exception as e:
                    self.logger.debug(f"获取xtquant连接统计失败: {e}")

            report['sources'][name] = source_report

        return report

force_switch

force_switch(usage: str, target_source: str) -> bool

强制切换数据源

参数：

名称	类型	描述	默认
usage	str	用途类型	必需
target_source	str	目标数据源名称	必需

返回：

类型	描述
bool	切换是否成功

源代码位于： health/health_manager.py

def force_switch(self, usage: str, target_source: str) -> bool:
    """
    强制切换数据源

    Args:
        usage: 用途类型
        target_source: 目标数据源名称

    Returns:
        切换是否成功
    """
    with self.lock:
        if target_source not in self.adapters:
            self.logger.error(f"目标数据源不存在: {target_source}")
            return False

        adapter = self.adapters[target_source]

        # 优先使用 roles 格式，回退到 use_for 格式（向后兼容）
        roles = adapter.config.get('roles', {})
        if roles:
            if usage not in roles:
                self.logger.error(f"{target_source}不支持{usage}用途")
                return False
        else:
            if usage not in adapter.config.get('use_for', []):
                self.logger.error(f"{target_source}不支持{usage}用途")
                return False

        old_source = self.active_sources.get(usage)
        self.active_sources[usage] = target_source

        self.logger.info(f"强制切换{usage}数据源: {old_source} -> {target_source}")

        # 记录切换历史
        self.switch_history.append({
            'time': datetime.now(),
            'usage': usage,
            'from': old_source,
            'to': target_source,
            'reason': '手动强制切换'
        })

        return True