数据自愈模块冗余分析报告

日期：2026-02-23
审计范围：src/utils/data_healing/ 全部 5 个文件 + 调用方 _run_data_healing()

一、代码规模

二、死代码

2.1 config.py — TIMEFRAME_MINUTES 字典

# config.py
TIMEFRAME_MINUTES = {
    '1m': 1, '5m': 5, '15m': 15,
    '1h': 60, '4h': 240, '1d': 1440
}

代码库中已有 timeframe_to_minutes() 函数封装，TIMEFRAME_MINUTES 字典定义后从未被直接引用。

可删除：~10 行

2.2 dataclass 字段 — 从未被任何调用方读取

HealingResult（orchestrator.py）

调用方验证：_run_data_healing() 读取 result.status、result.data、result.quality，从未访问 warmup_mode 或 repair_summary。

QualityReport（quality_assessor.py）

Diagnosis（orchestrator.py）

冗余字段合计：9 个

三、过度封装（单调用转发方法）

repair_executor.py 中 4 个方法各自只包装一个外部调用，无额外逻辑：

# 17 行：仅转发到 calculate_zscore_ols()
def _compute_zscore(self, alt_klines, base_klines) -> Optional[float]:
    try:
        return calculate_zscore_ols(base_klines, alt_klines, ...)
    except Exception as e:
        logger.debug(f"zscore计算失败: {e}")
        return None

# 11 行：仅转发到 calculate_correlation()
def _compute_correlation(self, alt_klines, base_klines) -> Optional[float]:
    try:
        return calculate_correlation(base_klines, alt_klines)
    except Exception:
        return None

# 12 行：仅转发到 analysis_repo.batch_insert()
def _insert_records(self, records: List[Dict]) -> int:
    if not records:
        return 0
    try:
        return self.analysis_repo.batch_insert(records)
    except Exception as e:
        logger.error(f"批量写入失败: {e}", exc_info=True)
        raise

# 26 行：仅构造一个字典
@staticmethod
def _build_analysis_record(missing_time, symbol, ...) -> Dict:
    return { 'analysis_time': missing_time, ... }

4 个方法共 66 行，均可直接内联到 _repair_from_klines()。

四、重复计算

4.1 completeness_pct 计算两次

# ContinuityChecker.check_continuity()（continuity_checker.py:53）
completeness_pct = (len(records) / expected_count) * 100

# QualityAssessor.assess()（quality_assessor.py:65）
completeness_pct = (actual_count / expected_count) * 100

两处独立计算同一指标，调用链上每轮迭代都执行两次。

4.2 健康状态判断两次

# orchestrator.py _diagnose()（行 266）
is_healthy = is_continuous and len(records) >= required_count and is_fresh

# orchestrator.py _final_assessment()（行 322）
再次调用 checker.check_continuity() + assessor.assess()
# 不使用 Diagnosis.is_healthy，重新执行整个评估流程

_diagnose() 的健康判断结果存入 Diagnosis.is_healthy，但 _final_assessment() 忽略它，重新计算。

五、SRP 违反（职责混合）

DataHealingOrchestrator（557 行 / 16 个方法）

混合了 5 个独立职责：

RepairExecutor（372 行 / 11 个方法）

混合了 4 个独立职责：

六、附：超时机制缺陷（关联问题）

详见 数据自愈超时机制失效分析.md。

except Exception 吞掉 TimeoutError 是冗余内层封装带来的副作用——封装越多，异常穿透路径越长，越容易在某一层被错误捕获。

七、量化汇总

八、清理建议（按优先级）

高优先级（零功能风险）

1. 删除 HealingResult.warmup_mode、HealingResult.repair_summary 两个未读字段
2. 删除 QualityReport.missing_count、Diagnosis.is_continuous、Diagnosis.staleness_minutes 未读字段
3. 删除 config.py 中 TIMEFRAME_MINUTES 字典
4. 内联 _compute_zscore()、_compute_correlation()、_insert_records()、_build_analysis_record() 四个转发方法

预计清理：~100 行，复杂度降低，不改变任何行为。

中优先级（需测试验证）

5. 合并 completeness_pct 重复计算：由 check_continuity() 返回，assess() 直接复用
6. 让 _final_assessment() 使用 Diagnosis.is_healthy，消除重复评估流程
7. 删除 Diagnosis dataclass，将三个 target 列表直接合并后传给 executor.repair()

预计清理：~150 行