#!/usr/bin/env python3 """ AntiObesityHTAWatch-Kor (안티오베시티에이치티에이워치코어) - 항비만 약물 multi-jurisdictional HTA 결정 모니터링 - Cost-effectiveness ICER/QALY threshold 호환성 + BIA - ESMO-MCBS analog for obesity + persistence/discontinuation - 학회 abstract + 한국 시장 진입 strategy alert - 한국어 weekly HTA digest """ from __future__ import annotations import argparse import json import os import sys from collections import Counter, defaultdict from dataclasses import dataclass from datetime import datetime from statistics import mean from typing import Any BASE_DIR = os.path.dirname(os.path.abspath(__file__)) DATA_DIR = os.path.join(BASE_DIR, "data") JURISDICTIONS = [ "NICE", "CADTH", "ICER", "MSAC", "PBAC", "G-BA", "HAS", "KFDA", "HIRA", "NECA", "REA", "JP", ] DISCLAIMER = ( "본 도구는 연구·교육·참고용입니다. 실제 HTA 결정·보험급여 정보는 NICE/CADTH/HIRA 등 " "공식 기관 데이터를 반드시 확인하십시오. Mock data로 동작하며 실제 결정 정보가 아닙니다. " "처방 결정은 임상의·약사·환자와의 협의 하에 이루어져야 합니다." ) # ----------------------------- 데이터 로딩 ----------------------------- def _load_json(name: str) -> dict[str, Any]: path = os.path.join(DATA_DIR, name) with open(path, "r", encoding="utf-8") as f: return json.load(f) def load_all() -> dict[str, Any]: return { "hta": _load_json("hta_decisions.json"), "icer": _load_json("icer_qaly.json"), "mcbs": _load_json("mcbs_obesity.json"), "abstracts": _load_json("abstracts.json"), } # ----------------------------- 디덥 + 중요도 ----------------------------- def dedup_decisions(decisions: list[dict]) -> list[dict]: """(drug_id, jurisdiction, agency_id) 기준 디덥, 가장 최근 결정 우선.""" by_key: dict[tuple, dict] = {} for d in decisions: key = (d.get("drug_id"), d.get("jurisdiction"), d.get("agency_id")) existing = by_key.get(key) if existing is None: by_key[key] = d continue # 최신 decision_date 우선 if d.get("decision_date", "") > existing.get("decision_date", ""): by_key[key] = d return list(by_key.values()) DECISION_WEIGHT = { "recommended": 1.0, "conditional": 0.85, "restricted": 0.7, "resubmission": 0.55, "not_recommended": 0.5, "withdrawn": 0.3, "pending": 0.6, } JURISDICTION_WEIGHT = { "HIRA": 1.6, "KFDA": 1.4, "NECA": 1.4, # 한국 우대 "NICE": 1.1, "ICER": 1.05, "G-BA": 1.05, "REA": 1.05, "CADTH": 1.0, "HAS": 1.0, "PBAC": 0.95, "MSAC": 0.9, "JP": 0.95, } def score_decision(d: dict, now: datetime | None = None) -> float: """중요도 스코어: jurisdiction × decision_type × 최근성 × BIA 규모.""" now = now or datetime(2026, 5, 14) j = JURISDICTION_WEIGHT.get(d.get("jurisdiction"), 1.0) dt = DECISION_WEIGHT.get(d.get("decision_type"), 0.5) # 최근성: 1년 이내 1.0, 2년 이내 0.7, 그 이상 0.4 date_str = d.get("decision_date", "2000-01-01") try: dd = datetime.strptime(date_str, "%Y-%m-%d") days = (now - dd).days except ValueError: days = 9999 if days < 365: recency = 1.0 elif days < 730: recency = 0.7 else: recency = 0.4 bia = d.get("budget_impact_5yr_local") or 0 # log-ish scaling to dampen outliers (단위 미상이라 jurisdiction normalize 생략, 상한만) bia_score = min(0.4, (bia / 1_000_000_000) * 0.02) return round(j * dt * recency + bia_score, 3) # ----------------------------- ICER threshold 호환성 ----------------------------- def evaluate_icer(d: dict, thresholds: dict) -> dict: """단일 결정의 ICER/QALY가 해당 jurisdiction의 임계값을 만족하는지 평가.""" icer = d.get("ICER_QALY") cur = d.get("ICER_currency") j = d.get("jurisdiction") result: dict[str, Any] = { "drug_id": d["drug_id"], "jurisdiction": j, "ICER_QALY": icer, "currency": cur, "threshold": None, "verdict": "N/A", "note": "", } if icer is None: result["note"] = "pending/no ICER reported" return result t = thresholds if j == "NICE": if icer <= t["NICE_standard"]["high_GBP"]: result["verdict"] = "WITHIN_THRESHOLD" elif icer <= t["NICE_end_of_life"]["high_GBP"]: result["verdict"] = "WITHIN_EOL_RANGE" else: result["verdict"] = "ABOVE_THRESHOLD" result["threshold"] = t["NICE_standard"]["label"] elif j == "ICER": if icer <= t["ICER_US"]["high_USD"]: result["verdict"] = "WITHIN_THRESHOLD" else: result["verdict"] = "ABOVE_THRESHOLD" result["threshold"] = t["ICER_US"]["label"] elif j == "CADTH": result["verdict"] = "WITHIN_THRESHOLD" if icer <= t["CADTH"]["threshold_CAD"] else "ABOVE_THRESHOLD" result["threshold"] = t["CADTH"]["label"] elif j in ("PBAC", "MSAC"): result["verdict"] = "WITHIN_THRESHOLD" if icer <= t["PBAC"]["threshold_AUD"] else "ABOVE_THRESHOLD" result["threshold"] = t["PBAC"]["label"] elif j in ("HIRA", "NECA"): if icer <= t["HIRA_1x_GDP"]["threshold_KRW"]: result["verdict"] = "WITHIN_THRESHOLD" elif icer <= t["HIRA_2x_GDP"]["threshold_KRW"]: result["verdict"] = "WITHIN_RARE_EXCEPTION" else: result["verdict"] = "ABOVE_THRESHOLD" result["threshold"] = t["HIRA_1x_GDP"]["label"] elif j in ("G-BA", "HAS", "REA"): # 정량 임계값 없음 — 정성적으로 NICE high_GBP×currency 비유 result["verdict"] = "WITHIN_RANGE" if icer <= 35000 else "ABOVE_RANGE" result["threshold"] = "EUR ~30k qualitative" elif j == "JP": result["verdict"] = "WITHIN_THRESHOLD" if icer <= t["JP_C2H"]["threshold_JPY"] else "ABOVE_THRESHOLD" result["threshold"] = t["JP_C2H"]["label"] elif j == "KFDA": result["note"] = "KFDA는 허가 기관 — HTA threshold 없음 (HIRA 별도 평가)" return result # ----------------------------- 한국 BIA 시뮬레이션 ----------------------------- KOREA_OBESITY_ELIGIBLE = 8_000_000 # 800만 처방 가능 인구 KOREA_HIRA_GDP_PER_CAPITA_KRW = 38_000_000 def simulate_bia_korea(annual_cost_KRW: int | None, uptake_pcts: list[float] | None = None) -> dict: """한국 NHIS 관점 5년·10년 BIA 시뮬레이션.""" if annual_cost_KRW is None: return {"available": False, "note": "annual_cost_KRW missing"} uptakes = uptake_pcts or [0.05, 0.10, 0.20] results = [] for u in uptakes: patients = int(KOREA_OBESITY_ELIGIBLE * u) annual_burden = patients * annual_cost_KRW results.append({ "uptake_pct": round(u * 100, 1), "patients": patients, "annual_KRW": annual_burden, "5yr_KRW": annual_burden * 5, "10yr_KRW": annual_burden * 10, }) return {"available": True, "scenarios": results, "eligible_pool": KOREA_OBESITY_ELIGIBLE} # ----------------------------- 비만 합병증 stratification ----------------------------- COMORBIDITY_TAGS = [ "T2DM", "MASLD", "MASH", "HFpEF", "CKD", "sleep apnea", "sarcopenic obesity", "청소년", "gestational", ] def stratify_population(decision: dict) -> list[str]: pop = (decision.get("indication_population") or "") + " " + (decision.get("indication") or "") hits = [] pop_lower = pop.lower() for tag in COMORBIDITY_TAGS: if tag.lower() in pop_lower: hits.append(tag) return hits # ----------------------------- 한국 시장 진입 strategy alert ----------------------------- @dataclass class StrategyAlert: drug_id: str brand: str headline: str detail: str estimated_patient_copay_KRW_month: int | None timeline_hint: str priority: int # 1=high def generate_korea_strategy_alerts(data: dict) -> list[StrategyAlert]: alerts: list[StrategyAlert] = [] icer_drugs = {d["drug_id"]: d for d in data["icer"]["drugs"]} # 1) Wegovy KFDA허가 + HIRA 미급여 → 비급여 본인부담 추정 weg = icer_drugs.get("semaglutide_2.4mg") if weg and weg.get("annual_cost_KRW_estimate"): copay_month = weg["annual_cost_KRW_estimate"] // 12 alerts.append(StrategyAlert( drug_id="semaglutide_2.4mg", brand="Wegovy", headline="KFDA 허가완료·HIRA 비급여 — 100% 본인부담 시장 진입", detail=( "노보 노디스크 코리아 2025-04 허가완료. HIRA 2025-11 ICER 5,200만원/QALY로 " "비급여 결정. 시장은 비급여 약 70만원/월 가격대 형성. BMI>=30+T2DM 환자에서 " "당뇨 적응증 Ozempic(급여)으로 cross-indication 처방 발생 가능." ), estimated_patient_copay_KRW_month=copay_month, timeline_hint="HIRA 재평가 2027 예상, 산정특례 가능성 낮음", priority=1, )) # 2) Zepbound NECA 신의료기술평가 진행 중 alerts.append(StrategyAlert( drug_id="tirzepatide_15mg", brand="Zepbound/Mounjaro", headline="NECA 신의료기술평가 진행 중 (NECA-2026-OB-01)", detail=( "한국릴리, 2026-05 NECA 비교효과연구 진입. BMI>=35+합병증 또는 한국 기준 " "BMI>=30 적응증. PBAC/G-BA에서 recommended 결과 다수 — HIRA 등재 협상 leverage. " "당뇨 적응증 Mounjaro는 2026-01 KFDA 허가 후 부분급여 시도 중." ), estimated_patient_copay_KRW_month=800_000, timeline_hint="NECA 결과 2026 H2, HIRA 등재 협상 2027 H1 예상", priority=1, )) # 3) 부분급여 50/30/20% 시나리오 (KOPS-2026-OBE-013 abstract 기반) alerts.append(StrategyAlert( drug_id="multi", brand="GLP-1RA 군", headline="HIRA 부분급여 50/30/20% 시나리오 토론 부상", detail=( "KOPS 2026 abstract에 따라 HIRA가 비만 약물 단계적 부분급여 도입을 검토할 가능성. " "50% 본인부담 시나리오가 NHIS 재정/접근성 최적 균형으로 분석. " "BMI>=35+합병증 1차 지정, 청소년·산정특례 별도 트랙 가능성." ), estimated_patient_copay_KRW_month=350_000, timeline_hint="HIRA 약제급여평가위원회 2026 H2~2027 H1 안건 채택 추적", priority=2, )) # 4) Setmelanotide 산정특례 alerts.append(StrategyAlert( drug_id="setmelanotide", brand="Imcivree", headline="희귀유전성비만 산정특례 진입 (HIRA-2026-RARE-02)", detail=( "POMC/LEPR/PCSK1 결손 환자 대상 conditional 결정. 본인부담 10%. " "유전자검사 의무, 신청·심사 procedure 가이드 필요." ), estimated_patient_copay_KRW_month=2_700_000, # 320M/yr * 0.10 / 12 timeline_hint="2026-03 이미 결정, 신규 환자 등록 절차 가이드 필요", priority=3, )) # 5) 한국 RWE persistence 낮음 → outcome-linked rebate 가능성 alerts.append(StrategyAlert( drug_id="multi", brand="GLP-1RA 군", headline="한국 RWE 1년 지속률 38% — outcome-linked RSA 전략 부상", detail=( "KSSO 2026 abstract: 한국 1년 지속률 38% (서구 50% 대비 낮음). " "체중 재증가 8.2kg 평균. HIRA managed entry agreement(outcome-linked rebate) " "협상 카드로 활용 가능 — 6개월·1년 체중 감소 미달 시 환급." ), estimated_patient_copay_KRW_month=None, timeline_hint="2026~2027 신규 등재 협상 시 표준 조항 검토", priority=2, )) return alerts # ----------------------------- 출력 헬퍼 ----------------------------- def _fmt_money(amount: int | float | None, currency: str | None = None) -> str: if amount is None: return "—" if currency == "KRW": if abs(amount) >= 1e12: return f"{amount/1e12:.2f}조원" if abs(amount) >= 1e8: return f"{amount/1e8:.1f}억원" if abs(amount) >= 1e4: return f"{amount/1e4:.0f}만원" return f"{amount:,.0f}원" if currency == "JPY": if abs(amount) >= 1e8: return f"¥{amount/1e8:.2f}억" return f"¥{amount:,.0f}" symbol = {"GBP": "£", "USD": "$", "CAD": "C$", "AUD": "A$", "EUR": "€"}.get(currency or "", "") if abs(amount) >= 1e9: return f"{symbol}{amount/1e9:.2f}B" if abs(amount) >= 1e6: return f"{symbol}{amount/1e6:.1f}M" return f"{symbol}{amount:,.0f}" def _print_sep(ch: str = "=", width: int = 78) -> None: print(ch * width) # ----------------------------- 커맨드: --summary ----------------------------- def cmd_summary(data: dict) -> None: decisions = dedup_decisions(data["hta"]["decisions"]) by_juri = Counter(d["jurisdiction"] for d in decisions) by_dec = Counter(d["decision_type"] for d in decisions) drugs = {d["drug_id"] for d in decisions} abstracts = data["abstracts"]["abstracts"] _print_sep() print("AntiObesityHTAWatch-Kor — Summary") _print_sep() print(f"총 HTA 결정: {len(decisions)}건 (디덥 후)") print(f"포함 약물: {len(drugs)}종 — {', '.join(sorted(drugs))}") print(f"포함 jurisdiction: {len(by_juri)}개") print() print("Jurisdiction별 분포:") for j, n in by_juri.most_common(): print(f" - {j:7s}: {n}건") print() print("결정 유형 분포:") for d, n in by_dec.most_common(): print(f" - {d:18s}: {n}건") print() print(f"학회 abstract: {len(abstracts)}건") print(f"MCBS 평가 약물: {len(data['mcbs']['drugs'])}종") print() print(f"디스클레이머: {DISCLAIMER}") # ----------------------------- 커맨드: --digest ----------------------------- def cmd_digest(data: dict, top: int) -> None: decisions = dedup_decisions(data["hta"]["decisions"]) scored = sorted( [(score_decision(d), d) for d in decisions], key=lambda x: -x[0], ) abstracts = sorted( data["abstracts"]["abstracts"], key=lambda a: -a.get("korea_relevance_score", 0), ) alerts = generate_korea_strategy_alerts(data) alerts_sorted = sorted(alerts, key=lambda a: a.priority) print() _print_sep() print(f"한국어 Weekly HTA Digest — 발행일 mock 2026-05-14") print(f"발행: AntiObesityHTAWatch-Kor") _print_sep() print(f"\n[1] Top {min(top*3, len(scored))} HTA 결정 (중요도 스코어 순)") print("-" * 78) for i, (sc, d) in enumerate(scored[: top * 3], 1): icer = _fmt_money(d.get("ICER_QALY"), d.get("ICER_currency")) bia = _fmt_money(d.get("budget_impact_5yr_local"), d.get("ICER_currency")) strat = stratify_population(d) print(f"{i:2d}. [{d['jurisdiction']:5s}] {d['brand']} — {d['decision_type']}") print(f" 날짜: {d['decision_date']} | 적응증: {d['indication']} ({d['indication_population']})") print(f" ICER/QALY: {icer} | 5yr BIA: {bia} | 본인부담: {d.get('patient_co_pay_pct', '—')}%") if strat: print(f" 합병증 stratification: {', '.join(strat)}") print(f" Notes: {d.get('notes', '')}") print(f" 중요도 score: {sc}") print() print(f"\n[2] Top {min(top*2, len(abstracts))} 학회 Abstract (한국 관련성 순)") print("-" * 78) for i, a in enumerate(abstracts[: top * 2], 1): print(f"{i:2d}. [{a['conference']}] {a['title']}") print(f" 저자: {a['lead_author']} | 일자: {a['date']} | 장소: {a['venue']}") print(f" 핵심: {a['key_finding']}") print(f" 한국 관련성: {a.get('korea_relevance_score', 0)}/10 | 약물: {', '.join(a.get('drug_ids', []))}") print() print(f"\n[3] Top {min(top, len(alerts_sorted))} 한국 시장 진입 Strategy Alert") print("-" * 78) for i, al in enumerate(alerts_sorted[:top], 1): copay = _fmt_money(al.estimated_patient_copay_KRW_month, "KRW") if al.estimated_patient_copay_KRW_month else "—" print(f"{i:2d}. [P{al.priority}] {al.headline}") print(f" 약물: {al.brand} ({al.drug_id})") print(f" {al.detail}") print(f" 예상 환자 본인부담: {copay}/월 | 타임라인: {al.timeline_hint}") print() print(f"\n[4] Top {min(top, 5)} ICER/QALY threshold 호환성 시뮬레이션") print("-" * 78) thresholds = data["icer"]["thresholds"] quant = [d for d in decisions if d.get("ICER_QALY") is not None] quant.sort(key=lambda d: -score_decision(d)) for i, d in enumerate(quant[:top], 1): ev = evaluate_icer(d, thresholds) print(f"{i:2d}. [{ev['jurisdiction']:5s}] {d['brand']} — {_fmt_money(ev['ICER_QALY'], ev['currency'])}") print(f" 임계값: {ev['threshold']} | 판정: {ev['verdict']}") print() print(f"\n[5] 한국 NHIS BIA 요약 (800만 처방 가능 인구 기준)") print("-" * 78) icer_drugs = data["icer"]["drugs"] targeted = ["tirzepatide_15mg", "semaglutide_2.4mg", "liraglutide_3.0mg"] for did in targeted: drug = next((x for x in icer_drugs if x["drug_id"] == did), None) if not drug: continue ann = drug.get("annual_cost_KRW_estimate") bia = simulate_bia_korea(ann, [0.05, 0.10]) print(f" - {drug['brand']} (연간 비용 {_fmt_money(ann, 'KRW')}/환자)") if bia["available"]: for s in bia["scenarios"]: print(f" uptake {s['uptake_pct']}%: 환자 {s['patients']:,}명, 5yr {_fmt_money(s['5yr_KRW'], 'KRW')}, 10yr {_fmt_money(s['10yr_KRW'], 'KRW')}") print() _print_sep() print(f"디스클레이머: {DISCLAIMER}") _print_sep() # ----------------------------- 커맨드: --jurisdiction ----------------------------- def cmd_jurisdiction(data: dict, juri: str, top: int) -> None: decisions = dedup_decisions(data["hta"]["decisions"]) if juri != "ALL": decisions = [d for d in decisions if d["jurisdiction"] == juri] decisions.sort(key=lambda d: -score_decision(d)) print(f"\n[Jurisdiction={juri}] {len(decisions)}건") print("-" * 78) for i, d in enumerate(decisions[:top], 1): icer = _fmt_money(d.get("ICER_QALY"), d.get("ICER_currency")) bia = _fmt_money(d.get("budget_impact_5yr_local"), d.get("ICER_currency")) print(f"{i:2d}. [{d['jurisdiction']:5s}] {d['brand']} — {d['decision_type']}") print(f" {d['decision_date']} | ICER {icer} | BIA {bia} | {d.get('notes', '')}") print() # ----------------------------- 커맨드: --drug ----------------------------- def cmd_drug(data: dict, drug_query: str) -> None: decisions = dedup_decisions(data["hta"]["decisions"]) q = drug_query.lower() matched = [ d for d in decisions if q in (d.get("drug_id") or "").lower() or q in (d.get("brand") or "").lower() ] if not matched: print(f"매칭되는 약물 없음: {drug_query}") return matched.sort(key=lambda d: d["jurisdiction"]) drug_name = matched[0]["brand"] print(f"\n[Cross-jurisdiction 비교] {drug_name} ({matched[0]['drug_id']})") print("-" * 78) print(f"{'Jurisdiction':12s} {'Date':12s} {'Decision':16s} {'ICER/QALY':18s} {'Co-pay':8s}") print("-" * 78) for d in matched: icer = _fmt_money(d.get("ICER_QALY"), d.get("ICER_currency")) copay = f"{d.get('patient_co_pay_pct')}%" if d.get("patient_co_pay_pct") is not None else "—" print(f"{d['jurisdiction']:12s} {d['decision_date']:12s} {d['decision_type']:16s} {icer:18s} {copay:8s}") print() # MCBS / ICER 데이터 mcbs = next((m for m in data["mcbs"]["drugs"] if m["drug_id"] == matched[0]["drug_id"]), None) icer_d = next((x for x in data["icer"]["drugs"] if x["drug_id"] == matched[0]["drug_id"]), None) if mcbs: print(f"ESMO-MCBS analog score: {mcbs.get('total_score', 'N/A')} ({mcbs.get('label', '')})") if icer_d: print(f"평균 체중 감소: {icer_d.get('weight_loss_pct_mean', '—')}%") print(f"MACE HR: {icer_d.get('MACE_HR', '—')} | T2DM prevention HR: {icer_d.get('T2DM_prevention_HR', '—')}") print(f"MASLD remission OR: {icer_d.get('MASLD_remission_OR', '—')} | HFpEF KCCQ delta: {icer_d.get('HFpEF_KCCQ_delta', '—')}") print(f"근감소 안전성: {icer_d.get('sarcopenia_safety', '—')}") print() # ----------------------------- 커맨드: --icer-eval ----------------------------- def cmd_icer_eval(data: dict, top: int) -> None: decisions = dedup_decisions(data["hta"]["decisions"]) thresholds = data["icer"]["thresholds"] rows = [] for d in decisions: if d.get("ICER_QALY") is None: continue ev = evaluate_icer(d, thresholds) rows.append((score_decision(d), d, ev)) rows.sort(key=lambda r: -r[0]) print(f"\n[ICER/QALY threshold 호환성 평가] {len(rows)}건 (수치 있는 결정)") print("-" * 78) print(f"{'#':3s} {'Juri':6s} {'Brand':28s} {'ICER/QALY':16s} {'Verdict':22s}") print("-" * 78) for i, (sc, d, ev) in enumerate(rows[:top], 1): icer = _fmt_money(ev["ICER_QALY"], ev["currency"]) print(f"{i:<3d} {ev['jurisdiction']:6s} {d['brand'][:28]:28s} {icer:16s} {ev['verdict']:22s}") print() # 요약 verdicts = Counter(r[2]["verdict"] for r in rows) print("판정 분포:") for v, n in verdicts.most_common(): print(f" - {v:24s}: {n}건") print() # ----------------------------- 커맨드: --bia ----------------------------- def cmd_bia(data: dict) -> None: print(f"\n[한국 NHIS BIA 시뮬레이션] 800만 처방 가능 인구 기준") print("-" * 78) drugs = data["icer"]["drugs"] for drug in drugs: ann = drug.get("annual_cost_KRW_estimate") if ann is None: continue bia = simulate_bia_korea(ann, [0.05, 0.10, 0.20]) if not bia["available"]: continue print(f"\n {drug['brand']} ({drug['drug_id']}) — 연간 비용 {_fmt_money(ann, 'KRW')}/환자") for s in bia["scenarios"]: within_1x = "✓" if (ann / 0.149 if drug["drug_id"]=='semaglutide_2.4mg' else 0) <= KOREA_HIRA_GDP_PER_CAPITA_KRW else "" print( f" uptake {s['uptake_pct']:>4.1f}%: " f"환자 {s['patients']:>9,}명 | " f"연 {_fmt_money(s['annual_KRW'], 'KRW'):>10s} | " f"5yr {_fmt_money(s['5yr_KRW'], 'KRW'):>10s} | " f"10yr {_fmt_money(s['10yr_KRW'], 'KRW'):>10s}" ) print() print(f"HIRA 1배 GDP per capita 임계값: {_fmt_money(KOREA_HIRA_GDP_PER_CAPITA_KRW, 'KRW')}/QALY") print() # ----------------------------- 커맨드: --mcbs ----------------------------- def cmd_mcbs(data: dict, top: int) -> None: drugs = sorted( data["mcbs"]["drugs"], key=lambda x: -(x.get("total_score") or -1), ) print(f"\n[ESMO-MCBS analog for Obesity — 종합 점수]") print("-" * 78) print(f"{'#':3s} {'Drug':28s} {'Score':8s} {'Label':30s}") print("-" * 78) for i, d in enumerate(drugs[:top], 1): sc = d.get("total_score") sc_str = f"{sc:.1f}" if sc is not None else "pend" print(f"{i:<3d} {d['drug_id'][:28]:28s} {sc_str:8s} {d.get('label', '')[:30]:30s}") print() rwe = data["mcbs"]["discontinuation_RWE_1yr"] print("Persistence / discontinuation RWE (1yr):") print(f" - GLP-1RA 비만 적응증 중단률: {rwe['GLP-1RA_obesity_indication_pct']}%") print(f" - GLP-1RA T2DM 적응증 중단률: {rwe['GLP-1RA_T2DM_indication_pct']}%") print(f" - STEP-4 drug-off regain 1yr: {rwe['STEP-4_drug_off_regain_1yr_pct']}%") print(f" - SURMOUNT-4 drug-off regain 1yr: {rwe['SURMOUNT-4_drug_off_regain_1yr_pct']}%") print() # ----------------------------- argparse ----------------------------- def build_parser() -> argparse.ArgumentParser: p = argparse.ArgumentParser( prog="anti-obesity-hta-watch-kor", description="항비만 약물 multi-jurisdictional HTA 결정 모니터링 + 한국 시장 진입 strategy alert (mock data)", ) p.add_argument("--digest", action="store_true", help="한국어 weekly HTA digest (기본)") p.add_argument("--summary", action="store_true", help="데이터셋 요약") p.add_argument( "--jurisdiction", choices=JURISDICTIONS + ["ALL"], help="특정 jurisdiction 필터", ) p.add_argument("--drug", help="특정 약물 cross-jurisdiction 비교 (brand 또는 drug_id)") p.add_argument("--icer-eval", action="store_true", help="ICER/QALY threshold 호환성 평가") p.add_argument("--bia", action="store_true", help="한국 NHIS BIA 5년/10년 시뮬레이션") p.add_argument("--mcbs", action="store_true", help="ESMO-MCBS analog for obesity 점수") p.add_argument("--top", type=int, default=5, help="상위 N개 (기본 5)") return p def main(argv: list[str] | None = None) -> int: args = build_parser().parse_args(argv) data = load_all() # 라우팅: 명시 옵션 우선, 아니면 digest if args.summary: cmd_summary(data) elif args.jurisdiction: cmd_jurisdiction(data, args.jurisdiction, args.top) elif args.drug: cmd_drug(data, args.drug) elif getattr(args, "icer_eval"): cmd_icer_eval(data, args.top * 4) elif args.bia: cmd_bia(data) elif args.mcbs: cmd_mcbs(data, args.top * 3) else: # default: digest cmd_digest(data, args.top) return 0 if __name__ == "__main__": sys.exit(main())