edc-data-validation

EDC 数据验证 — 流水线技能

概述

你将为临床数据验证流水线生成确定性的 Python 代码。 当用户提供 EDC 导出数据或 SDTM 数据集时，请按照以下方案生成验证代码。生成代码前，务必确认输入格式和治疗领域。

用法

• 使用范围检查、表单间一致性检查和部分日期推算来验证 EDC 导出数据
• 根据 CDISC 受控术语和结构规则检查 SDTM 数据集
• 生成 define.xml 和按优先级排序的验证报告

核心概念

---

响应格式

• 首先给出用户需要的命令或代码，再进行解释
• 按以下结构组织：确认输入 → 可运行代码 → 解释关键参数 → 注意事项
• 每项任务提供一个完整的可运行示例；不要展示所有替代方案
• 代码注释应简洁实用（说明做什么，而非为什么存在）
• 目标：50–100 行代码，并附简短说明

---

1. 验证优先级决策树

使用此决策树对发现的问题进行分类并分配查询优先级：

Issue Detected
├── Missing/invalid RFSTDTC or DSSTDTC? → CRITICAL (blocks submission)
├── AE date outside study period (before consent or after completion)? → CRITICAL
├── CT violation on required codelist (SEX, AEOUT, AESER)? → CRITICAL
├── Out-of-range vital sign WITHOUT data comment? → HIGH
├── Missing AESER when AEOUT='FATAL' or SAE reported? → HIGH
├── Lab value triggers Hy's Law flag (ALT/AST >3×ULN + BILI >2×ULN)? → HIGH
├── Minor CT mismatch (case difference, trailing space)? → MEDIUM
├── Missing optional variable (AEENDTC, AESEV)? → MEDIUM
├── Formatting issue (date separators, decimal precision)? → LOW
└── Case sensitivity only (e.g., "Male" vs "M")? → LOW

严重程度 → 操作映射：

• CRITICAL → 自动发起查询，阻止数据集锁定
• HIGH → 48 小时内发起查询，审核者会标记
• MEDIUM → 在下次数据截点批量发起查询
• LOW → 留待最终清理，无需发起查询

---

2. 治疗领域决策树

根据研究类型选择范围表：

What is the therapeutic area?
├── Oncology
│   ├── Use RECIST response values: CR, PR, SD, PD, NE
│   ├── Tumor measurement: 0–300 mm (flag >200 as HIGH)
│   └── ECOG valid values: 0, 1, 2, 3, 4, 5
├── Cardiology
│   ├── QTcF: normal <450ms, borderline 450–480ms, prolonged >480ms, critical >500ms
│   ├── SYSBP: 70–200 mmHg (tighter than general)
│   └── HR: 40–150 beats/min (tighter than general)
├── Pediatric (age <18)
│   ├── SYSBP: multiply adult low by 0.7, high by 0.8
│   ├── HR: multiply adult high by 1.3 (higher resting HR)
│   ├── WEIGHT: 2–120 kg
│   └── HEIGHT: 30–200 cm
└── General / Geriatric (age ≥65)
    ├── Use standard ranges (§3 below)
    └── Flag HR >180 as CRITICAL (not just HIGH)

---

3. 范围检查 — 生命体征和实验室检查

import pandas as pd

VS_RANGES = {
    "SYSBP":  {"low": 60, "high": 250, "unit": "mmHg"},
    "DIABP":  {"low": 30, "high": 150, "unit": "mmHg"},
    "HR":     {"low": 30, "high": 220, "unit": "beats/min"},
    "TEMP":   {"low": 34.0, "high": 42.0, "unit": "C"},
    "WEIGHT": {"low": 20, "high": 300, "unit": "kg"},
    "HEIGHT": {"low": 50, "high": 250, "unit": "cm"},
    "RESP":   {"low": 6, "high": 60, "unit": "breaths/min"},
}

LB_RANGES = {
    "ALT":   {"low": 0, "high": 500, "unit": "U/L"},
    "AST":   {"low": 0, "high": 500, "unit": "U/L"},
    "BILI":  {"low": 0, "high": 30, "unit": "mg/dL"},
    "CREAT": {"low": 0, "high": 20, "unit": "mg/dL"},
    "HGB":   {"low": 3, "high": 25, "unit": "g/dL"},
    "WBC":   {"low": 0, "high": 100, "unit": "10^9/L"},
    "PLAT":  {"low": 0, "high": 1500, "unit": "10^9/L"},
    "GLUC":  {"low": 10, "high": 1000, "unit": "mg/dL"},
}

