clinicaltrials-database-search

ClinicalTrials.gov 数据库 — 临床试验搜索

概述

查询 ClinicalTrials.gov API v2（公开，无需身份验证），搜索和检索全球临床试验数据。支持按疾病、干预措施、地点、申办方和状态搜索；通过 NCT ID 检索详细的研究信息；对大型结果集进行分页；以及导出为 CSV。

适用场景

• 搜索针对特定疾病或病症、正在招募受试者的临床试验
• 查找测试特定药物、设备或干预措施的试验
• 定位特定地理区域内的试验，以便转介患者
• 跟踪申办方或机构的临床试验项目组合
• 检索特定试验的详细入选标准、结局和联系人信息
• 分析某个治疗领域的临床试验趋势（阶段、招募人数、时间线）
• 导出试验数据，用于系统综述或荟萃分析
• 监测试验状态变化和结果发布情况
• 化合物生物活性数据请使用 chembl-database-bioactivity；已发表文献请使用 pubmed-database

前置条件

uv pip install requests pandas

API 详情：

• 基础 URL：https://clinicaltrials.gov/api/v2
• 身份验证：无需身份验证（公开 API）
• 速率限制：每个 IP 每分钟约 50 次请求
• 响应格式：JSON（默认）、CSV
• 最大页面大小：每次请求 1000 项研究
• 日期格式：ISO 8601；文本字段使用 CommonMark Markdown

快速入门

import requests
import time

CT_API = "https://clinicaltrials.gov/api/v2"

def ct_search(params):
    """Reusable helper for ClinicalTrials.gov searches."""
    response = requests.get(f"{CT_API}/studies", params=params, timeout=30)
    response.raise_for_status()
    return response.json()

# Search for recruiting breast cancer trials
results = ct_search({
    "query.cond": "breast cancer",
    "filter.overallStatus": "RECRUITING",
    "pageSize": 10,
    "sort": "LastUpdatePostDate:desc"
})
print(f"Found {results['totalCount']} trials")
for study in results['studies'][:3]:
    nct = study['protocolSection']['identificationModule']['nctId']
    title = study['protocolSection']['identificationModule']['briefTitle']
    print(f"  {nct}: {title}")

核心概念

响应数据结构

ClinicalTrials.gov 返回深度嵌套的 JSON。关键导航路径：

数据	路径
NCT ID	study['protocolSection']['identificationModule']['nctId']
标题	study['protocolSection']['identificationModule']['briefTitle']
状态	study['protocolSection']['statusModule']['overallStatus']
阶段	study['protocolSection']['designModule']['phases']
招募人数	study['protocolSection']['designModule']['enrollmentInfo']['count']
入选标准	study['protocolSection']['eligibilityModule']
地点	study['protocolSection']['contactsLocationsModule']['locations']
干预措施	study['protocolSection']['armsInterventionsModule']['interventions']
结果	study.get('resultsSection')（未发布结果时为 None）

研究状态值

状态	说明
RECRUITING	当前正在招募受试者
NOT_YET_RECRUITING	已获批，但尚未开始招募
ENROLLING_BY_INVITATION	仅通过邀请招募
ACTIVE_NOT_RECRUITING	研究进行中，已停止招募
SUSPENDED	暂时中止
TERMINATED	提前终止
COMPLETED	研究已结束
WITHDRAWN	招募前撤回

研究阶段值

阶段	说明
EARLY_PHASE1	早期 1 期（原称 0 期）
PHASE1	1 期——安全性和剂量
PHASE2	2 期——疗效和副作用
PHASE3	3 期——大规模疗效评估
PHASE4	4 期——上市后监测
NA	不适用（非药物研究）

查询参数参考

参数	类型	描述	示例
query.cond	string	病症／疾病	lung cancer
query.intr	string	干预措施／药物	Pembrolizumab
query.locn	string	地理位置	New York
query.spons	string	申办方名称	National Cancer Institute
query.term	string	全文搜索	immunotherapy
filter.overallStatus	string	状态筛选（逗号分隔）	RECRUITING,COMPLETED
filter.phase	string	阶段筛选	PHASE2,PHASE3
filter.ids	string	NCT ID 筛选	NCT04852770
sort	string	排序顺序	LastUpdatePostDate:desc
pageSize	int	每页结果数（最多 1000 条）	100
pageToken	string	分页令牌	（来自上一条响应）
format	string	响应格式	json 或 csv

