⚡

细胞基因普查Skill cellxgene-census

这个技能提供程序化访问CZ CELLxGENE普查中的标准化单细胞基因组数据，支持按细胞类型、组织、疾病等条件高效查询和分析6100万+细胞，可集成scanpy和PyTorch进行大规模单细胞分析、机器学习和数据可视化。关键词：单细胞分析、基因表达、细胞普查、scanpy、PyTorch、数据查询、生物信息学。

数据分析 0 次安装 0 次浏览 更新于 3/20/2026

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

CZ CELLxGENE普查

概述

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

普查包括：

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

何时使用此技能

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

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

安装和设置

安装普查API：

uv pip install cellxgene-census

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

uv 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的Ensembl ID
  • 检查数据集存在矩阵以查看基因是否已测量
  • 某些基因可能在普查构建期间被过滤

版本不一致

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