基因突变VCF分析系统

项目背景

ANU研究生专题项目，给定VCF基因突变文件和评分算法，写Python脚本分析致病风险，开发包括数据库、ETL管道、API服务和前端看板的数据平台。

GitHub 仓库：https://github.com/LuckLuffy/gene-mutation-platform

项目流程

读取VCF文件并解析每个文件里记录的基因突变信息，然后用指定算法计算每个患者的疾病风险评分，最后用 Z-Score 分出是健康还是患病。

核心问题

1. 解析并记录VCF文件数据
2. 实现查筛改比等功能
3. 开发网页前端
4. 检查数据质量，记录运行结果

关键词：ETL、数据库、API、前端

整体结构

├── 数据库层 (db/) ← 把结果存进数据库 ├── ETL管道 (etl/) ← 把处理流程拆成 Extract → Transform → Load ├── API服务 (api/) ← 用FastAPI暴露接口，浏览器就能访问 ├── 前端看板 (frontend/) ← 用图表展示数据 └── Docker部署 ← 用docker-compose一键启动所有服务

第1步：数据存储

解析VCF文件数据，结果存到数据库里。

-- 6张表的ER关系patients-- 每个VCF文件对应一个患者├── gene_mutations-- 一个患者有多条基因突变记录 (外键 → patients)├── predictions-- 一个患者有一条预测结果 (外键 → patients)└── subnetwork_burdens-- 基因互作网络中的突变负荷 (外键 → patients)data_quality_logs-- 每次质量检查的记录 (独立表)etl_pipeline_logs-- 管道每次运行的记录 (独立表)

为什么分6张表而不是1张大宽表

如果不分表，把所有信息塞一起，患者的名字在每一行突变里都要重复一遍。更新一个患者信息要改N个地方。

分表之后，每张表只描述一个实体：患者是谁、有哪些突变、预测结果是什么、质量检查怎么样。表之间通过外键关联。

支持三种数据库

开发的时候用 SQLite — 不需要装任何东西，一个文件就是数据库。生产环境可以切到 PostgreSQL，同时做了 MySQL 的兼容，因为 Flink CDC 演示需要 MySQL 作为源端。

# db/connection.py 的核心逻辑ifdb_type=='sqlite':engine_url=f"sqlite:///{database}.db"elifdb_type=='postgresql':engine_url=f"postgresql://{user}:{password}@{host}:{port}/{database}"

SQLAlchemy 做的抽象层，切数据库只改配置文件，不用动业务代码。

第2步：ETL管道

实现三层分离的ETL管道：

Extract Transform Load ┌──────────┐ ┌──────────────────┐ ┌──────────┐ │ 读VCF文件 │ → │ 1.清洗 (去脏) │ → │ 写patients│ │ 解析INFO │ │ 2.校验 (质量检查) │ │ 写mutations│ │ 返回字典 │ │ 3.特征 (算评分) │ │ 写predictions│ └──────────┘ └──────────────────┘ │ 写质量日志 │ └──────────┘

对应代码结构：

etl/ ├── extractors/vcf_extractor.py → Extract ├── transformers/data_cleaner.py → Transform: 清洗 ├── transformers/data_validator.py → Transform: 校验 ├── transformers/feature_engineer.py → Transform: 特征工程 ├── loaders/db_loader.py → Load └── pipeline.py → 编排这5步

YAML 配置驱动

致病阈值放到pipeline_config.yaml：

scoring:cadd_threshold:20sift_threshold:0.05polyphen_threshold:0.85primateai_threshold:0.7

想调试模型效果改YAML就行，不用碰代码。

第3步：数据质量

设置DataValidator，对清洗后的数据做5项检查：

