#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ DMDeviceRegWatch-Kor (디엠디바이스레그워치코어) ================================================= 한국 당뇨 의료기기·디지털치료기기·AI/ML SaMD multi-jurisdictional 규제 등재 + ISO/IEC/AAMI 표준 + 보험급여 결정 + 학회 abstract 통합 모니터링 도구. - Mock data 기반, 오프라인 동작 (외부 네트워크 호출 없음) - Python 3.11+ 표준 라이브러리만 사용 - 한국어 출력 (weekly regulatory digest) ⚠️ 의학적 디스클레이머 본 도구는 연구·교육·참고용입니다. 실제 규제 결정은 FDA/KFDA/EMA/MHRA/PMDA/NMPA/Health Canada/TGA 등 공식 기관 데이터를 반드시 확인하십시오. Mock data로 동작하며 실제 등재 정보가 아닙니다. """ from __future__ import annotations import argparse import json import sys from collections import Counter, defaultdict from dataclasses import dataclass from datetime import datetime from pathlib import Path from typing import Any # ----------------------------------------------------------------------------- # 상수 정의 # ----------------------------------------------------------------------------- DATA_DIR = Path(__file__).resolve().parent / "data" JURISDICTIONS = ["FDA", "EU", "KFDA", "MHRA", "PMDA", "NMPA", "HC", "TGA"] CATEGORIES = ["CGM", "AID", "PEN", "SaMD", "DTx", "WEARABLE", "STD", "HTA", "ABSTRACT"] # 기기 카테고리 priority (당뇨 임상적 영향도 기준) CATEGORY_PRIORITY = { "AID": 100, # 인공췌장/closed-loop 최우선 "CGM": 80, "PEN": 55, # smart insulin pen "SaMD": 70, # AI 진단(망막 등) "DTx": 60, # 디지털치료기기 "WEARABLE": 25, # non-invasive wearable } # 기기 class priority (위험도 기반) CLASS_PRIORITY = { "III": 60, "IV": 60, "IIb": 45, "3등급": 55, "4등급": 60, "II": 35, "IIa": 35, "2등급": 30, "I": 15, "1등급": 15, } # SaMD risk class (IMDRF) SAMD_PRIORITY = { "Class IV (IMDRF)": 40, "Class III (IMDRF)": 30, "Class II (IMDRF)": 15, "Class I (IMDRF)": 5, } # 한국 시장 / 닥터앤서 3.0 alignment 가산 KOR_ALIGNMENT_KEYWORDS = [ "메디웨일", "닥터누", "뷰노", "VUNO", "웰트", "WELT", "아이센스", "i-SENS", "이오플로우", "EOFlow", "노을", "miLab", "셀바스", "Selvas", "휴이노", "HUINNO", "순천향대", "닥터앤서", ] DISCLAIMER_KOR = ( "⚠️ 본 도구는 연구·교육·참고용입니다. 실제 규제 결정은 " "FDA/KFDA/EMA/MHRA 등 공식 기관 데이터를 반드시 확인하십시오. " "Mock data로 동작하며 실제 등재 정보가 아닙니다." ) # ----------------------------------------------------------------------------- # 데이터 로딩 # ----------------------------------------------------------------------------- @dataclass class DataStore: devices: list[dict[str, Any]] standards: list[dict[str, Any]] hta: list[dict[str, Any]] abstracts: list[dict[str, Any]] def load_data() -> DataStore: """data/ 폴더의 JSON 파일을 로드한다.""" def _load(name: str) -> list[dict[str, Any]]: path = DATA_DIR / name if not path.exists(): print(f"[WARN] 데이터 파일 없음: {path}", file=sys.stderr) return [] with path.open("r", encoding="utf-8") as f: return json.load(f) return DataStore( devices=_load("devices.json"), standards=_load("standards.json"), hta=_load("hta.json"), abstracts=_load("abstracts.json"), ) # ----------------------------------------------------------------------------- # 디덥 # ----------------------------------------------------------------------------- def dedup_devices(devices: list[dict[str, Any]]) -> list[dict[str, Any]]: """(jurisdiction, device_id) 기준 디덥.""" seen: set[tuple[str, str]] = set() out = [] for d in devices: key = (d.get("jurisdiction", ""), d.get("device_id", "")) if key in seen: continue seen.add(key) out.append(d) return out # ----------------------------------------------------------------------------- # 중요도 스코어링 # ----------------------------------------------------------------------------- def score_device(d: dict[str, Any]) -> tuple[float, dict[str, float]]: """디바이스 등재 항목의 중요도 점수를 계산한다.""" breakdown: dict[str, float] = {} # 카테고리 priority cat = d.get("device_category", "") breakdown["category"] = float(CATEGORY_PRIORITY.get(cat, 10)) # device class cls = str(d.get("class", "")).strip() cls_score = 0.0 for key, val in CLASS_PRIORITY.items(): if key in cls: cls_score = max(cls_score, float(val)) breakdown["class"] = cls_score # SaMD class samd = d.get("samd_classification", "") breakdown["samd"] = float(SAMD_PRIORITY.get(samd, 0)) # 결정 유형 가산 (PMA/De Novo/혁신의료기기 우대) dec = str(d.get("decision_type", "")) bonus = 0.0 if "PMA" in dec: bonus += 15 if "De Novo" in dec or "혁신" in dec: bonus += 12 if "SAKIGAKE" in dec or "Breakthrough" in dec: bonus += 10 if "DiGA" in dec: bonus += 8 breakdown["decision_bonus"] = bonus # 한국 시장 진입 가능성 / 닥터앤서 3.0 alignment haystack = " ".join( str(d.get(k, "")) for k in ("sponsor", "device_name", "label_change", "indication") ) kor_bonus = 0.0 if d.get("jurisdiction") == "KFDA": kor_bonus += 25 # KFDA 등재는 한국 시장 직접 진입 for kw in KOR_ALIGNMENT_KEYWORDS: if kw in haystack: kor_bonus += 8 break # 중복 방지 if "닥터앤서" in haystack: kor_bonus += 12 breakdown["kor_alignment"] = kor_bonus # 최신성: 결정일이 최근일수록 가산 (오늘 기준 -90일 이내 +6, -180일 이내 +3) try: ddate = datetime.strptime(d.get("decision_date", ""), "%Y-%m-%d") today = datetime(2026, 5, 14) delta = (today - ddate).days if 0 <= delta <= 90: breakdown["recency"] = 6.0 elif 0 <= delta <= 180: breakdown["recency"] = 3.0 else: breakdown["recency"] = 0.0 except (ValueError, TypeError): breakdown["recency"] = 0.0 total = sum(breakdown.values()) return total, breakdown def score_hta(h: dict[str, Any]) -> float: score = 0.0 if h.get("country") == "KR": score += 40 # 한국 결정 최우선 if "급여" in str(h.get("decision", "")) or "등재" in str(h.get("decision", "")): score += 20 if "확대" in str(h.get("decision", "")): score += 15 if "혁신" in str(h.get("decision", "")) or "선별급여" in str(h.get("decision", "")): score += 12 # 카테고리 가산 for rel in h.get("relevance", []): score += CATEGORY_PRIORITY.get(rel, 0) * 0.3 return score def score_standard(s: dict[str, Any]) -> float: score = 0.0 text = (s.get("title", "") + s.get("scope", "") + s.get("impact", "")).upper() if "AI" in text or "ML" in text or "SaMD" in text.upper(): score += 25 if "ISO 15197" in s.get("std_id", "") or "IEC 62304" in s.get("std_id", ""): score += 20 if "KFDA" in s.get("organization", "") or "식약처" in s.get("organization", ""): score += 30 if "보안" in s.get("scope", "") or "cybersecurity" in s.get("scope", "").lower(): score += 15 # 한국 시장 노트 가산 if s.get("kor_market_note"): score += 10 return score def score_abstract(a: dict[str, Any]) -> float: score = 0.0 conf = a.get("conference", "") if "KDA" in conf or "KSAD" in conf: score += 35 # 한국 학회 우선 if "ATTD" in conf or "DTM" in conf: score += 20 # 당뇨테크 전문 학회 if "ADA" in conf or "EASD" in conf: score += 15 # 한국인 데이터 / 닥터앤서 가산 haystack = a.get("presenter", "") + a.get("title", "") + a.get("kor_relevance_score_note", "") if any(kw in haystack for kw in KOR_ALIGNMENT_KEYWORDS) or "한국" in haystack or "Korean" in haystack: score += 18 # session_type if a.get("session_type", "").startswith("Oral"): score += 8 # 카테고리 for rel in a.get("relevance", []): score += CATEGORY_PRIORITY.get(rel, 0) * 0.25 return score # ----------------------------------------------------------------------------- # 필터링 # ----------------------------------------------------------------------------- def filter_devices( devices: list[dict[str, Any]], jurisdiction: str | None = None, category: str | None = None, ) -> list[dict[str, Any]]: out = devices if jurisdiction and jurisdiction != "ALL": out = [d for d in out if d.get("jurisdiction") == jurisdiction] if category and category != "ALL": out = [d for d in out if d.get("device_category") == category] return out # ----------------------------------------------------------------------------- # Timeline (cross-jurisdiction) # ----------------------------------------------------------------------------- def device_timeline(devices: list[dict[str, Any]], device_query: str) -> list[dict[str, Any]]: """device_id 또는 device_name 일부 매칭으로 jurisdiction별 timeline 정렬.""" q = device_query.lower() matched = [ d for d in devices if q in d.get("device_id", "").lower() or q in d.get("device_name", "").lower() ] # 동일 product family를 묶기 위해 sponsor + device_category로도 그룹 if matched: seeds = {(m.get("sponsor"), m.get("device_category"), m.get("device_name", "").split()[0]) for m in matched} for d in devices: sig = (d.get("sponsor"), d.get("device_category"), d.get("device_name", "").split()[0]) if sig in seeds and d not in matched: matched.append(d) matched.sort(key=lambda d: d.get("decision_date", "")) return matched # ----------------------------------------------------------------------------- # 출력 포맷 # ----------------------------------------------------------------------------- def hline(char: str = "─", n: int = 78) -> str: return char * n def render_device_row(d: dict[str, Any], score: float | None = None) -> str: score_str = f" [점수 {score:5.1f}]" if score is not None else "" return ( f" • [{d.get('jurisdiction'):5}] {d.get('decision_date'):10} " f"{d.get('device_category'):8} {d.get('class'):8}\n" f" {d.get('device_name')} — {d.get('sponsor')}{score_str}\n" f" 결정: {d.get('decision_type')} | ID: {d.get('device_id')}\n" f" 적응증: {d.get('indication')}\n" f" 변경/특이사항: {d.get('label_change')}\n" ) def render_hta_row(h: dict[str, Any], score: float | None = None) -> str: score_str = f" [점수 {score:5.1f}]" if score is not None else "" return ( f" • [{h.get('country'):3}] {h.get('decision_date'):10} {h.get('agency')}{score_str}\n" f" 대상: {h.get('device_or_intervention')}\n" f" 적응증: {h.get('indication')}\n" f" 결정: {h.get('decision')}\n" f" 한국 시장 영향: {h.get('kor_market_impact')}\n" ) def render_standard_row(s: dict[str, Any], score: float | None = None) -> str: score_str = f" [점수 {score:5.1f}]" if score is not None else "" return ( f" • [{s.get('published_date'):10}] {s.get('std_id')} ({s.get('organization')}){score_str}\n" f" {s.get('title')}\n" f" 영향: {s.get('impact')}\n" f" 한국 시장 노트: {s.get('kor_market_note')}\n" ) def render_abstract_row(a: dict[str, Any], score: float | None = None) -> str: score_str = f" [점수 {score:5.1f}]" if score is not None else "" return ( f" • [{a.get('conference')}] {a.get('session_date'):10} {a.get('session_type')}{score_str}\n" f" {a.get('title')}\n" f" 발표자: {a.get('presenter')}\n" f" 핵심: {a.get('key_finding')}\n" f" 한국 의의: {a.get('kor_relevance_score_note')}\n" ) # ----------------------------------------------------------------------------- # 주요 액션 # ----------------------------------------------------------------------------- def action_digest(store: DataStore, top: int = 20, jurisdiction: str | None = None, category: str | None = None) -> None: """한국어 weekly regulatory digest.""" print(hline("═")) print(" DMDeviceRegWatch-Kor — 주간 규제 다이제스트 (2026-05-14 발행)") print(hline("═")) print(f" 도메인: DM (당뇨) 카테고리: 연구 알림") if jurisdiction: print(f" 필터: jurisdiction={jurisdiction}") if category: print(f" 필터: category={category}") print() # 1. Top device clearance devices = dedup_devices(store.devices) devices = filter_devices(devices, jurisdiction, category if category not in (None, "ALL", "STD", "HTA", "ABSTRACT") else None) scored_devices = [(d, *score_device(d)) for d in devices] scored_devices.sort(key=lambda x: x[1], reverse=True) n_dev = min(top, len(scored_devices)) print(hline()) print(f" [1] 디바이스 등재 Top {n_dev} (multi-jurisdictional clearance)") print(hline()) if not scored_devices: print(" (데이터 없음)\n") for d, score, _bd in scored_devices[:n_dev]: print(render_device_row(d, score)) # 2. 보험급여 결정 htas = store.hta scored_hta = [(h, score_hta(h)) for h in htas] scored_hta.sort(key=lambda x: x[1], reverse=True) n_hta = min(max(top // 2, 5), len(scored_hta)) print(hline()) print(f" [2] 보험급여/HTA 결정 Top {n_hta} (HIRA·NICE·CADTH·G-BA·HAS·ICER·MSAC·PBAC)") print(hline()) if not scored_hta: print(" (데이터 없음)\n") for h, score in scored_hta[:n_hta]: print(render_hta_row(h, score)) # 3. ISO/IEC/AAMI 표준 + DTx 인증 프레임워크 stds = store.standards scored_std = [(s, score_standard(s)) for s in stds] scored_std.sort(key=lambda x: x[1], reverse=True) n_std = min(max(top // 2, 5), len(scored_std)) print(hline()) print(f" [3] ISO/IEC/AAMI 표준 + DTx 인증 프레임워크 Top {n_std}") print(hline()) if not scored_std: print(" (데이터 없음)\n") for s, score in scored_std[:n_std]: print(render_standard_row(s, score)) # 4. 학회 abstract abss = store.abstracts scored_abs = [(a, score_abstract(a)) for a in abss] scored_abs.sort(key=lambda x: x[1], reverse=True) n_abs = min(max(top // 4, 5), len(scored_abs)) print(hline()) print(f" [4] 학회 abstract Top {n_abs} (ATTD·DTM·ADA·EASD·KDA·KSAD)") print(hline()) if not scored_abs: print(" (데이터 없음)\n") for a, score in scored_abs[:n_abs]: print(render_abstract_row(a, score)) # 5. 한국 시장 진입 strategy alert print(hline()) print(" [5] 한국 시장 진입 Strategy Alerts") print(hline()) alerts = build_korea_alerts(store) if not alerts: print(" (특이 알림 없음)\n") for alert in alerts: print(f" • {alert}") print() # Footer print(hline("═")) print(f" {DISCLAIMER_KOR}") print(hline("═")) def build_korea_alerts(store: DataStore) -> list[str]: """KFDA 등재 가능성·HIRA 급여 신호·닥터앤서 3.0 alignment alert.""" alerts: list[str] = [] # KFDA 최근 등재 카테고리 통계 kfda = [d for d in store.devices if d.get("jurisdiction") == "KFDA"] cat_counter = Counter(d.get("device_category") for d in kfda) if cat_counter: top_cat = cat_counter.most_common(1)[0] alerts.append( f"KFDA 최근 등재 최다 카테고리: {top_cat[0]} ({top_cat[1]}건) — 동일 카테고리 후속 신청 경쟁 격화 예상" ) # FDA 등재 후 KFDA 미등재 디바이스 (한국 시장 진입 후보) fda_devs = {(d.get("device_name", "").split()[0], d.get("device_category")) for d in store.devices if d.get("jurisdiction") == "FDA"} kfda_devs = {(d.get("device_name", "").split()[0], d.get("device_category")) for d in store.devices if d.get("jurisdiction") == "KFDA"} fda_only = fda_devs - kfda_devs if fda_only: sample = list(fda_only)[:3] sample_str = ", ".join(f"{n[0]}({n[1]})" for n in sample if n[0]) alerts.append( f"FDA 등재 후 KFDA 미등재 디바이스 {len(fda_only)}건 — 예: {sample_str} — 한국 진입 잠재력 모니터링 필요" ) # HIRA 한국 결정 건수 kr_hta = [h for h in store.hta if h.get("country") == "KR"] if kr_hta: alerts.append( f"최근 HIRA/NECA/복지부 결정 {len(kr_hta)}건 발생 — 보험 환경 빠르게 재편 중 (특히 CGM·AID·DTx)" ) # 닥터앤서 3.0 alignment da3 = [d for d in store.devices if any(kw in " ".join(str(v) for v in d.values()) for kw in ["닥터앤서", "메디웨일", "뷰노", "VUNO", "노을", "miLab"])] if da3: alerts.append( f"닥터앤서 3.0 컨소시엄 연계 디바이스 {len(da3)}건 — 정부과제 alignment 적극 활용 권장" ) # DTx 트렌드 dtx = [d for d in store.devices if d.get("device_category") == "DTx"] if dtx: alerts.append( f"전체 DTx 등재 {len(dtx)}건 — 한국 DTx 보험연계(WELT-DM 등) 글로벌 동향 추적 권장" ) return alerts def action_timeline(store: DataStore, device_query: str) -> None: print(hline("═")) print(f" Timeline Cross-Comparison: '{device_query}'") print(hline("═")) matches = device_timeline(dedup_devices(store.devices), device_query) if not matches: print(f" (매칭 결과 없음 — device_id 또는 device_name 일부로 검색)") print(hline("═")) print(f" {DISCLAIMER_KOR}") return by_juris: dict[str, list[dict[str, Any]]] = defaultdict(list) for d in matches: by_juris[d.get("jurisdiction", "?")].append(d) print(f"\n 매칭 항목: {len(matches)}건, jurisdiction {len(by_juris)}개\n") for j in JURISDICTIONS: if j not in by_juris: continue print(f" [{j}]") for d in by_juris[j]: score, _ = score_device(d) print(render_device_row(d, score)) # 미진입 jurisdiction missing = [j for j in JURISDICTIONS if j not in by_juris] if missing: print(hline()) print(f" 미진입 jurisdiction: {', '.join(missing)}") if "KFDA" in missing: print(" ⚠️ KFDA 미진입 — 한국 시장 진입 검토 필요") print() print(hline("═")) print(f" {DISCLAIMER_KOR}") def action_summary(store: DataStore) -> None: print(hline("═")) print(" DMDeviceRegWatch-Kor — 데이터 요약 통계") print(hline("═")) devices = dedup_devices(store.devices) print(f"\n • 총 디바이스 등재: {len(devices)}건 (디덥 후)") # Jurisdiction별 j_counter = Counter(d.get("jurisdiction") for d in devices) print("\n Jurisdiction별:") for j in JURISDICTIONS: cnt = j_counter.get(j, 0) bar = "█" * cnt print(f" {j:6} {cnt:3d} {bar}") # Category별 c_counter = Counter(d.get("device_category") for d in devices) print("\n 카테고리별:") for cat, cnt in c_counter.most_common(): bar = "█" * cnt print(f" {cat:10} {cnt:3d} {bar}") # Decision type dec_counter = Counter() for d in devices: dec = str(d.get("decision_type", "")) # 정규화 (PMA, 510(k), De Novo, KFDA, CE/UKCA, PMDA, NMPA, DiGA 등) if "510(k)" in dec: dec_counter["FDA 510(k)"] += 1 elif "PMA" in dec: dec_counter["FDA PMA"] += 1 elif "De Novo" in dec: dec_counter["FDA De Novo"] += 1 elif "KFDA" in dec or "식약처" in dec: dec_counter["KFDA"] += 1 elif "CE" in dec or "MDR" in dec: dec_counter["EU MDR/CE"] += 1 elif "DiGA" in dec: dec_counter["DiGA"] += 1 elif "UKCA" in dec or "MHRA" in dec: dec_counter["UKCA/MHRA"] += 1 elif "PMDA" in dec or "承認" in dec: dec_counter["PMDA"] += 1 elif "NMPA" in dec: dec_counter["NMPA"] += 1 elif "Health Canada" in dec: dec_counter["Health Canada"] += 1 elif "TGA" in dec: dec_counter["TGA"] += 1 else: dec_counter["기타"] += 1 print("\n 결정 유형별:") for k, v in dec_counter.most_common(): print(f" {k:18} {v:3d}") # 다른 데이터셋 print(f"\n • ISO/IEC/AAMI 표준 update: {len(store.standards)}건") print(f" • HTA/보험급여 결정: {len(store.hta)}건") print(f" • 학회 abstract: {len(store.abstracts)}건") # 한국 alignment kor_aligned = sum( 1 for d in devices if any(kw in " ".join(str(v) for v in d.values()) for kw in KOR_ALIGNMENT_KEYWORDS) or d.get("jurisdiction") == "KFDA" ) print(f"\n • 한국 시장 alignment 디바이스: {kor_aligned}건 ({kor_aligned/max(len(devices),1)*100:.1f}%)") print() print(hline("═")) print(f" {DISCLAIMER_KOR}") print(hline("═")) # ----------------------------------------------------------------------------- # CLI 엔트리 # ----------------------------------------------------------------------------- def build_argparser() -> argparse.ArgumentParser: p = argparse.ArgumentParser( prog="dm-device-reg-watch-kor", description=( "DMDeviceRegWatch-Kor (디엠디바이스레그워치코어) — " "한국 당뇨 의료기기·DTx·SaMD multi-jurisdictional 규제 모니터링 CLI" ), epilog="⚠️ 본 도구는 연구·교육·참고용입니다. 실제 규제 결정은 공식 기관 데이터 확인 필수.", formatter_class=argparse.RawDescriptionHelpFormatter, ) p.add_argument("--digest", action="store_true", help="(기본) 한국어 weekly regulatory digest 출력") p.add_argument("--jurisdiction", choices=JURISDICTIONS + ["ALL"], default=None, help="Jurisdiction 필터") p.add_argument("--category", choices=CATEGORIES + ["ALL"], default=None, help="카테고리 필터 (CGM/AID/PEN/SaMD/DTx/WEARABLE/STD/HTA/ABSTRACT)") p.add_argument("--top", type=int, default=20, help="상위 N개 (기본 20)") p.add_argument("--timeline", type=str, default=None, help="특정 device의 jurisdiction별 timeline 비교 (device_id 또는 device_name 일부)") p.add_argument("--summary", action="store_true", help="데이터 요약 통계 출력") return p def main(argv: list[str] | None = None) -> int: parser = build_argparser() args = parser.parse_args(argv) try: store = load_data() except json.JSONDecodeError as e: print(f"[ERROR] JSON 파싱 실패: {e}", file=sys.stderr) return 2 except OSError as e: print(f"[ERROR] 데이터 파일 읽기 실패: {e}", file=sys.stderr) return 2 # 라우팅 (timeline > summary > digest 기본) if args.timeline: action_timeline(store, args.timeline) elif args.summary: action_summary(store) else: # digest 기본 action_digest( store, top=args.top, jurisdiction=args.jurisdiction, category=args.category, ) return 0 if __name__ == "__main__": sys.exit(main())