细胞基因普查技能Skill cellxgene-census

---

name: cellxgene-census description: “查询CZ CELLxGENE普查数据（超过6100万个细胞）。按细胞类型/组织/疾病筛选，检索表达数据，与scanpy/PyTorch集成，用于群体规模单细胞分析。”

CZ CELLxGENE普查

概述

CZ CELLxGENE普查提供了对来自CZ CELLxGENE Discover的标准化单细胞基因组数据的全面、版本化集合的程序化访问。此技能能够高效查询和分析跨越数千个数据集的数百万个细胞。

普查包括：

• 超过6100万个细胞来自人类和小鼠
• 标准化元数据（细胞类型、组织、疾病、供体）
• 原始基因表达矩阵
• 预计算嵌入和统计信息
• 与PyTorch、scanpy和其他分析工具的集成

何时使用此技能

此技能应在以下情况下使用：

• 按细胞类型、组织或疾病查询单细胞表达数据
• 探索可用的单细胞数据集和元数据
• 在单细胞数据上训练机器学习模型
• 执行大规模跨数据集分析
• 将普查数据与scanpy或其他分析框架集成
• 计算数百万个细胞的统计信息
• 访问预计算的嵌入或模型预测

安装和设置

安装普查API：

pip install cellxgene-census

对于机器学习工作流，安装额外依赖：

pip install cellxgene-census[experimental]

核心工作流模式

1. 打开普查

始终使用上下文管理器以确保资源正确清理：

import cellxgene_census

# 打开最新稳定版本
with cellxgene_census.open_soma() as census:
    # 使用普查数据

# 打开特定版本以实现可重复性
with cellxgene_census.open_soma(census_version="2023-07-25") as census:
    # 使用普查数据

关键点：

• 使用上下文管理器（with语句）进行自动清理
• 指定census_version以实现可重复分析
• 默认打开最新“稳定”版本

2. 探索普查信息

在查询表达数据之前，探索可用数据集和元数据。

访问摘要信息：

# 获取摘要统计信息
summary = census["census_info"]["summary"].read().concat().to_pandas()
print(f"总细胞数：{summary['total_cell_count'][0]}")

# 获取所有数据集
datasets = census["census_info"]["datasets"].read().concat().to_pandas()

# 按条件筛选数据集
covid_datasets = datasets[datasets["disease"].str.contains("COVID", na=False)]

查询细胞元数据以了解可用数据：

# 获取组织中唯一的细胞类型
cell_metadata = cellxgene_census.get_obs(
    census,
    "homo_sapiens",
    value_filter="tissue_general == 'brain' and is_primary_data == True",
    column_names=["cell_type"]
)
unique_cell_types = cell_metadata["cell_type"].unique()
print(f"在大脑中找到了{len(unique_cell_types)}个细胞类型")

# 按组织计数细胞
tissue_counts = cell_metadata.groupby("tissue_general").size()

重要： 始终筛选is_primary_data == True以避免重复计数细胞，除非专门分析重复项。

3. 查询表达数据（小到中等规模）

对于返回适合内存的<100k细胞的查询，使用get_anndata()：

# 基本查询，带细胞类型和组织筛选器
adata = cellxgene_census.get_anndata(
    census=census,
    organism="Homo sapiens",  # 或 "Mus musculus"
    obs_value_filter="cell_type == 'B cell' and tissue_general == 'lung' and is_primary_data == True",
    obs_column_names=["assay", "disease", "sex", "donor_id"],
)

# 查询特定基因，带多个筛选器
adata = cellxgene_census.get_anndata(
    census=census,
    organism="Homo sapiens",
    var_value_filter="feature_name in ['CD4', 'CD8A', 'CD19', 'FOXP3']",
    obs_value_filter="cell_type == 'T cell' and disease == 'COVID-19' and is_primary_data == True",
    obs_column_names=["cell_type", "tissue_general", "donor_id"],
)

筛选语法：

• 使用obs_value_filter进行细胞筛选
• 使用var_value_filter进行基因筛选
• 用and、or组合条件
• 用in处理多个值：tissue in ['lung', 'liver']
• 用obs_column_names选择仅需要的列

单独获取元数据：

# 查询细胞元数据
cell_metadata = cellxgene_census.get_obs(
    census, "homo_sapiens",
    value_filter="disease == 'COVID-19' and is_primary_data == True",
    column_names=["cell_type", "tissue_general", "donor_id"]
)

# 查询基因元数据
gene_metadata = cellxgene_census.get_var(
    census, "homo_sapiens",
    value_filter="feature_name in ['CD4', 'CD8A']",
    column_names=["feature_id", "feature_name", "feature_length"]
)

4. 大规模查询（外部核心处理）

对于超过可用RAM的查询，使用axis_query()进行迭代处理：

import tiledbsoma as soma

