FinDataHub 数据管道架构设计

一、现状审视：我们已经拥有什么

在进入架构设计之前，先梳理项目的现有资产。FinDataHub 已经构建了相当完善的基础设施，理解现状是设计未来的前提。

1.1 数据采集基础设施

1.2 现有校验能力

两个 fetcher 各自内建了 DataValidator，校验维度包括：

1.3 现有数据预处理 DONE

1.4 现有数据调用方式 DONE

数据使用层已实现四层递进接入架构：

• Python SDK：findatahub-client 提供 11 个方法（daily/weekly/monthly/fina_indicator/search_stock/export 等）
• RESTful API：FastAPI 服务，14 个端点（数据查询 + 质量分 + 健康监控）
• Skills 层：skill.md 结构化描述 5 表 schema，AI 智能体可自主查询
• MCP 服务：7 个 MCP 工具，支持自然语言直达数据（stdio 协议）

二、问题定义：三个核心问题

将所有需求归结起来，FinDataHub 要解决的不过是三个问题：

TuShare 在其 AI 适配指南^[1]中提出了三层接入架构（SDK 层 → Skills 层 → MCP 层），覆盖了从"AI 辅助编程取数"到"AI 自主取数"的完整演进路径。这正是 FinDataHub 数据调用框架的核心参考模型。

三、整体架构：三层解耦设计

基于以上分析，FinDataHub 的数据管道采用三层解耦架构，每一层有明确的职责边界和接口契约：

flowchart TB
    subgraph L1["🔵 第一层：数据采集层 Data Collection"]
        direction LR
        TS[TuShare API]
        AK[AKShare API]
        DB1[(MySQL)]
        VAL[DataValidator]
        REP[SyncReport]
        TS -->|pro API| F1[DataFetcher]
        AK -->|free API| F2[DataFetcher]
        F1 --> DB1
        F2 --> DB1
        DB1 --> VAL
        VAL --> REP
    end

    subgraph L2["🟣 第二层：数据预处理层 Data Preprocessing"]
        direction LR
        CLEAN[数据清洗]
        OUT[异常值检测]
        STAT[统计量分析]
        CORR[相关性分析]
        PCA[主成分分析]
        NORM[归一化]
        DB2[(MySQL)]
        CLEAN --> OUT --> STAT --> CORR --> PCA --> NORM
        NORM --> DB2
    end

    subgraph L3["🟢 第三层：数据使用层 Data Access"]
        direction LR
        SDK[Python SDK]
        REST[RESTful HTTP API]
        SKILL[Skills 层]
        MCP[MCP 服务]
    end

    L1 -->|raw data| L2
    L2 -->|clean data| L3

    style L1 fill:#eff6ff,stroke:#2563eb,stroke-width:2px
    style L2 fill:#f5f3ff,stroke:#7c3aed,stroke-width:2px
    style L3 fill:#f0fdf4,stroke:#059669,stroke-width:2px
  

3.1 层间接口契约

每一层通过明确定义的接口与相邻层通信：

3.2 Medallion 架构：三数据库分层

上述三层在物理上映射为三个 MySQL 数据库，形成业界标准的 Medallion 架构（Bronze → Silver → Gold）：

四、第一层：数据采集层

4.1 数据源

FinDataHub 采用双数据源互补策略：

TuShare Pro 作为主数据源（标准化程度高，"几乎零清洗"），AKShare 作为互补数据源（覆盖更广的另类和宏观数据）和交叉验证源（双源对比校验）。这一策略在知乎的量化数据获取指南中得到了印证^[4]，该文也推荐组合使用多数据源（yfinance + TuShare + AKShare）。

4.2 现状优势

当前的数据采集层设计已经相当成熟，核心组件包括：

4.3 增量更新策略

当前 fetcher.py 已实现基于日期列的智能增量更新：

sequenceDiagram
    participant S as Scheduler
    participant F as DataFetcher
    participant C as SyncCache
    participant A as TuShare API
    participant D as MySQL

    S->>F: 触发同步
    F->>C: 查询上次同步日期
    C-->>F: last_date = 2026-08-05
    F->>F: 计算增量起始 = last_date + 1d
    F->>A: 请求数据 (start=20260806)
    A-->>F: DataFrame
    F->>D: UPSERT (幂等写入)
    F->>C: 更新 last_date
    F->>D: 更新 _table_meta
  

4.4 数据采集层的质量保障

参考 TuShare 的数据质量理念^[1]，数据采集层的核心保障包括：

• 非爬取数据源：TuShare 和 AKShare 都是经过社区采集、整理、入库和质量控制后的标准化数据，非原始爬虫数据
• 字段规范化：所有接口遵循统一的字段命名规则（ts_code、trade_date等）和数据类型定义
• 幂等写入：基于唯一键的 INSERT ... ON DUPLICATE KEY UPDATE，重复同步不会产生脏数据
• SQL 注入防护：safe_identifier() 白名单校验所有动态表名/列名
• 同步后校验：每次同步后自动执行完整性校验并生成报告

