CAR-T 肿瘤临床试验监测平台：工程化架构与生物信息学应用指南¶

概述¶

CAR-T 肿瘤临床试验监测平台（onc-car-t-trial-monitor.streamlit.app）是一个实时的交互式数据看板，专门追踪血液肿瘤和实体瘤领域的 CAR-T、CAR-NK、CAAR-T 以及 CAR-γδ T 临床试验。数据源自公开注册库 ClinicalTrials.gov，平台提供过滤后的试验列表、三层疾病层级分类（Branch → Category → Entity）、级联筛选、靶点-抗原分类、细胞疗法模态标注、已批准产品叠加、全球地理映射（含国家下钻）、战略景观分析、出版级图表及带溯源标签的 CSV 导出，以及自动生成的方法学段落。该工具定位为研究与教育资源，并非医疗、监管或决策支持系统。

从工程化角度看，平台在设计上强调实时性、可重复性、可维护性，适用于 Linux 环境下的部署与二次开发。本文聚焦其工程化架构与生物信息学应用，帮助开发者和研究人员理解如何构建类似的数据监测系统。

核心知识点¶

1. 数据流水线与架构设计¶

1.1 Live-First 加载与缓存机制¶

平台采用“Live-first”策略：当每日首次访问时，后端从 ClinicalTrials.gov v2 API 实时拉取最新数据，并将结果缓存 24 小时。后续同一日内的所有用户请求均直接从缓存获取，极大降低外部 API 调用次数，同时保证数据时效性。缓存实现依赖于 Streamlit 的 @st.cache_data 装饰器，可通过 ttl=86400 控制过期时间。

import streamlit as st
import requests
from datetime import datetime, timedelta

# 中文注释：使用 Streamlit 缓存数据，有效期 24 小时（86400 秒）
@st.cache_data(ttl=86400, show_spinner="正在获取 ClinicalTrials.gov 最新数据...")
def fetch_trials_data(api_url: str) -> dict:
    """从 ClinicalTrials.gov v2 API 获取 CAR-T 相关试验数据"""
    response = requests.get(api_url)
    response.raise_for_status()
    return response.json()

data = fetch_trials_data("https://clinicaltrials.gov/api/v2/studies?query.cond=CAR-T&pageSize=1000")

1.2 可重复性快照（Reproducibility Pinning）¶

为满足学术引用的严格可重复性要求，平台提供“快照冻结”功能。用户通过侧边栏展开器选择特定日期，系统便加载对应日期版本的数据集（通过 Zenodo 发布的带 DOI 备份实现），而不是实时 API。这一设计使得论文审稿人或读者可以精确复现某次分析。

工程实现要点： - 存储每日数据快照至对象存储（如 AWS S3），路径中包含日期。 - 根据用户选择的日期加载对应快照文件，或回退到实时 API（当快照不存在时提示）。 - DOI 由 Zenodo 自动生成，指向整套数据集的静态版本。

# 中文注释：根据用户选择的日期加载对应快照
def load_snapshot(selected_date: str) -> pd.DataFrame:
    try:
        path = f"s3://car-t-snapshots/{selected_date}.parquet"
        df = pd.read_parquet(path)
        return df
    except Exception:
        st.warning("快照不可用，将回退至实时数据。")
        return fetch_live_data()

1.3 三层疾病本体论（Three-tier Ontology）¶

数据清洗阶段构建了标准化分类体系： - Branch（分支）：血液肿瘤（Heme-onc）、实体瘤（Solid-onc）、混合（Mixed）、未知（Unknown） - Category（20 个二级分桶）：如 B-ALL、淋巴瘤、肺癌、黑色素瘤等 - Entity（约 70 个三级叶子节点）：具体疾病亚型，如 “弥漫大 B 细胞淋巴瘤（DLBCL）”

工程挑战：篮式试验（Basket trials）常包含多个实体，且原始 ClinicalTrials.gov 数据中的 disease condition 字段语义模糊。平台采用基于规则的 NLP 后处理与人工校验队列结合的方式，实现高准确度分类。处理流程为： 1. 从原始条件字段提取关键词。 2. 使用正则表达式与词典映射至初步 Category。 3. 对无法确定者加入审核队列（curation queue），通过众包或领域专家验证。 4. 定期计算分类一致率（κ 系数），监控质量。

2. 前端工程化与交互设计¶

2.1 多标签页架构¶

