tooluniverse-data-wrangling

数据整理：通用访问模式

从任意来源下载和解析科学数据的参考指南。每一步都通过 Bash 编写并运行 Python 代码。

何时使用

• ToolUniverse 工具返回了元数据或搜索结果，但你需要原始数据或批量数据
• 数据采用工具无法解析的格式（VCF、h5ad、BAM、SDF、GCT）
• 你需要执行多步 API 工作流（搜索 -> 筛选 -> 下载 -> 解析）
• 数据源完全没有对应的 ToolUniverse 工具
• 你需要数千条记录，而不是工具返回的 10-100 条

决策：使用工具还是代码

情况	使用方式
查询单条记录、简单搜索、结果少于 100 条	ToolUniverse 工具（execute_tool）
批量下载、自定义筛选、格式转换	编写 Python 代码
工具存在，但返回结果被截断	使用该工具封装的同一 API 编写代码
此数据源没有对应的工具	直接编写代码

---

A 部分：格式指南

表格数据

import pandas as pd, io

df = pd.read_csv("data.csv")                                # CSV
df = pd.read_csv("data.tsv", sep="\t")                      # TSV
df = pd.read_sas(io.BytesIO(content), format="xport")       # SAS Transport (XPT) — NHANES, CDC
df = pd.read_sas("data.sas7bdat", format="sas7bdat")        # SAS native
df = pd.read_stata("data.dta")                               # Stata — ICPSR, HRS
df = pd.read_parquet("data.parquet")                         # Parquet — MIMIC-IV
df = pd.read_excel("data.xlsx")                              # Excel
df = pd.read_spss("data.sav")                                # SPSS
df = pd.read_fwf("data.dat")                                 # Fixed-width — legacy surveys

基因组学

from Bio import SeqIO
records = list(SeqIO.parse("seqs.fasta", "fasta"))           # FASTA
records = list(SeqIO.parse("reads.fastq", "fastq"))          # FASTQ

# VCF (no cyvcf2 needed)
vcf_lines = [l for l in open("vars.vcf") if not l.startswith("##")]
df = pd.read_csv(io.StringIO("".join(vcf_lines)), sep="\t")

df = pd.read_csv("genes.gff3", sep="\t", comment="#",        # GFF/GTF
     names=["seqid","source","type","start","end","score","strand","phase","attrs"])
df = pd.read_csv("regions.bed", sep="\t", header=None,       # BED
     names=["chrom","start","end","name","score","strand"])

import pysam                                                  # BAM (requires pysam)
bam = pysam.AlignmentFile("aligned.bam", "rb")
for read in bam.fetch("chr1", 1000, 2000): print(read.query_name)

结构数据

from Bio.PDB import PDBParser, MMCIFParser
parser = PDBParser(QUIET=True)
structure = parser.get_structure("prot", "structure.pdb")     # PDB

parser = MMCIFParser(QUIET=True)
structure = parser.get_structure("prot", "structure.cif")     # mmCIF

from rdkit import Chem                                        # SDF/MOL (requires rdkit)
supplier = Chem.SDMolSupplier("compounds.sdf")
mols = [m for m in supplier if m is not None]

组学矩阵

import anndata
adata = anndata.read_h5ad("expression.h5ad")                 # AnnData (scRNA-seq, spatial)

import scipy.io
mat = scipy.io.mmread("matrix.mtx")                          # 10X Genomics MTX
barcodes = pd.read_csv("barcodes.tsv", header=None)[0].tolist()
features = pd.read_csv("features.tsv", sep="\t", header=None)[1].tolist()

df = pd.read_csv("expression.gct", sep="\t", skiprows=2)     # GCT (gene expression)

import loompy                                                 # Loom (legacy single-cell)
ds = loompy.connect("data.loom")

质谱与流式细胞术

from pyteomics import mzml                                    # mzML (proteomics, requires pyteomics)
spectra = list(mzml.read("spectra.mzML"))

import fcsparser                                              # FCS (flow cytometry, requires fcsparser)
meta, data = fcsparser.parse("sample.fcs", reformat_meta=True)

神经影像

import nibabel as nib                                         # NIfTI (requires nibabel)
img = nib.load("brain.nii.gz")
data = img.get_fdata()  # 3D/4D numpy array

# DICOM (requires pydicom)
import pydicom
dcm = pydicom.dcmread("scan.dcm")
pixel_data = dcm.pixel_array