4.5 需要增强的部分 DONE

五、第二层：数据预处理层

5.1 预处理 Pipeline 设计

flowchart LR
    RAW[(Raw Tables)] --> STEP1

    subgraph PP["数据预处理 Pipeline"]
        STEP1[🔍 数据清洗
类型修正/去重/格式统一]
        STEP2[📏 缺失值处理
插值/填充/标记]
        STEP3[⚠️ 异常值检测
3σ/IQR/MAD]
        STEP4[📊 统计量分析
mean/std/skew/kurtosis]
        STEP5[🔗 相关性分析
Pearson/Spearman矩阵]
        STEP6[📐 主成分分析
PCA降维]
        STEP7[📏 归一化
MinMax/Z-score/Robust]
    end

    STEP1 --> STEP2 --> STEP3 --> STEP4 --> STEP5 --> STEP6 --> STEP7
    STEP4 -.->|统计报告| REPORT[📋 Stats Report]
    STEP5 -.->|相关性矩阵| CORR_RPT[📋 Corr Matrix]
    STEP7 --> CLEAN[(Clean Tables)]

    style PP fill:#f5f3ff,stroke:#7c3aed,stroke-width:2px
  

5.2 各步骤详细设计

步骤 1：数据清洗

步骤 2：缺失值处理

步骤 3：异常值检测

金融数据的异常值有三种典型来源，需要区别处理：

步骤 4：统计量分析

对每张核心表自动生成统计报告，包含：

• 基本统计量：count, mean, std, min, 25%, 50%, 75%, max
• 分布特征：skewness（偏度）, kurtosis（峰度）
• 缺失率：每列的缺失比例
• 唯一值数量：判断是否为类别变量

步骤 5：相关性分析

生成 Pearson 和 Spearman 相关性矩阵，用于：

• 识别高度相关的特征对（冗余特征检测）
• 发现潜在的因子共线性问题
• 为后续降维和特征选择提供依据

步骤 6-7：PCA + 归一化

这两个步骤是可选模块，仅在用户需要进行机器学习建模时激活：

• PCA：对高维财务指标进行降维，保留 95% 方差的主成分
• 归一化：支持 MinMaxScaler（值域压缩到 [0,1]）、StandardScaler（Z-score 标准化）、RobustScaler（基于中位数和 IQR，抗异常值）

5.3 预处理层模块结构

Services/preprocessing/
├── config/
│   └── pipeline.yaml          # Pipeline 配置（启用/禁用各步骤）
├── src/preprocessing/
│   ├── __init__.py
│   ├── pipeline.py            # PreprocessingPipeline 主控器
│   ├── cleaner.py             # 步骤1：数据清洗
│   ├── imputer.py             # 步骤2：缺失值处理
│   ├── outlier.py             # 步骤3：异常值检测
│   ├── stats.py               # 步骤4：统计量分析
│   ├── correlation.py         # 步骤5：相关性分析
│   ├── pca.py                 # 步骤6：主成分分析
│   ├── normalizer.py          # 步骤7：归一化
│   └── report.py              # 预处理报告生成器
└── pyproject.toml

5.4 Pipeline 配置示例

# pipeline.yaml
pipeline:
  tables:
    - stock_daily
    - stock_weekly
    - stock_monthly
  steps:
    cleaning:
      enabled: true
      drop_duplicates: true
      standardize_columns: true
    imputation:
      enabled: true
      strategy: ffill       # ffill / linear / mean / median
      add_flag_column: true
    outlier_detection:
      enabled: true
      method: iqr            # iqr / zscore / mad
      threshold: 3.0
    statistics:
      enabled: true
      output: reports/stats_report.json
    correlation:
      enabled: true
      method: pearson        # pearson / spearman
      output: reports/corr_matrix.json
    pca:
      enabled: false         # 按需启用
      n_components: 0.95
    normalization:
      enabled: false         # 按需启用
      method: zscore         # minmax / zscore / robust
  output:
    target_schema: fin_data_clean    # 清洗后数据的数据库名
    create_views: true               # 同时创建 SQL VIEW

5.5 质量门控：如何判断数据合格

答案是一个四级门控链（Quality Gate Chain）。每一级有明确的阻断策略——不是所有问题都会阻止数据入库，但所有问题都会被记录。