平台使用 Streamlit 的多页面应用模式，通过侧边栏导航组织内容： - Overview（概览）：一目了然的摘要面板，包含分支芯片、疾病层级旭日图、按类别/靶点/阶段/年份/平台分组的图表、顶级申办方、招募热点等。 - Geography/Map（地理分布）：区域聚合（中国/北美/欧盟/亚太/中东/拉美），国家选择器展示该国阶段与申办方分布，世界地图站点级散点图，多城市下钻表。 - Data（数据表）：可过滤的试验列表，支持列内搜索，每条记录可展开查看详情。 - Deep Dive（深度分析）：四个子标签：By disease（队列焦点）、By target（抗原焦点 + 成熟度热图 + 时间轨迹 + 共靶向 + 申办方承诺 + 空白覆盖）、By product（逐产品组合 + 甘特图 + 活动率）、Strategic landscape（4 项交叉分析）。 - Publication Figures（出版级图表）：11 张经肿瘤学调优的图表（图 1‑10 + 图 12；图 11 为 PRISMA 流程图，位于 Methods 中），所有图表导出时附带来源标签 CSV。 - Methods & Appendix（方法与附录）：自动生成方法叙述、本体论、排除列表、审核队列、分类一致率 κ 等，收纳在二级子标签行中。 - About（关于）：引用信息、免责声明、联系方式。

2.2 级联过滤与状态管理¶

侧边栏提供级联过滤器：首先选择 Branch，随后 Category 下拉框仅显示该分支下的选项，选择 Category 后 Entity 再相应更新。这种 UI 依赖 Streamlit 的 session_state 实现，确保过滤状态在 rerun 时保持。 - 额外过滤器包括：Phase（试验阶段）、Antigen target（抗原靶点）、Cell therapy modality（细胞疗法模式）等。 - 所有过滤器的选择会即时更新数据表与图表。

2.3 导出与视觉定制¶

“Display options”展开器允许全局设置： - 图表格式：PNG（5× 分辨率，用于幻灯片）或 SVG（矢量格式，用于期刊/Illustrator 编辑）。 - 调色板：基于 Tableau‑20 的高对比度调色板，兼顾色盲友好。

3. 生物信息学应用亮点¶

虽然该平台本身不进行高通量数据分析，但其在生物信息学中的价值体现在结构化数据整合与可视化的工程化交付： - 抗原靶点分类：将所有试验的靶点归入标准化抗原字典（CD19, BCMA, CD22, GPC3 等），支持共靶向网络分析。 - 成熟度热图：根据试验阶段与招募状态描绘靶点开发的成熟度。 - 时间轨迹：展现特定靶点或疾病领域试验启动数量的年度变化，反映研发热度趋势。 - 策略景观（Strategic Landscape）：结合产品管线、地理分布、申办方关系进行交叉分析，识别红海与蓝海区域。

这些分析本质上依赖于底层 ETL 管道对 ClinicalTrials.gov 原始数据的清洗、统一、转换，以及将结果映射为适合 Plotly 或 Altair 图形库的数据框。工程化的数据处理使得科研人员无需重复造轮子，即可快速获得可复现的洞察。

代码实操：构建最小化监测管道¶

以下展示如何在 Linux 环境下用 Python 快速搭建一个剥离的监测管道骨架，涵盖数据获取、本体映射与缓存。

import pandas as pd
import requests
import streamlit as st
from datetime import datetime

# ========== 常量定义 ==========
CACHE_TTL = 86400  # 24 小时
API_BASE = "https://clinicaltrials.gov/api/v2/studies"
QUERY = "CAR-T OR CAAR-T OR CAR-NK OR CAR-gamma-delta"

# 中文注释：疾病本体映射字典（仅示例部分）
DISEASE_ONTOLOGY = {
    "Branch": {"DLBCL": "Heme-onc", "Multiple Myeloma": "Heme-onc",
               "NSCLC": "Solid-onc", "Hepatocellular": "Solid-onc",},
    "Category": {"DLBCL": "Lymphoma", "Multiple Myeloma": "Plasma Cell Neoplasm",
                 "NSCLC": "Lung Cancer", "Hepatocellular": "Liver Cancer"},
    "Entity": {"DLBCL": "Diffuse Large B-Cell Lymphoma",
               "Multiple Myeloma": "Multiple Myeloma",
               "NSCLC": "Non-Small Cell Lung Cancer",
               "Hepatocellular": "Hepatocellular Carcinoma"}
}