classDataValidator:defvalidate(self,data):checks={'null_check':self._check_nulls(data),# CADD等字段是否缺失'range_check':self._check_ranges(data),# AF是否在[0,1]、CADD是否≥0'duplicate_check':self._check_duplicates(data),# 有没有重复突变记录'distribution_check':...,# 各字段分布是否合理'completeness_check':...,# 必要字段是否齐全}

检查结果

-- 每次检查的结果写入 data_quality_logs 表，永久可追溯INSERTINTOdata_quality_logs(table_name,check_type,failed_rows,total_rows,pass_rate)VALUES('gene_mutations','null_check',3,1616,0.998);

跑完管道后在数据库里能看到：哪次检查、查了什么、多少条没过、通过率多少。管道运行日志同理，记录到etl_pipeline_logs表里。实现数据溯源。

第4步：API服务

用 FastAPI 暴露 RESTful 接口后，浏览器打开http://localhost:8000/docs就能调所有功能。

api/ ├── main.py ← FastAPI入口，注册路由 ├── routers/ ← 每个资源一个router文件 │ ├── patients.py → /api/v1/patients │ ├── mutations.py → /api/v1/mutations │ ├── predictions.py → /api/v1/predictions │ └── data_quality.py → /api/v1/data-quality ├── schemas/ ← Pydantic数据模型（自动校验请求格式） └── services/ ← 业务逻辑（预测服务、查询服务）

为什么分 router 和 service 两层

router 只负责收HTTP请求、调service、返回响应。service 负责具体的业务逻辑。好处是——

• 测预测逻辑时不需要启动服务器，直接调prediction_service.py
• 改预测算法也只改 service，不影响 router 的接口格式

Swagger 文档

FastAPI 自带 Swagger UI，访问/docs就能看到所有接口，支持在线调试，自动生成接口文档，无需手动维护。

第5步：前端看板

纯静态 HTML/CSS/JS，不需要构建工具，直接浏览器打开。5个Tab标签页：

怎么和后端通信

前端通过 HTTP 请求调 FastAPI 的接口：

// frontend/js/api.jsconstAPI_BASE='http://localhost:8000';asyncfunctiongetPatients(page=1,limit=20){consturl=`${API_BASE}/api/v1/patients/?page=${page}&limit=${limit}`;constresponse=awaitfetch(url);constdata=awaitresponse.json();returndata;}

因为是跨域请求（file://访问http://localhost:8000），后端需要加 CORS 中间件，否则浏览器会拦截。

第6步：Docker部署

用 Docker Compose 把整套环境打包：

# docker-compose.yml 核心服务services:api:# FastAPI 应用frontend:# Nginx 托管前端静态文件postgres:# 数据仓库mysql:# CDC 源端（演示用）

docker-composeup-d# 一条命令启动所有服务

Dockerfile实现：装 Python 依赖、复制代码、暴露端口。

核心算法简述

评分算法由老师指定。

1. 共识评分

四个预测工具（CADD、PolyPhen、SIFT、PrimateAI）各自给一个分数，如果四个都认为有害，总分×3：

consensus=0.25*CADD_norm+0.25*PolyPhen+0.25*(1-SIFT)+0.25*PrimateAIif四个都达标:consensus*=3.0

SIFT 要反转（1-SIFT）是因为它"越小越有害"，和其他三个方向相反。

2. 线性回归

score=0.05*(1-AF)+0.12*(10-constraint)+0.28*consensus+1.0*phenotypic

表型(phenotypic)的系数最大(1.0)，因为"已知致病"优先级最高。

3. Z-Score 分类

不设绝对阈值，而是算这个人的分数离哪组（健康/疾病）更近：

z_healthy=|个人分-健康组均值|/健康组标准差 z_disease=|个人分-疾病组均值|/疾病组标准差# 哪个Z-score小就分到哪组

技术栈清单

项目反思

1. ETL三层分离开发便利后期改动— 换个数据源只改提取层，调算法只改转换层
2. YAML配置外置— 改阈值不用翻代码
3. FastAPI + Pydantic— 自动校验减少逻辑冗余
4. Docker Compose— 打包便于移动

如果对你有帮助可以给个 Star ⭐，欢迎提 Issue 讨论。

GitHub：https://github.com/LuckLuffy/gene-mutation-platform