flowchart TB
    RAW[(Raw Tables)] --> G1

    subgraph Gates["四级质量门控"]
        G1["🚦 Gate 1: 采集门控
非空检查 · 行数阈值 · API返回码"]
        G2["🚦 Gate 2: 清洗门控
OHLC关系 · 列名标准 · 日期格式"]
        G3["🚦 Gate 3: 入库门控
唯一键 · 日期连续性 · Schema验证"]
        G4["🚦 Gate 4: 后验门控
跨源交叉验证 · 质量评分"]
    end

    G1 -->|PASS| G2
    G2 -->|PASS| G3
    G3 -->|PASS| CLEAN[(fin_data_clean)]
    CLEAN --> G4

    G1 -.->|FAIL → 重试3次| ALERT[告警 + 人工介入]
    G2 -.->|WARN → 标记| ISSUES[写入 _data_issues 表]
    G3 -.->|FAIL → 退回| RETRY[重试队列]
    G4 -.->|INFO → 记录| METRICS[写入 _quality_metrics]

    style Gates fill:#fef3c7,stroke:#d97706,stroke-width:2px
  

每一级的判定标准

数据质量评分公式

每张表在通过 G3 之前，会被赋予一个0-1 的量化质量分数，综合六个维度：

5.6 双源交叉验证

这是之前报告中标记为 TODO 的高优先级项，也是 L1→L2 之间最重要的质量防线。

验证策略

验证结果处理

• 差异 < 0.5%：两源一致，直接入库，标记 source='merged'
• 差异 0.5%-2%：记录到 _cross_validation_log 表，以 TuShare 为准入库
• 差异 > 2%：标记 is_validated=FALSE，写入 _data_issues 表等待人工复核

5.7 入库决策流程图

flowchart TD
    START([采集完成]) --> G1_CHECK{G1: 采集门控}
    G1_CHECK -->|FAIL| RETRY[重试机制
最多3次
指数退避]
    RETRY -->|仍失败| ALERT[🚨 告警通知
人工介入]
    RETRY -->|成功| G1_CHECK
    G1_CHECK -->|PASS| G2_CHECK{G2: 清洗门控}

    G2_CHECK -->|OHLC异常| TAG_WARN[标记 _is_flagged
写入 _data_issues]
    G2_CHECK -->|价格负值| TAG_BLOCK[标记 _is_error
该行不入 clean 库]
    G2_CHECK -->|格式错误| FIX[自动修正
日期/列名/类型]
    G2_CHECK -->|通过| G3_CHECK{G3: 入库门控}

    TAG_WARN --> G3_CHECK
    FIX --> G3_CHECK

    G3_CHECK -->|唯一键冲突| DEDUP[去重
保留最新]
    G3_CHECK -->|质量分 ≥ 0.6| WRITE[✅ 写入 fin_data_clean
标记 is_validated=TRUE]
    G3_CHECK -->|质量分 < 0.6| REJECT[❌ 退回收容表
_rejected_records]

    DEDUP --> G3_CHECK
    WRITE --> G4_CHECK{G4: 后验门控}
    G4_CHECK --> CROSS[双源交叉验证]
    CROSS --> SCORE[计算质量分]
    SCORE --> TREND[更新质量趋势]
    TREND --> DONE([完成])
  

5.8 清洗策略画像（Strategy Profile）

为什么停牌填充在不同策略下效果不同

ffill 后周三→周五的收益率 = (10.20 - 10.50) / 10.50 = -2.86%。但实际上这三天股票根本没有交易，这个收益率是计算幻觉。对于持仓 1-3 天的高频策略，基于假数据做出的交易决策可能导致真实亏损。而对于持仓数月的趋势策略，3 天假数据在 200 天序列中影响微乎其微。

策略画像定义

# pipeline_profiles.yaml — 不同策略画像使用不同清洗参数
profiles:
  high_freq:           # 高频/短线策略
    imputation:
      strategy: drop       # 不填充，删除停牌日
    outlier:
      method: mad          # MAD 更鲁棒，保留极端波动
      threshold: 4.0       # 更宽松的阈值
    normalization:
      enabled: false       # 不做归一化，保留原始价格

  trend:               # 趋势/中长线策略
    imputation:
      strategy: ffill      # 前向填充，保证序列对齐
      add_flag_column: true
    outlier:
      method: iqr
      threshold: 2.5
    normalization:
      enabled: false

  ml_model:            # 机器学习建模
    imputation:
      strategy: linear     # 线性插值，更平滑
    outlier:
      method: mad
      threshold: 3.0
    pca:
      enabled: true        # 启用 PCA 降维
      n_components: 0.95
    normalization:
      enabled: true        # 必须归一化
      method: robust       # RobustScaler，抗异常值

调用时指定画像即可：

# 不同策略拿到不同清洗程度的同一份数据
df_hf  = pipeline.run("stock_daily", profile="high_freq")
df_tr  = pipeline.run("stock_daily", profile="trend")
df_ml  = pipeline.run("stock_daily", profile="ml_model")

5.9 pct_change 收益率的归属

知乎文章中把 pct_change 放在清洗阶段，但不建议放进 L2 预处理层。原因：

pct_change 本质上是查询视图，不是数据转换。它不属于 Bronze → Silver 的精炼过程，而是 Silver → Gold 的消费行为。放在 SDK 层最合适：

# findatahub-client SDK 内置
client.daily("000001.SZ", start="20260101", with_returns=True)
# 返回的 DataFrame 自动包含 pct_chg, return_1d, return_5d 等衍生列

六、第三层：数据使用层

参考 TuShare 的 AI 适配三层架构^[1]，FinDataHub 的数据使用层同样设计了三层递进的接入方式：

flowchart TB
    subgraph Users["用户群体"]
        DEV[Python 开发者
量化工程师]
        AI_DEV[AI Coding 用户
Cursor/Claude用户]
        AI_NATIVE[AI 原生用户
零代码用户]
    end

    subgraph L3["🟢 第三层：数据使用层"]
        SDK[Python SDK
findatahub-client]
        API[RESTful HTTP API
FastAPI]
        SKILL[Skills 层
findatahub-skill]
        MCP[MCP 服务
Model Context Protocol]
    end

    DB[(Clean MySQL)]

    DB --> SDK
    DB --> API
    API --> SKILL
    API --> MCP

    DEV -->|pip install| SDK
    DEV -->|HTTP requests| API
    AI_DEV -->|AI 辅助调用| SKILL
    AI_NATIVE -->|自然语言| MCP

    style L3 fill:#f0fdf4,stroke:#059669,stroke-width:2px
  

6.1 SDK 层：Python SDK + RESTful HTTP

Python SDK 设计

# pip install findatahub-client
from findatahub import Client

client = Client(
    host="localhost",
    port=3306,
    user="findata",
    password="***",
    database="fin_data_clean"
)

# 获取日线数据
df = client.daily(
    ts_code="000001.SZ",
    start_date="20260101",
    end_date="20260808",
    fields=["open", "high", "low", "close", "vol", "amount"]
)

# 获取财务指标
df = client.fina_indicator(
    ts_code="000001.SZ",
    start_date="20250101",
    fields=["roe", "roa", "eps", "bps"]
)

# 高级查询：按行业获取估值数据
df = client.query("""
    SELECT b.ts_code, b.name, f.pe, f.pb, f.roe
    FROM stock_basic b
    JOIN fina_indicator f ON b.ts_code = f.ts_code
    WHERE b.industry = '银行' AND f.end_date = '20260630'
""")

RESTful HTTP API 设计

# 启动 API 服务
findatahub serve --port 8000

# --- Endpoints ---
GET  /api/v1/health                    # 健康检查
GET  /api/v1/tables                    # 列出可用数据表
GET  /api/v1/{table}/schema            # 获取表结构
GET  /api/v1/{table}/data              # 查询数据
     ?ts_code=000001.SZ
     &start_date=20260101
     &end_date=20260808
     &fields=open,high,low,close
     &limit=1000
POST /api/v1/query                     # 高级 SQL 查询（参数化）

6.2 Skills 层：AI 智能体"读懂"数据

参考 TuShare 的 tushare-data Skill 设计^[1]：

• 接口文档结构化：将所有数据表的字段说明、参数定义、调用示例转化为 AI 可理解的结构化描述
• 调用逻辑内置：AI 无需用户告知具体表名和字段名，只需理解自然语言意图即可自主匹配
• 错误处理指引：包含常见错误场景和处理策略，降低 AI 调用失败率

实际效果：用户说"帮我提取平安银行过去 5 年的 ROE 和归母净利润"，AI 智能体可自主完成：识别对应表 (fina_indicator, income) → 匹配字段 (roe, n_income_attr_p) → 构造查询参数 → 执行查询 → 格式化输出。

6.3 MCP 层：AI 原生数据通道

MCP 服务的核心能力：

FinDataHub MCP 服务的 Tool 定义示例：

// MCP Tools
{
  "tools": [
    {
      "name": "get_daily_data",
      "description": "获取A股日线行情数据。支持按股票代码、日期范围查询。",
      "parameters": {
        "ts_code": "股票代码，如 000001.SZ",
        "start_date": "起始日期 YYYYMMDD",
        "end_date": "结束日期 YYYYMMDD",
        "fields": ["open", "high", "low", "close", "vol", "amount"]
      }
    },
    {
      "name": "get_financial_indicator",
      "description": "获取财务指标数据，包括ROE、ROA、EPS、PE、PB等。",
      "parameters": {
        "ts_code": "股票代码",
        "start_date": "起始报告期 YYYYMMDD",
        "end_date": "结束报告期 YYYYMMDD",
        "indicators": ["roe", "roa", "eps", "pe", "pb"]
      }
    },
    {
      "name": "search_stock",
      "description": "搜索股票，按名称、行业、地区筛选。",
      "parameters": {
        "keyword": "搜索关键词",
        "industry": "行业（可选）",
        "market": "市场（可选）:主板/创业板/科创板"
      }
    }
  ]
}

6.4 三层对比总结

七、模块化重构与流水线编排

7.1 现状问题：模块重复与 ad-hoc 脚本

对所有 Python 脚本的分析揭示出以下系统性问题：

7.2 解决方案：findatahub-core 统一公共核心

将上述所有重复模块收敛到单一公共包，各 fetcher 和脚本通过 pip install 引用，不再各自维护一份。

Services/findatahub-core/           # 🆕 统一公共核心包
├── config/
│   ├── pipeline.yaml               # Pipeline 默认配置
│   └── profiles.yaml               # 策略画像配置
├── src/findatahub_core/
│   ├── __init__.py                 # 公共 API 导出
│   ├── context.py                  # PipelineContext + RerouteRequest
│   ├── pipeline.py                 # DAG 编排器（核心）
│   ├── gate.py                     # 四级质量门控
│   ├── cleaner.py                  # 数据清洗（待迁移）
│   ├── imputer.py                  # 缺失值填充（待迁移）
│   ├── outlier.py                  # 异常值检测（待迁移）
│   ├── validator.py                # 统一 DataValidator（合并 tushare + akshare 两版）
│   ├── reporter.py                 # 统一 ReportGenerator（合并两版）
│   ├── cache.py                    # 统一 SyncCache（合并两版）
│   ├── cross_validator.py          # 双源交叉验证
│   ├── db.py                       # 统一 DatabaseManager（合并两版）
│   ├── config.py                   # 统一配置加载（合并三套）
│   ├── logging.py                  # 统一结构化日志
│   └── schema.py                   # Schema 注册表 + 迁移
├── flows/
│   └── stock_daily_flow.py         # 端到端流水线定义
└── pyproject.toml

7.3 模块单一职责契约

每个模块通过 PipelineContext 作为唯一数据载体，签名统一为 (PipelineContext) → PipelineContext：

7.4 Pipeline 编排器：DAG + 并行 + 反馈回路

Pipeline 编排器是整个框架的核心组件。它将各个 Step 组织为有向无环图（DAG），自动处理并行执行和依赖管理。

flowchart TB
    subgraph Parallel["⚡ 并行分支"]
        F1[fetch_tushare]
        F2[fetch_akshare]
    end

    F1 --> G1[gate1_collection]
    F2 --> G1

    G1 --> CL[clean] --> G2[gate2_cleaning] --> IM[impute] --> G3[gate3_ingestion] --> ST[store] --> G4[gate4_post]

    G3 -.->|"🔄 发现 Gate2 漏检
Reroute 回传"| G2

    style Parallel fill:#eff6ff,stroke:#2563eb,stroke-width:1.5px
    style G3 fill:#fffbeb,stroke:#f59e0b,stroke-width:1.5px
  

核心特性

使用示例

from findatahub_core import Pipeline, Step, PipelineContext
from findatahub_core.pipeline import make_fetch_step, make_gate_step, make_clean_step

# 构建 DAG
pipeline = Pipeline("stock_daily_sync")
pipeline.add_step(Step("fetch_tushare", make_fetch_step(tushare_fetcher), depends_on=[]))
pipeline.add_step(Step("fetch_akshare",  make_fetch_step(akshare_fetcher),  depends_on=[]))
pipeline.add_step(Step("gate1", make_gate_step("gate1_collection"), depends_on=["fetch_tushare"]))
pipeline.add_step(Step("clean", make_clean_step(cleaner), depends_on=["gate1"]))
pipeline.add_step(Step("gate2", make_gate_step("gate2_cleaning"), depends_on=["clean"], max_reroute=3))
# ... 后续步骤 ...

# 执行
ctx = PipelineContext(batch_id="...", source="tushare", table_name="stock_daily")
result = pipeline.run(ctx)  # 自动 DAG 执行，支持并行 + reroute

7.5 未来扩展点

八、实施路线图

8.1 分阶段推进

8.2 关键里程碑与验收标准

九、架构规范

以下规范定义了 FinDataHub 数据管道的开发法则。所有代码变更必须遵循。

9.1 依赖规则（铁律）

依赖方向：采集层 → 核心层 → 使用层。禁止反向引用。

9.2 组件契约

9.3 可扩展性模式

9.4 代码复检清单

1. ✅ core 层不 import fetcher
2. ✅ pyproject.toml 声明所有跨包依赖
3. ✅ 无裸 except Exception
4. ✅ SQL 标识符经 safe_identifier 白名单校验
5. ✅ 无硬编码密码
6. ✅ 生产代码使用 structlog
7. ✅ 异常值/缺失值用标记列，不删除原始数据
8. ✅ 使用 INSERT ON DUPLICATE KEY UPDATE（幂等写入）

9.5 测试策略

十、附录A：预处理核心代码

以下是实现预处理层最关键模块的参考代码骨架。每段代码对应前文设计中的一个具体组件。

9.1 数据清洗器

# cleaner.py — 数据清洗器
import pandas as pd
import numpy as np

class DataCleaner:
    """数据清洗器：类型修正、去重、列名标准化、空值标记、单位统一."""

    # TuShare ↔ AKShare 列名映射表
    COLUMN_MAP = {
        '成交额': 'amount', '成交量': 'vol', '收盘价': 'close',
        '开盘价': 'open', '最高价': 'high', '最低价': 'low',
        '涨跌幅': 'pct_chg', '股票代码': 'ts_code', '日期': 'trade_date',
    }

    # 空值表示形式
    NULL_PATTERNS = ['', '--', 'N/A', 'null', 'None', 'nan']

    def clean(self, df: pd.DataFrame, source: str = 'tushare') -> pd.DataFrame:
        """执行完整清洗流程，返回清洗后的 DataFrame."""
        df = self._fix_types(df, source)
        df = self._drop_duplicates(df)
        df = self._standardize_columns(df)
        df = self._mark_nulls(df)
        df = self._unify_units(df, source)
        return df

    def _fix_types(self, df, source):
        """日期转 DATE，数字转 numeric."""
        if 'trade_date' in df.columns:
            df['trade_date'] = pd.to_datetime(df['trade_date']).dt.strftime('%Y%m%d')
        numeric_cols = ['open', 'high', 'low', 'close', 'vol', 'amount', 'pct_chg']
        for col in numeric_cols:
            if col in df.columns:
                df[col] = pd.to_numeric(df[col], errors='coerce')
        return df

    def _drop_duplicates(self, df):
        """基于 (ts_code, trade_date) 去重."""
        if 'ts_code' in df.columns and 'trade_date' in df.columns:
            return df.drop_duplicates(subset=['ts_code', 'trade_date'], keep='last')
        return df

    def _standardize_columns(self, df):
        """中文列名 → 英文列名."""
        df = df.rename(columns=self.COLUMN_MAP)
        df.columns = df.columns.str.strip().str.lower().str.replace(r'[^a-z0-9_]', '_', regex=True)
        return df

    def _mark_nulls(self, df):
        """空字符串/占位符 → NaN."""
        for pat in self.NULL_PATTERNS:
            df = df.replace(pat, np.nan)
        return df

    def _unify_units(self, df, source):
        """单位统一：AKShare amount 可能是元，TuShare 是千元，统一为元."""
        if source == 'tushare' and 'amount' in df.columns:
            df['amount'] = df['amount'] * 1000  # 千元 → 元
        return df

9.2 缺失值处理器

# imputer.py — 缺失值处理器
import pandas as pd
import numpy as np

class DataImputer:
    """缺失值处理：前向填充、线性插值、均值填充，所有填充均标记 _is_imputed."""

    def impute(self, df: pd.DataFrame, strategy: str = 'ffill') -> pd.DataFrame:
        """对数值列执行缺失值处理."""
        df = df.copy()
        numeric_cols = df.select_dtypes(include=[np.number]).columns.tolist()
        # 排除标记列，避免递归标记
        numeric_cols = [c for c in numeric_cols if not c.startswith('_is_')]

        for col in numeric_cols:
            null_mask = df[col].isna()
            if null_mask.sum() == 0:
                continue

            # 初始化标记列
            flag_col = f'_is_{col}_imputed'
            if flag_col not in df.columns:
                df[flag_col] = False

            if strategy == 'ffill':
                df[col] = df[col].ffill()
            elif strategy == 'linear':
                df[col] = df[col].interpolate(method='linear')
            elif strategy == 'median':
                median_val = df[col].median()
                df[col] = df[col].fillna(median_val)

            # 标记哪些行被填充了
            df.loc[null_mask, flag_col] = True

        return df

9.3 异常值检测器

# outlier.py — 异常值检测器
import pandas as pd
import numpy as np

class OutlierDetector:
    """异常值检测：IQR / MAD / 绝对阈值，区分录入错误 vs 统计异常 vs 真实事件."""

    def detect(self, df: pd.DataFrame, method: str = 'iqr',
               threshold: float = 3.0) -> pd.DataFrame:
        """检测异常值，返回带 _is_outlier 标记的 DataFrame."""
        df = df.copy()
        price_cols = ['open', 'high', 'low', 'close', 'vol']

        for col in price_cols:
            if col not in df.columns:
                continue
            flag_col = f'_is_{col}_outlier'
            df[flag_col] = False

            # Level 1: 绝对阈值（录入错误）
            if col in ['open', 'high', 'low', 'close']:
                df.loc[df[col] <= 0, flag_col] = True  # 价格不能为负

            # Level 2: 统计异常（IQR 或 MAD）
            if method == 'iqr':
                Q1 = df[col].quantile(0.25)
                Q3 = df[col].quantile(0.75)
                IQR = Q3 - Q1
                lower, upper = Q1 - threshold * IQR, Q3 + threshold * IQR
                stat_outliers = (df[col] < lower) | (df[col] > upper)
            elif method == 'mad':
                median = df[col].median()
                mad = np.median(np.abs(df[col] - median))
                mad_norm = mad * 1.4826
                lower, upper = median - threshold * mad_norm, median + threshold * mad_norm
                stat_outliers = (df[col] < lower) | (df[col] > upper)

            df.loc[stat_outliers, flag_col] = True

        # Level 3: 涨跌停豁免（真实事件，不标记）
        if 'pct_chg' in df.columns:
            df.loc[df['pct_chg'].abs() > 9.5, f'_is_close_outlier'] = False

        return df

9.4 质量门控器

# quality_gate.py — 质量门控器（最核心组件）
from dataclasses import dataclass, field
from typing import Optional
import pandas as pd
import numpy as np

@dataclass
class GateResult:
    """单级门控结果."""
    gate_name: str
    passed: bool = True
    checks: list[dict] = field(default_factory=list)

class QualityGate:
    """四级质量门控链式执行器."""

    def __init__(self, batch_name: str):
        self.batch_name = batch_name
        self.results: list[GateResult] = []

    def gate1_collection(self, df, expected_count: int) -> GateResult:
        """Gate 1: 采集门控."""
        result = GateResult(gate_name="Gate1-采集门控")
        # 非空
        if df is None or len(df) == 0:
            result.checks.append({"check": "non_empty", "passed": False, "severity": "block"})
            result.passed = False
        else:
            result.checks.append({"check": "non_empty", "passed": True, "severity": "block"})
        # 行数阈值
        if len(df) < expected_count * 0.5:
            result.checks.append({"check": "min_rows", "passed": False, "severity": "block",
                                  "detail": f"got {len(df)}, expected >= {int(expected_count*0.5)}"})
            result.passed = False
        else:
            result.checks.append({"check": "min_rows", "passed": True, "severity": "block"})
        self.results.append(result)
        return result

    def gate2_cleaning(self, df) -> GateResult:
        """Gate 2: 清洗门控 — OHLC 关系、价格非负、涨跌幅."""
        result = GateResult(gate_name="Gate2-清洗门控")
        # OHLC 关系
        if all(c in df.columns for c in ['open', 'high', 'low', 'close']):
            ohlc_ok = (
                (df['high'] >= df[['open', 'close']].max(axis=1)) &
                (df['low'] <= df[['open', 'close']].min(axis=1))
            )
            fail_count = (~ohlc_ok).sum()
            result.checks.append({
                "check": "ohlc_relation", "passed": fail_count == 0,
                "severity": "warn", "detail": f"{fail_count} rows violate OHLC rule"
            })
        # 价格非负
        for col in ['open', 'close']:
            if col in df.columns:
                neg_count = (df[col] <= 0).sum()
                result.checks.append({
                    "check": f"{col}_non_negative", "passed": neg_count == 0,
                    "severity": "block" if neg_count > 0 else "info",
                    "detail": f"{neg_count} rows have {col} <= 0"
                })
                if neg_count > 0:
                    result.passed = False
        self.results.append(result)
        return result

    def gate3_ingestion(self, df, quality_score: float) -> GateResult:
        """Gate 3: 入库门控 — 唯一键、质量分."""
        result = GateResult(gate_name="Gate3-入库门控")
        # 唯一键
        dup_count = df.duplicated(subset=['ts_code', 'trade_date']).sum()
        result.checks.append({
            "check": "unique_key", "passed": dup_count == 0,
            "severity": "block", "detail": f"{dup_count} duplicate rows"
        })
        # 质量分
        result.checks.append({
            "check": "quality_score", "passed": quality_score >= 0.60,
            "severity": "block", "detail": f"score={quality_score:.2f}, threshold=0.60"
        })
        if quality_score < 0.60:
            result.passed = False
        self.results.append(result)
        return result

    def is_blocked(self) -> bool:
        """是否有阻断级错误."""
        for r in self.results:
            for c in r.checks:
                if c['severity'] == 'block' and not c['passed']:
                    return True
        return False

    def can_write(self) -> bool:
        """是否允许写入 clean 库."""
        return not self.is_blocked()

9.5 Pipeline 主控器

# pipeline.py — 预处理 Pipeline 主控器
from cleaner import DataCleaner
from imputer import DataImputer
from outlier import OutlierDetector
from quality_gate import QualityGate

class PreprocessingPipeline:
    """预处理 Pipeline — 串联所有步骤，驱动质量门控."""

    def __init__(self, config: dict, db_raw, db_clean):
        self.config = config
        self.db_raw = db_raw       # tushare_data / akshare_data
        self.db_clean = db_clean   # fin_data_clean
        self.cleaner = DataCleaner()
        self.imputer = DataImputer()
        self.outlier = OutlierDetector()

    def run(self, table_name: str, source: str = 'tushare'):
        """对一张表执行完整预处理流程."""
        # Step 0: 从 raw 库读取
        df = self.db_raw.read_table(table_name)

        # Gate 1: 采集门控
        gate = QualityGate(f"{table_name}_{source}")
        g1 = gate.gate1_collection(df, expected_count=5000)
        if gate.is_blocked():
            self._alert(f"Gate 1 FAILED for {table_name}")
            return None

        # Step 1: 数据清洗
        df = self.cleaner.clean(df, source)

        # Gate 2: 清洗门控
        g2 = gate.gate2_cleaning(df)

        # Step 2: 缺失值处理
        strategy = self.config.get('imputation', {}).get('strategy', 'ffill')
        df = self.imputer.impute(df, strategy)

        # Step 3: 异常值检测
        method = self.config.get('outlier_detection', {}).get('method', 'iqr')
        threshold = self.config.get('outlier_detection', {}).get('threshold', 3.0)
        df = self.outlier.detect(df, method, threshold)

        # Gate 3: 入库门控
        quality_score = self._compute_quality_score(df)
        g3 = gate.gate3_ingestion(df, quality_score)
        if gate.is_blocked():
            self.db_clean.write_rejected(table_name, df)
            self._alert(f"Gate 3 FAILED for {table_name}, score={quality_score:.2f}")
            return None

        # Step 4: 写入 clean 库
        self.db_clean.upsert(table_name, df)
        self.db_clean.update_quality_metrics(table_name, quality_score)

        return {"table": table_name, "rows": len(df), "score": quality_score, "gate_results": gate.results}

    def _compute_quality_score(self, df) -> float:
        """计算数据质量综合分."""
        scores = {}
        key_cols = ['open', 'high', 'low', 'close', 'vol']
        # 完整性
        scores['completeness'] = df[key_cols].notna().all(axis=1).mean()
        # 有效性
        if all(c in df.columns for c in ['high', 'low', 'open', 'close']):
            valid = (df['high'] >= df['low']) & (df['high'] >= df[['open','close']].max(axis=1))
            scores['validity'] = valid.mean()
        else:
            scores['validity'] = 1.0
        scores['accuracy'] = scores['validity']  # 简化
        scores['uniqueness'] = 1 - df.duplicated(subset=['ts_code', 'trade_date']).sum() / len(df)
        # 加权
        weights = {'completeness': 0.25, 'accuracy': 0.35, 'validity': 0.25, 'uniqueness': 0.15}
        return sum(scores[k] * weights[k] for k in weights)

    def _alert(self, message: str):
        """发送告警（日志 + 通知）."""
        import structlog
        logger = structlog.get_logger(__name__)
        logger.error("quality_gate_blocked", message=message)

9.6 质量追踪辅助表

-- 在 fin_data_clean 库中创建辅助表
CREATE TABLE IF NOT EXISTS `_data_issues` (
    `id` BIGINT AUTO_INCREMENT PRIMARY KEY,
    `table_name` VARCHAR(64) NOT NULL,
    `ts_code` VARCHAR(16),
    `trade_date` VARCHAR(10),
    `issue_type` VARCHAR(32) NOT NULL COMMENT 'missing/outlier/ohlc_violation/cross_validation',
    `severity` VARCHAR(16) NOT NULL COMMENT 'block/warn/info',
    `detail` JSON COMMENT '详细信息',
    `created_at` DATETIME DEFAULT CURRENT_TIMESTAMP,
    KEY `idx_table` (`table_name`),
    KEY `idx_type` (`issue_type`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

CREATE TABLE IF NOT EXISTS `_quality_metrics` (
    `id` BIGINT AUTO_INCREMENT PRIMARY KEY,
    `table_name` VARCHAR(64) NOT NULL,
    `batch_time` DATETIME NOT NULL,
    `completeness` DOUBLE,
    `accuracy` DOUBLE,
    `timeliness` DOUBLE,
    `consistency` DOUBLE,
    `uniqueness` DOUBLE,
    `validity` DOUBLE,
    `total_score` DOUBLE NOT NULL,
    `row_count` INT,
    `created_at` DATETIME DEFAULT CURRENT_TIMESTAMP,
    KEY `idx_table_time` (`table_name`, `batch_time`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

CREATE TABLE IF NOT EXISTS `_rejected_records` (
    `id` BIGINT AUTO_INCREMENT PRIMARY KEY,
    `table_name` VARCHAR(64) NOT NULL,
    `ts_code` VARCHAR(16),
    `trade_date` VARCHAR(10),
    `reject_reason` VARCHAR(128),
    `original_row` JSON COMMENT '被拒绝的原始行数据',
    `rejected_at` DATETIME DEFAULT CURRENT_TIMESTAMP,
    KEY `idx_table` (`table_name`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;