系统发育学与系统生物学

from Bio import Phylo                                         # Newick/Nexus (BioPython)
tree = Phylo.read("tree.nwk", "newick")
tree = Phylo.read("tree.nex", "nexus")

import libsbml                                                # SBML (systems biology, requires python-libsbml)
reader = libsbml.SBMLReader()
doc = reader.readSBML("model.xml")
model = doc.getModel()

序列化数据

import json, xml.etree.ElementTree as ET, h5py

data = json.load(open("data.json"))                           # JSON
df = pd.read_json("records.json")                             # JSON -> DataFrame
tree = ET.parse("data.xml"); root = tree.getroot()            # XML
f = h5py.File("data.h5", "r"); dataset = f["group/data"][:]   # HDF5

压缩数据

df = pd.read_csv("data.csv.gz")                              # gzip (pandas auto-detects)
df = pd.read_csv("data.tsv.gz", sep="\t")                    # gzip TSV

import zipfile
with zipfile.ZipFile(io.BytesIO(content)) as z:               # ZIP
    df = pd.read_csv(z.open(z.namelist()[0]))

import tarfile
with tarfile.open("archive.tar.gz") as t:                     # tar.gz
    f = t.extractfile(t.getnames()[0])
    df = pd.read_csv(f)

---

B 部分：按领域划分的 API 模式

每个类别都会列出已有的 ToolUniverse 工具，以及如何通过直接调用 API 进一步扩展。

1. NCBI E-utilities（Gene、Nucleotide、Protein、SRA、GEO）

工具：NCBIGene_search、NCBI_search_nucleotide、SRA_search_experiments、geo_search_datasets

import requests
base = "https://eutils.ncbi.nlm.nih.gov/entrez/eutils"
# Search -> get IDs -> fetch records in batches
ids = requests.get(f"{base}/esearch.fcgi?db=gene&term=BRCA1+AND+human&retmax=500&retmode=json").json()
id_list = ids["esearchresult"]["idlist"]
# Fetch in batches of 500
for i in range(0, len(id_list), 500):
    batch = ",".join(id_list[i:i+500])
    data = requests.get(f"{base}/efetch.fcgi?db=gene&id={batch}&retmode=xml").text

2. EBI API（UniProt、PDBe、ChEMBL、Ensembl、InterPro）

工具：UniProt_search、PDBe_*、ChEMBL_*、Ensembl_*、InterPro_*

# UniProt bulk TSV download with cursor pagination
url = "https://rest.uniprot.org/uniprotkb/search?query=organism_id:9606+AND+keyword:kinase&format=tsv&size=500"
all_rows = []
while url:
    resp = requests.get(url)
    all_rows.append(resp.text)
    url = resp.headers.get("Link", "").split(";")[0].strip("<>") if "Link" in resp.headers else None

3. NCI GDC（TCGA/TARGET 癌症数据）

工具：GDC_search_cases、GDC_list_files、GDC_get_clinical_data

# Bulk clinical data with filters
filters = {"op":"and","content":[
    {"op":"=","content":{"field":"project.project_id","value":"TCGA-BRCA"}},
    {"op":"=","content":{"field":"demographic.vital_status","value":"Dead"}}
]}
cases = requests.post("https://api.gdc.cancer.gov/cases", json={
    "filters": filters, "fields": "demographic.vital_status,diagnoses.days_to_death",
    "size": 1000, "from": 0
}).json()["data"]["hits"]

4. CDC 健康调查（NHANES、BRFSS、WONDER）

工具：NHANES_download_and_parse、cdc_data_search_datasets

# Direct NHANES XPT download (any cycle, any component)
cycle, component = "2017-2018", "DEMO_J"
url = f"https://wwwn.cdc.gov/Nchs/Data/Nhanes/Public/2017/DataFiles/{component}.XPT"
df = pd.read_sas(io.BytesIO(requests.get(url).content), format="xport")

5. GWAS 与遗传学（GWAS Catalog、gnomAD、ClinVar）

工具：gwas_search_associations、gnomAD_*、ClinVar_*

# GWAS Catalog full download (37MB TSV, all associations)
url = "https://www.ebi.ac.uk/gwas/api/search/downloads/alternative"
df = pd.read_csv(url, sep="\t")
# Filter locally
hits = df[df["DISEASE/TRAIT"].str.contains("diabetes", case=False, na=False)]

6. 化学（PubChem、ChEMBL、KEGG）