排序选项：LastUpdatePostDate、EnrollmentCount、StartDate、StudyFirstPostDate——每个选项均可使用 :asc 或 :desc。

核心 API

1. 按病症搜索

results = ct_search({
    "query.cond": "type 2 diabetes",
    "filter.overallStatus": "RECRUITING",
    "pageSize": 20,
    "sort": "LastUpdatePostDate:desc"
})
print(f"Found {results['totalCount']} recruiting diabetes trials")
for study in results['studies'][:5]:
    proto = study['protocolSection']
    nct = proto['identificationModule']['nctId']
    title = proto['identificationModule']['briefTitle']
    print(f"  {nct}: {title}")

2. 按干预措施／药物搜索

# Find Phase 3 trials testing Pembrolizumab
results = ct_search({
    "query.intr": "Pembrolizumab",
    "filter.overallStatus": "RECRUITING,ACTIVE_NOT_RECRUITING",
    "filter.phase": "PHASE3",
    "pageSize": 50
})
print(f"Phase 3 Pembrolizumab trials: {results['totalCount']}")

3. 按地点搜索

results = ct_search({
    "query.cond": "cancer",
    "query.locn": "New York",
    "filter.overallStatus": "RECRUITING",
    "pageSize": 20
})

# Extract location details
for study in results['studies'][:3]:
    locs = study['protocolSection'].get('contactsLocationsModule', {}).get('locations', [])
    for loc in locs:
        if 'New York' in loc.get('city', ''):
            print(f"  {loc.get('facility')}: {loc['city']}, {loc.get('state', '')}")

4. 按申办方搜索

results = ct_search({
    "query.spons": "National Cancer Institute",
    "pageSize": 20
})

for study in results['studies'][:5]:
    sponsor_mod = study['protocolSection']['sponsorCollaboratorsModule']
    lead = sponsor_mod['leadSponsor']['name']
    collabs = [c['name'] for c in sponsor_mod.get('collaborators', [])]
    print(f"  Lead: {lead}, Collaborators: {collabs}")

5. 按 NCT ID 获取研究详情

nct_id = "NCT04852770"
response = requests.get(f"{CT_API}/studies/{nct_id}", timeout=30)
response.raise_for_status()
study = response.json()

# Extract key information
proto = study['protocolSection']
print(f"Title: {proto['identificationModule']['briefTitle']}")
print(f"Status: {proto['statusModule']['overallStatus']}")

# Eligibility criteria
elig = proto.get('eligibilityModule', {})
print(f"Ages: {elig.get('minimumAge')} - {elig.get('maximumAge')}")
print(f"Sex: {elig.get('sex')}")
print(f"Criteria:\n{elig.get('eligibilityCriteria', 'N/A')[:300]}")

6. 对大量结果集进行分页

all_studies = []
page_token = None
max_pages = 10

for page in range(max_pages):
    params = {
        "query.cond": "cancer",
        "filter.overallStatus": "RECRUITING",
        "pageSize": 1000,
    }
    if page_token:
        params["pageToken"] = page_token

    results = ct_search(params)
    all_studies.extend(results['studies'])
    page_token = results.get('nextPageToken')

    if not page_token:
        break
    time.sleep(1.5)  # respect rate limits

print(f"Retrieved {len(all_studies)} studies across {page + 1} pages")

7. 导出为 CSV

response = requests.get(f"{CT_API}/studies", params={
    "query.cond": "heart disease",
    "filter.overallStatus": "RECRUITING",
    "format": "csv",
    "pageSize": 1000
}, timeout=60)

with open("heart_disease_trials.csv", "w") as f:
    f.write(response.text)
print("Exported to heart_disease_trials.csv")

常见工作流

工作流 1：多条件临床试验发现

import requests, time

CT_API = "https://clinicaltrials.gov/api/v2"

def ct_search(params):
    response = requests.get(f"{CT_API}/studies", params=params, timeout=30)
    response.raise_for_status()
    return response.json()

# Step 1: Search with multiple filters
results = ct_search({
    "query.cond": "lung cancer",
    "query.intr": "immunotherapy",
    "query.locn": "California",
    "filter.overallStatus": "RECRUITING,NOT_YET_RECRUITING",
    "pageSize": 100,
    "sort": "LastUpdatePostDate:desc"
})
print(f"Total matches: {results['totalCount']}")

# Step 2: Filter by phase
phase23 = [
    s for s in results['studies']
    if any(p in ['PHASE2', 'PHASE3']
           for p in s['protocolSection'].get('designModule', {}).get('phases', []))
]
print(f"Phase 2/3 trials: {len(phase23)}")

