pyhealth

PyHealth

概要

PyHealth は、EHR データを使った医療 ML のエンドツーエンドパイプラインを提供します。データの読み込み → 医療コードの処理 → 患者レベルのデータセット構築 → モデルのトレーニング → 評価までを実行できます。MIMIC-III、MIMIC-IV、eICU-CRD、OMOP-CDM の構造化データベースをネイティブにサポートし、それぞれに固有のデータ形式を処理します。医療コード（ICD-9、ICD-10、ATC、NDC、SNOMED）は、コードレベルの埋め込みとオントロジー間のマッピングに対応する階層型コードシステムに整理されています。死亡率予測、薬剤推薦、再入院、入院期間、診断コード予測などの事前構築済みタスクは、数行のコードでインスタンス化できます。カスタムタスクは標準化されたインターフェースに従います。

使用する場面

• MIMIC-III または MIMIC-IV から臨床転帰予測モデル（死亡率、再入院、LOS）をトレーニングする場合
• ATC コード階層を使用して、薬剤推薦モデルや薬物相互作用予測モデルを構築する場合
• 医療機関の EHR システムから取得した OMOP-CDM 形式のデータを ML 用に処理する場合
• 医療ベンチマークで、事前学習済みの臨床モデル（RETAIN、GRASP、MedBERT）をベースラインとして使用する場合
• RNN／Transformer モデル向けに、時間構造を持つ患者の受診シーケンスを構築する場合
• 適切な指標（AUROC、AUPRC、F1、Jaccard）を使って臨床予測モデルを評価する場合
• ML を使わずに EHR の前処理のみを行う場合は FIDDLE を使用し、臨床記録の NLP には clinical-longformer を使用してください

前提条件

• Python パッケージ：pyhealth、torch、pandas、scikit-learn
• データ要件：MIMIC-III／IV の CSV ファイル（PhysioNet の認証が必要）、eICU、または OMOP-CDM データベース
• MIMIC へのアクセス：physionet.org で申請してください（無料。CITI のトレーニングが必要で、約 1 週間かかります）

pip install pyhealth torch pandas scikit-learn
# Download MIMIC-III: https://physionet.org/content/mimiciii/
# Download MIMIC-IV: https://physionet.org/content/mimiciv/

クイックスタート

from pyhealth.datasets import MIMIC3Dataset

# Load MIMIC-III (specify path to downloaded CSV files)
dataset = MIMIC3Dataset(
    root="path/to/mimic-iii/",
    tables=["DIAGNOSES_ICD", "PRESCRIPTIONS", "PROCEDURES_ICD"],
    code_mapping={"ICD9CM": "CCSCM"},  # map ICD-9 codes to CCS multi-level
    dev=True,  # dev=True uses 1% of data for fast testing
)

print(f"Patients: {dataset.stat()['num_patients']}")
print(f"Visits: {dataset.stat()['num_visits']}")

コア API

モジュール 1：データセットの読み込み

MIMIC-III、MIMIC-IV、eICU、OMOP-CDM のデータセットを読み込みます。

from pyhealth.datasets import MIMIC3Dataset, MIMIC4Dataset, eICUDataset

# MIMIC-III
mimic3 = MIMIC3Dataset(
    root="data/mimic-iii/",
    tables=["DIAGNOSES_ICD", "PRESCRIPTIONS", "PROCEDURES_ICD", "LABEVENTS"],
    code_mapping={"ICD9CM": "CCSCM", "NDC": "ATC3"},  # standardize codes
    dev=False,
)
stats = mimic3.stat()
print(f"MIMIC-III: {stats['num_patients']} patients, {stats['num_visits']} visits")

# MIMIC-IV
mimic4 = MIMIC4Dataset(
    root="data/mimic-iv/",
    tables=["diagnoses_icd", "prescriptions", "procedures_icd"],
    code_mapping={"ICD10CM": "CCSCM"},
    dev=True,
)

# eICU
eicu = eICUDataset(
    root="data/eicu/",
    tables=["diagnosis", "medication", "treatment"],
    dev=True,
)
print(f"eICU loaded: {eicu.stat()}")

# Explore dataset structure
patient_id = list(mimic3.patients.keys())[0]
patient = mimic3.patients[patient_id]
print(f"Patient {patient_id}: {len(patient.visits)} visits")

