edc-data-validation

EDC データ検証 — パイプラインスキル

概要

臨床データ検証パイプライン向けの決定論的な Python コードを生成します。 ユーザーから EDC エクスポートまたは SDTM データセットが提供されたら、以下のレシピに従って検証コードを作成します。生成する前に、必ず入力形式と治療領域を確認してください。

使用方法

• EDC エクスポートを、範囲チェック、フォーム間の整合性チェック、部分日付の補完で検証します
• SDTM データセットを CDISC 管理用語および構造ルールに照らしてチェックします
• define.xml と優先順位付けされた検証レポートを生成します

基本概念

---

応答形式

• ユーザーが必要とするコマンドまたはコードを最初に提示し、その後に説明します
• 次の順序で構成します：入力の確認 → 動作するコード → 主要なパラメーターの説明 → 注意点
• タスクごとに、完全に動作する例を 1 つ提示します。あらゆる代替案は示しません
• コードコメントは最小限かつ実用的な内容にします（存在理由ではなく、内容を説明します）
• 目安：コードは 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 beats/min	小児：高値側は ×1.3。高齢者：>180 は CRITICAL としてフラグ
LB ALT/AST	0–500 U/L	Hy's Law のため、>3×ULN（ALT ULN=40、AST ULN=37）にフラグ
LB BILI	0–30 mg/dL	Hy's Law：ALT/AST の上昇を伴う >2×ULN（ULN=1.2）
QTcF（心臓病学）	<450 正常	450–480 境界、>480 延長、>500 CRITICAL
RECIST（腫瘍学）	CR/PR/SD/PD/NE	それ以外の値は CT 違反としてフラグ
部分日付の補完	開始→最も早い日付、終了→最も遅い日付	常に DTYPE 列を設定
重大度レベル	CRITICAL/HIGH/MEDIUM/LOW	クエリ優先度 P1–P4 に対応付け

---

よくある間違い

• 誤り： 年齢を考慮せず、小児被験者に成人のバイタルサイン範囲を適用する 正しい方法： 年齢層に応じて範囲を調整します（例：小児の収縮期血圧の上限＝成人の値 × 0.8、心拍数の上限＝成人の値 × 1.3） 理由： 小児の正常値は成人と大きく異なります。心拍数130の子どもを異常としてフラグ付けすると、データ管理の時間を無駄にします。

• 誤り： 部分日付を補完ルールに従って処理せず、エラーとしてフラグ付けする 正しい方法： 開始日は可能な限り早い日付（01-JAN-YYYY）に、終了日は可能な限り遅い日付に補完し、DTYPE = "DERIVED" を設定します。 理由： 臨床データでは部分日付が想定されています。これを却下すると、有効なレコードが失われ、SDTM の補完規則にも違反します。

• 誤り： 試験で使用する CT バージョンではなく、ハードコードされたリストに照らして SDTM の統制用語を検証する 正しい方法： 必ず、試験の 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（外観上の問題）。 理由： 優先度の低いクエリを施設に大量に送ると、安全性に関わる重大な問題の解決が遅れます。