# Pediatric multipliers (apply to VS_RANGES when age < 18)
PEDIATRIC_MULT = {"SYSBP_low": 0.7, "SYSBP_high": 0.8, "HR_high": 1.3}

# Hy's Law: ALT or AST >3×ULN AND BILI >2×ULN (ULN: ALT=40, AST=37, BILI=1.2)
HYS_LAW = {"ALT_ULN": 40, "AST_ULN": 37, "BILI_ULN": 1.2, "ALT_MULT": 3, "BILI_MULT": 2}

def check_ranges(df: pd.DataFrame, testcd_col: str, value_col: str,
                 ranges: dict, domain: str) -> pd.DataFrame:
    issues = []
    for testcd, limits in ranges.items():
        subset = df[df[testcd_col] == testcd]
        oob = subset[(subset[value_col] < limits["low"]) | (subset[value_col] > limits["high"])]
        for _, row in oob.iterrows():
            severity = "HIGH" if testcd in ("SYSBP", "HR", "ALT", "AST", "BILI") else "MEDIUM"
            issues.append({"USUBJID": row.get("USUBJID"), "DOMAIN": domain,
                           "TESTCD": testcd, "VALUE": row[value_col],
                           "EXPECTED": f"{limits['low']}–{limits['high']} {limits['unit']}",
                           "SEVERITY": severity})
    return pd.DataFrame(issues)

---

4. 表单间一致性检查（8 项检查）

def cross_form_checks(datasets: dict[str, pd.DataFrame]) -> pd.DataFrame:
    ae, ds, dm = datasets.get("AE", pd.DataFrame()), datasets.get("DS", pd.DataFrame()), datasets.get("DM", pd.DataFrame())
    sv, cm, ex = datasets.get("SV", pd.DataFrame()), datasets.get("CM", pd.DataFrame()), datasets.get("EX", pd.DataFrame())
    issues = []
    to_dt = lambda s: pd.to_datetime(s, errors="coerce")

    # 1. AE start < informed consent → CRITICAL
    if not ae.empty and "RFICDTC" in dm.columns:
        ic = dm.set_index("USUBJID")["RFICDTC"].apply(to_dt)
        for _, r in ae.assign(AESTDTC=to_dt(ae["AESTDTC"])).iterrows():
            ic_dt = ic.get(r["USUBJID"])
            if pd.notna(ic_dt) and pd.notna(r["AESTDTC"]) and r["AESTDTC"] < ic_dt:
                issues.append({"USUBJID": r["USUBJID"], "CHECK": "AE_BEFORE_CONSENT", "SEVERITY": "CRITICAL"})

    # 2. Fatal AE without DS death record → CRITICAL
    if not ae.empty and not ds.empty and "DSDECOD" in ds.columns:
        death_subj = set(ds[ds["DSDECOD"] == "DEATH"]["USUBJID"])
        for _, r in ae[ae["AEOUT"] == "FATAL"].iterrows():
            if r["USUBJID"] not in death_subj:
                issues.append({"USUBJID": r["USUBJID"], "CHECK": "FATAL_AE_NO_DEATH", "SEVERITY": "CRITICAL"})

    # 3. Disposition date > last visit + 7 days → HIGH
    if not ds.empty and not sv.empty:
        last_sv = sv.groupby("USUBJID")["SVSTDTC"].max().apply(to_dt)
        for _, r in ds.assign(DSSTDTC=to_dt(ds["DSSTDTC"])).iterrows():
            lv = last_sv.get(r["USUBJID"])
            if pd.notna(lv) and pd.notna(r["DSSTDTC"]) and r["DSSTDTC"] > lv + pd.Timedelta(days=7):
                issues.append({"USUBJID": r["USUBJID"], "CHECK": "DS_AFTER_LAST_VISIT", "SEVERITY": "HIGH"})

    # 4. Lab collection on non-visit day → MEDIUM
    if not datasets.get("LB", pd.DataFrame()).empty and not sv.empty:
        sv_set = set(zip(sv["USUBJID"], to_dt(sv["SVSTDTC"]).dt.date))
        lb = datasets["LB"].assign(LBDTC=to_dt(datasets["LB"].get("LBDTC")))
        for _, r in lb.dropna(subset=["LBDTC"]).iterrows():
            if (r["USUBJID"], r["LBDTC"].date()) not in sv_set:
                issues.append({"USUBJID": r["USUBJID"], "CHECK": "LB_NO_VISIT", "SEVERITY": "MEDIUM"})

    # 5. Conmed start after study completion → MEDIUM
    if not cm.empty and "RFPENDTC" in dm.columns:
        end = dm.set_index("USUBJID")["RFPENDTC"].apply(to_dt)
        for _, r in cm.assign(CMSTDTC=to_dt(cm.get("CMSTDTC"))).dropna(subset=["CMSTDTC"]).iterrows():
            e = end.get(r["USUBJID"])
            if pd.notna(e) and r["CMSTDTC"] > e:
                issues.append({"USUBJID": r["USUBJID"], "CHECK": "CM_AFTER_COMPLETION", "SEVERITY": "MEDIUM"})

    # 6. Randomization without signed consent → CRITICAL
    if not dm.empty and "RFICDTC" in dm.columns and "ARMCD" in dm.columns:
        for _, r in dm.iterrows():
            if pd.notna(r.get("ARMCD")) and r["ARMCD"] != "SCRNFAIL" and pd.isna(r.get("RFICDTC")):
                issues.append({"USUBJID": r["USUBJID"], "CHECK": "RAND_NO_CONSENT", "SEVERITY": "CRITICAL"})

    # 7. Unscheduled visit without comment → MEDIUM
    if not sv.empty and "SVUPDES" in sv.columns:
        unsched = sv[sv["VISITNUM"].astype(str).str.contains(r"\.", na=False)]
        no_comment = unsched[unsched["SVUPDES"].isna() | (unsched["SVUPDES"].str.strip() == "")]
        for _, r in no_comment.iterrows():
            issues.append({"USUBJID": r["USUBJID"], "CHECK": "UNSCHED_NO_COMMENT", "SEVERITY": "MEDIUM"})

    # 8. Dose modification without corresponding AE → HIGH
    if not ex.empty and not ae.empty and "EXADJ" in ex.columns:
        ae_subj = set(zip(ae["USUBJID"], to_dt(ae.get("AESTDTC")).dt.date))
        dose_mod = ex[ex["EXADJ"].notna() & (ex["EXADJ"] != "")].assign(EXSTDTC=to_dt(ex["EXSTDTC"]))
        for _, r in dose_mod.dropna(subset=["EXSTDTC"]).iterrows():
            if (r["USUBJID"], r["EXSTDTC"].date()) not in ae_subj:
                issues.append({"USUBJID": r["USUBJID"], "CHECK": "DOSE_MOD_NO_AE", "SEVERITY": "HIGH"})

    return pd.DataFrame(issues)