工具：PubChem_*、ChEMBL_*、KEGG_*

# PubChem batch property retrieval (up to 100 CIDs at once)
cids = "2244,5988,3672"  # aspirin, sucrose, ibuprofen
url = f"https://pubchem.ncbi.nlm.nih.gov/rest/pug/compound/cid/{cids}/property/MolecularWeight,XLogP,TPSA/JSON"
props = requests.get(url).json()["PropertyTable"]["Properties"]

7. 表达数据（GEO、ArrayExpress、GTEx）

工具：geo_search_datasets、arrayexpress_search_experiments

# GEO series matrix direct download
geo_id = "GSE12345"
url = f"https://ftp.ncbi.nlm.nih.gov/geo/series/{geo_id[:5]}nnn/{geo_id}/matrix/{geo_id}_series_matrix.txt.gz"
df = pd.read_csv(url, sep="\t", comment="!", index_col=0)

# GTEx bulk expression (median TPM per tissue)
url = "https://storage.googleapis.com/adult-gtex/bulk-gex/v8/rna-seq/GTEx_Analysis_2017-06-05_v8_RNASeQCv1.1.9_gene_median_tpm.gct.gz"
df = pd.read_csv(url, sep="\t", skiprows=2)

8. 临床（ClinicalTrials.gov、FDA/OpenFDA、FAERS）

工具：search_clinical_trials、OpenFDA_*

# ClinicalTrials.gov v2 API with pagination
all_studies = []
token = None
while True:
    params = {"query.cond": "lung cancer", "query.intr": "immunotherapy", "pageSize": 100}
    if token: params["pageToken"] = token
    resp = requests.get("https://clinicaltrials.gov/api/v2/studies", params=params).json()
    all_studies.extend(resp.get("studies", []))
    token = resp.get("nextPageToken")
    if not token: break

9. 文献（PubMed、PMC、EuropePMC）

工具：PubMed_search_articles、EuropePMC_search_articles

# EuropePMC full-text search with cursor
cursor = "*"
all_results = []
while cursor:
    resp = requests.get("https://www.ebi.ac.uk/europepmc/webservices/rest/search",
        params={"query": "BRCA1 AND resistance", "format": "json", "pageSize": 100, "cursorMark": cursor}).json()
    all_results.extend(resp.get("resultList", {}).get("result", []))
    cursor = resp.get("nextCursorMark") if len(all_results) < resp.get("hitCount", 0) else None

10. 数据存储库（Zenodo、Figshare、Dryad、DataCite）

工具：DataCite_search_dois、Zenodo_search_records、Dryad_search_datasets

# Zenodo: search + download files
record = requests.get("https://zenodo.org/api/records", params={"q": "proteomics cancer", "size": 5}).json()["hits"]["hits"][0]
for f in record["files"]:
    content = requests.get(f["links"]["self"]).content  # download each file

11-24. 专业领域

对于以下 14 个其他领域，在需要具体 API 模式时，请阅读 references/specialized-domains.md：

#	领域	主要 API/工具	阅读时机
11	蛋白质组学	PRIDE、MassIVE、ProteomeXchange	下载质谱数据
12	代谢组学	MetaboLights、Metabolomics Workbench、HMDB	代谢物/光谱数据
13	微生物组	MGnify、GMREPO	宏基因组谱
14	生态学	GBIF、iNaturalist、OBIS	物种出现数据
15	模式生物	FlyBase、WormBase、ZFIN、RGD	非人类物种的基因数据
16	通路与网络	Reactome、STRING、BioGRID	导出网络/通路
17	本体	OLS、GO、HPO	遍历术语层级
18	免疫学	IEDB、VDJdb、ImmPort	表位/受体数据
19	药物与制药	PharmGKB、DGIdb、SIDER	药物-基因相互作用
20	成像与图谱	TCIA、HPA、Allen Brain Atlas	成像数据集
21	蛋白质结构	RCSB PDB、AlphaFold	下载 PDB/CIF 文件
22	临床基因组学	ClinVar、ClinGen、CIViC	变异解读批量数据
23	单细胞	cellxgene、ARCHS4	scRNA-seq 数据门户
24	毒理学	CTD、EPA CompTox	化学物质-基因-疾病

---

C 部分：受限／未覆盖的数据源

这些数据源需要注册，或没有对应的 ToolUniverse 工具。对于每个数据源，表格列出了访问要求，以及获得凭据后以编程方式获取数据的方法。