# Step 3: Extract summaries
for study in phase23[:5]:
    proto = study['protocolSection']
    nct = proto['identificationModule']['nctId']
    title = proto['identificationModule']['briefTitle']
    enrollment = proto.get('designModule', {}).get('enrollmentInfo', {}).get('count', 'N/A')
    print(f"  {nct}: {title} (n={enrollment})")

工作流 2：已完成试验及结果分析

# Step 1: Find completed trials with posted results
results = ct_search({
    "query.cond": "alzheimer disease",
    "filter.overallStatus": "COMPLETED",
    "pageSize": 100,
    "sort": "LastUpdatePostDate:desc"
})

with_results = [s for s in results['studies'] if s.get('hasResults', False)]
print(f"Completed with results: {len(with_results)} / {len(results['studies'])}")

# Step 2: Get detailed results for top trial
if with_results:
    nct = with_results[0]['protocolSection']['identificationModule']['nctId']
    detail = requests.get(f"{CT_API}/studies/{nct}", timeout=30).json()

    if 'resultsSection' in detail:
        outcomes = detail['resultsSection'].get('outcomeMeasuresModule', {})
        measures = outcomes.get('outcomeMeasures', [])
        for m in measures[:3]:
            print(f"  Outcome: {m.get('title')}")
            print(f"  Type: {m.get('type')}")

工作流 3：申办方项目组合比较

sponsors = ["Pfizer", "Novartis", "Roche"]
for sponsor in sponsors:
    results = ct_search({
        "query.spons": sponsor,
        "filter.overallStatus": "RECRUITING",
        "pageSize": 1
    })
    print(f"{sponsor}: {results['totalCount']} recruiting trials")
    time.sleep(1.5)

常见用法

用法：限速批量搜索

def ct_search_with_retry(params, max_retries=3):
    for attempt in range(max_retries):
        try:
            response = requests.get(f"{CT_API}/studies", params=params, timeout=30)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.HTTPError as e:
            if e.response.status_code == 429:
                wait = 60
                print(f"Rate limited. Waiting {wait}s...")
                time.sleep(wait)
            else:
                raise
        except requests.exceptions.RequestException:
            if attempt == max_retries - 1:
                raise
            time.sleep(2 ** attempt)
    raise Exception("Max retries exceeded")

用法：提取研究摘要

def extract_summary(study):
    proto = study.get('protocolSection', {})
    ident = proto.get('identificationModule', {})
    status = proto.get('statusModule', {})
    design = proto.get('designModule', {})
    return {
        'nct_id': ident.get('nctId'),
        'title': ident.get('officialTitle') or ident.get('briefTitle'),
        'status': status.get('overallStatus'),
        'phases': design.get('phases', []),
        'enrollment': design.get('enrollmentInfo', {}).get('count'),
        'last_update': status.get('lastUpdatePostDateStruct', {}).get('date')
    }

# Usage
for study in results['studies'][:3]:
    s = extract_summary(study)
    print(f"{s['nct_id']}: {s['status']} | Phase: {s['phases']} | n={s['enrollment']}")

用法：安全访问字段

def safe_get(study, *keys, default='N/A'):
    """Navigate nested study JSON safely."""
    current = study
    for key in keys:
        if isinstance(current, dict):
            current = current.get(key)
        else:
            return default
        if current is None:
            return default
    return current

# Usage — handles missing fields gracefully
nct = safe_get(study, 'protocolSection', 'identificationModule', 'nctId')
phases = safe_get(study, 'protocolSection', 'designModule', 'phases', default=[])
enrollment = safe_get(study, 'protocolSection', 'designModule', 'enrollmentInfo', 'count')

关键参数

参数	端点	默认值	描述
query.cond	search	—	病症／疾病搜索词
query.intr	search	—	干预措施／药物搜索词
query.locn	search	—	地理位置筛选
query.spons	search	—	申办方／组织筛选
query.term	search	—	全文搜索
filter.overallStatus	search	all	逗号分隔的状态值
filter.phase	search	all	逗号分隔的阶段值
pageSize	search	10	每页结果数（最多 1000 条）
sort	search	relevance	{field}:{asc|desc}
format	both	json	json 或 csv
timeout	（客户端）	30s	在 requests 调用中设置

故障排除