# 创建轴查询
query = census["census_data"]["homo_sapiens"].axis_query(
    measurement_name="RNA",
    obs_query=soma.AxisQuery(
        value_filter="tissue_general == 'brain' and is_primary_data == True"
    ),
    var_query=soma.AxisQuery(
        value_filter="feature_name in ['FOXP2', 'TBR1', 'SATB2']"
    )
)

# 以块为单位遍历表达矩阵
iterator = query.X("raw").tables()
for batch in iterator:
    # batch是一个pyarrow.Table，列包括：
    # - soma_data：表达值
    # - soma_dim_0：细胞（obs）坐标
    # - soma_dim_1：基因（var）坐标
    process_batch(batch)

计算增量统计信息：

# 示例：计算平均表达
n_observations = 0
sum_values = 0.0

iterator = query.X("raw").tables()
for batch in iterator:
    values = batch["soma_data"].to_numpy()
    n_observations += len(values)
    sum_values += values.sum()

mean_expression = sum_values / n_observations

5. 使用PyTorch进行机器学习

用于训练模型，使用实验性PyTorch集成：

from cellxgene_census.experimental.ml import experiment_dataloader

with cellxgene_census.open_soma() as census:
    # 创建数据加载器
dataloader = experiment_dataloader(
        census["census_data"]["homo_sapiens"],
        measurement_name="RNA",
        X_name="raw",
        obs_value_filter="tissue_general == 'liver' and is_primary_data == True",
        obs_column_names=["cell_type"],
        batch_size=128,
        shuffle=True,
    )

    # 训练循环
    for epoch in range(num_epochs):
        for batch in dataloader:
            X = batch["X"]  # 基因表达张量
            labels = batch["obs"]["cell_type"]  # 细胞类型标签

            # 前向传播
            outputs = model(X)
            loss = criterion(outputs, labels)

            # 反向传播
            optimizer.zero_grad()
            loss.backward()
            optimizer.step()

训练/测试分割：

from cellxgene_census.experimental.ml import ExperimentDataset

# 从实验创建数据集
dataset = ExperimentDataset(
    experiment_axis_query,
    layer_name="raw",
    obs_column_names=["cell_type"],
    batch_size=128,
)

# 分割为训练和测试
train_dataset, test_dataset = dataset.random_split(
    split=[0.8, 0.2],
    seed=42
)

6. 与Scanpy集成

无缝将普查数据与scanpy工作流集成：

import scanpy as sc

# 从普查加载数据
adata = cellxgene_census.get_anndata(
    census=census,
    organism="Homo sapiens",
    obs_value_filter="cell_type == 'neuron' and tissue_general == 'cortex' and is_primary_data == True",
)

# 标准scanpy工作流
sc.pp.normalize_total(adata, target_sum=1e4)
sc.pp.log1p(adata)
sc.pp.highly_variable_genes(adata, n_top_genes=2000)

# 降维
sc.pp.pca(adata, n_comps=50)
sc.pp.neighbors(adata)
sc.tl.umap(adata)

# 可视化
sc.pl.umap(adata, color=["cell_type", "tissue", "disease"])

7. 多数据集集成

查询和集成多个数据集：

# 策略1：分别查询多个组织
tissues = ["lung", "liver", "kidney"]
adatas = []

for tissue in tissues:
    adata = cellxgene_census.get_anndata(
        census=census,
        organism="Homo sapiens",
        obs_value_filter=f"tissue_general == '{tissue}' and is_primary_data == True",
    )
    adata.obs["tissue"] = tissue
    adatas.append(adata)

# 合并
combined = adatas[0].concatenate(adatas[1:])

# 策略2：直接查询多个数据集
adata = cellxgene_census.get_anndata(
    census=census,
    organism="Homo sapiens",
    obs_value_filter="tissue_general in ['lung', 'liver', 'kidney'] and is_primary_data == True",
)

关键概念和最佳实践

始终筛选主数据

除非分析重复项，否则始终在查询中包含is_primary_data == True以避免多次计数细胞：

obs_value_filter="cell_type == 'B cell' and is_primary_data == True"

指定普查版本以实现可重复性

始终在生产分析中指定普查版本：

census = cellxgene_census.open_soma(census_version="2023-07-25")

在加载前估计查询大小

对于大查询，首先检查细胞数以避免内存问题：

# 获取细胞计数
metadata = cellxgene_census.get_obs(
    census, "homo_sapiens",
    value_filter="tissue_general == 'brain' and is_primary_data == True",
    column_names=["soma_joinid"]
)
n_cells = len(metadata)
print(f"查询将返回{n_cells:,}个细胞")

# 如果太大（>100k），使用外部核心处理

使用tissue_general进行更广泛分组

tissue_general字段提供比tissue更粗的类别，适用于跨组织分析：

# 更广泛的分组
obs_value_filter="tissue_general == 'immune system'"

# 特定组织
obs_value_filter="tissue == 'peripheral blood mononuclear cell'"

只选择需要的列

