#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ MASHGraveyardReframe-Kor (매시그레이브야드리프레임코어) 도메인: MASLD/MASH | 카테고리: 연구 아이디어 생성 (가설 생성·문헌 갭 분석·재설계) phase 2/3에서 endpoint/design 사유로 실패한 MASH trial을 '표적 무효(target-invalidity) vs trial 설계 결함(trial-design)'으로 귀속하고, "표적 폐기 vs subtype 재포지셔닝 vs endpoint 재설계(NIT 기반) vs 조합" 재설계 가설을 attribution 근거와 함께 ranked 생성하는 standalone CLI. 표준 라이브러리만으로 완전 동작 (streamlit/networkx 불필요). 외부 네트워크 호출 없음. 모든 데이터는 로컬 curated/synthetic. 면책: 참고용·연구용 가설 생성 도구이며 임상 의사결정 도구가 아님. """ import argparse import json import os import sys DATA_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "data", "mash_failures.json") DISCLAIMER = ( "※ 면책: 본 도구는 공개된 역사적 임상시험 사실 기반의 연구용 가설 생성 데모입니다. " "attribution 확률·placebo_response 등 세부 수치는 교육/연구용 합성·근사값이며 " "임상 의사결정·규제 판단의 근거가 아닙니다." ) # -------------------------------------------------------------------------- # 데이터 로드 # -------------------------------------------------------------------------- def load_registry(path=DATA_PATH): """MASH 실패 레지스트리(JSON) 로드.""" if not os.path.exists(path): raise FileNotFoundError("데이터 파일을 찾을 수 없습니다: %s" % path) with open(path, "r", encoding="utf-8") as f: data = json.load(f) return data # -------------------------------------------------------------------------- # 귀속 엔진: 표적 무효 vs trial 설계 결함 # -------------------------------------------------------------------------- def attribution_verdict(drug): """ 표적-vs-trial 귀속 판정. p_target_invalid / p_trial_killed (+ 선택적 p_safety_killed)을 비교하여 범주형 판정과 정규화 확률을 반환. """ a = drug.get("attribution", {}) p_target = float(a.get("p_target_invalid", 0.0)) p_trial = float(a.get("p_trial_killed", 0.0)) p_safety = float(a.get("p_safety_killed", 0.0)) total = p_target + p_trial + p_safety if total <= 0: norm = {"target_invalid": 0.0, "trial_killed": 0.0, "safety_killed": 0.0} verdict = "unknown" else: norm = { "target_invalid": round(p_target / total, 3), "trial_killed": round(p_trial / total, 3), "safety_killed": round(p_safety / total, 3), } # 우세 원인 결정 dominant = max(norm, key=norm.get) margin = sorted(norm.values(), reverse=True) spread = margin[0] - margin[1] if len(margin) > 1 else margin[0] if spread < 0.12: verdict = "mixed(%s)" % dominant else: verdict = dominant return { "verdict": verdict, "p_target_invalid": p_target, "p_trial_killed": p_trial, "p_safety_killed": p_safety, "normalized": norm, "evidence": drug.get("attribution_evidence", []), } def verdict_label_kor(verdict): base = verdict.split("(")[0] mapping = { "target_invalid": "표적 무효(폐기 후보)", "trial_killed": "trial이 죽임(재설계 후보)", "safety_killed": "안전성이 죽임(차세대 분자 후보)", "unknown": "판정 불가", } label = mapping.get(base, base) if verdict.startswith("mixed"): label = "혼합 원인 / " + label return label # -------------------------------------------------------------------------- # 재설계 가설 생성 # -------------------------------------------------------------------------- def generate_redesign_hypotheses(drug): """ 귀속 판정 -> 재설계 가설 목록. 표적유효+trial결함 -> endpoint/population/placebo/조합 재설계 표적무효 -> 폐기 권고 + 인접표적 제안 안전성 -> 차세대 분자/조합/용량 """ verd = attribution_verdict(drug) base = verd["verdict"].split("(")[0] hyps = [] levers = drug.get("redesign_levers", []) if base == "target_invalid": hyps.append({ "kind": "deprecate", "text": "표적(%s) 폐기 권고: 독립 endpoint/적응증에서 일관 무반응. " "인접 경로/대체 표적 탐색 권장." % drug["target"], "evidence": verd["evidence"], }) for rt in drug.get("reposition_targets", []): hyps.append({ "kind": "adjacent_target", "text": "인접/대체 방향: %s" % rt, "evidence": [], }) elif base == "safety_killed": hyps.append({ "kind": "next_gen_molecule", "text": "표적(%s) 효능 신호는 존재하나 분자 안전성이 실패 원인. " "차세대(개선된 안전성 프로파일) 분자 또는 저용량·조합 재도전." % drug["target"], "evidence": verd["evidence"], }) for lev in levers: hyps.append({"kind": "redesign_lever", "text": _lever_kor(lev), "evidence": []}) else: # trial_killed / mixed / unknown -> 재설계 위주 for lev in levers: hyps.append({"kind": "redesign_lever", "text": _lever_kor(lev), "evidence": []}) if drug.get("reposition_targets"): hyps.append({ "kind": "reposition", "text": "재포지셔닝 탐색 가치 있음: %s" % ", ".join(drug["reposition_targets"]), "evidence": [], }) if not hyps: hyps.append({"kind": "insufficient", "text": "재설계 가설 도출 근거 부족.", "evidence": []}) return verd, hyps _LEVER_KOR = { "earlier_stage_population": "더 이른 섬유화 단계(F2 중심) population 재정의", "earlier_F2_F3_population": "F2-F3 조기 단계로 population 재정의(가역성 확보)", "NIT_or_MRI_PDFF_endpoint": "조직학 대신 NIT(혈청)/MRI-PDFF 1차 평가변수로 재설계", "NIT_endpoint": "NIT 기반(ALT/MRI-PDFF/cT1) 1차 평가변수 재설계", "MRI_PDFF_primary_endpoint": "MRI-PDFF를 1차 평가변수로 채택", "MRI_PDFF_endpoint": "MRI-PDFF 평가변수 채택", "combination_with_metabolic_target": "대사 표적과의 조합요법", "combination_with_FXR(tropifexor 시도 전례)": "FXR과의 조합(tropifexor 전례)", "enrich_inflammatory_subtype": "염증 우세 subtype enrichment", "reduce_placebo_response_run_in": "run-in/생활습관 표준화로 placebo response 완화", "lower_placebo_via_lifestyle_standardization": "생활습관 표준화로 placebo NASH-resolution 완화", "lower_dose_combination": "저용량 조합으로 안전성 확보", "next_gen_nonsteroidal_FXR": "비스테로이드/비담즙산 차세대 FXR 분자", "statin_co_administration_for_LDL": "LDL 상승 상쇄 위해 statin 병용", "metabolic_high_responder_population": "대사 고반응 subtype 환자 선택", "lower_placebo_via_lifestyle_standardization ": "생활습관 표준화", "larger_N_power": "표본수 확대로 검정력 확보", "diabetic_subtype_enrichment": "당뇨 동반 subtype enrichment", "longer_duration": "추적기간 연장(조직학 검출 확보)", "dose_optimization": "용량/노출 최적화", "early_inflammatory_subtype_only": "초기 염증 subtype만 선별", "cholestatic_pruritus_subtype": "담즙정체성 소양증 subtype으로 전환", "dose_titration_pruritus_management": "용량 적정+소양증 관리로 유효용량 도달", "combination_backbone": "조합요법 백본으로 활용", "powered_phase3_combination": "검정력 확보된 phase3 조합 설계", "earlier_stage": "조기 단계 population", "TG_management_for_ACC": "ACC 부작용(고TG) 관리 병용", "combination_to_offset_TG": "혈청 TG 상승 상쇄 조합", "fish_oil_or_fibrate_coadmin": "오메가-3/fibrate 병용으로 TG 관리", "maintain_phase3": "phase3 지속(유효 표적)", "fibrosis_coprimary": "섬유화 공동-1차평가변수 설정", "lower_dose_combination ": "저용량 조합", } def _lever_kor(lever): return _LEVER_KOR.get(lever, _LEVER_KOR.get(lever.strip(), lever)) # -------------------------------------------------------------------------- # 가설 ranking # -------------------------------------------------------------------------- def rank_drugs(registry): """ 재설계 가능성 점수로 약물 정렬. score = plausibility(재설계 성공 가능성) 기반. 표적무효(폐기)는 점수 낮음, trial-killed/safety는 재설계 여지 높음. """ drugs = registry["drugs"] scored = [] for d in drugs: verd = attribution_verdict(d) base = verd["verdict"].split("(")[0] plaus = float(d.get("plausibility", 0.0)) # 재설계 임팩트 가중: trial_killed/safety는 재설계 여지 가산 if base == "trial_killed": impact = 1.0 elif base == "safety_killed": impact = 0.9 elif verd["verdict"].startswith("mixed"): impact = 0.7 elif base == "target_invalid": impact = 0.2 else: impact = 0.5 score = round(plaus * 0.6 + impact * 0.4, 3) scored.append((score, d, verd)) scored.sort(key=lambda x: x[0], reverse=True) return scored def minimal_redesign_design(drug, verd): """최소 phase 2 재설계 초안(endpoint·population·N) 생성.""" base = verd["verdict"].split("(")[0] if base == "target_invalid": return None # 폐기 후보는 재설계 초안 없음 pr = drug.get("placebo_response_pct", 15) # placebo response가 높을수록 더 큰 N / NIT endpoint 권고 endpoint = "NIT 기반(MRI-PDFF ≥30% 상대감소 + cT1) 1차, 조직학은 2차" if pr >= 17: n = "양군 합산 ~250 (높은 placebo response 보정)" placebo_note = "run-in 4주 + 생활습관 표준화로 placebo response 완화" else: n = "양군 합산 ~180" placebo_note = "표준 생활습관 통제" pop = drug.get("population", "F2_F3") pop_kor = "조기 단계(F2 중심)로 재정의" if "F3" in pop or "cirrhosis" in pop.lower() else pop return { "endpoint": endpoint, "population": pop_kor, "duration": "24-48주", "N": n, "placebo_strategy": placebo_note, "primary_attribution_basis": verd["verdict"], } # -------------------------------------------------------------------------- # 출력 헬퍼 # -------------------------------------------------------------------------- def _bar(p, width=20): n = int(round(p * width)) return "█" * n + "·" * (width - n) def print_summary(registry): meta = registry["_meta"] drugs = registry["drugs"] print("=" * 72) print("MASHGraveyardReframe-Kor — MASH 실패 trial 재설계 가설 생성기") print("도메인: %s | 약물 레지스트리: %d종" % (meta["domain"], len(drugs))) print("=" * 72) print("맥락: %s" % meta.get("context", "")) print() # 귀속 유형별 집계 counts = {} for d in drugs: v = attribution_verdict(d)["verdict"].split("(")[0] counts[v] = counts.get(v, 0) + 1 print("[귀속 유형별 집계]") for k in ("target_invalid", "trial_killed", "safety_killed", "unknown"): if k in counts: print(" - %-16s : %d종 (%s)" % (k, counts[k], verdict_label_kor(k))) print() print("[실패 유형(failure_type) 분포]") ftc = {} for d in drugs: ftc[d["failure_type"]] = ftc.get(d["failure_type"], 0) + 1 for k, v in sorted(ftc.items(), key=lambda x: -x[1]): print(" - %-26s : %d" % (k, v)) print() print(DISCLAIMER) def print_list(registry): print("ID | 약물 / 표적 / phase / 실패유형") print("-" * 72) for d in registry["drugs"]: print("%-19s | %s | %s | p%s | %s" % (d["id"], d["drug"], d["target"], d["phase"], d["failure_type"])) def print_drug_detail(drug): verd, hyps = generate_redesign_hypotheses(drug) print("=" * 72) print("■ %s" % drug["drug"]) print(" 표적: %s | 기전: %s" % (drug["target"], drug["mechanism"])) print(" sponsor: %s | phase %s | trials: %s" % (drug.get("sponsor", "?"), drug["phase"], ", ".join(drug.get("trials", [])))) print(" 결과: %s" % drug["result_summary"]) print(" placebo response: %s%% | treatment: %s%% | endpoint: %s" % (drug.get("placebo_response_pct", "?"), drug.get("treatment_response_pct", "?"), drug.get("histology_endpoint", "?"))) print("-" * 72) print(" [귀속 판정] %s" % verdict_label_kor(verd["verdict"])) nm = verd["normalized"] print(" 표적무효 %s %.0f%%" % (_bar(nm["target_invalid"]), nm["target_invalid"] * 100)) print(" trial결함 %s %.0f%%" % (_bar(nm["trial_killed"]), nm["trial_killed"] * 100)) if nm["safety_killed"] > 0: print(" 안전성 %s %.0f%%" % (_bar(nm["safety_killed"]), nm["safety_killed"] * 100)) if verd["evidence"]: print(" 근거(audit):") for e in verd["evidence"]: print(" • %s" % e) print("-" * 72) print(" [재설계 가설]") for i, h in enumerate(hyps, 1): print(" %d. (%s) %s" % (i, h["kind"], h["text"])) design = minimal_redesign_design(drug, verd) if design: print("-" * 72) print(" [최소 phase 2 재설계 초안]") print(" endpoint : %s" % design["endpoint"]) print(" population : %s" % design["population"]) print(" duration : %s" % design["duration"]) print(" N : %s" % design["N"]) print(" placebo : %s" % design["placebo_strategy"]) print() def print_ranked(registry, top): scored = rank_drugs(registry) print("=" * 72) print("재설계 가설 RANKED (재설계 성공 plausibility × 귀속 임팩트)") print("=" * 72) for rank, (score, d, verd) in enumerate(scored[:top], 1): verd_lbl = verdict_label_kor(verd["verdict"]) print("%2d. [%.3f] %-32s | %s" % (rank, score, d["drug"], verd_lbl)) # 대표 재설계 가설 1개 _, hyps = generate_redesign_hypotheses(d) if hyps: print(" → %s" % hyps[0]["text"]) print() print(DISCLAIMER) def print_reposition(registry): print("=" * 72) print("재포지셔닝 탐색 — 실패 MASH 약물 표적의 인접 적응증 매핑") print("=" * 72) print("(전례: elafibranor/PPAR — MASH 실패 후 PBC 재포지셔닝 승인)") print("-" * 72) for d in registry["drugs"]: rts = d.get("reposition_targets", []) if not rts: continue print("■ %s (%s)" % (d["drug"], d["target"])) for rt in rts: print(" → %s" % rt) print() print(DISCLAIMER) # -------------------------------------------------------------------------- # CLI # -------------------------------------------------------------------------- def build_parser(): p = argparse.ArgumentParser( prog="main.py", description="MASHGraveyardReframe-Kor: MASH 실패 trial을 " "'표적 무효 vs trial 설계 결함'으로 귀속하고 " "재설계/재포지셔닝 가설을 ranked 생성하는 standalone 도구 " "(표준 라이브러리만으로 동작, 오프라인).", epilog="예) python3 main.py --summary | --top 8 | " "--drug selonsertib --attribution | --reposition | --list", formatter_class=argparse.RawDescriptionHelpFormatter, ) p.add_argument("--summary", action="store_true", help="레지스트리 요약(귀속 유형·실패 유형 집계) 출력") p.add_argument("--list", action="store_true", help="전체 실패 약물 레지스트리 목록 출력") p.add_argument("--top", type=int, metavar="N", help="재설계 가설 ranked 상위 N개 출력") p.add_argument("--drug", type=str, metavar="NAME", help="특정 약물(id 또는 이름 일부) 상세: 귀속+재설계 가설+초안") p.add_argument("--attribution", action="store_true", help="(--drug와 함께/단독) 귀속 판정 상세 강조") p.add_argument("--reposition", action="store_true", help="재포지셔닝 탐색(인접 적응증 매핑) 출력") p.add_argument("--data", type=str, default=DATA_PATH, help="데이터 JSON 경로 override (기본: data/mash_failures.json)") return p def find_drug(registry, name): name_l = name.lower() for d in registry["drugs"]: if d["id"].lower() == name_l or name_l in d["drug"].lower() \ or name_l in d["id"].lower() or name_l in d["target"].lower(): return d return None def main(argv=None): parser = build_parser() args = parser.parse_args(argv) try: registry = load_registry(args.data) except Exception as e: print("ERROR: 데이터 로드 실패 — %s" % e, file=sys.stderr) return 2 did_something = False if args.summary: print_summary(registry) did_something = True if args.list: print_list(registry) did_something = True if args.drug: d = find_drug(registry, args.drug) if d is None: print("해당 약물을 찾을 수 없습니다: %s" % args.drug, file=sys.stderr) print("사용 가능한 id: %s" % ", ".join(x["id"] for x in registry["drugs"]), file=sys.stderr) return 1 print_drug_detail(d) if args.attribution: verd = attribution_verdict(d) print(" [귀속 raw] target_invalid=%.2f trial_killed=%.2f safety_killed=%.2f -> %s" % (verd["p_target_invalid"], verd["p_trial_killed"], verd["p_safety_killed"], verd["verdict"])) did_something = True elif args.attribution: # --drug 없이 --attribution: 전체 귀속 요약 테이블 print("전체 귀속 판정 요약") print("-" * 72) for d in registry["drugs"]: v = attribution_verdict(d) print("%-19s | %-28s | %s" % (d["id"], d["target"], verdict_label_kor(v["verdict"]))) did_something = True if args.top is not None: print_ranked(registry, max(1, args.top)) did_something = True if args.reposition: print_reposition(registry) did_something = True if not did_something: # 기본 동작: 요약 + top 5 print_summary(registry) print() print_ranked(registry, 5) return 0 if __name__ == "__main__": sys.exit(main())