for visit in patient.visits[:2]:
    print(f"  Visit {visit.visit_id}:")
    print(f"    Diagnoses: {visit.get_code_list('CCSCM')[:5]}")
    print(f"    Medications: {visit.get_code_list('ATC3')[:3]}")

モジュール 2：タスクの構築

生のデータセットを ML で使用可能なタスクデータセットに変換します。

from pyhealth.tasks import mortality_prediction_mimic3_fn
from pyhealth.datasets import SampleDataset

# Mortality prediction task
# Each sample: patient's visit history → binary mortality label
mortality_dataset = SampleDataset(
    dataset=mimic3,
    task_fn=mortality_prediction_mimic3_fn,
)

print(f"Task: mortality prediction")
print(f"Samples: {len(mortality_dataset)}")

# Inspect a sample
sample = mortality_dataset[0]
print(f"Sample keys: {list(sample.keys())}")
print(f"Conditions (ICD codes): {sample['conditions'][:3]}")
print(f"Drugs (ATC codes): {sample['drugs'][:3]}")
print(f"Label (mortality): {sample['label']}")

# Custom task: 30-day readmission prediction
def readmission_30day_fn(patient):
    """Custom task function: predict 30-day readmission after discharge."""
    samples = []
    for i, visit in enumerate(patient.visits[:-1]):
        next_visit = patient.visits[i + 1]
        # Compute days between discharge and next admission
        days_gap = (next_visit.encounter_time - visit.discharge_time).days
        label = int(days_gap <= 30)

        samples.append({
            "visit_id": visit.visit_id,
            "patient_id": patient.patient_id,
            "conditions": visit.get_code_list("CCSCM"),
            "drugs": visit.get_code_list("ATC3"),
            "procedures": visit.get_code_list("ICD9PROC"),
            "label": label,
        })
    return samples

readmission_dataset = SampleDataset(dataset=mimic3, task_fn=readmission_30day_fn)
print(f"Readmission samples: {len(readmission_dataset)}")
pos_rate = sum(s["label"] for s in readmission_dataset) / len(readmission_dataset)
print(f"Positive rate (30-day readmission): {pos_rate:.2%}")

モジュール 3：医療コードシステム

ICD、ATC、NDC などの階層型医療コードシステムを扱います。

from pyhealth.medcode import InnerMap

# ICD-9-CM diagnosis codes
icd9 = InnerMap.load("ICD9CM")
code = "428.0"  # Heart failure, unspecified
print(f"Code: {code}")
print(f"Description: {icd9.lookup(code)}")
print(f"Ancestors: {icd9.get_ancestors(code)}")
print(f"Children: {icd9.get_children(code)[:5]}")

# ATC drug classification
atc = InnerMap.load("ATC")
drug_code = "A10BA02"  # Metformin
print(f"\nATC: {drug_code}")
print(f"Drug: {atc.lookup(drug_code)}")
print(f"L1 class: {atc.get_ancestors(drug_code)}")

# Cross-ontology code mapping
from pyhealth.medcode import CrossMap

# Map NDC (drug product codes) to ATC level 3
ndc_to_atc = CrossMap.load("NDC", "ATC3")
ndc_code = "0069-2587-30"  # example NDC
atc3_codes = ndc_to_atc.map(ndc_code)
print(f"NDC {ndc_code} → ATC3: {atc3_codes}")

# Map ICD-9 to ICD-10
icd9_to_icd10 = CrossMap.load("ICD9CM", "ICD10CM")
icd10 = icd9_to_icd10.map("428.0")
print(f"ICD-9 428.0 → ICD-10: {icd10}")

モジュール 4：モデルのトレーニング

実装済みの臨床 ML モデルをトレーニングします。

from pyhealth.models import Transformer, RETAIN
from pyhealth.datasets import split_by_patient, get_dataloader
import torch

# Train/val/test split (patient-level, no leakage)
train_ds, val_ds, test_ds = split_by_patient(mortality_dataset, [0.7, 0.1, 0.2])

train_loader = get_dataloader(train_ds, batch_size=32, shuffle=True)
val_loader   = get_dataloader(val_ds,   batch_size=64, shuffle=False)
test_loader  = get_dataloader(test_ds,  batch_size=64, shuffle=False)

print(f"Train: {len(train_ds)} | Val: {len(val_ds)} | Test: {len(test_ds)}")