注意：ToolUniverse 提供 2300 多种工具 — 使用 find_tools("your topic") 可发现上文未列出的工具。B 部分介绍了最常见的 API 模式；还有许多数据库使用相同的模式（例如，所有 EBI 数据库都遵循第 2 条中的 EBI REST 模式）。

数据源	访问方式	等待时间	格式	内容
UK Biobank	受限（机构申请）	2-6 个月	CSV/Bulk	50 万名参与者的遗传学、影像和健康记录
dbGaP	受控（PI 申请）	1-3 个月	SRA/VCF/phenotype	数千项研究中的 GWAS 基因型和表型数据
MIMIC-IV	需凭据（PhysioNet）	1-2 周	CSV/Parquet	ICU 临床数据，30 多万次入院记录
ICPSR	注册	1-3 天	Stata/CSV	社会科学／健康科学档案（1 万多项研究）
HRS	注册	1-3 天	Stata	健康与退休研究，涵盖 2 万多名美国老年人，每两年开展一次
ELSA	注册	1-3 天	Stata/SPSS	英国老龄化纵向研究
SHARE	注册	1-2 周	Stata	欧洲健康、老龄化与退休调查（28 个国家）
Materials Project	免费 API 密钥	即时	JSON	15 万多种计算所得的材料属性
Human Cell Atlas	开放	即时	h5ad/loom	涵盖人体组织的单细胞图谱
ADNI	申请	1-2 个月	DICOM/CSV	阿尔茨海默病神经影像、生物标志物和认知数据
OpenNeuro	开放	即时	NIfTI/BIDS	800 多个神经影像数据集
CIBERSORTx	免费注册	即时	GCT/TSV	基于整体表达数据进行细胞类型去卷积
FlowRepository	开放	即时	FCS	流式细胞术实验
SynBioHub	开放	即时	SBOL/GenBank	合成生物学元件和设计

对于受限数据源：在文献中（PubMed）搜索使用该数据集发表的分析。论文会注明其数据来源，并且通常会将衍生数据存入公共存储库（GEO、SRA、Zenodo）。

---

D 部分：通用模式

分页

# Pattern 1: offset + limit (most REST APIs)
all_records = []
offset = 0
while True:
    resp = requests.get(f"{api_url}?offset={offset}&limit=500", timeout=30).json()
    batch = resp.get("data", resp.get("results", resp.get("hits", [])))
    if not batch: break
    all_records.extend(batch)
    offset += len(batch)

# Pattern 2: cursor/token (EuropePMC, ClinicalTrials.gov, UniProt)
token = None
while True:
    params = {"pageSize": 100}
    if token: params["pageToken"] = token
    resp = requests.get(api_url, params=params).json()
    all_records.extend(resp["results"])
    token = resp.get("nextPageToken")
    if not token: break

速率限制与重试

import time
def fetch_with_retry(url, max_retries=3, **kwargs):
    for attempt in range(max_retries):
        resp = requests.get(url, timeout=30, **kwargs)
        if resp.status_code == 200: return resp
        if resp.status_code == 429:  # rate limited
            wait = int(resp.headers.get("Retry-After", 2 ** attempt))
            time.sleep(wait)
        else:
            time.sleep(2 ** attempt)
    raise RuntimeError(f"Failed after {max_retries} retries: {url}")

身份验证

import os
# API key in header (most common)
headers = {"Authorization": f"Bearer {os.environ.get('API_KEY', '')}"}
# API key as query param
params = {"api_key": os.environ.get("API_KEY", "")}
# No auth needed for most scientific APIs (NCBI, EBI, PubChem, GDC, CDC)

使用流式传输批量下载

def download_large_file(url, output_path):
    with requests.get(url, stream=True, timeout=300) as r:
        r.raise_for_status()
        with open(output_path, "wb") as f:
            for chunk in r.iter_content(chunk_size=8192):
                f.write(chunk)

错误处理

resp = requests.get(url, timeout=30)
if resp.status_code != 200:
    raise ValueError(f"HTTP {resp.status_code}: {resp.text[:200]}")
# Guard against HTML error pages (CDC, NCBI return 200 with HTML for missing files)
if resp.content[:5] in (b"<!DOC", b"<html"):
    raise ValueError(f"Server returned HTML error page for {url}")
data = resp.json()  # raises JSONDecodeError if not valid JSON