问题	原因	解决方案
429 Too Many Requests	超出速率限制（约 50 次／分钟）	等待 60s；将 pageSize 设为最大值 1000；实现指数退避
studies 数组为空	没有试验符合筛选条件	放宽搜索条件（移除状态／阶段筛选）；检查拼写
400 Bad Request	参数值无效	确认状态／阶段值与枚举完全匹配（例如，使用 RECRUITING 而非 recruiting）
缺少 resultsSection	试验没有已发布的结果	访问结果前检查 study['hasResults']
嵌套字段出现 KeyError	并非所有试验都包含所有模块	使用带默认值的 .get() 或 safe_get 辅助函数（见“常见用法”）
分页提前停止	缺少 nextPageToken	已获取全部结果；检查 totalCount 与已收集数量
CSV 格式与 JSON 不同	字段结构不同	CSV 会扁平化嵌套结构；程序化访问请使用 JSON
大型导出超时	CSV 包含大量结果	增加超时时间；改用 pageSize=1000 分页获取

最佳实践

• 使用最大页面大小（1000）进行批量检索，以尽量减少请求次数，避免触及速率限制
• 访问 resultsSection 前务必检查 hasResults——大多数试验没有已公布的结果
• 通过 .get() 链安全导航——并非所有试验都会填充所有模块（尤其是 contactsLocationsModule 和 armsInterventionsModule）
• 使用逗号指定多个状态值（例如 RECRUITING,NOT_YET_RECRUITING）——不要针对每个状态分别发送请求
• 默认使用 sort=LastUpdatePostDate:desc——按最近更新的试验优先返回
• 日期解读：lastUpdatePostDateStruct.date 是 ISO 8601 字符串；type 字段表示 ACTUAL 还是 ESTIMATED

相关技能

• pubmed-database — 发布文献搜索，可补充试验注册数据
• chembl-database-bioactivity — 研究中药物的化合物生物活性数据
• bioservices-multi-database — 通过统一的 Python 接口访问其他数据库

参考资料

• ClinicalTrials.gov API 文档：https://clinicaltrials.gov/data-api/api
• API 迁移指南（v1→v2）：https://clinicaltrials.gov/data-api/about-api/api-migration
• ClinicalTrials.gov 主页：https://clinicaltrials.gov/
• OpenAPI 规范：https://clinicaltrials.gov/data-api/about-api/api-spec

捆绑资源

自包含条目。原始总计：866 行（SKILL.md 507 行 + api_reference.md 359 行）。脚本：216 行（query_clinicaltrials.py）。

原始文件处理方式：

• SKILL.md（507 行）→ 核心 API 模块 1-7（条件、干预、地点、申办方、详情、分页、CSV 导出）。“核心功能”第 1-10 节已合并：按条件搜索 → 模块 1，按干预搜索 → 模块 2，地理搜索 → 模块 3，按申办方搜索 → 模块 4，获取研究详情 → 模块 5，分页 → 模块 6，数据导出 → 模块 7，组合查询 → 工作流 1，提取摘要 → 配方。“资源”部分的占位内容已移除，相关内容已合并至正文。按使用场景的处理方式：患者匹配 → “适用场景”要点 + 工作流 1；研究分析 → “适用场景” + 工作流 2；药物追踪 → “适用场景” + 模块 2；地理搜索 → 模块 3；申办方追踪 → 模块 4 + 工作流 3；数据导出 → 模块 7；试验监控 → “适用场景”要点；资格筛查 → 模块 5
• references/api_reference.md（359 行）→ 已全部合并至正文：端点参数 → “关键概念”中的“查询参数参考”表格；状态/阶段值 → “关键概念”表格；响应结构 → “关键概念”中的“响应数据结构”表格；HTTP 错误代码 → “故障排除”表格；速率限制指南 → “先决条件” + “最佳实践”；使用场景 → 与主 SKILL.md 示例重复的内容，已并入核心 API；数据标准（ISO 8601、CommonMark）→ “先决条件”注释。错误处理模式 → “配方”中的“受速率限制的批量搜索”
• scripts/query_clinicaltrials.py（216 行）→ 辅助函数模式：search_studies() → “快速入门”中的 ct_search() 辅助函数；get_study_details() → 模块 5 内联内容；search_with_all_results() → 模块 6 分页模式；extract_study_summary() → “配方”中的“提取研究摘要”。采用了精简封装方案——每个函数都是对 requests.get() 的简单封装

保留比例：约 465 行 / 原始 866 行（不含脚本）= 约 54%。