# Transformer model for EHR sequence modeling
model = Transformer(
    dataset=mortality_dataset,
    feature_keys=["conditions", "drugs", "procedures"],
    label_key="label",
    mode="binary",        # binary classification
    embedding_dim=128,
    num_heads=4,
    num_layers=2,
    dropout=0.1,
)

print(f"Model parameters: {sum(p.numel() for p in model.parameters()):,}")

# RETAIN: Reverse Time Attention model (interpretable clinical ML)
retain_model = RETAIN(
    dataset=mortality_dataset,
    feature_keys=["conditions", "drugs"],
    label_key="label",
    mode="binary",
    embedding_dim=64,
)

モジュール 5：トレーニングと評価

from pyhealth.trainer import Trainer

trainer = Trainer(
    model=model,
    metrics=["pr_auc", "roc_auc", "f1"],  # PR-AUC, ROC-AUC, F1
)

trainer.train(
    train_dataloader=train_loader,
    val_dataloader=val_loader,
    epochs=20,
    optimizer_params={"lr": 1e-3},
    weight_decay=1e-5,
    monitor="pr_auc",           # early stopping metric
    monitor_criterion="max",
    load_best_model_after_train=True,
)

# Evaluate on test set
results = trainer.evaluate(test_loader)
print("Test results:")
for metric, value in results.items():
    print(f"  {metric}: {value:.4f}")

モジュール 6：薬剤推薦

受診履歴に基づいて、患者に処方すべき薬剤を予測します。

from pyhealth.tasks import drug_recommendation_mimic3_fn
from pyhealth.models import GAMENet
from pyhealth.datasets import SampleDataset, split_by_patient, get_dataloader

# Drug recommendation task
drug_dataset = SampleDataset(
    dataset=mimic3,
    task_fn=drug_recommendation_mimic3_fn,
)

train_ds, val_ds, test_ds = split_by_patient(drug_dataset, [0.7, 0.1, 0.2])

# GAMENet: graph-augmented memory network for drug recommendation
gamenet = GAMENet(
    dataset=drug_dataset,
    feature_keys=["conditions", "procedures"],
    label_key="drugs",
    mode="multilabel",     # recommend multiple drugs per visit
    embedding_dim=64,
)

train_loader = get_dataloader(train_ds, batch_size=16, shuffle=True)

trainer = Trainer(model=gamenet, metrics=["jaccard", "f1", "prauc"])
trainer.train(train_dataloader=train_loader,
              val_dataloader=get_dataloader(val_ds, 32),
              epochs=30, monitor="jaccard")
print("Drug recommendation model trained")

主要な概念

患者・受診・イベントの階層構造

PyHealth は、EHR データを Patient → Visit → 医療コードの階層構造で整理します。各 Visit には、複数のテーブル（診断、薬剤、処置、検査）にまたがるタイムスタンプ付きのイベントが含まれます。ML モデルは、各患者を受診のシーケンス、各受診を医療コードの集合として扱い、疾患の時間的な進行を捉えます。

コードのマッピングと標準化

生の EHR コード（ICD-9、NDC）は非常に細分化されており、種類も膨大です。PyHealth の code_mapping パラメーターは、コードをより粗いオントロジー（CCSCM は約 260 カテゴリーで、ICD-9 コードは約 15,000 種類）に自動変換し、語彙サイズを削減して、データセット間の転移を可能にします。

一般的なワークフロー

ワークフロー 1：死亡率予測の全パイプライン

from pyhealth.datasets import MIMIC3Dataset, SampleDataset, split_by_patient, get_dataloader
from pyhealth.tasks import mortality_prediction_mimic3_fn
from pyhealth.models import Transformer
from pyhealth.trainer import Trainer

# 1. Load data
dataset = MIMIC3Dataset(
    root="data/mimic-iii/",
    tables=["DIAGNOSES_ICD", "PRESCRIPTIONS", "PROCEDURES_ICD"],
    code_mapping={"ICD9CM": "CCSCM", "NDC": "ATC3"},
    dev=True,
)

# 2. Build task dataset
task_ds = SampleDataset(dataset, task_fn=mortality_prediction_mimic3_fn)
print(f"Samples: {len(task_ds)}, Positive rate: {sum(s['label'] for s in task_ds)/len(task_ds):.2%}")

# 3. Split
train_ds, val_ds, test_ds = split_by_patient(task_ds, [0.7, 0.1, 0.2])

