""" DMTrialSupplyChain-Kor — 합성 데모 데이터 생성 스크립트. 당뇨 cold-chain 주사제 IMP(인슐린·GLP-1RA·tirzepatide) 임상시험의 약물 공급망 시뮬레이션/추적 데모용 CSV들을 생성한다. 생성 파일 (모두 data/ 폴더에): - lots.csv : depot 보유 lot별 재고/만료 원장 (FEFO 엔진 입력) - temperature_log.csv : 배송/보관 온도 로그 (excursion 추적 입력) - enrollment_log.csv : site별 주차별 등록 로그 (resupply forecasting 입력) - accountability.csv : drug accountability 거래 원장 (dispensed/returned/destroyed) - sites.csv : site 마스터 (현재 재고·par level) 모든 데이터는 합성이며 외부 네트워크/API를 사용하지 않는다. 난수 시드 고정으로 재현 가능. """ import os import numpy as np import pandas as pd SEED = 20260531 rng = np.random.default_rng(SEED) HERE = os.path.dirname(os.path.abspath(__file__)) def _path(name: str) -> str: return os.path.join(HERE, name) # --------------------------------------------------------------------------- # 공통 마스터: site / IMP(arm) # --------------------------------------------------------------------------- SITES = [ {"site_id": "KR-01", "site_name": "순천향대부천병원", "region": "경기"}, {"site_id": "KR-02", "site_name": "서울대병원", "region": "서울"}, {"site_id": "KR-03", "site_name": "삼성서울병원", "region": "서울"}, {"site_id": "KR-04", "site_name": "분당서울대병원", "region": "경기"}, {"site_id": "KR-05", "site_name": "부산대병원", "region": "부산"}, ] # IMP: 당뇨 cold-chain 주사제. dose/주, 안정성 budget(누적 허용 일탈 시간, 시간 단위) IMPS = [ {"imp": "insulin_glargine", "label": "인슐린 글라진(주사 펜)", "doses_per_week": 7, "kit_doses": 30, "excursion_budget_h": 720}, # 30일 실온 허용(예시) {"imp": "semaglutide", "label": "세마글루타이드 GLP-1RA(펜)", "doses_per_week": 1, "kit_doses": 4, "excursion_budget_h": 1344}, # 56일 실온 허용(예시) {"imp": "tirzepatide", "label": "터제파타이드(바이알)", "doses_per_week": 1, "kit_doses": 4, "excursion_budget_h": 504}, # 21일 실온 허용(예시) ] START_DATE = pd.Timestamp("2026-01-05") # 첫 등록 주(월요일) N_WEEKS = 24 # --------------------------------------------------------------------------- # 1) lots.csv — depot lot 원장 # --------------------------------------------------------------------------- def gen_lots() -> pd.DataFrame: rows = [] lot_counter = 1 for imp in IMPS: # IMP당 lot 4~5개, 만료일을 분산시켜 FEFO/폐기 시연 n_lots = 5 for i in range(n_lots): lot_id = f"LOT-{imp['imp'][:3].upper()}-{lot_counter:03d}" lot_counter += 1 # 만료일: 시험 시작 기준 3~14개월 후로 분산 (일부는 임박) expiry_offset_days = int(rng.integers(90, 430)) expiry = START_DATE + pd.Timedelta(days=expiry_offset_days) qty_kits = int(rng.integers(40, 160)) rows.append({ "lot_id": lot_id, "imp": imp["imp"], "imp_label": imp["label"], "expiry_date": expiry.strftime("%Y-%m-%d"), "qty_kits": qty_kits, "kit_doses": imp["kit_doses"], "received_date": (START_DATE - pd.Timedelta(days=int(rng.integers(5, 40)))).strftime("%Y-%m-%d"), "storage": "2-8C", "status": "available", }) return pd.DataFrame(rows) # --------------------------------------------------------------------------- # 2) temperature_log.csv — 배송/보관 온도 로그 # --------------------------------------------------------------------------- def gen_temperature_log(lots: pd.DataFrame) -> pd.DataFrame: """ 각 lot에 대해 배송 이벤트별 시계열 온도 로그를 생성. 일부 구간에 의도적 excursion(>8C 또는 <2C)을 삽입. """ rows = [] # lot 일부만 배송 로그 보유 (depot→site shipment) shipped = lots.sample(n=min(10, len(lots)), random_state=SEED).reset_index(drop=True) for idx, lot in shipped.iterrows(): site = SITES[idx % len(SITES)]["site_id"] ship_start = START_DATE + pd.Timedelta(days=int(rng.integers(0, 120))) # 48시간 배송, 1시간 간격 측정 n_points = 48 # 약 30% 배송건에 excursion 삽입 has_excursion = rng.random() < 0.4 exc_start = int(rng.integers(10, 30)) if has_excursion else None exc_len = int(rng.integers(3, 12)) if has_excursion else 0 for h in range(n_points): ts = ship_start + pd.Timedelta(hours=h) base_temp = rng.normal(5.0, 0.8) # 정상 2-8C 범위 if has_excursion and exc_start is not None and exc_start <= h < exc_start + exc_len: # 상온 일탈 base_temp = rng.normal(18.0, 3.0) temp = round(float(base_temp), 2) rows.append({ "lot_id": lot["lot_id"], "imp": lot["imp"], "shipment_id": f"SHIP-{idx+1:03d}", "site_id": site, "timestamp": ts.strftime("%Y-%m-%d %H:%M"), "temp_c": temp, }) return pd.DataFrame(rows) # --------------------------------------------------------------------------- # 3) enrollment_log.csv — site별 주차별 등록 로그 # --------------------------------------------------------------------------- def gen_enrollment_log() -> pd.DataFrame: rows = [] arms = [imp["imp"] for imp in IMPS] for site in SITES: # site별 등록 속도(주당 평균) 상이 rate = rng.uniform(0.6, 2.2) for w in range(N_WEEKS): week_start = START_DATE + pd.Timedelta(weeks=w) # 초반 ramp-up ramp = min(1.0, (w + 1) / 8.0) n_enrolled = rng.poisson(rate * ramp) for _ in range(int(n_enrolled)): arm = arms[int(rng.integers(0, len(arms)))] rows.append({ "site_id": site["site_id"], "week_start": week_start.strftime("%Y-%m-%d"), "week_index": w, "subject_id": f"{site['site_id']}-{rng.integers(1000,9999)}", "arm": arm, }) df = pd.DataFrame(rows) return df.sort_values(["week_index", "site_id"]).reset_index(drop=True) # --------------------------------------------------------------------------- # 4) accountability.csv — drug accountability 거래 원장 # --------------------------------------------------------------------------- def gen_accountability(lots: pd.DataFrame) -> pd.DataFrame: """ site별 IMP별 dispensed/returned/destroyed 거래. 일부 site에 의도적 불일치(balance mismatch) 삽입. """ rows = [] txn_id = 1 for site in SITES: for imp in IMPS: imp_lots = lots[lots["imp"] == imp["imp"]]["lot_id"].tolist() if not imp_lots: continue shipped_kits = int(rng.integers(10, 40)) # shipment(입고) lot = imp_lots[int(rng.integers(0, len(imp_lots)))] base_date = START_DATE + pd.Timedelta(days=int(rng.integers(0, 60))) rows.append({ "txn_id": f"TXN-{txn_id:04d}", "site_id": site["site_id"], "imp": imp["imp"], "lot_id": lot, "txn_type": "shipped_to_site", "qty_kits": shipped_kits, "date": base_date.strftime("%Y-%m-%d"), }); txn_id += 1 # dispensed dispensed = int(shipped_kits * rng.uniform(0.5, 0.85)) rows.append({ "txn_id": f"TXN-{txn_id:04d}", "site_id": site["site_id"], "imp": imp["imp"], "lot_id": lot, "txn_type": "dispensed", "qty_kits": dispensed, "date": (base_date + pd.Timedelta(days=int(rng.integers(7, 40)))).strftime("%Y-%m-%d"), }); txn_id += 1 # returned (사용 후/미사용 반납) returned = int(dispensed * rng.uniform(0.1, 0.3)) rows.append({ "txn_id": f"TXN-{txn_id:04d}", "site_id": site["site_id"], "imp": imp["imp"], "lot_id": lot, "txn_type": "returned", "qty_kits": returned, "date": (base_date + pd.Timedelta(days=int(rng.integers(40, 80)))).strftime("%Y-%m-%d"), }); txn_id += 1 # destroyed (온도일탈/만료 폐기) destroyed = int(rng.integers(0, 4)) if destroyed > 0: rows.append({ "txn_id": f"TXN-{txn_id:04d}", "site_id": site["site_id"], "imp": imp["imp"], "lot_id": lot, "txn_type": "destroyed", "qty_kits": destroyed, "date": (base_date + pd.Timedelta(days=int(rng.integers(50, 90)))).strftime("%Y-%m-%d"), }); txn_id += 1 return pd.DataFrame(rows) # --------------------------------------------------------------------------- # 5) sites.csv — site 마스터 + 현재 재고 / par level # --------------------------------------------------------------------------- def gen_sites() -> pd.DataFrame: rows = [] for site in SITES: for imp in IMPS: rows.append({ "site_id": site["site_id"], "site_name": site["site_name"], "region": site["region"], "imp": imp["imp"], "on_hand_kits": int(rng.integers(2, 25)), "par_level_kits": int(rng.integers(8, 20)), # 보급 trigger 기준선 "lead_time_days": int(rng.integers(3, 10)), }) return pd.DataFrame(rows) def main(): lots = gen_lots() lots.to_csv(_path("lots.csv"), index=False) temp = gen_temperature_log(lots) temp.to_csv(_path("temperature_log.csv"), index=False) enroll = gen_enrollment_log() enroll.to_csv(_path("enrollment_log.csv"), index=False) acct = gen_accountability(lots) acct.to_csv(_path("accountability.csv"), index=False) sites = gen_sites() sites.to_csv(_path("sites.csv"), index=False) print("Generated demo CSVs:") for name, df in [("lots.csv", lots), ("temperature_log.csv", temp), ("enrollment_log.csv", enroll), ("accountability.csv", acct), ("sites.csv", sites)]: print(f" {name:24s} shape={df.shape}") if __name__ == "__main__": main()