---

5. 部分日期推算

import calendar

def impute_partial_date(dtc: str, position: str = "start") -> tuple[str, str]:
    """Impute partial ISO 8601 dates. Returns (imputed_date, DTYPE).
    position: 'start' imputes early, 'end' imputes late."""
    if pd.isna(dtc) or dtc.strip() == "":
        return (None, None)
    parts = dtc.split("-")
    if len(parts) == 3 and len(parts[2]) >= 2:
        return (dtc, None)  # Complete date, no imputation
    year = int(parts[0])
    if len(parts) == 1:  # YYYY only
        if position == "start":
            return (f"{year}-01-01", "FIRST")
        return (f"{year}-12-31", "LAST")
    month = int(parts[1])
    if len(parts) == 2:  # YYYY-MM only
        if position == "start":
            return (f"{year}-{month:02d}-01", "FIRST")
        last_day = calendar.monthrange(year, month)[1]
        return (f"{year}-{month:02d}-{last_day}", "LAST")
    return (dtc, None)

---

6. 汇总报告和 define.xml

from lxml import etree

def generate_validation_report(*issue_dfs, output_path="validation_report.csv"):
    """Combine all issues into a single prioritized report."""
    all_issues = pd.concat([df for df in issue_dfs if not df.empty], ignore_index=True)
    sev_order = {"CRITICAL": 0, "HIGH": 1, "MEDIUM": 2, "LOW": 3}
    all_issues = all_issues.sort_values(by="SEVERITY", key=lambda s: s.map(sev_order).fillna(4))
    all_issues.to_csv(output_path, index=False)
    for sev in ["CRITICAL", "HIGH", "MEDIUM", "LOW"]:
        print(f"  {sev}: {(all_issues['SEVERITY'] == sev).sum()}")
    return all_issues