# 4. Model
model = Transformer(
    dataset=task_ds,
    feature_keys=["conditions", "drugs"],
    label_key="label",
    mode="binary",
    embedding_dim=128,
)

# 5. Train
trainer = Trainer(model=model, metrics=["pr_auc", "roc_auc"])
trainer.train(
    train_dataloader=get_dataloader(train_ds, 32, shuffle=True),
    val_dataloader=get_dataloader(val_ds, 64),
    epochs=15,
    monitor="pr_auc",
)

# 6. Evaluate
results = trainer.evaluate(get_dataloader(test_ds, 64))
print(f"Test PR-AUC: {results['pr_auc']:.4f}")
print(f"Test ROC-AUC: {results['roc_auc']:.4f}")

ワークフロー 2：モデル比較ベンチマーク

from pyhealth.models import Transformer, RETAIN, MedBERT
from pyhealth.trainer import Trainer
from pyhealth.datasets import get_dataloader
import pandas as pd

models = {
    "Transformer": Transformer(task_ds, ["conditions", "drugs"], "label", "binary"),
    "RETAIN":      RETAIN(task_ds, ["conditions", "drugs"], "label", "binary"),
}

results_list = []
for name, model in models.items():
    trainer = Trainer(model=model, metrics=["pr_auc", "roc_auc", "f1"])
    trainer.train(
        train_dataloader=get_dataloader(train_ds, 32, shuffle=True),
        val_dataloader=get_dataloader(val_ds, 64),
        epochs=10,
        monitor="pr_auc",
    )
    test_results = trainer.evaluate(get_dataloader(test_ds, 64))
    test_results["model"] = name
    results_list.append(test_results)
    print(f"{name}: PR-AUC={test_results['pr_auc']:.4f}, ROC-AUC={test_results['roc_auc']:.4f}")

results_df = pd.DataFrame(results_list).set_index("model")
results_df.to_csv("model_comparison.csv")
print(results_df)

主要なパラメーター

パラメーター	モジュール／クラス	デフォルト	範囲／オプション	効果
tables	MIMIC3Dataset	—	MIMIC テーブル名のリスト	読み込む EHR テーブルを指定します。テーブルが多いほど特徴量が豊富になりますが、読み込みに時間がかかります
code_mapping	MIMIC3Dataset	{}	{"ICD9CM": "CCSCM"}	生のコードを標準化されたオントロジーにマッピングします
dev	MIMIC3Dataset	False	True／False	True にすると、開発を高速化するためにデータの 1% を使用します
feature_keys	すべてのモデル	—	コードタイプ文字列のリスト	モデル入力として使用する医療コードの種類を指定します
embedding_dim	Transformer、RETAIN	128	64–512	コードと患者の埋め込みの隠れ層のサイズです
num_heads	Transformer	4	1–16	マルチヘッドアテンションのヘッド数です（embedding_dim で割り切れる必要があります）
num_layers	Transformer	2	1–6	Transformer エンコーダーレイヤーの数です
dropout	すべてのモデル	0.1	0–0.5	正則化のためのドロップアウト率です
mode	すべてのモデル	—	"binary"、"multiclass"、"multilabel"	予測タスクの種類です
monitor	Trainer.train	"loss"	"pr_auc"、"roc_auc"、"f1"	早期停止とモデル選択に使用する指標です

ベストプラクティス

1. 必ず split_by_patient を使用し、ランダム分割は避けてください：訪問単位でランダムに分割すると、データリークが発生します。同じ患者が異なる訪問を通じて訓練セットとテストセットの両方に含まれる可能性があります。PyHealth の split_by_patient は、患者単位で厳密に分離します。

2. 開発中は dev=True を使用してください：MIMIC-III には約 46,000 人の患者が含まれています。データセット全体の読み込みには数分かかりますが、開発モードでは約 460 人の患者を数秒で読み込めます。最終的な訓練を実行する場合にのみ、dev=False に切り替えてください。

3. コードマッピングを適用して、データセット間で標準化してください：生の ICD-9 コードには約 15,000 種類の固有値がありますが、CCSCM を使うと臨床的に意味のある約 260 のグループにまとめられます。これにより、モデルの語彙が大幅に削減され、意味のあるコード埋め込みが可能になります。特に小規模なデータセットで重要です。