通过指定仅需的元数据列最小化数据传输：

obs_column_names=["cell_type", "tissue_general", "disease"]  # 不是所有列

检查基因特定查询的数据集存在性

分析特定基因时，验证哪些数据集测量了它们：

presence = cellxgene_census.get_presence_matrix(
    census,
    "homo_sapiens",
    var_value_filter="feature_name in ['CD4', 'CD8A']"
)

两步工作流：先探索再查询

首先探索元数据以了解可用数据，然后查询表达：

# 步骤1：探索可用内容
metadata = cellxgene_census.get_obs(
    census, "homo_sapiens",
    value_filter="disease == 'COVID-19' and is_primary_data == True",
    column_names=["cell_type", "tissue_general"]
)
print(metadata.value_counts())

# 步骤2：基于发现进行查询
adata = cellxgene_census.get_anndata(
    census=census,
    organism="Homo sapiens",
    obs_value_filter="disease == 'COVID-19' and cell_type == 'T cell' and is_primary_data == True",
)

可用的元数据字段

细胞元数据（obs）

关键筛选字段：

• cell_type, cell_type_ontology_term_id
• tissue, tissue_general, tissue_ontology_term_id
• disease, disease_ontology_term_id
• assay, assay_ontology_term_id
• donor_id, sex, self_reported_ethnicity
• development_stage, development_stage_ontology_term_id
• dataset_id
• is_primary_data（布尔值：True = 唯一细胞）

基因元数据（var）

• feature_id（Ensembl基因ID，例如“ENSG00000161798”）
• feature_name（基因符号，例如“FOXP2”）
• feature_length（基因长度，以碱基对计）

参考文档

此技能包括详细参考文档：

references/census_schema.md

全面文档化：

• 普查数据结构与组织
• 所有可用元数据字段
• 值筛选语法和操作符
• SOMA对象类型
• 数据包含标准

何时阅读： 当需要详细模式信息、完整元数据字段列表或复杂筛选语法时。

references/common_patterns.md

示例和模式：

• 探索性查询（仅元数据）
• 小到中等查询（AnnData）
• 大查询（外部核心处理）
• PyTorch集成
• Scanpy集成工作流
• 多数据集集成
• 最佳实践和常见陷阱

何时阅读： 当实现特定查询模式、寻找代码示例或解决常见问题时。

常见用例

用例1：探索组织中的细胞类型

with cellxgene_census.open_soma() as census:
    cells = cellxgene_census.get_obs(
        census, "homo_sapiens",
        value_filter="tissue_general == 'lung' and is_primary_data == True",
        column_names=["cell_type"]
    )
    print(cells["cell_type"].value_counts())

用例2：查询标记基因表达

with cellxgene_census.open_soma() as census:
    adata = cellxgene_census.get_anndata(
        census=census,
        organism="Homo sapiens",
        var_value_filter="feature_name in ['CD4', 'CD8A', 'CD19']",
        obs_value_filter="cell_type in ['T cell', 'B cell'] and is_primary_data == True",
    )

用例3：训练细胞类型分类器

from cellxgene_census.experimental.ml import experiment_dataloader

with cellxgene_census.open_soma() as census:
    dataloader = experiment_dataloader(
        census["census_data"]["homo_sapiens"],
        measurement_name="RNA",
        X_name="raw",
        obs_value_filter="is_primary_data == True",
        obs_column_names=["cell_type"],
        batch_size=128,
        shuffle=True,
    )

    # 训练模型
    for epoch in range(epochs):
        for batch in dataloader:
            # 训练逻辑
            pass

用例4：跨组织分析

with cellxgene_census.open_soma() as census:
    adata = cellxgene_census.get_anndata(
        census=census,
        organism="Homo sapiens",
        obs_value_filter="cell_type == 'macrophage' and tissue_general in ['lung', 'liver', 'brain'] and is_primary_data == True",
    )

    # 分析巨噬细胞跨组织差异
    sc.tl.rank_genes_groups(adata, groupby="tissue_general")

故障排除

查询返回太多细胞

• 添加更具体的筛选器以减少范围
• 使用tissue而不是tissue_general以获取更细粒度
• 如果已知，筛选特定dataset_id
• 对于大查询，切换到外部核心处理

内存错误

• 通过更具限制性的筛选器减少查询范围
• 用var_value_filter选择更少基因
• 使用axis_query()进行外部核心处理
• 分批处理数据

结果中出现重复细胞

• 始终在筛选器中包含is_primary_data == True
• 检查是否故意跨多个数据集查询

找不到基因

• 验证基因名称拼写（区分大小写）
• 尝试使用feature_id而不是feature_name
• 检查数据集存在性矩阵以查看是否测量了基因
• 一些基因可能在普查构建过程中被筛选掉

版本不一致

• 始终明确指定census_version
• 在所有分析中使用相同版本
• 检查版本特定更改的发布说明