@st.cache_data(ttl=CACHE_TTL)
def fetch_studies(query: str, page_size: int = 1000) -> pd.DataFrame:
    """从 ClinicalTrials.gov API 获取研究数据并返回 DataFrame"""
    studies = []
    page_token = None
    while True:
        params = {"query.cond": query, "pageSize": page_size}
        if page_token:
            params["pageToken"] = page_token
        resp = requests.get(API_BASE, params=params)
        resp.raise_for_status()
        data = resp.json()
        studies.extend(data.get("studies", []))
        page_token = data.get("nextPageToken")
        if not page_token:
            break
    return pd.json_normalize(studies)  # 展平半结构化数据

def map_ontology(condition: str) -> dict:
    """中文注释：根据条件字符串映射 Branch, Category, Entity"""
    for key in DISEASE_ONTOLOGY["Entity"].keys():
        if key.lower() in condition.lower():
            return {
                "Branch": DISEASE_ONTOLOGY["Branch"][key],
                "Category": DISEASE_ONTOLOGY["Category"][key],
                "Entity": DISEASE_ONTOLOGY["Entity"][key]
            }
    # 未知归于 Unknown
    return {"Branch": "Unknown", "Category": "Unknown", "Entity": "Unknown"}

def main():
    st.title("CAR-T 试验监测管道示例")
    df = fetch_studies(QUERY)

    # 中文注释：应用本体映射
    onto_df = df["protocolSection.conditionsModule.conditions"].apply(
        lambda conds: map_ontology(conds[0] if isinstance(conds, list) else conds)
    )
    df = pd.concat([df, pd.json_normalize(onto_df.to_list())], axis=1)

    st.dataframe(df[[
        "protocolSection.identificationModule.nctId",
        "Branch", "Category", "Entity",
        "protocolSection.statusModule.overallStatus"
    ]])

if __name__ == "__main__":
    main()

该示例展示了从 API 拉取数据、使用本体字典进行分类的基本流程，便于后续扩展出级联过滤、图表绘制等功能。实际生产环境还需加入错误处理、日志记录、数据版本控制等工程化实践。

常见问题¶

Q1: 数据多久更新一次？能否强制刷新？¶

A: 默认缓存 24 小时，每日首次访问自动拉取最新数据。如需即时更新，可在网址末尾添加查询参数 ?clear_cache=1（仅对管理员生效），或通过侧边栏手动清除缓存按钮。

Q2: 如何引用特定日期的平台数据？¶

A: 使用 Reproducibility pinning 功能：展开 “Snapshot” 选项，选择目标日期（如 2024-03-15），平台将加载该日的静态快照，并显示对应的 DOI。您可以在论文中引用该 DOI 和快照日期。

Q3: 为什么某些试验被标记为 “Unknown” 分支？¶

A: 平台的本体规则无法覆盖所有条件描述，尤其是罕见疾病或描述极为笼统（如 “advanced solid tumor”）。这类试验会被归为 “Unknown”，并自动进入审核队列，由人工定期纠正。您若发现分类错误，可通过 About 页面联系团队反馈。

Q4: 导出的图表是矢量图吗？¶

A: 在 “Display options” 中选择 SVG 格式即可导出矢量图，适合期刊投稿或 Illustrator 编辑；选择 PNG（5× resolution）则适用于幻灯片，边缘清晰。

Q5: 如何将平台集成到自己的分析流水线？¶

A: 可通过直接下载 Data 标签页中带溯源标签的 CSV 文件，或通过 Zenodo 上的版本化数据集 (DOI: 10.5281/zenodo.19738097) 在脚本中读取。数据集结构固定，适合自动化分析。

速查表¶

Tab 结构与核心功能¶

级联过滤选项¶

出版物图表索引¶

Fig 11 保留为 PRISMA 筛选流程图，位于 Methods 页面。

扩展阅读¶

• 平台源码与架构细节可参阅姊妹项目 Rheumatology CAR-T Trials Monitor。
• 数据快照与 DOI 引用请访问 Zenodo 记录。
• 若需在 Linux 服务器上部署类似应用，推荐使用 Docker 容器化 Streamlit 服务，通过 Nginx 反向代理并配置 HTTPS。

本平台将复杂的临床试验数据工程化、标准化，为肿瘤免疫治疗研究提供了可重复、可交互的分析入口。掌握其架构设计思想，有助于快速构建面向其他疾病领域的类似监测系统。