4. ROC-AUC と併せて PR-AUC で評価してください：臨床データセットは通常、クラスの不均衡が非常に大きくなります（例：死亡率 10%）。不均衡なタスクでは、PR-AUC のほうが ROC-AUC より有用な情報を提供します。すべてのサンプルを陰性と予測するモデルの ROC-AUC は約 0.5 ですが、PR-AUC は陽性率（約 0.1）に近づきます。

よく使われるレシピ

レシピ：患者の埋め込みをエクスポートする

import torch
from pyhealth.datasets import get_dataloader
import numpy as np

model.eval()
embeddings = []
labels = []

with torch.no_grad():
    for batch in get_dataloader(test_ds, batch_size=64):
        # Get patient-level representation (before classification head)
        hidden = model.get_patient_representation(batch)
        embeddings.append(hidden.cpu().numpy())
        labels.append(batch["label"].numpy())

embeddings = np.concatenate(embeddings, axis=0)
labels = np.concatenate(labels, axis=0)
np.save("patient_embeddings.npy", embeddings)
np.save("patient_labels.npy", labels)
print(f"Embeddings: {embeddings.shape}")

レシピ：重み付きサンプラーによるクラス不均衡への対処

import torch
from torch.utils.data import WeightedRandomSampler
from pyhealth.datasets import get_dataloader

# Compute class weights for imbalanced mortality prediction
labels = [sample["label"] for sample in train_ds]
pos = sum(labels)
neg = len(labels) - pos
weights = [1.0/neg if l == 0 else 1.0/pos for l in labels]
sampler = WeightedRandomSampler(weights, num_samples=len(weights), replacement=True)

# Use sampler in dataloader
balanced_loader = torch.utils.data.DataLoader(
    train_ds, batch_size=32, sampler=sampler,
    collate_fn=train_ds.collate_fn
)
print(f"Balanced loader: {len(balanced_loader)} batches")

トラブルシューティング

問題	原因	解決策
データセットの読み込み時に FileNotFoundError が発生する	root のパスが誤っている、または MIMIC の CSV ファイルがない	root に正しい CSV ファイルが含まれていることを確認します。ls data/mimic-iii/*.csv で内容を一覧表示してください
feature_keys のコードタイプで KeyError が発生する	コードタイプが読み込まれていない、またはマッピングに失敗している	tables に必要なテーブルが含まれ、code_mapping が期待されるオントロジーにマッピングされていることを確認します
タスクデータセットのすべてのサンプルで label=0 になる	タスク関数のロジックエラー、またはデータの問題	1 人の患者に対して task_fn(patient) を出力し、ラベルの導出ロジックを確認します
MIMIC 全体の読み込み時にメモリエラーが発生する	データセットが大きく、RAM が不足している	開発中は dev=True を使用します。チャンク単位で読み込むか、tables のリストを減らしてください
テストセットの ROC-AUC が約 0.5 になる	深刻なクラス不均衡により、単純な予測器になっている	WeightedRandomSampler を使用し、代わりに PR-AUC を報告して、分類閾値を下げてください
訓練中に CUDA のメモリ不足が発生する	バッチサイズが大きすぎる	batch_size を 32 から 8 または 16 に減らし、勾配累積を使用してください
コードマッピングが空のリストを返す	コードがクロスマップに見つからない	コード形式（例：小数点あり／なしの ICD-9）を確認します。InnerMap.load("ICD9CM").lookup(code) を試してください

関連スキル

• statsmodels-statistical-modeling — 臨床アウトカム予測のためのロジスティック回帰ベースライン
• scikit-learn-machine-learning — PyHealth の特徴量を使った従来型 ML のベースライン（ランダムフォレスト、勾配ブースティング）
• clinical-decision-support-documents — 臨床 ML モデルの出力を意思決定支援ツールに変換する方法

参考資料

• PyHealth ドキュメント — API リファレンス、チュートリアル、タスクの説明
• PyHealth 論文：Zhao et al.（2021）、arXiv — ライブラリの設計とベンチマークタスク
• MIMIC-III 論文：Johnson et al.（2016）、Nature Scientific Data — データセットの説明
• MIMIC-IV 論文：Johnson et al.（2023）、Nature Scientific Data — 更新されたデータセット
• RETAIN 論文：Choi et al.（2016）、NeurIPS — 解釈可能な臨床予測モデル
• PyHealth GitHub — ソースコードと使用例