""" app.py — MASLDNITSchedule-Kor Streamlit UI MASLD/MASH 종단 관찰연구 NIT 측정 스케줄 설계 계산기. 실행: streamlit run app.py Streamlit 미설치 시: python3 main.py --demo (CLI로 동일 계산 가능) 면책: 참고용·연구용 도구. 실제 연구 설계는 생물통계학자 검토 필수. NIT 변동성·진행률·단가는 문헌 기반 추정치. """ import sys try: import streamlit as st except ImportError: print("Streamlit이 설치되어 있지 않습니다.") print("설치: pip install --user streamlit") print("또는 CLI 사용: python3 main.py --demo") sys.exit(1) import core st.set_page_config(page_title="MASLDNITSchedule-Kor", layout="wide") lib = core.load_nit_library() NIT_KEYS = core.list_nits(lib) def nit_label(k): return core.get_nit(lib, k).get("label", k) # --------------------------------------------------------------------------- st.title("MASLDNITSchedule-Kor — MASLD/MASH NIT 종단 측정 스케줄 설계") st.caption("비침습검사(NIT) 측정 변동성·MDC·진행률 기반 관찰연구 측정 스케줄 설계 도구") st.warning(core.DISCLAIMER) # --- 사이드바: 공통 설계 입력 --- st.sidebar.header("연구 설계 입력") sel_nit = st.sidebar.selectbox("대상 NIT", NIT_KEYS, format_func=nit_label) n_subjects = st.sidebar.number_input("표본 수 (명)", 10, 5000, 120, step=10) n_followups = st.sidebar.number_input("추적 visit 수", 1, 12, 2) interval = st.sidebar.number_input("visit 간격 (개월)", 1, 36, 12) n_repeat = st.sidebar.number_input("visit당 반복측정 횟수", 1, 5, 1) window_weeks = st.sidebar.slider("visit window (±주)", 1, 16, 4) missing_rate = st.sidebar.slider("가정 결측률", 0.0, 0.5, 0.15, 0.05) subgroup_opts = ["estimate_default"] + sorted( set().union(*[set(core.get_nit(lib, k)["progression"].get("subgroups", {}).keys()) for k in NIT_KEYS]) ) subgroup = st.sidebar.selectbox("진행률 하위집단", subgroup_opts) budget = st.sidebar.number_input("가용 예산 (원, 0=미설정)", 0, 10_000_000_000, 50_000_000, step=1_000_000) n_sim = st.sidebar.select_slider("Monte-Carlo 반복 수", [100, 200, 300, 500, 800], 300) sched = core.build_schedule(0, n_followups, interval) visit_months = sched["visit_months"] nit_cfg = core.get_nit(lib, sel_nit) sem, unit, baseline = core.resolve_sem(nit_cfg) sg = None if subgroup == "estimate_default" else ( subgroup if subgroup in nit_cfg["progression"].get("subgroups", {}) else None) tabs = st.tabs([ "1. 스케줄 빌더", "2. MDC 산출기", "3. 기간 vs 진행률", "4. visit window·결측", "5. 비용효율 비교", "6. 설계 리포트", ]) # --- 탭 1: 스케줄 빌더 --- with tabs[0]: st.subheader("NIT 측정 스케줄") st.write(f"**선택 NIT:** {nit_label(sel_nit)} · 단위 `{unit}` · " f"범주 {nit_cfg.get('category','-')}") c1, c2, c3 = st.columns(3) c1.metric("총 visit 수", f"{sched['n_visits']}회") c2.metric("총 추적기간", f"{sched['total_months']:.0f}개월") c3.metric("visit window", f"±{window_weeks}주") st.write("**visit 시점:**", ", ".join(f"{m:.0f}개월" for m in visit_months)) chart_data = {"visit 시점(개월)": visit_months, "측정 차수": list(range(1, len(visit_months) + 1))} st.bar_chart(chart_data, x="visit 시점(개월)", y="측정 차수") qc = nit_cfg.get("quality_criteria", {}) if qc.get("rule"): st.info(f"검사 품질기준: {qc['rule']}") # --- 탭 2: MDC --- with tabs[1]: st.subheader("최소검출변화(MDC) 산출") st.caption("공식: MDC = 1.96 × √2 × SEM / √(반복횟수) (95% 신뢰)") tbl = core.mdc_table_for_nit(lib, sel_nit, repeats=(1, 2, 3)) st.write(f"측정-재측정 SEM(추정): **{sem:.3f} {unit}**") st.dataframe([{"반복측정 횟수": r["n_repeat"], f"MDC ({unit})": round(r["mdc"], 3)} for r in tbl["rows"]], use_container_width=True) cur = core.compute_mdc(sem, n_repeat=n_repeat) rt = lib.get("responder_thresholds", {}) mmap = {"VCTE_LSM": rt.get("VCTE_LSM_meaningful_kPa"), "MRE": rt.get("MRE_meaningful_kPa"), "FIB-4": rt.get("FIB-4_meaningful_index"), "ELF": rt.get("ELF_meaningful_score"), "MRI_PDFF": None} is_warn, msg = core.mdc_warning(cur["mdc"], mmap.get(sel_nit), unit) st.metric(f"현재 설계 MDC (반복 {n_repeat}회)", f"{cur['mdc']:.3f} {unit}") (st.error if is_warn else st.success)(msg) # --- 탭 3: tradeoff --- with tabs[2]: st.subheader("연구기간 vs 자연 진행률 tradeoff") tc = core.tradeoff_curve(lib, sel_nit, subgroup=sg, n_repeat=n_repeat, max_months=120, step_months=6) st.write(f"가정 진행률(추정): **{tc['progression_per_year']:+.3f} {unit}/년** " f"(하위집단: {sg or '기본값'})") st.line_chart({"개월": tc["months"], "누적 기대 변화": tc["expected_change"], "MDC 기준선": [tc["mdc"]] * len(tc["months"])}, x="개월") ttm = tc["time_to_mdc_months"] if ttm == float("inf"): st.warning("진행률 0 가정 — MDC 도달기간 산출 불가") else: st.metric("MDC 만큼 변화 도달 예상기간", f"약 {ttm:.1f}개월") if ttm > sched["total_months"]: st.error(f"계획 추적기간({sched['total_months']:.0f}개월) < " f"MDC 도달기간 — 추적기간 연장 권장") else: st.success(f"계획 추적기간({sched['total_months']:.0f}개월) 내 검출 가능 예상") # --- 탭 4: visit window·결측 --- with tabs[3]: st.subheader("visit window 최적화 + 결측 영향 Monte-Carlo") slope = nit_cfg["progression"]["estimate_per_year"] if sg and sg in nit_cfg["progression"].get("subgroups", {}): slope = nit_cfg["progression"]["subgroups"][sg] with st.spinner("시뮬레이션 중..."): sim = core.simulate_missingness(visit_months, n_subjects, sem, slope, window_weeks=window_weeks, missing_rate=missing_rate, n_sim=n_sim) opt = core.optimize_visit_window(visit_months, n_subjects, sem, slope, window_candidates=(2, 4, 8, 12), missing_rate=missing_rate, n_sim=min(n_sim, 300)) st.markdown("**(5-1) 집단 평균 진행 검정력** — 표본이 크면 포화되어 스케줄에 둔감") c1, c2, c3 = st.columns(3) c1.metric("이상적 검정력", f"{sim['power_analytic_ideal']*100:.1f}%") c2.metric("결측 반영 검정력", f"{sim['power_simulated']*100:.1f}%") c3.metric("검정력 손실", f"{sim['power_loss']*100:.1f}%p") st.markdown("**(5-2) 피험자 단위 변화 검출률** — 스케줄 품질 핵심 지표") d1, d2, d3 = st.columns(3) d1.metric("이상적 변화검출률", f"{sim['subject_detect_ideal']*100:.1f}%") d2.metric("결측 반영 변화검출률", f"{sim['subject_detect_rate']*100:.1f}%") d3.metric("엔드포인트 완전성", f"{sim['endpoint_completeness']*100:.1f}%") st.write("**visit window 후보별 비교:**") st.dataframe([{"window(±주)": r["window_weeks"], "변화검출률(%)": round(r["detect_rate"]*100, 1), "검출률손실(%p)": round(r["detect_rate_loss"]*100, 1), "엔드포인트완전성(%)": round(r["endpoint_completeness"]*100, 1), "집단검정력(%)": round(r["power"]*100, 1)} for r in opt["results"]], use_container_width=True) st.success(f"권장 visit window: ±{opt['recommended_window_weeks']}주 " f"(변화검출률 {opt['recommended_detect_rate']*100:.1f}%)") if sim["detect_rate_loss"] > 0.05: st.error("결측·window로 인한 변화 검출률 손실 5%p 초과 — " "결측 최소화/visit window 단축 검토") if sim["power_simulated"] < 0.80: st.warning("집단 검정력 80% 미만 — 표본 수 증대/추적기간 연장 필요") # --- 탭 5: 비용효율 --- with tabs[4]: st.subheader("NIT 조합·비용 비교 (비용효율 frontier)") cef = core.cost_efficiency_frontier(lib, NIT_KEYS, n_subjects, visit_months, n_repeat=n_repeat) rows = sorted(cef["rows"], key=lambda r: r["cost_per_subject"]) st.dataframe([{"NIT": r["label"], "범주": r["category"], "1인당 검사비(원)": f"{r['cost_per_subject']:,}", f"전체({n_subjects}명, 원)": f"{r['cost_total']:,}", "검정력(%)": round(r["power"]*100, 1), "frontier": "O" if r["nit"] in cef["frontier_keys"] else ""} for r in rows], use_container_width=True) st.line_chart({"1인당 검사비(원)": [r["cost_per_subject"] for r in rows], "검정력": [r["power"] for r in rows]}, x="1인당 검사비(원)") fl = [nit_label(k) for k in cef["frontier_keys"]] st.info(f"비용효율 frontier (낮은 비용 대비 높은 검정력): {', '.join(fl)}") # --- 탭 6: 설계 리포트 --- with tabs[5]: st.subheader("관찰연구 프로토콜 측정계획 섹션 리포트") report = core.generate_design_report( lib, sel_nit, n_subjects=n_subjects, n_followups=n_followups, interval_months=interval, subgroup=(sg or "T2DM_or_MASH"), n_repeat=n_repeat, window_weeks=window_weeks, missing_rate=missing_rate, budget_krw=(budget if budget > 0 else None), n_sim=n_sim) st.code(report, language="text") st.download_button("리포트 다운로드 (.txt)", report, file_name=f"masld_nit_design_{sel_nit}.txt") st.divider() st.caption("출처: AASLD MASLD Practice Guidance 2023 · " "EASL-EASD-EASO MASLD CPG 2024 (NIT 권고 참조). " "변동성·진행률·단가는 문헌 기반 추정치.")