def generate_define_xml(datasets: dict[str, pd.DataFrame], study_oid: str,
                        output_path: str = "define.xml") -> None:
    """Generate minimal define.xml 2.0 skeleton from SDTM datasets."""
    NS = "http://www.cdisc.org/ns/odm/v1.3"
    DEF = "http://www.cdisc.org/ns/def/v2.0"
    nsmap = {None: NS, "def": DEF, "xlink": "http://www.w3.org/1999/xlink"}
    root = etree.Element("ODM", nsmap=nsmap, FileOID=f"define_{study_oid}",
                          FileType="Snapshot", ODMVersion="1.3.2")
    study = etree.SubElement(root, "Study", OID=study_oid)
    gd = etree.SubElement(study, "GlobalVariables")
    etree.SubElement(gd, "StudyName").text = study_oid
    etree.SubElement(gd, "ProtocolName").text = study_oid
    mdv = etree.SubElement(study, "MetaDataVersion", OID="MDV.1", Name="SDTM Metadata")
    for domain_name, df in sorted(datasets.items()):
        igd = etree.SubElement(mdv, "ItemGroupDef", OID=f"IG.{domain_name}",
                               Name=domain_name, Repeating="Yes" if domain_name != "DM" else "No")
        for col in df.columns:
            dtype = "integer" if df[col].dtype in ("int64", "float64") else "text"
            etree.SubElement(mdv, "ItemDef", OID=f"IT.{domain_name}.{col}",
                             Name=col, DataType=dtype)
            etree.SubElement(igd, "ItemRef", ItemOID=f"IT.{domain_name}.{col}",
                             Mandatory="Yes" if col in ("STUDYID", "USUBJID") else "No")
    etree.ElementTree(root).write(output_path, xml_declaration=True,
                                   encoding="UTF-8", pretty_print=True)

---

7. 参数参考

参数	默认值	说明
VS SYSBP 范围	60–250 mmHg	儿科：下限 ×0.7，上限 ×0.8
VS HR 范围	30–220 次/分钟	儿科：上限 ×1.3；老年：>180 时标记为 CRITICAL
LB ALT/AST	0–500 U/L	>3×ULN 时标记（ALT ULN=40，AST ULN=37），用于 Hy's Law
LB BILI	0–30 mg/dL	Hy's Law：>2×ULN（ULN=1.2）且 ALT/AST 升高
QTcF（心脏病学）	<450 正常	450–480 临界，>480 延长，>500 为 CRITICAL
RECIST（肿瘤学）	CR/PR/SD/PD/NE	任何其他值均标记为 CT 违规
部分日期推算	开始→最早，结束→最晚	始终设置 DTYPE 列
严重程度级别	CRITICAL/HIGH/MEDIUM/LOW	映射到查询优先级 P1–P4

---

常见错误

• **错误：**未按年龄调整，就将成人生命体征范围应用于儿科受试者
  **正确：**按年龄组调整范围（例如，儿科 SBP 上限 = 成人值 × 0.8，HR 上限 = 成人值 × 1.3）
  **原因：**儿科正常值与成人存在显著差异——将儿童 HR 130 标记为异常，会浪费数据管理时间

• **错误：**将不完整日期标记为错误，而不是应用插补规则
  **正确：**将开始日期插补为最早可能日期（01-JAN-YYYY），将结束日期插补为最晚可能日期，然后设置 DTYPE = "DERIVED"
  **原因：**临床数据中常有不完整日期；拒绝这些日期会丢失有效记录，并违反 SDTM 插补规范

• **错误：**根据硬编码列表验证 SDTM 受控术语，而不是依据研究使用的 CT 版本
  **正确：**始终根据研究 define.xml 中声明的特定 CDISC CT 版本进行验证
  **原因：**CT 会随版本演进——CT 2023-12-15 中的有效术语，在 CT 2022-09-30 中可能不存在，反之亦然

• **错误：**执行跨表单日期检查时，没有考虑时区或访视窗口
  **正确：**跨表单日期比较时允许 ±1 天的容差，并在验证计划中记录该容差
  **原因：**受试者跨越时区或在夜间访视，会导致合理的 1 天偏差，这并非数据错误

• **错误：**未检查单位，就为每个超出范围的实验室检查结果生成质疑
  **正确：**在检查范围前先统一单位（例如，使用 LBORRESU/LBSTRESU 将肌酐单位在 mg/dL ↔ µmol/L 之间转换）
  **原因：**88 µmol/L 的肌酐值属于正常范围，但如果范围检查假定单位为 mg/dL，就会被标记为危急值（正常范围：0.6–1.2）

• **错误：**将所有验证发现视为同等优先级
  **正确：**按临床影响分类：CRITICAL（患者安全）、HIGH（主要终点）、MEDIUM（次要终点）、LOW（外观问题）
  **原因：**向研究中心大量发送低优先级质疑，会延误安全关键问题的解决