分析层模块

services/analysis/omics/ 包含各组学数据分析功能。

---

模块列表

模块	文件	功能	状态
微生物组	microbiome.py	肠道菌群分析、疾病关联	✅
微生物功能	microbiome_function.py	通路预测、功能分析	✅
基因组学	genomics.py	SNP/CNV 分析、解读	✅
代谢组学	metabolomics.py	代谢物分析、通路富集	✅
蛋白组学	proteomics.py	蛋白表达分析	✅
心血管	cardiovascular.py	心血管风险评估	✅
转录组学	transcriptomics.py	基因表达分析	✅
表观组学	epigenomics.py	DNA 甲基化分析	✅
免疫组学	immunology.py	免疫功能评估	✅
基因芯片	array.py	SNP芯片分析	✅

---

1. 微生物组分析 (microbiome.py)

功能

分析肠道菌群组成，评估健康状态和疾病风险。

流程

样本数据 → 序列处理 → 物种注释 → 多样性分析 → 疾病关联 → 食物推荐

核心函数

from services.analysis.omics.microbiome import (
    build_microbiome_interpretation,
    recommend_foods,
    classify_enterotype,
    get_disease_associations
)

# 构建解读报告
interpretation = build_microbiome_interpretation(
    taxonomy={"Bacteroides": 0.4, "Prevotella": 0.3, "Faecalibacterium": 0.2},
    diversity={"shannon": 3.5, "simpson": 0.85}
)

# 肠型分类
enterotype = classify_enterotype(gut_text=" Bacteroides dominant...")

# 食物推荐
foods = recommend_foods(
    microbiome_analysis=interpretation,
    limit=10
)

# 疾病关联
diseases = get_disease_associations("type_2_diabetes")

输出结构

{
    "enterotype": "Bacteroides",
    "diversity": {"shannon": 3.5, "simpson": 0.85},
    "dominant_species": ["Bacteroides vulgatus", "Bacteroides fragilis"],
    "deficient_species": ["Faecalibacterium prausnitzii"],
    "disease_risks": [
        {"disease": "type_2_diabetes", "risk": "medium", "confidence": 0.75},
        {"disease": "cardiovascular", "risk": "low", "confidence": 0.60}
    ],
    "food_recommendations": [
        {"food": "全谷物", "benefit": "增加Faecalibacterium"},
        {"food": "发酵食品", "benefit": "增加益生菌"}
    ]
}

---

2. 基因组学分析 (genomics.py)

功能

分析基因组变异，生成临床解读报告。

流程

VCF 文件 → 变异注释 → 功能解读 → 药物响应 → 疾病风险 → 报告生成

核心函数

from services.analysis.omics.genomics import (
    build_genomics_interpretation,
    parse_genomics_text,
    extract_abnormalities
)

# 构建基因组解读
interpretation = build_genomics_interpretation(
    vcf_data={"variants": [...], "coverage": 30},
    patient_info={"age": 45, "gender": "male"}
)

# 解析基因组文本
parsed = parse_genomics_text(text="检出 BRCA1 c.68_69delAG 致病突变...")

# 提取异常
abnormalities = extract_abnormalities(text="发现以下致病变异...")

---

3. 代谢组学分析 (metabolomics.py)

功能

分析代谢物谱，评估代谢健康状态。

流程

质谱数据 → 峰检测 → 化合物鉴定 → 通路分析 → 代谢评分 → 报告生成

核心函数

from services.analysis.omics.metabolomics import (
    build_metabolomics_interpretation,
    parse_metabolomics_text,
    extract_abnormalities
)

# 构建代谢组解读
interpretation = build_metabolomics_interpretation(
    metabolites={"glucose": 5.5, "insulin": 25.0, "hba1c": 6.2},
    pathway_data={"glycolysis": 0.8, "tca_cycle": 0.6}
)

---

4. 心血管分析 (cardiovascular.py)

功能

心血管健康风险评估和指标解读。

流程

体检数据 → 指标提取 → 风险计算 → 分层评估 → 建议生成

核心函数

from services.analysis.omics.cardiovascular import build_cardiovascular_interpretation

# 构建心血管报告
report = build_cardiovascular_interpretation(
    blood_pressure={"systolic": 130, "diastolic": 85},
    lipids={"ldl": 3.5, "hdl": 1.2, "triglycerides": 2.0},
    age=50,
    gender="male"
)

---

5. 其他组学模块

转录组学 (transcriptomics.py)

from services.analysis.omics.transcriptomics import parse_transcriptomics_text

# 解析转录组数据
parsed = parse_transcriptomics_text(
    text="基因表达谱分析报告...",
    patient_info="患者信息..."
)

表观组学 (epigenomics.py)

from services.analysis.omics.epigenomics import parse_epigenomics_text

parsed = parse_epigenomics_text(
    text="甲基化分析结果..."
)

免疫组学 (immunology.py)

from services.analysis.omics.immunology import parse_immunology_text

parsed = parse_immunology_text(
    text="免疫指标分析报告...",
    patient_info="年龄：45，性别：男"
)

蛋白组学 (proteomics.py)

from services.analysis.omics.proteomics import build_proteomics_interpretation

report = build_proteomics_interpretation(
    proteins={"albumin": 40, "globulin": 30}
)

基因芯片 (array.py)

from services.analysis.omics.array import run_array_pipeline, build_array_interpretation

# 运行芯片分析
qc = run_array_pipeline(vcf_path="data/sample.vcf", sample_id="S001")

# 构建解读
interpretation = build_array_interpretation(
    qc=qc,
    high_risk=["BRCA1"],
    drug_resp=["CYP2C19"],
    disease_risks=["type_2_diabetes"]
)

---

通用分析流程

各组学模块遵循统一的分析流程：

原始数据 → 数据清洗 → 特征提取 → 统计检验 → 解读生成 → 报告输出

异常提取

所有组学模块都支持从文本中提取异常指标：

def extract_abnormalities(text: str) -> List[Dict]:
    """从文本报告提取异常指标"""
    # 实现细节...

输入验证

def validate_sample_data(data: Dict) -> bool:
    """验证样本数据格式"""
    required_fields = ["sample_id", "data_type"]
    return all(field in data for field in required_fields)