""" 합성 다기관 MASH RCT NIT/biopsy 판독 데이터 생성 스크립트. 본 스크립트는 streamlit에 의존하지 않으며 pandas/numpy만으로 동작한다. 표준 환경(python3 + pandas + numpy)에서 직접 실행하면 data/ 폴더에 CSV가 생성된다. 생성 데이터 (오프라인, 완전 합성): - measurements.csv : 환자별/방문별 NIT(VCTE LSM/CAP, MRI-PDFF, MRE, FIB-4, ELF) + VCTE IQR/median·유효측정수, central reader/site/scanner ID - biopsy.csv : baseline / EOT biopsy NAS/fibrosis stage, 2명의 central pathologist 동시 판독 (일치도 추적용) - sites.csv : 사이트 메타데이터 삽입된 품질 시그널 (검수/데모용): - READER_DRIFT : reader 'R03' 가 trial 진행에 따라 LSM 을 systematic 하게 과대평가 - SITE_BIAS : site 'S07' 의 VCTE 스캐너가 일관되게 +bias (calibration outlier) - LOW_QUALITY : site 'S05' 의 VCTE 측정이 자주 품질기준 위반 (높은 IQR/median) - KAPPA_DRIFT : trial 후반 구간에서 두 pathologist 간 fibrosis stage 일치도 저하 - IMPLAUSIBLE : 소수 환자에 생물학적으로 불가능한 종단 급변/비단조 패턴 삽입 면책: 본 데이터는 전적으로 합성이며 실제 환자·시험 데이터가 아니다. 수치 기준은 예시·교육용이다. """ import os import numpy as np import pandas as pd SEED = 20260519 RNG = np.random.default_rng(SEED) # ---------------------------------------------------------------------- # 시험 설계 파라미터 # ---------------------------------------------------------------------- N_SITES = 8 N_READERS = 6 N_SCANNERS_PER_SITE = 1 PATIENTS_PER_SITE = 20 # 총 160 환자 VISITS = ["BL", "W24", "W48", "W72"] # baseline + 3 longitudinal VISIT_WEEK = {"BL": 0, "W24": 24, "W48": 48, "W72": 72} SITE_IDS = [f"S{idx:02d}" for idx in range(1, N_SITES + 1)] READER_IDS = [f"R{idx:02d}" for idx in range(1, N_READERS + 1)] # 품질 시그널 대상 (검수 시 기대 플래그 확인용) DRIFT_READER = "R03" BIAS_SITE = "S07" LOWQ_SITE = "S05" # ---------------------------------------------------------------------- # 사이트 메타데이터 # ---------------------------------------------------------------------- def build_sites(): rows = [] for sid in SITE_IDS: rows.append( { "site_id": sid, "scanner_id": f"{sid}-VCTE-1", "country": RNG.choice(["KOR", "USA", "DEU", "JPN"]), "enrollment_target": PATIENTS_PER_SITE, } ) return pd.DataFrame(rows) # ---------------------------------------------------------------------- # 측정(NIT) 데이터 # ---------------------------------------------------------------------- def build_measurements(): rows = [] pid_counter = 0 for sid in SITE_IDS: for _ in range(PATIENTS_PER_SITE): pid_counter += 1 patient_id = f"P{pid_counter:04d}" # 환자 baseline 특성 (case-mix) age = float(RNG.normal(54, 10)) bmi = float(RNG.normal(32, 5)) sex = RNG.choice(["M", "F"]) # baseline 질병 중증도 base_lsm = float(RNG.normal(12.0, 4.0)) # kPa base_lsm = max(3.0, base_lsm) base_cap = float(RNG.normal(320, 35)) # dB/m base_pdff = float(RNG.normal(16.0, 6.0)) # % base_pdff = max(3.0, base_pdff) base_mre = float(RNG.normal(3.0, 0.8)) # kPa base_mre = max(1.5, base_mre) base_fib4 = float(RNG.normal(1.6, 0.8)) base_fib4 = max(0.3, base_fib4) base_elf = float(RNG.normal(9.5, 1.0)) # ELF score # 치료군: 절반은 활성약(시간경과로 호전), 절반은 위약 arm = RNG.choice(["ACTIVE", "PLACEBO"]) # 활성약의 진성(true) 종단 호전율(주당) if arm == "ACTIVE": lsm_slope = RNG.normal(-0.045, 0.02) # kPa/week pdff_slope = RNG.normal(-0.10, 0.04) # %/week mre_slope = RNG.normal(-0.010, 0.005) else: lsm_slope = RNG.normal(-0.005, 0.015) pdff_slope = RNG.normal(-0.02, 0.03) mre_slope = RNG.normal(-0.002, 0.004) # 이 환자를 담당할 central reader (방문별로 무작위 배정) for visit in VISITS: week = VISIT_WEEK[visit] reader_id = RNG.choice(READER_IDS) # --- 진성 값 (true longitudinal) --- lsm = base_lsm + lsm_slope * week cap = base_cap + RNG.normal(-0.15, 0.1) * week pdff = base_pdff + pdff_slope * week mre = base_mre + mre_slope * week fib4 = base_fib4 + RNG.normal(-0.003, 0.004) * week elf = base_elf + RNG.normal(-0.004, 0.004) * week # --- 측정 잡음 --- lsm += RNG.normal(0, 0.9) cap += RNG.normal(0, 12) pdff += RNG.normal(0, 1.0) mre += RNG.normal(0, 0.15) fib4 += RNG.normal(0, 0.12) elf += RNG.normal(0, 0.15) # --- VCTE 품질 메트릭 (기본: 양호) --- valid_measurements = int(RNG.integers(10, 18)) iqr_median_ratio = float(abs(RNG.normal(0.14, 0.05))) # --- 시그널 1: READER DRIFT (LSM systematic 과대) --- if reader_id == DRIFT_READER: # trial 진행(week)이 커질수록 reader가 점점 더 과대평가 drift = 0.06 * week # week 72 -> +4.3 kPa lsm += drift # --- 시그널 2: SITE/SCANNER BIAS (일관된 +bias) --- if sid == BIAS_SITE: lsm += 3.2 # 사이트 전반에 걸친 calibration offset pdff += 2.5 # --- 시그널 3: LOW QUALITY SITE (VCTE 품질 불량) --- if sid == LOWQ_SITE: iqr_median_ratio = float(abs(RNG.normal(0.38, 0.08))) valid_measurements = int(RNG.integers(5, 11)) # 물리적 하한 lsm = max(2.5, lsm) cap = float(np.clip(cap, 100, 400)) pdff = max(1.0, pdff) mre = max(1.0, mre) fib4 = max(0.2, fib4) elf = max(6.0, elf) rows.append( { "patient_id": patient_id, "site_id": sid, "scanner_id": f"{sid}-VCTE-1", "reader_id": reader_id, "visit": visit, "week": week, "arm": arm, "age": round(age, 1), "sex": sex, "bmi": round(bmi, 1), "vcte_lsm_kpa": round(lsm, 2), "vcte_cap_dbm": round(cap, 1), "vcte_iqr_median_ratio": round(iqr_median_ratio, 3), "vcte_valid_measurements": valid_measurements, "mri_pdff_pct": round(pdff, 2), "mre_kpa": round(mre, 2), "fib4": round(fib4, 2), "elf": round(elf, 2), } ) df = pd.DataFrame(rows) # --- 시그널 5: IMPLAUSIBLE 종단 변화 삽입 (소수 환자) --- impl_patients = RNG.choice(df["patient_id"].unique(), size=4, replace=False) for pid in impl_patients: mask = df["patient_id"] == pid sub = df[mask].sort_values("week") if len(sub) >= 3: # W48 LSM 을 비현실적으로 급감시켜 비단조 패턴 생성 idx = sub.index[2] df.loc[idx, "vcte_lsm_kpa"] = round( max(2.5, df.loc[idx, "vcte_lsm_kpa"] * 0.25), 2 ) # W24 PDFF 를 비현실적으로 급감 idx2 = sub.index[1] df.loc[idx2, "mri_pdff_pct"] = round( max(1.0, df.loc[idx2, "mri_pdff_pct"] * 0.2), 2 ) return df # ---------------------------------------------------------------------- # Biopsy 데이터 (두 central pathologist 동시 판독) # ---------------------------------------------------------------------- def build_biopsy(measurements): """baseline / EOT(W72) biopsy. 두 pathologist(PA, PB)가 같은 슬라이드 판독. KAPPA_DRIFT 시그널: trial 후반 등록(enrollment_block 후반) 환자에서 두 pathologist 간 fibrosis stage 불일치가 증가하도록 설계. """ rows = [] patients = ( measurements[["patient_id", "site_id", "arm"]] .drop_duplicates() .reset_index(drop=True) ) n = len(patients) for i, prow in patients.iterrows(): pid = prow["patient_id"] # 등록 구간: 0(초반)~1(후반) enroll_frac = i / max(1, n - 1) enroll_block = "EARLY" if enroll_frac < 0.5 else "LATE" # 진성 fibrosis stage (0-4), NAS (0-8) true_fib_bl = int(np.clip(RNG.integers(1, 5), 0, 4)) true_nas_bl = int(np.clip(RNG.integers(3, 8), 0, 8)) # 치료 효과: ACTIVE 군은 EOT 에서 호전 경향 if prow["arm"] == "ACTIVE": fib_change = RNG.choice([-1, 0, 0, -1], p=[0.35, 0.3, 0.2, 0.15]) nas_change = RNG.choice([-2, -1, 0, -3], p=[0.3, 0.35, 0.2, 0.15]) else: fib_change = RNG.choice([0, 1, -1, 0], p=[0.45, 0.25, 0.15, 0.15]) nas_change = RNG.choice([0, -1, 1, 0], p=[0.4, 0.25, 0.2, 0.15]) true_fib_eot = int(np.clip(true_fib_bl + fib_change, 0, 4)) true_nas_eot = int(np.clip(true_nas_bl + nas_change, 0, 8)) for visit, true_fib, true_nas in [ ("BL", true_fib_bl, true_nas_bl), ("EOT", true_fib_eot, true_nas_eot), ]: # pathologist A : 안정적 판독 pa_fib = int(np.clip(true_fib + RNG.choice([-1, 0, 0, 0, 1]), 0, 4)) pa_nas = int(np.clip(true_nas + RNG.choice([-1, 0, 0, 1]), 0, 8)) # pathologist B : LATE 구간에서 불일치 증가 (KAPPA_DRIFT) if enroll_block == "LATE": fib_noise = RNG.choice([-2, -1, 0, 1, 2], p=[0.15, 0.25, 0.2, 0.25, 0.15]) nas_noise = RNG.choice([-2, -1, 0, 1, 2], p=[0.15, 0.25, 0.2, 0.25, 0.15]) else: fib_noise = RNG.choice([-1, 0, 0, 0, 1]) nas_noise = RNG.choice([-1, 0, 0, 1]) pb_fib = int(np.clip(true_fib + fib_noise, 0, 4)) pb_nas = int(np.clip(true_nas + nas_noise, 0, 8)) rows.append( { "patient_id": pid, "site_id": prow["site_id"], "arm": prow["arm"], "visit": visit, "enroll_block": enroll_block, "pathologist_a_fibrosis_stage": pa_fib, "pathologist_b_fibrosis_stage": pb_fib, "pathologist_a_nas": pa_nas, "pathologist_b_nas": pb_nas, } ) return pd.DataFrame(rows) def main(): here = os.path.dirname(os.path.abspath(__file__)) sites = build_sites() measurements = build_measurements() biopsy = build_biopsy(measurements) sites_path = os.path.join(here, "sites.csv") meas_path = os.path.join(here, "measurements.csv") biopsy_path = os.path.join(here, "biopsy.csv") sites.to_csv(sites_path, index=False) measurements.to_csv(meas_path, index=False) biopsy.to_csv(biopsy_path, index=False) print(f"[OK] sites.csv rows={len(sites):5d} -> {sites_path}") print(f"[OK] measurements.csv rows={len(measurements):5d} -> {meas_path}") print(f"[OK] biopsy.csv rows={len(biopsy):5d} -> {biopsy_path}") print() print("삽입 시그널: DRIFT_READER=%s BIAS_SITE=%s LOWQ_SITE=%s" % (DRIFT_READER, BIAS_SITE, LOWQ_SITE)) if __name__ == "__main__": main()