AI 헬스케어 부트캠프

1-1. 변수와 데이터 타입

변수는 데이터를 저장하는 상자입니다. 상자에 이름표를 붙여서 나중에 꺼내 쓸 수 있어요.

name = "김개발"       # 문자열(str) - 글자를 저장
age = 25              # 정수(int) - 숫자를 저장
height = 175.5        # 실수(float) - 소수점 숫자
is_student = True     # 불린(bool) - 참/거짓

변수 이름 짓는 법 (snake_case)

# 좋은 예
patient_name = "김환자"
blood_pressure = "120/80"

# 나쁜 예
a = "김환자"         # 의미 불분명
2patient = "김환자"  # 숫자로 시작 불가
patient-name = "..."  # 하이픈 불가

타입 확인 및 변환

type(name)             # <class 'str'>  타입 확인
int("25")              # 문자열 → 정수
float("175.5")         # 문자열 → 실수
str(95)                # 숫자 → 문자열
bool(0)                # False (0만 False, 나머지 True)

f-string (문자열 포매팅)

name = "이민수"
age = 28
height = 180.5489
print(f"이름: {name}, 나이: {age}세")
print(f"키: {height:.1f}cm")       # 소수점 1자리 → 180.5
print(f"내년 나이: {age + 1}세")   # 계산도 가능!

연산자 모음

1-2. 자료구조

리스트 (List) — 순서 있고 변경 가능

medications = ["아스피린", "메트포르민", "리시노프릴"]

# 인덱싱 (하나 꺼내기)
medications[0]     # "아스피린" (첫 번째)
medications[-1]    # "리시노프릴" (마지막)

# 슬라이싱 (여러 개 꺼내기)
medications[0:2]   # ["아스피린", "메트포르민"] (0~1번)
medications[::-1]  # 전체 역순

# 요소 추가/삭제
medications.append("인슐린")        # 맨 뒤에 추가
medications.insert(1, "타이레놀")   # 1번 위치에 삽입
medications.remove("아스피린")      # 값으로 삭제
medications.pop(0)                 # 인덱스로 삭제 (삭제된 값 반환)

딕셔너리 (Dictionary) — 키-값 쌍

patient = {
    "name": "김환자",
    "age": 65,
    "diagnosis": "고혈압",
    "medications": ["아스피린", "메트포르민"]
}
patient["age"]                  # 65 (값 접근)
patient["phone"] = "010-1234"   # 새 키-값 추가
patient.get("email", "없음")    # 키 없으면 "없음" 반환 (에러 안남)

튜플 (Tuple) — 순서 있고 변경 불가

# 튜플 생성
vital_signs = ("혈압", "맥박", "체온", "호흡수")
vital_signs[0]    # "혈압"
len(vital_signs)  # 4

# 튜플 언패킹 — 여러 변수에 한번에 할당
name, age, diagnosis = ("김환자", 65, "고혈압")
print(f"{name}({age}세): {diagnosis}")

# 함수에서 여러 값 반환할 때 튜플 사용
def get_patient_summary(data):
    return data["name"], data["age"], data["bp"]  # 자동으로 튜플

name, age, bp = get_patient_summary({"name": "이환자", "age": 50, "bp": 130})

# * 연산자로 나머지 묶기
first, *rest = [1, 2, 3, 4, 5]
# first = 1, rest = [2, 3, 4, 5]

집합 (Set) — 중복 제거 & 집합 연산

# 중복 제거 (가장 많이 쓰는 용도!)
all_symptoms = ["발열", "기침", "두통", "발열", "기침"]
unique_symptoms = set(all_symptoms)  # {"발열", "기침", "두통"}
unique_list = list(unique_symptoms)  # 다시 리스트로

# 집합 연산 (벤다이어그램 떠올리기!)
doctors_morning = {"김의사", "이의사", "박의사", "최의사"}
doctors_afternoon = {"박의사", "최의사", "정의사", "한의사"}

# 합집합: 오전 OR 오후 근무 의사
all_doctors = doctors_morning | doctors_afternoon
# {"김의사", "이의사", "박의사", "최의사", "정의사", "한의사"}

# 교집합: 오전 AND 오후 모두 근무
both_shifts = doctors_morning & doctors_afternoon
# {"박의사", "최의사"}

# 차집합: 오전만 근무 (오후 제외)
morning_only = doctors_morning - doctors_afternoon
# {"김의사", "이의사"}

# 포함 여부 확인 (in 연산이 리스트보다 훨씬 빠름!)
if "김의사" in doctors_morning:
    print("오전 근무 확인")

문자열 메서드

# 대소문자 변환
"Hello World".lower()     # "hello world"
"Hello World".upper()     # "HELLO WORLD"

# 공백 제거 (입력값 정리에 필수!)
user_input = "  김환자  "
user_input.strip()        # "김환자"

# 문자열 교체
diagnosis = "Type 2 Diabetes"
diagnosis.replace("Type 2", "제2형")  # "제2형 Diabetes"

# 분할 & 합치기
"120/80".split("/")       # ["120", "80"]
"/".join(["120", "80"])   # "120/80"

# 시작/끝 확인
"ICD-10-A00".startswith("ICD")  # True
"report.pdf".endswith(".pdf")   # True

# 검색
"고혈압, 당뇨병, 고지혈증".find("당뇨병")  # 4 (시작 인덱스)
"고혈압, 당뇨병, 고지혈증".count(",")      # 2 (쉼표 개수)

딕셔너리 활용 — 리스트 of 딕셔너리 패턴

# 실무에서 가장 많이 쓰는 패턴: 환자 목록
patients_db = [
    {"id": "P001", "name": "김철수", "age": 45, "bp": 120, "diagnosis": "정상"},
    {"id": "P002", "name": "이영희", "age": 67, "bp": 155, "diagnosis": "고혈압"},
    {"id": "P003", "name": "박민수", "age": 32, "bp": 110, "diagnosis": "정상"},
    {"id": "P004", "name": "최지은", "age": 58, "bp": 165, "diagnosis": "고혈압"},
]

# 고혈압 환자만 필터링
high_bp = [p for p in patients_db if p["bp"] >= 140]
print(f"고혈압 환자: {len(high_bp)}명")

# 평균 나이 계산
avg_age = sum(p["age"] for p in patients_db) / len(patients_db)

# 딕셔너리 메서드
patient = patients_db[0]
patient.keys()       # dict_keys(['id', 'name', 'age', 'bp', 'diagnosis'])
patient.values()     # dict_values(['P001', '김철수', 45, 120, '정상'])
patient.items()      # dict_items([('id','P001'), ('name','김철수'), ...])

# 딕셔너리 컴프리헨션
bp_dict = {p["name"]: p["bp"] for p in patients_db}
# {"김철수": 120, "이영희": 155, "박민수": 110, "최지은": 165}

자료구조 비교표

1-3. 제어문 (조건문)

# if-elif-else: 하나만 실행
if bmi < 18.5:
    status = "저체중"
elif bmi < 25:
    status = "정상"
elif bmi < 30:
    status = "과체중"
else:
    status = "비만"

# 삼항 연산자 (한 줄 조건문)
fever = "발열" if temperature >= 37.5 else "정상"

# 포함 여부 확인
if "아스피린" in patient_allergies:
    print("알레르기 주의!")

1-4. 반복문

# for문: 횟수/대상이 명확할 때
for name in patient_names:
    print(f"{name}님, 진료실로 들어오세요!")

# enumerate: 인덱스와 값 함께
for i, patient in enumerate(patients):
    print(f"{i+1}번째: {patient}")

# zip: 여러 리스트 동시 순회
for name, age in zip(names, ages):
    print(f"{name}({age}세)")

# range: 숫자 범위 생성
for i in range(1, 6):   # 1, 2, 3, 4, 5
    print(i)

# while문: 조건이 참인 동안 반복
count = 0
while count < 5:
    count += 1

# break: 즉시 종료 / continue: 건너뛰기
for temp in temperatures:
    if temp >= 38.5:
        print("응급!")
        break

리스트 컴프리헨션 (한 줄로 리스트 만들기)

high_bp = [bp for bp in pressures if bp >= 140]              # 필터링
categories = ["성인" if age >= 18 else "미성년" for age in ages]  # 변환

1-5. 함수

# 함수 정의
def calculate_bmi(weight, height):
    """BMI 계산 함수"""
    return weight / (height ** 2)

result = calculate_bmi(70, 1.75)  # 함수 호출

# 기본값 매개변수
def greet(name, greeting="안녕하세요"):
    return f"{greeting}, {name}님!"

# 여러 값 반환 (튜플)
def analyze_bp(systolic, diastolic):
    category = "고혈압" if systolic >= 140 else "정상"
    risk = systolic - 120
    return category, risk

cat, risk = analyze_bp(145, 90)

# *args: 개수 불정 인자 (튜플로 받음)
def average(*values):
    return sum(values) / len(values)

average(120, 135, 128)         # 127.67
average(90, 88, 95, 100, 92)   # 93.0

# **kwargs: 키워드 인자 (딕셔너리로 받음)
def create_patient(**info):
    return info

patient = create_patient(name="김환자", age=65, bp=140)
# {"name": "김환자", "age": 65, "bp": 140}

# 람다 함수 (한 줄 함수)
bmi = lambda w, h: w / (h ** 2)

# 람다 + sorted (정렬 기준 지정)
patients.sort(key=lambda p: p["age"])        # 나이순 정렬
patients.sort(key=lambda p: p["bp"], reverse=True)  # 혈압 내림차순

# 람다 + filter (조건 필터링)
high_risk = list(filter(lambda p: p["bp"] >= 140, patients))

# 람다 + map (변환)
names = list(map(lambda p: p["name"], patients))

# sorted로 원본 유지하면서 정렬
by_age = sorted(patients, key=lambda p: p["age"])
by_bp = sorted(patients, key=lambda p: -p["bp"])  # 내림차순

1-6. 객체지향 프로그래밍 (OOP)

클래스 = 설계도 (붕어빵 틀), 객체 = 실체 (붕어빵)

class Patient:
    def __init__(self, name, age, diagnosis):
        self.name = name           # 인스턴스 변수
        self.age = age
        self.diagnosis = diagnosis

    def get_info(self):            # 메서드
        return f"{self.name}({self.age}세) - {self.diagnosis}"

patient1 = Patient("김환자", 65, "고혈압")  # 객체 생성
print(patient1.get_info())

# 상속: 부모 클래스의 기능을 물려받기
class EmergencyPatient(Patient):
    def __init__(self, name, age, diagnosis, level):
        super().__init__(name, age, diagnosis)  # 부모 생성자 호출
        self.level = level

    def get_info(self):  # 메서드 오버라이딩 (재정의)
        return f"[응급{self.level}] {super().get_info()}"

ep = EmergencyPatient("박환자", 40, "골절", 2)
print(ep.get_info())  # [응급2] 박환자(40세) - 골절

실전 OOP — 의료 스태프 상속 계층

class MedicalStaff:
    """의료진 기본 클래스"""
    def __init__(self, name, employee_id):
        self.name = name
        self.employee_id = employee_id

    def introduce(self):
        return f"{self.name} ({self.employee_id})"

class Doctor(MedicalStaff):
    def __init__(self, name, employee_id, specialty):
        super().__init__(name, employee_id)
        self.specialty = specialty
        self.patients = []

    def add_patient(self, patient_name):
        self.patients.append(patient_name)

    def introduce(self):
        return f"[의사] {super().introduce()} - {self.specialty}"

class Nurse(MedicalStaff):
    def __init__(self, name, employee_id, ward):
        super().__init__(name, employee_id)
        self.ward = ward

    def introduce(self):
        return f"[간호사] {super().introduce()} - {self.ward}병동"

# 사용
doc = Doctor("김닥터", "D001", "순환기내과")
doc.add_patient("김환자")
nurse = Nurse("이너스", "N001", "3")

for staff in [doc, nurse]:  # 다형성: 같은 메서드, 다른 동작
    print(staff.introduce())
# [의사] 김닥터 (D001) - 순환기내과
# [간호사] 이너스 (N001) - 3병동

1-7. 파일처리 & 예외처리

# 파일 읽기/쓰기 (with문 → 자동으로 닫아줌)
with open("data.txt", "w", encoding="utf-8") as f:
    f.write("내용 쓰기")

with open("data.txt", "r", encoding="utf-8") as f:
    content = f.read()         # 전체 읽기

# CSV 파일 처리
import csv
with open("patients.csv", "r", encoding="utf-8") as f:
    reader = csv.DictReader(f)
    for row in reader:
        print(row["name"])

# JSON 파일 처리
import json
with open("data.json", "r", encoding="utf-8") as f:
    data = json.load(f)       # JSON → 딕셔너리

# 예외처리 (에러가 나도 프로그램 안 죽게)
try:
    result = 10 / 0
except ZeroDivisionError:
    print("0으로 나눌 수 없습니다")
finally:
    print("항상 실행됨")

1-8. 실전 종합 예제 — 혈압 분류 & 응급 분류

예제 1: 환자 혈압 통계 & 분류

# 환자 데이터 (리스트 of 딕셔너리)
patients = [
    {"name": "김철수", "age": 45, "systolic": 120, "diastolic": 80},
    {"name": "이영희", "age": 67, "systolic": 155, "diastolic": 95},
    {"name": "박민수", "age": 32, "systolic": 110, "diastolic": 70},
    {"name": "최지은", "age": 58, "systolic": 165, "diastolic": 100},
    {"name": "정도영", "age": 71, "systolic": 145, "diastolic": 88},
]

# 혈압 분류 함수
def classify_bp(systolic, diastolic):
    if systolic < 120 and diastolic < 80:
        return "정상"
    elif systolic < 130 and diastolic < 80:
        return "주의"
    elif systolic < 140 or diastolic < 90:
        return "고혈압 전단계"
    elif systolic < 160 or diastolic < 100:
        return "고혈압 1기"
    else:
        return "고혈압 2기"

# 모든 환자 분류
for p in patients:
    grade = classify_bp(p["systolic"], p["diastolic"])
    print(f"{p['name']}({p['age']}세): {p['systolic']}/{p['diastolic']} → {grade}")

# 통계 계산
bp_values = [p["systolic"] for p in patients]
avg_bp = sum(bp_values) / len(bp_values)
max_bp = max(bp_values)
min_bp = min(bp_values)
high_count = len([bp for bp in bp_values if bp >= 140])

print(f"\n--- 혈압 통계 ---")
print(f"평균: {avg_bp:.1f}mmHg")
print(f"최고: {max_bp}mmHg / 최저: {min_bp}mmHg")
print(f"고혈압(≥140) 환자: {high_count}명 / 전체 {len(patients)}명")

예제 2: 응급실 환자 분류 (Triage)

# 응급도 분류 시스템 (KTAS 기반 간소화)
def triage(temperature, heart_rate, consciousness):
    """
    응급도 분류:
    Level 1 (즉시): 의식 없음
    Level 2 (긴급): 고열 + 빈맥
    Level 3 (응급): 고열 또는 빈맥
    Level 4 (준응급): 미열
    Level 5 (비응급): 정상 범위
    """
    if consciousness == "무반응":
        return 1, "즉시 (Resuscitation)"
    elif temperature >= 39.0 and heart_rate >= 120:
        return 2, "긴급 (Emergency)"
    elif temperature >= 38.5 or heart_rate >= 110:
        return 3, "응급 (Urgent)"
    elif temperature >= 37.5:
        return 4, "준응급 (Less Urgent)"
    else:
        return 5, "비응급 (Non-Urgent)"

# 응급실 환자 데이터
er_patients = [
    {"name": "환자A", "temp": 36.5, "hr": 75,  "conscious": "명료"},
    {"name": "환자B", "temp": 39.5, "hr": 130, "conscious": "명료"},
    {"name": "환자C", "temp": 38.8, "hr": 95,  "conscious": "명료"},
    {"name": "환자D", "temp": 37.0, "hr": 80,  "conscious": "무반응"},
]

# 분류 및 정렬
results = []
for p in er_patients:
    level, desc = triage(p["temp"], p["hr"], p["conscious"])
    results.append({**p, "level": level, "desc": desc})

# 응급도순 정렬 (Level 1이 가장 긴급)
results.sort(key=lambda x: x["level"])

for r in results:
    print(f"Level {r['level']} {r['desc']}: {r['name']} "
          f"(체온 {r['temp']}°C, 맥박 {r['hr']}bpm)")

2-1. 스칼라, 벡터, 행렬

데이터를 표현하는 3가지 방식입니다. 쉽게 말하면:

# 스칼라 (Scalar) - 숫자 하나
patient_age = 45
patient_weight = 70.5

# 벡터 (Vector) - 한 환자의 여러 정보
patient_vector = np.array([45, 70.5, 175, 120, 80])
# → shape: (5,), ndim: 1

# 행렬 (Matrix) - 여러 환자 × 여러 특성
patients_matrix = np.array([
    [45, 70.5, 175, 120, 80],   # 환자1
    [32, 55.0, 162, 110, 75],   # 환자2
    [67, 82.3, 168, 140, 90],   # 환자3
    [28, 48.7, 155, 105, 70],   # 환자4
])
# → shape: (4, 5), ndim: 2, size: 20

2-2. NumPy 기초

파이썬 리스트보다 수천 배 빠른 수학 연산 라이브러리입니다. Pandas, Scikit-learn의 내부 연산 엔진이에요.

배열 생성 방법

import numpy as np

# 리스트 → 배열
arr = np.array([45, 70.5, 175])

# 특수 배열 생성
zeros = np.zeros((10, 5))        # 10행 5열 영행렬 (전부 0)
ones = np.ones(100) * 36.5       # 100명 정상체온 (전부 36.5)
nums = np.arange(1, 101)         # 1~100 연속 숫자 (환자 ID)

# dtype 지정
arr_int = np.array([45, 70.5, 175], dtype=int)     # 정수로 강제
arr_float = np.array([45, 70.5, 175], dtype=float)  # 실수로 강제

배열 속성 확인

patient = np.array([45, 70.5, 175, 120, 80])
print(f"shape: {patient.shape}")   # (5,) → 1차원, 원소 5개
print(f"ndim:  {patient.ndim}")    # 1 → 차원 수
print(f"dtype: {patient.dtype}")   # float64
print(f"size:  {patient.size}")    # 5 → 전체 원소 개수

인덱싱 (하나 꺼내기)

patients = np.array([
    [45, 70.5, 175, 120, 80],   # 환자1 (김씨)
    [32, 55.0, 162, 110, 75],   # 환자2 (이씨)
    [67, 82.3, 168, 140, 90],   # 환자3 (박씨)
    [28, 48.7, 155, 105, 70],   # 환자4 (최씨)
])

# 1D 인덱싱
patient_vector = np.array([45, 70.5, 175, 120, 80])
patient_vector[0]     # 45.0 (첫 번째)
patient_vector[-1]    # 80.0 (마지막)

# 2D 인덱싱
patients[0]           # [45, 70.5, 175, 120, 80] → 환자1 전체
patients[-1]          # [28, 48.7, 155, 105, 70] → 마지막 환자
patients[0, 0]        # 45.0 → 환자1의 나이
patients[-1, 1]       # 48.7 → 마지막 환자의 체중

슬라이싱 (여러 개 꺼내기)

# 모든 환자의 나이 (첫 번째 열)
all_ages = patients[:, 0]          # [45, 32, 67, 28]

# 모든 환자의 체중 (두 번째 열)
all_weights = patients[:, 1]       # [70.5, 55, 82.3, 48.7]

# 앞쪽 3명 환자만
first_three = patients[:3, :]

# 혈압 데이터만 (마지막 2열)
blood_pressure = patients[:, -2:]
# [[120, 80], [110, 75], [140, 90], [105, 70]]

조건부 인덱싱 (Boolean Indexing) ⭐

조건에 맞는 데이터만 골라내는 강력한 기능입니다.

# 수축기 혈압 ≥ 130인 환자 찾기
systolic = patients[:, 3]           # [120, 110, 140, 105]
mask = systolic >= 130              # [False, False, True, False]
high_bp = patients[mask]            # [[67, 82.3, 168, 140, 90]]

# 40세 이상 환자
ages = patients[:, 0]
elderly = patients[ages >= 40]      # 환자1(45세), 환자3(67세)

# 체중 70kg 이상 환자
weights = patients[:, 1]
heavy = patients[weights >= 70]     # 환자1(70.5kg), 환자3(82.3kg)

벡터화 연산 (for문이 필요 없다!)

NumPy는 배열 전체를 한번에 계산합니다. for문보다 훨씬 빠릅니다.

# cm → m 변환 (전체 한번에!)
heights_cm = patients[:, 2]       # [175, 162, 168, 155]
heights_m = heights_cm / 100      # [1.75, 1.62, 1.68, 1.55]

# 모든 환자 내년 나이
current_ages = patients[:, 0]     # [45, 32, 67, 28]
next_year = current_ages + 1      # [46, 33, 68, 29]

# kg → lb 변환 (1kg = 2.205lb)
weights_kg = patients[:, 1]
weights_lb = weights_kg * 2.205   # [155.5, 121.3, 181.5, 107.4]

# 속도 비교 실험
import time, random

heights_list = [random.uniform(150, 200) for _ in range(10000)]
heights_numpy = np.array(heights_list)

# Python for문 방식
start = time.time()
result_list = [h / 100 for h in heights_list]
print(f"Python: {time.time()-start:.6f}초")

# NumPy 벡터화 방식
start = time.time()
result_numpy = heights_numpy / 100
print(f"NumPy:  {time.time()-start:.6f}초")  # 훨씬 빠름!

2-3. 기초 통계 함수

# 10명 환자의 혈압 데이터
blood_pressure = np.array([120, 135, 118, 142, 128, 156, 110, 139, 125, 148])

# 기본 통계
print(f"최솟값: {np.min(blood_pressure)}")     # 110
print(f"최댓값: {np.max(blood_pressure)}")     # 156
print(f"평  균: {np.mean(blood_pressure):.1f}") # 132.1
print(f"표준편차: {np.std(blood_pressure):.1f}") # 13.7

# argmin/argmax: "몇 번째" 환자인지 찾기
patient_ids = np.array(['P001','P002','P003','P004','P005',
                        'P006','P007','P008','P009','P010'])

min_idx = np.argmin(blood_pressure)   # 6 (인덱스)
max_idx = np.argmax(blood_pressure)   # 5

print(f"최저 혈압 환자: {patient_ids[min_idx]} ({blood_pressure[min_idx]})")
# → 최저 혈압 환자: P007 (110)
print(f"최고 혈압 환자: {patient_ids[max_idx]} ({blood_pressure[max_idx]})")
# → 최고 혈압 환자: P006 (156)

축(axis) 개념 — 2차원 통계의 핵심 ⭐

# 5명 환자, 3개 지표: [나이, 수축기BP, 혈당]
patients_data = np.array([
    [45, 120, 95],    # 환자1
    [67, 156, 180],   # 환자2
    [32, 110, 88],    # 환자3
    [58, 142, 102],   # 환자4
    [29, 118, 92]     # 환자5
])

# axis=0 → 지표별(열별) 통계
np.min(patients_data, axis=0)    # [29, 110, 88]  ← 각 지표의 최솟값
np.max(patients_data, axis=0)    # [67, 156, 180] ← 각 지표의 최댓값
np.mean(patients_data, axis=0)   # [46.2, 129.2, 111.4]

# axis=1 → 환자별(행별) 통계
np.mean(patients_data, axis=1)
# [86.7, 134.3, 76.7, 100.7, 79.7] ← 각 환자의 평균 수치

2-4. 내적(Dot Product)과 행렬곱

AI 예측의 핵심 연산: 환자 데이터 × 가중치 = 예측값

# 내적 계산 3가지 방법 (모두 같은 결과)
a = np.array([1, 2, 3])
b = np.array([4, 5, 6])

np.dot(a, b)     # 32
a @ b            # 32 (@ 연산자, 권장)
a.dot(b)         # 32

실전: 당뇨 위험도 예측

# 환자 정보: [나이, 체중, BMI]
patient_info = np.array([45, 85, 25])

# AI가 학습한 가중치 (각 정보의 중요도)
weights = np.array([0.02, 0.015, 0.08])

# 위험 점수 = 내적
risk_score = np.dot(patient_info, weights)
# = 45×0.02 + 85×0.015 + 25×0.08
# = 0.9 + 1.275 + 2.0 = 4.175

# 위험 분류
if risk_score < 2.0:    level = "낮음"
elif risk_score < 4.0:  level = "보통"
elif risk_score < 6.0:  level = "높음"
else:                    level = "매우 높음"
print(f"위험 점수: {risk_score:.2f} → {level}")  # 높음

행렬곱: 여러 환자를 동시에 예측

# 3명 환자 데이터 (행렬)
patients_data = np.array([
    [45, 85, 25],   # 환자1
    [32, 60, 22],   # 환자2
    [67, 90, 28],   # 환자3
])
weights = np.array([0.02, 0.015, 0.08])

# 행렬곱 → 3명 동시 계산!
all_risks = np.dot(patients_data, weights)
# [4.175, 3.3, 4.93]

# 가장 위험한 환자 찾기
most_risky = np.argmax(all_risks)  # 2 (환자3)
print(f"최고 위험: 환자{most_risky+1} ({all_risks[most_risky]:.2f}점)")

행렬 전치 (Transpose)

# 원본: (2행, 3열)
data = np.array([[45, 70, 25],
                 [32, 55, 22]])

# 전치: (3행, 2열) - 행↔열 뒤집기
data.T
# [[45, 32],
#  [70, 55],
#  [25, 22]]

# 행렬곱에서 차원 맞출 때 필요!

2-5. 가중합과 선형방정식

AI가 예측하는 기본 공식입니다.

# 실전: 칼로리 소모 예측
activity = np.array([45, 8, 30, 2, 6])
# [나이, 수면시간, 운동시간, 계단횟수, 걷기시간]

weights = np.array([0.8, 1.5, 0.6, 2.0, 0.3])
basal_rate = 1400  # 편향 (기초대사율)

# 가중합 + 편향
total_calories = np.dot(activity, weights) + basal_rate
# = (45×0.8 + 8×1.5 + 30×0.6 + 2×2.0 + 6×0.3) + 1400
# = 71.8 + 1400 = 1471.8 kcal

실전: 맞춤형 약물 투여량 계산

# 3명 환자: [체중, 나이]
patients_info = np.array([
    [70, 45],   # 환자1
    [55, 32],   # 환자2
    [85, 60]    # 환자3
])
dosage_weights = np.array([0.5, 0.1])  # [체중계수, 나이계수]
base_dosage = 10  # 기본 투여량 (편향)

# 모든 환자 투여량 한번에 계산
dosages = np.dot(patients_info, dosage_weights) + base_dosage
# 환자1: 70×0.5 + 45×0.1 + 10 = 49.5mg
# 환자2: 55×0.5 + 32×0.1 + 10 = 40.7mg
# 환자3: 85×0.5 + 60×0.1 + 10 = 58.5mg

AI 학습이란? = 최적의 가중치 찾기

# 정답이 45mg인 환자 (체중65, 나이50)
actual_patient = np.array([65, 50])
correct_dosage = 45
base = 10

# AI가 여러 가중치를 시도 (학습)
trials = [
    np.array([0.3, 0.2]),   # 시도1 → 39.5mg (오차 5.5)
    np.array([0.5, 0.1]),   # 시도2 → 47.5mg (오차 2.5) ✅ 가장 좋음!
    np.array([0.6, 0.05]),  # 시도3 → 51.5mg (오차 6.5)
]

for i, w in enumerate(trials):
    pred = np.dot(actual_patient, w) + base
    error = abs(correct_dosage - pred)
    print(f"시도{i+1}: 예측={pred:.1f}mg, 오차={error:.1f}")

2-6. 데이터 타입 분류

데이터 타입별 분석 가능 여부

데이터 타입별 시각화 방법

3-1. Pandas 기초

엑셀처럼 표 형태 데이터를 다루는 파이썬 라이브러리입니다. Series(1열)와 DataFrame(표)이 핵심 구조입니다.

DataFrame 생성 & 기본 확인

import pandas as pd

# 딕셔너리로 DataFrame 직접 생성
df = pd.DataFrame({
    "이름": ["김환자", "이환자", "박환자", "최환자", "정환자"],
    "나이": [45, 32, 67, 28, 55],
    "체중": [70.5, 55.0, 82.3, 48.7, 65.0],
    "혈압": [120, 110, 145, 105, 138],
    "당뇨": ["정상", "정상", "당뇨", "정상", "당뇨"]
})

# CSV 파일에서 로드
df = pd.read_csv("patients.csv")
df = pd.read_csv("patients.csv", encoding="cp949")  # 한글 인코딩

# 기본 정보 확인 (EDA의 첫 단계!)
df.shape           # (행수, 열수) → (5, 5)
df.head()          # 상위 5행 미리보기
df.tail(3)         # 하위 3행
df.info()          # 컬럼명, 타입, 결측치 수 요약
df.describe()      # 수치형 기술통계 (count, mean, std, min, 25%, 50%, 75%, max)
df.dtypes          # 각 컬럼의 데이터 타입

컬럼 & 행 선택

# 컬럼 선택
df["나이"]                       # Series (1개 컬럼)
df[["나이", "혈압"]]             # DataFrame (여러 컬럼)

# 행 선택: loc(라벨 기반) vs iloc(위치 기반)
df.loc[0]                        # 인덱스 이름이 0인 행
df.loc[0:2]                      # 인덱스 0, 1, 2 (끝 포함!)
df.iloc[0:3]                     # 0, 1, 2번째 행 (끝 미포함!)
df.loc[0, "나이"]                # 특정 행의 특정 컬럼 값

# 조건부 필터링 ⭐ (가장 많이 사용!)
high_bp = df[df["혈압"] >= 140]
young_diabetic = df[(df["나이"] < 40) & (df["당뇨"] == "당뇨")]  # AND
risk = df[(df["혈압"] >= 140) | (df["나이"] >= 65)]               # OR

# isin / query (복합 조건을 깔끔하게)
df[df["당뇨"].isin(["당뇨", "전단계"])]
df.query("나이 >= 40 and 혈압 >= 130")

컬럼 추가 & 수정

# 새 컬럼 만들기
df["BMI"] = df["체중"] / (df["키"]/100)**2

# 조건부 컬럼 생성
df.loc[df["혈압"] >= 140, "위험도"] = "고위험"
df.loc[df["혈압"] < 140, "위험도"] = "정상"

# apply: 복잡한 함수 적용
def classify_age(age):
    if age < 30: return "청년"
    elif age < 60: return "중년"
    else: return "노년"

df["연령대"] = df["나이"].apply(classify_age)

# 빈도 분석
df["당뇨"].value_counts()                  # 절대빈도 (개수)
df["당뇨"].value_counts(normalize=True)     # 상대빈도 (비율 0~1)

데이터 정렬 & 합치기

# 정렬
df.sort_values("나이", ascending=False)           # 나이 내림차순
df.sort_values(["당뇨", "나이"], ascending=[True, False])  # 복합 정렬

# 합치기
pd.concat([df1, df2])                              # 행 기준 합치기
pd.merge(df1, df2, on="환자ID", how="left")       # 열 기준 (SQL JOIN)

3-2. 탐색적 데이터 분석 (EDA)

데이터를 분석하기 전에 데이터의 특성을 파악하는 과정입니다. 시각화 + 통계를 같이 봐야 합니다.

중심경향성 & 산포도

df["BMI"].mean()        # 평균 (이상치에 민감!)
df["BMI"].median()      # 중앙값 (이상치에 강함, 권장)
df["BMI"].mode()[0]     # 최빈값 (가장 많이 나온 값)
df["BMI"].std()         # 표준편차 (값들이 평균에서 얼마나 퍼져있는지)
df["BMI"].var()         # 분산 (표준편차의 제곱)

# 변동계수(CV) = 표준편차/평균 × 100
cv = (df["BMI"].std() / df["BMI"].mean()) * 100
# CV < 10%: 매우 안정 / < 30%: 보통 / ≥ 30%: 불안정

# 사분위수 & IQR
Q1 = df["BMI"].quantile(0.25)    # 25번째 백분위
Q3 = df["BMI"].quantile(0.75)    # 75번째 백분위
IQR = Q3 - Q1                   # 사분위간 범위

GroupBy (그룹별 분석) ⭐

# 당뇨 여부별 BMI 평균
df.groupby("당뇨")["BMI"].mean()
# 당뇨    28.5
# 정상    23.2

# 여러 통계량 한번에
df.groupby("당뇨")["BMI"].agg(["count", "mean", "median", "std"])

# 여러 컬럼에 대해 그룹별 통계
df.groupby("당뇨")[["BMI", "혈압", "혈당"]].mean()

# 피벗 테이블 (더 유연한 GroupBy)
pd.pivot_table(df, values="BMI", index="성별", columns="당뇨", aggfunc="mean")

GroupBy 심화 — 람다 & 다중 집계

# 람다 함수로 비율 계산
obesity_rate = df.groupby("당뇨")["BMI"].apply(
    lambda x: (x >= 30).mean() * 100
)
# 당뇨    62.3%  (당뇨 환자 중 비만 비율)
# 정상    28.1%

# 그룹별 결측치 대체 (매우 유용!)
df["중성지방"] = df["중성지방"].fillna(
    df.groupby("연령대")["중성지방"].transform("mean")
)

# 교차표 (Cross Tabulation)
crosstab = pd.crosstab(df["성별"], df["당뇨"], margins=True)
crosstab_pct = pd.crosstab(df["성별"], df["당뇨"], normalize="index") * 100
print(crosstab_pct.round(1))
#         당뇨    정상
# 남      18.5   81.5
# 여      14.2   85.8

피벗 테이블 (Pivot Table)

# 행: 성별, 열: 당뇨, 값: BMI 평균
pivot = pd.pivot_table(df, values="BMI", index="성별", columns="당뇨", aggfunc="mean")

# 여러 집계함수
pivot = pd.pivot_table(df, values="BMI",
                       index="연령대", columns="당뇨",
                       aggfunc=["mean", "count"])

피처 엔지니어링 (새 변수 만들기)

# pd.cut: 연속형 → 범주형 (구간 분할)
bins = [0, 18.5, 25, 30, np.inf]
labels = ["저체중", "정상", "과체중", "비만"]
df["BMI그룹"] = pd.cut(df["BMI"], bins=bins, labels=labels)

# apply로 복잡한 분류 함수 적용
def classify_risk(row):
    if row["혈압"] >= 160 or row["혈당"] >= 200:
        return "고위험"
    elif row["혈압"] >= 140 or row["혈당"] >= 126:
        return "위험"
    else:
        return "정상"

df["위험등급"] = df.apply(classify_risk, axis=1)  # axis=1 → 행 단위 적용

# 로그 변환 (왜곡된 분포 정규화)
df["BMI_log"] = np.log1p(df["BMI"])  # log(1+x), 0도 안전하게 처리

# 날짜에서 파생 변수 추출
df["Date"] = pd.to_datetime(df["Date"])
df["연도"] = df["Date"].dt.year
df["월"] = df["Date"].dt.month
df["요일"] = df["Date"].dt.day_name()

상관관계 분석

# 상관관계 행렬 계산
medical_vars = ["Diabetes", "BMI", "HighBP", "Stroke", "PhysActivity", "Age"]
corr = df[medical_vars].corr()

# 타겟 변수와의 상관계수 정렬
target_corr = corr["Diabetes"].drop("Diabetes").abs().sort_values(ascending=False)
print(target_corr)

# 상관관계 해석 자동화
for var, val in target_corr.items():
    original = corr["Diabetes"][var]
    direction = "양의 상관" if original > 0 else "음의 상관"
    strength = "강한" if val > 0.3 else "보통" if val > 0.1 else "약한"
    print(f"{var}: {original:.3f} ({strength} {direction})")

# |r| ≥ 0.7 → 강한 상관 / 0.3~0.7 → 중간 / < 0.3 → 약한 상관

3-3. 데이터 시각화

데이터를 그래프로 표현하여 패턴, 이상치, 분포를 한눈에 파악합니다.

기본 설정 (한글 폰트)

import matplotlib.pyplot as plt
import seaborn as sns

# 한글 폰트 설정 (반드시!)
plt.rc("font", family="NanumBarunGothic")          # 또는 "Malgun Gothic" (Windows)
plt.rcParams["axes.unicode_minus"] = False          # 마이너스 기호 깨짐 방지
sns.set_style("whitegrid")                          # 배경 스타일

Matplotlib 기본 구조 (Figure & Axes)

# 방법1: plt 직접 사용 (간단할 때)
plt.figure(figsize=(8, 5))
plt.plot([1,2,3,4], [10,20,25,30])
plt.xlabel("시간")
plt.ylabel("혈당")
plt.title("혈당 변화 추이")
plt.grid(True)
plt.show()

# 방법2: Figure + Axes (여러 그래프, 권장)
fig, axes = plt.subplots(1, 2, figsize=(12, 5))
axes[0].hist(df["BMI"], bins=20, color="skyblue", edgecolor="black")
axes[0].set_title("BMI 분포")
axes[1].scatter(df["BMI"], df["혈압"], alpha=0.6)
axes[1].set_title("BMI vs 혈압")
plt.tight_layout()
plt.show()

Seaborn 주요 그래프 ⭐

Matplotlib 실전 — 꺾은선 & 막대 그래프

# 꺾은선 그래프: 시간에 따른 혈압 변화
days = [1, 2, 3, 4, 5, 6, 7]
blood_pressure = [140, 138, 135, 132, 130, 128, 125]

plt.figure(figsize=(8, 5))
plt.plot(days, blood_pressure,
         marker="o", color="red", linewidth=2, markersize=8)
plt.title("환자 혈압 변화 추이", fontsize=14)
plt.xlabel("치료 일수", fontsize=12)
plt.ylabel("수축기 혈압 (mmHg)", fontsize=12)
plt.axhline(y=130, color="green", linestyle="--", label="목표 혈압")
plt.legend()
plt.grid(True, alpha=0.5)
plt.show()

# 막대 그래프: 진료과별 환자 수
departments = ["내과", "외과", "소아과", "산부인과"]
patient_counts = [45, 32, 28, 38]

plt.figure(figsize=(8, 5))
bars = plt.bar(departments, patient_counts,
               color=["skyblue", "lightgreen", "salmon", "gold"], alpha=0.8)
plt.ylabel("환자 수 (명)")
plt.title("진료과별 환자 수")

# 막대 위에 숫자 표시
for bar in bars:
    plt.text(bar.get_x() + bar.get_width()/2, bar.get_height() + 0.5,
             str(int(bar.get_height())), ha="center", fontweight="bold")
plt.show()

Seaborn 실전 — 그룹 비교 시각화

# 히스토그램: 두 그룹 BMI 분포 비교
plt.figure(figsize=(8, 5))
sns.histplot(data=df, x="BMI", hue="당뇨",
             bins=30, alpha=0.7, palette="Reds", kde=True)
plt.title("당뇨병 여부별 BMI 분포")
plt.xlabel("BMI (체질량지수)")
plt.show()

# 박스플롯: 두 변수 나란히 비교
fig, axes = plt.subplots(1, 2, figsize=(12, 5))

sns.boxplot(data=df, x="당뇨", y="BMI",
            hue="당뇨", palette="coolwarm", legend=False, ax=axes[0])
axes[0].set_title("당뇨별 BMI 분포")

sns.boxplot(data=df, x="당뇨", y="혈압",
            hue="당뇨", palette="pastel", legend=False, ax=axes[1])
axes[1].set_title("당뇨별 혈압 분포")

plt.tight_layout()
plt.show()

# 바이올린 플롯: 분포 형태까지 보여줌
sns.violinplot(data=df, x="당뇨", y="BMI", palette="Set2")
plt.title("바이올린 플롯 — 박스플롯 + 밀도 곡선")
plt.show()

hue — 범주 분리의 핵심 ⭐

hue로 범주형 변수를 색상으로 분리하면 그룹 간 차이를 한눈에 볼 수 있습니다.

# hue: 색상으로 범주 분리 (가장 중요한 옵션!)
sns.scatterplot(data=df, x="BMI", y="혈압", hue="당뇨")      # 색상으로 구분
sns.scatterplot(data=df, x="BMI", y="혈압", hue="당뇨",
                style="성별", size="나이")                      # 색+모양+크기

# 히스토그램에 KDE(밀도곡선) 겹치기
sns.histplot(data=df, x="BMI", hue="당뇨", kde=True, alpha=0.5)

박스플롯 읽는 법 ⭐

히트맵 (상관관계 시각화) ⭐

# 상관관계 히트맵 (실전 완전판)
plt.figure(figsize=(10, 8))
medical_vars = ["Diabetes", "BMI", "HighBP", "Stroke",
                "HeartDisease", "PhysActivity", "Smoker", "Age"]
corr = df[medical_vars].corr()

# 삼각형 마스크 (중복 제거)
mask = np.triu(np.ones_like(corr, dtype=bool))

sns.heatmap(corr,
    mask=mask,            # 상삼각 숨김 (깔끔!)
    annot=True,           # 숫자 표시
    fmt=".2f",            # 소수점 2자리
    cmap="RdBu_r",        # 빨-파 색상 (양-음 직관적)
    center=0,             # 0 기준 색상
    square=True,          # 정사각형 셀
    linewidths=0.5,       # 셀 경계선
    cbar_kws={"label": "상관계수", "shrink": 0.8}
)
plt.title("의료 변수 간 상관관계", fontsize=16, pad=20)
plt.xticks(rotation=45, ha="right")
plt.tight_layout()
plt.show()

# 타겟과 상관계수 추출 & 정렬
diabetes_corr = corr["Diabetes"].drop("Diabetes").abs()
print(diabetes_corr.sort_values(ascending=False))

서브플롯 (여러 그래프 한번에)

# 방법1: plt.subplot (간단)
plt.figure(figsize=(15, 5))
plt.subplot(1, 3, 1)                                  # 1행 3열 중 1번
sns.histplot(data=df, x="BMI", kde=True)
plt.title("BMI 분포")

plt.subplot(1, 3, 2)                                  # 1행 3열 중 2번
sns.boxplot(data=df, x="당뇨", y="BMI")
plt.title("당뇨별 BMI")

plt.subplot(1, 3, 3)                                  # 1행 3열 중 3번
sns.scatterplot(data=df, x="BMI", y="혈압", hue="당뇨")
plt.title("BMI vs 혈압")

plt.tight_layout()
plt.show()

# 방법2: fig, axes (더 유연, 권장)
fig, axes = plt.subplots(2, 2, figsize=(12, 10))
sns.histplot(data=df, x="BMI", ax=axes[0,0])
sns.boxplot(data=df, x="당뇨", y="혈압", ax=axes[0,1])
sns.heatmap(corr, annot=True, ax=axes[1,0])
sns.countplot(data=df, x="당뇨", ax=axes[1,1])
plt.suptitle("EDA 종합 분석", fontsize=16, y=1.02)
plt.tight_layout()
plt.show()

3-4. 데이터 전처리

1. 결측치 처리

# 결측치 확인
df.isnull().sum()                          # 컬럼별 결측치 개수
df.isnull().sum() / len(df) * 100          # 결측 비율 (%)

# 시각화로 결측 패턴 확인
import missingno as msno
msno.matrix(df)                            # 결측 패턴 시각화

# 삭제 방식
df.dropna()                                # 결측 있는 행 전부 삭제
df.dropna(subset=["Age", "BMI"])           # 특정 컬럼 기준만

# 대체 방식 (권장)
df["Age"].fillna(df["Age"].mean())         # 평균으로 대체
df["Age"].fillna(df["Age"].median())       # 중앙값 대체 (이상치 있을 때)
df["BP"].ffill()                           # 앞의 값으로 채우기 (시계열)
df["성별"].fillna(df["성별"].mode()[0])      # 최빈값 대체 (범주형)

# 그룹별 결측치 대체 (가장 정확한 방법!)
# → 연령대별 평균으로 BMI 결측치 대체
df["BMI"] = df["BMI"].fillna(
    df.groupby("연령대")["BMI"].transform("mean")
)

# 안전한 숫자 변환 (오류값 → NaN)
df["BMI"] = pd.to_numeric(df["BMI"], errors="coerce")

# 의료 데이터 유효성 검증
medical_valid = df[
    (df["BMI"] >= 10) & (df["BMI"] <= 60) &       # BMI 합리적 범위
    (df["나이"] >= 0) & (df["나이"] <= 120) &      # 나이 합리적 범위
    (df["혈압"] >= 50) & (df["혈압"] <= 300)        # 혈압 합리적 범위
]
print(f"유효 데이터: {len(medical_valid)}/{len(df)}건")

2. 중복 제거

df.duplicated().sum()                                       # 전체 중복 행 수
df.duplicated(subset=["PatientID"]).sum()                    # 특정 컬럼 기준
df.drop_duplicates(subset=["PatientID"], keep="last")        # 마지막 것만 유지
df.drop_duplicates(keep="first", inplace=True)               # 첫번째 유지, 원본 수정

3. 이상치 탐지 & 처리

# 방법1: IQR (가장 많이 사용)
Q1 = df["BMI"].quantile(0.25)
Q3 = df["BMI"].quantile(0.75)
IQR = Q3 - Q1
lower = Q1 - 1.5 * IQR     # 하한선
upper = Q3 + 1.5 * IQR     # 상한선

# 이상치 확인
outliers = df[(df["BMI"] < lower) | (df["BMI"] > upper)]
print(f"이상치 {len(outliers)}건 발견")

# 이상치 제거 (.copy()로 원본 보호!)
df_clean = df[(df["BMI"] >= lower) & (df["BMI"] <= upper)].copy()

# 방법2: Z-score
from scipy import stats
z_scores = stats.zscore(df["BMI"])
df_clean = df[abs(z_scores) < 3].copy()    # z-score 3 초과 제거

# 시각화로 이상치 확인
sns.boxplot(data=df, y="BMI")              # 점으로 이상치 표시됨

4. 타입 변환

df["Age"] = df["Age"].astype(int)                         # 강제 변환
df["BMI"] = pd.to_numeric(df["BMI"], errors="coerce")     # 안전 변환 (실패→NaN)
df["Date"] = pd.to_datetime(df["Date"])                   # 날짜 변환
df["성별"] = df["성별"].astype("category")                  # 범주형 변환 (메모리 절약)

5. 스케일링 & 인코딩 (ML 전 필수)

from sklearn.preprocessing import StandardScaler, MinMaxScaler, LabelEncoder
from sklearn.preprocessing import OneHotEncoder

# 스케일링 (수치형)
scaler = StandardScaler()
df[["BMI", "혈압"]] = scaler.fit_transform(df[["BMI", "혈압"]])

# 인코딩 (범주형 → 숫자)
le = LabelEncoder()
df["성별_encoded"] = le.fit_transform(df["성별"])           # M=1, F=0

# 원-핫 인코딩
df_encoded = pd.get_dummies(df, columns=["혈액형"], drop_first=True)
# 혈액형_B, 혈액형_O, 혈액형_AB 컬럼 생성

EDA 퀵 레퍼런스

4-1. MySQL + PyMySQL

SQL은 관계형 데이터베이스(엑셀 표처럼 행·열로 구성)를 다루는 언어입니다.

import pymysql

# DB 연결
conn = pymysql.connect(
    host="localhost", user="root", password="test1234",
    db="health_ai", charset="utf8mb4",
    cursorclass=pymysql.cursors.DictCursor   # 결과를 딕셔너리로
)

# CRUD 기본
with conn.cursor() as cur:
    # CREATE
    cur.execute("""CREATE TABLE patient_info (
        id INT AUTO_INCREMENT PRIMARY KEY,
        name VARCHAR(50) NOT NULL,
        gender CHAR(1),
        birth_date DATE
    )""")

    # INSERT (%s로 파라미터 바인딩 → SQL Injection 방지!)
    cur.execute("INSERT INTO patient_info (name, gender) VALUES (%s, %s)",
                ("김환자", "M"))

    # SELECT
    cur.execute("SELECT * FROM patient_info WHERE gender=%s", ("M",))
    results = cur.fetchall()     # 전체 결과 리스트

    # UPDATE / DELETE
    cur.execute("UPDATE patient_info SET gender=%s WHERE id=%s", ("F", 1))
    cur.execute("DELETE FROM patient_info WHERE id=%s", (1,))

    conn.commit()   # 반드시 커밋!

SQL 주요 문법

# WHERE: 필터링
SELECT * FROM patient WHERE age >= 60 AND gender = 'M';

# GROUP BY + 집계 함수
SELECT department, COUNT(*) AS cnt, AVG(bmi) AS avg_bmi
FROM patient
GROUP BY department
HAVING COUNT(*) >= 10    -- GROUP BY 후 필터링
ORDER BY avg_bmi DESC;

# 연령대별 평균 BMI
SELECT FLOOR(age/10)*10 AS age_group,
       COUNT(*) AS cnt,
       ROUND(AVG(bmi), 2) AS avg_bmi
FROM patient
GROUP BY age_group
ORDER BY age_group;

JOIN (테이블 합치기) ⭐

# INNER JOIN: 환자 정보 + 진료 기록 (매칭되는 것만)
SELECT p.name, p.age, v.visit_date, v.diagnosis
FROM patient p
INNER JOIN visit v ON p.id = v.patient_id;

# LEFT JOIN: 모든 환자 + 진료 기록 (진료 없어도 포함)
SELECT p.name, v.visit_date
FROM patient p
LEFT JOIN visit v ON p.id = v.patient_id;

# 서브쿼리: 평균보다 BMI 높은 환자
SELECT name, bmi FROM patient
WHERE bmi > (SELECT AVG(bmi) FROM patient);

4-2. MongoDB + PyMongo

MongoDB는 문서형(NoSQL) 데이터베이스입니다. JSON 형태로 데이터를 저장합니다.

from pymongo import MongoClient

client = MongoClient("mongodb+srv://user:pw@cluster0.xxx.mongodb.net/")
db = client["sample_db"]
col = db["patients"]

# INSERT
col.insert_one({"name": "김환자", "age": 65})
col.insert_many([{"name": "이환자"}, {"name": "박환자"}])

# FIND
doc = col.find_one({"name": "김환자"})             # 1건
docs = col.find({"age": {"$gte": 60}})             # 60세 이상 전체
docs = col.find({}, {"_id": 0, "name": 1})         # name만 조회 (_id 제외)

# UPDATE / DELETE
col.update_one({"name": "김환자"}, {"$set": {"age": 66}})
col.delete_one({"name": "김환자"})

4-3. MongoDB Aggregation (집계 파이프라인)

여러 단계를 순서대로 거치며 데이터를 변환·분석합니다. SQL의 GROUP BY와 비슷합니다.

# 파이프라인 흐름: $match → $group → $project → $sort → $limit
col.aggregate([
    {"$match": {"systolic": {"$gte": 140}}},           # WHERE (필터)
    {"$group": {"_id": "$department",                   # GROUP BY
                "avg_bp": {"$avg": "$systolic"}}},
    {"$sort": {"avg_bp": -1}},                          # ORDER BY DESC
    {"$limit": 5}                                       # LIMIT
])

SQL vs MongoDB 대응표

MongoDB 비교 연산자

4-4. CSV → DB 파이프라인 (실전 패턴)

CSV 데이터를 정제해서 DB에 일괄 삽입하는 실무 패턴입니다.

import pandas as pd
import pymysql

# 1. CSV 읽기
df = pd.read_csv("health_train.csv")

# 2. 전처리
df.columns = [c.strip() for c in df.columns]           # 공백 제거
df = df[["환자ID", "나이", "키", "몸무게", "BMI", "혈압"]].copy()
df.columns = ["patient_id", "age", "height_cm", "weight_kg", "bmi", "bp"]
df = df.where(pd.notna(df), None)      # NaN → None (DB에서 NULL)

# 3. DB 삽입
sql = """INSERT INTO patient
    (patient_id, age, height_cm, weight_kg, bmi, bp)
    VALUES (%s, %s, %s, %s, %s, %s)"""

with conn.cursor() as cur:
    for idx in range(len(df)):
        cur.execute(sql, tuple(df.values[idx]))
    conn.commit()
    print(f"{len(df)}건 삽입 완료!")

4-5. MongoDB 복합 Aggregation 예제

# 실전: 고위험 환자 Top 3 찾기
# risk = LDL + 수축기혈압 + 혈당 → 높은 순 정렬
docs = col.aggregate([
    {"$project": {
        "name": 1,
        "ldl": 1, "systolic": 1, "glucose": 1,
        "_id": 0,
        "risk": {"$add": ["$ldl", "$systolic", "$glucose"]}
    }},
    {"$sort": {"risk": -1}},     # 위험도 높은 순
    {"$limit": 3}                # 상위 3명
])

# $project 산술 연산
# $add: 더하기 / $subtract: 빼기 / $multiply: 곱하기 / $divide: 나누기

# 복합 조건 검색 ($or)
docs = col.find({
    "$or": [
        {"glucose": {"$gte": 125}},
        {"systolic": {"$gte": 140}}
    ]
})

5-1. 머신러닝 개요

데이터에서 패턴을 스스로 학습하여 예측하는 알고리즘입니다. 알고리즘 모음이 아니라 "의사결정 파이프라인"입니다.

ML 워크플로우 (반드시 이 순서!)

scikit-learn 기본 워크플로우 (핵심 코드!)

import pandas as pd
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import accuracy_score, classification_report

# 1. 데이터 준비
df = pd.read_csv("diabetes.csv")
X = df.drop("Outcome", axis=1)    # 특성 (정답 제외)
y = df["Outcome"]                  # 타겟 (정답)

# 2. 데이터 분할 (stratify=y → 클래스 비율 유지!)
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42, stratify=y
)
print(f"학습: {X_train.shape}, 테스트: {X_test.shape}")

# 3. 스케일링 (Train 기준으로 fit!)
scaler = StandardScaler()
X_train_scaled = scaler.fit_transform(X_train)    # fit + transform
X_test_scaled = scaler.transform(X_test)           # transform만! (누수 방지)

# 4. 모델 생성 & 학습
model = LogisticRegression(max_iter=1000, random_state=42)
model.fit(X_train_scaled, y_train)

# 5. 예측 & 평가
y_pred = model.predict(X_test_scaled)
print(f"정확도: {accuracy_score(y_test, y_pred):.4f}")
print(classification_report(y_test, y_pred, target_names=["정상", "당뇨"]))

# 6. 모델 저장/로드
import joblib
joblib.dump(model, "model.pkl")
loaded = joblib.load("model.pkl")

5-2. 회귀 (Regression)

연속적인 숫자를 예측합니다. (예: 당뇨 진행도, 혈압 수치, 약물 투여량)

선형 회귀 (Linear Regression)

from sklearn.linear_model import LinearRegression
from sklearn.metrics import mean_squared_error, r2_score
import numpy as np

lr = LinearRegression()
lr.fit(X_train, y_train)
y_pred = lr.predict(X_test)

# 회귀 계수 해석
print(f"절편 (bias): {lr.intercept_:.4f}")
for name, coef in zip(feature_names, lr.coef_):
    print(f"  {name}: {coef:+.4f}")   # +양의 영향, -음의 영향

# 평가
mse = mean_squared_error(y_test, y_pred)
rmse = np.sqrt(mse)
r2 = r2_score(y_test, y_pred)
print(f"MSE: {mse:.4f}, RMSE: {rmse:.4f}, R²: {r2:.4f}")

Ridge / Lasso (정규화 회귀)

from sklearn.linear_model import Ridge, Lasso

# Ridge (L2): 큰 계수에 벌점 → 과적합 방지
ridge = Ridge(alpha=1.0)
ridge.fit(X_train, y_train)

# Lasso (L1): 불필요한 특성의 계수를 0으로 → 특성 선택 효과!
lasso = Lasso(alpha=0.01)
lasso.fit(X_train, y_train)
# 계수가 0인 특성 = 모델이 판단한 불필요한 특성
print("제거된 특성:", [n for n, c in zip(feature_names, lasso.coef_) if c == 0])

회귀 평가 지표

회귀 결과 시각화 — Actual vs Predicted

from sklearn.metrics import mean_squared_error, r2_score

y_pred = model.predict(X_test)

# 실제 vs 예측 산점도
plt.figure(figsize=(8, 6))
plt.scatter(y_test, y_pred, alpha=0.5, color="steelblue", edgecolors="k", s=40)
plt.plot([y_test.min(), y_test.max()], [y_test.min(), y_test.max()],
         "r--", lw=2, label="완벽한 예측 (y=x)")
plt.xlabel("실제값 (Actual)"); plt.ylabel("예측값 (Predicted)")
plt.title(f"Actual vs Predicted  |  R²={r2_score(y_test, y_pred):.4f}")
plt.legend(); plt.tight_layout()

# 잔차 플롯 (Residual Plot)
residuals = y_test - y_pred
plt.figure(figsize=(8, 4))
plt.scatter(y_pred, residuals, alpha=0.5, color="coral", edgecolors="k", s=30)
plt.axhline(y=0, color="black", linestyle="--")
plt.xlabel("예측값"); plt.ylabel("잔차 (실제-예측)")
plt.title("Residual Plot — 패턴이 없어야 좋은 모델!")
plt.tight_layout()

5-3. 분류 (Classification)

범주(카테고리)를 예측합니다. (예: 당뇨 여부, 질병 종류)

로지스틱 회귀 (Logistic Regression)

이름에 "회귀"가 붙지만 분류 모델입니다. Sigmoid 함수로 확률을 출력합니다. 가장 기본적인 baseline 모델.

from sklearn.linear_model import LogisticRegression
model = LogisticRegression(max_iter=1000, random_state=42)
model.fit(X_train_scaled, y_train)
y_pred = model.predict(X_test_scaled)

# 확률 출력 (0~1)
y_proba = model.predict_proba(X_test_scaled)[:, 1]  # 양성 클래스 확률
print(f"환자1 당뇨 확률: {y_proba[0]:.2%}")

KNN (K-Nearest Neighbors)

"가장 가까운 K개 이웃의 다수결"로 분류합니다. 스케일링 필수! (거리 기반)

from sklearn.neighbors import KNeighborsClassifier
knn = KNeighborsClassifier(n_neighbors=5)    # k=5 (홀수 권장)
knn.fit(X_train_scaled, y_train)
y_pred = knn.predict(X_test_scaled)
# 핵심 파라미터: n_neighbors (k값), weights ("uniform"/"distance")

SVM (Support Vector Machine) ⭐

데이터를 분류하는 최적의 경계선(결정 경계)을 찾는 알고리즘입니다. 마진을 최대화합니다.

from sklearn.svm import SVC
svm = SVC(kernel="rbf", C=1.0, gamma="scale", random_state=42)
svm.fit(X_train_scaled, y_train)    # 반드시 스케일링!
y_pred = svm.predict(X_test_scaled)
# C: 마진 vs 오류 trade-off (클수록 엄격)
# gamma: RBF 커널의 영향 범위 (클수록 복잡)

Decision Tree (의사결정나무)

특성값의 조건으로 트리처럼 분기하며 분류합니다. White-box 모델 (해석 가능!)이고 스케일링 불필요합니다.

from sklearn.tree import DecisionTreeClassifier, plot_tree
import matplotlib.pyplot as plt

dt = DecisionTreeClassifier(max_depth=4, min_samples_leaf=10, random_state=42)
dt.fit(X_train, y_train)    # 스케일링 불필요!

# 트리 시각화
plt.figure(figsize=(20, 10))
plot_tree(dt, feature_names=X.columns, class_names=["정상","당뇨"],
          filled=True, rounded=True, fontsize=10)
plt.title("의사결정나무")
plt.tight_layout()
plt.show()

# Feature Importance (특성 중요도)
importances = pd.Series(dt.feature_importances_, index=X.columns)
importances.sort_values(ascending=True).plot(kind="barh")
plt.title("특성 중요도")

분류 평가 지표 ⭐

혼동행렬 (Confusion Matrix)

from sklearn.metrics import classification_report, confusion_matrix
from sklearn.metrics import roc_auc_score, roc_curve

# 분류 리포트 (한번에 모든 지표)
print(classification_report(y_test, y_pred, target_names=["정상", "당뇨"]))

# 혼동행렬 시각화
import matplotlib.pyplot as plt
import seaborn as sns
cm = confusion_matrix(y_test, y_pred)
sns.heatmap(cm, annot=True, fmt="d", cmap="Blues",
            xticklabels=["정상","당뇨"], yticklabels=["정상","당뇨"])
plt.xlabel("예측"); plt.ylabel("실제"); plt.title("혼동행렬")

# ROC 곡선 & AUC
y_proba = model.predict_proba(X_test)[:, 1]   # 양성 클래스 확률
fpr, tpr, _ = roc_curve(y_test, y_proba)
auc = roc_auc_score(y_test, y_proba)
plt.plot(fpr, tpr, label=f"AUC={auc:.4f}")
plt.plot([0,1],[0,1], "k--")     # 대각선 (랜덤 기준)
plt.xlabel("FPR"); plt.ylabel("TPR"); plt.title("ROC Curve")
plt.legend()

Random Forest (랜덤 포레스트)

여러 개의 결정 트리를 만들어 다수결로 예측합니다. 스케일링 불필요, 과적합에 강합니다.

from sklearn.ensemble import RandomForestClassifier

rf = RandomForestClassifier(
    n_estimators=200,      # 트리 개수
    max_depth=6,           # 최대 깊이
    min_samples_leaf=10,   # 리프 최소 샘플 수
    random_state=42,
    n_jobs=-1              # 모든 CPU 사용
)
rf.fit(X_train, y_train)
y_pred = rf.predict(X_test)

# Feature Importance 시각화 (매우 중요!)
importances = pd.Series(rf.feature_importances_, index=X.columns)
top_features = importances.sort_values(ascending=False).head(10)

plt.figure(figsize=(10, 6))
top_features.sort_values().plot(kind="barh", color="steelblue")
plt.xlabel("중요도")
plt.title("Feature Importance (Top 10)")
plt.tight_layout()
plt.show()

Stacking (스태킹) — 앙상블의 앙상블

여러 모델의 예측을 메타 모델의 입력으로 사용합니다.

from sklearn.ensemble import StackingClassifier

# 기본 모델 (Level 0)
estimators = [
    ("rf", RandomForestClassifier(n_estimators=100, random_state=42)),
    ("xgb", XGBClassifier(n_estimators=200, eval_metric="logloss")),
    ("svm", SVC(kernel="rbf", probability=True))
]

# 메타 모델 (Level 1)
stacking = StackingClassifier(
    estimators=estimators,
    final_estimator=LogisticRegression(),
    cv=5                    # 기본 모델도 교차검증으로 예측
)
stacking.fit(X_train_scaled, y_train)
y_pred = stacking.predict(X_test_scaled)
print(f"Stacking 정확도: {accuracy_score(y_test, y_pred):.4f}")

5-4. 비지도학습

PCA (주성분 분석) — 차원 축소

많은 특성을 핵심 성분 몇 개로 압축합니다. 반드시 StandardScaler 적용 후 사용!

from sklearn.decomposition import PCA
from sklearn.preprocessing import StandardScaler

X_scaled = StandardScaler().fit_transform(X)

# 1) 전체 성분으로 분산 비율 확인 (Scree Plot)
pca_full = PCA().fit(X_scaled)
cumsum = np.cumsum(pca_full.explained_variance_ratio_)

plt.figure(figsize=(8, 4))
plt.bar(range(1, len(cumsum)+1), pca_full.explained_variance_ratio_,
        alpha=0.6, label="개별 분산")
plt.step(range(1, len(cumsum)+1), cumsum, where="mid",
         color="red", label="누적 분산")
plt.axhline(y=0.95, color="gray", linestyle="--", label="95% 기준선")
plt.xlabel("주성분 번호"); plt.ylabel("설명 분산 비율")
plt.title("Scree Plot"); plt.legend(); plt.tight_layout()

# 2) 2차원 축소 후 산점도
pca = PCA(n_components=2)
X_pca = pca.fit_transform(X_scaled)
print(f"PC1: {pca.explained_variance_ratio_[0]:.2%}")
print(f"PC2: {pca.explained_variance_ratio_[1]:.2%}")

plt.figure(figsize=(8, 6))
scatter = plt.scatter(X_pca[:, 0], X_pca[:, 1], c=y, cmap="coolwarm", alpha=0.6)
plt.colorbar(scatter, label="클래스")
plt.xlabel("PC1"); plt.ylabel("PC2")
plt.title("PCA 2D 시각화"); plt.tight_layout()

K-Means — 군집화

데이터를 K개 그룹으로 자동 분류합니다.

from sklearn.cluster import KMeans
from sklearn.metrics import silhouette_score

# 최적 K 탐색
for k in range(2, 11):
    km = KMeans(n_clusters=k, random_state=42, n_init=10)
    labels = km.fit_predict(X_scaled)
    score = silhouette_score(X_scaled, labels)  # 1에 가까울수록 좋음
    print(f"k={k}: {score:.4f}")

5-5. 앙상블 (심화)

여러 모델을 합쳐서 더 좋은 성능을 내는 방법입니다.

XGBoost 코드 패턴 (가장 많이 사용)

from xgboost import XGBClassifier
from sklearn.model_selection import GridSearchCV, StratifiedKFold

xgb = XGBClassifier(random_state=42, eval_metric="logloss")
param_grid = {
    "n_estimators": [200, 300],
    "learning_rate": [0.005, 0.01],
    "max_depth": [4, 5, 6],
    "subsample": [0.7, 0.8],
    "colsample_bytree": [0.7, 0.8]
}
grid = GridSearchCV(xgb, param_grid, cv=StratifiedKFold(5),
                    scoring="accuracy", n_jobs=-1)
grid.fit(X_train, y_train)
best_model = grid.best_estimator_

LightGBM (가장 빠른 부스팅)

from lightgbm import LGBMClassifier

lgbm = LGBMClassifier(
    n_estimators=300,
    learning_rate=0.01,
    max_depth=5,
    num_leaves=31,          # leaf-wise (LightGBM 핵심)
    subsample=0.8,
    colsample_bytree=0.8,
    random_state=42,
    verbose=-1              # 로그 끔
)
lgbm.fit(X_train, y_train,
         eval_set=[(X_val, y_val)],
         callbacks=[lgbm_early_stopping(10)])  # Early Stopping

Early Stopping (조기 종료)

xgb = XGBClassifier(**best_params, early_stopping_rounds=10)
xgb.fit(X_train, y_train, eval_set=[(X_val, y_val)], verbose=False)
# → 검증 성능이 10라운드 동안 개선 안 되면 자동 종료

5-6. 데이터 불균형 & SMOTE

소수 클래스 데이터가 너무 적으면 모델이 "대부분 0이야"라고 예측해도 정확도가 높게 나옵니다. 하지만 의료(질병 탐지)에서는 놓치는 것(FN)이 치명적!

해결 전략

from imblearn.pipeline import Pipeline
from imblearn.over_sampling import SMOTE

pipe = Pipeline([
    ("smote", SMOTE(random_state=42)),
    ("clf", LogisticRegression(max_iter=1000))
])
pipe.fit(X_train, y_train)       # SMOTE는 자동으로 Train에만 적용
pred = pipe.predict(X_test)

5-7. 비지도학습 심화 (군집화)

DBSCAN — 실무에서 가장 강력한 군집화

밀도 기반 군집화. 군집 개수를 미리 정하지 않아도 되고, 이상치(Noise)를 자동 분류합니다.

• Core Point: 반경 eps 안에 min_samples 이상 존재하는 핵심 점
• Border Point: 코어 점 근처에 있지만 자체는 코어가 아닌 점
• Noise: 어디에도 속하지 못한 점 (레이블 = -1)

from sklearn.cluster import DBSCAN
dbscan = DBSCAN(eps=0.5, min_samples=5)
labels = dbscan.fit_predict(X_scaled)

# K-Distance Plot으로 최적 eps 찾기 ⭐
from sklearn.neighbors import NearestNeighbors

neighbors = NearestNeighbors(n_neighbors=5)
neighbors.fit(X_scaled)
distances, _ = neighbors.kneighbors(X_scaled)
distances = np.sort(distances[:, 4])  # 5번째 이웃까지 거리

plt.figure(figsize=(8, 4))
plt.plot(distances)
plt.xlabel("데이터 포인트 (정렬됨)")
plt.ylabel("5번째 이웃까지 거리")
plt.title("K-Distance Plot — 꺾이는 지점 = 최적 eps!")
plt.axhline(y=0.55, color="r", linestyle="--", label="eps ≈ 0.55")
plt.legend(); plt.tight_layout()

# 시각화
plt.figure(figsize=(8, 6))
unique_labels = set(labels)
for label in unique_labels:
    mask = labels == label
    marker = "x" if label == -1 else "o"
    color = "gray" if label == -1 else None
    plt.scatter(X_pca[mask, 0], X_pca[mask, 1], marker=marker,
                s=40, alpha=0.6, label=f"{'Noise' if label==-1 else f'군집{label}'}")
plt.legend(); plt.title("DBSCAN 결과")

GMM (Gaussian Mixture Model) — 확률 기반 군집

K-means는 "이 점은 A군집!" (Hard 할당), GMM은 "이 점은 A일 확률 70%, B일 확률 30%" (Soft 할당)

from sklearn.mixture import GaussianMixture
gmm = GaussianMixture(n_components=3, random_state=42)
gmm_labels = gmm.fit_predict(X_scaled)
probs = gmm.predict_proba(X_scaled)  # 각 군집에 속할 확률

군집화 알고리즘 비교

5-8. 전처리 심화

스케일링 비교

범주형 인코딩

Pipeline — 전처리+모델을 하나로 묶기

데이터 누수를 방지하고 코드를 깔끔하게 만듭니다.

from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.decomposition import PCA
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import cross_val_score

pipe = Pipeline([
    ("scaler", StandardScaler()),
    ("pca", PCA(n_components=2)),
    ("clf", LogisticRegression(max_iter=1000))
])
scores = cross_val_score(pipe, X, y, cv=5, scoring="accuracy")
print(f"평균: {scores.mean():.4f}")

RFECV — 자동 특성 선택 (몇 개가 최적인지 찾기)

from sklearn.feature_selection import RFECV
from sklearn.svm import SVC

rfecv = RFECV(
    estimator=SVC(kernel="linear"),
    step=1,                            # 한번에 1개씩 제거
    cv=StratifiedKFold(5),
    scoring="accuracy",
    min_features_to_select=1
)
rfecv.fit(X_train_scaled, y_train)

print(f"최적 특성 수: {rfecv.n_features_}")
print(f"선택된 특성: {X.columns[rfecv.support_].tolist()}")

# 특성 수에 따른 성능 변화 시각화
plt.figure(figsize=(8, 4))
plt.plot(range(1, len(rfecv.cv_results_["mean_test_score"])+1),
         rfecv.cv_results_["mean_test_score"], marker="o")
plt.xlabel("특성 수"); plt.ylabel("정확도")
plt.title("RFECV — 특성 수 vs 성능"); plt.tight_layout()

5-9. 교차검증 & 하이퍼파라미터 튜닝

교차검증 (Cross Validation) ⭐

데이터를 K등분하여 K번 학습/평가를 반복합니다. 모든 데이터가 한번씩 검증에 사용되므로 신뢰도 높은 성능 추정이 가능합니다.

from sklearn.model_selection import cross_val_score, StratifiedKFold

# 기본 교차검증
scores = cross_val_score(model, X_scaled, y, cv=5, scoring="accuracy")
print(f"정확도: {scores.mean():.4f} ± {scores.std():.4f}")

# StratifiedKFold (분류에서 클래스 비율 유지 ⭐)
skf = StratifiedKFold(n_splits=5, shuffle=True, random_state=42)
scores = cross_val_score(model, X_scaled, y, cv=skf, scoring="f1")

GridSearchCV (최적 하이퍼파라미터 찾기)

from sklearn.model_selection import GridSearchCV

# 모든 조합을 시도 (시간 오래 걸림)
param_grid = {
    "n_estimators": [100, 200, 300],
    "max_depth": [3, 4, 5, 6],
    "learning_rate": [0.01, 0.05, 0.1],
    "subsample": [0.7, 0.8, 0.9]
}
grid = GridSearchCV(
    xgb, param_grid,
    cv=StratifiedKFold(5),
    scoring="f1",            # 최적화 기준
    n_jobs=-1,               # 모든 CPU 사용
    verbose=1
)
grid.fit(X_train, y_train)

print(f"최적 파라미터: {grid.best_params_}")
print(f"최고 점수: {grid.best_score_:.4f}")
best_model = grid.best_estimator_

Optuna — 베이지안 하이퍼파라미터 최적화

GridSearchCV보다 훨씬 효율적입니다. TPE 알고리즘으로 좋은 영역을 집중 탐색합니다.

import optuna
from xgboost import XGBClassifier
from sklearn.model_selection import cross_val_score

def objective(trial):
    """Optuna 목적 함수: 최대화할 지표 반환"""
    params = {
        "n_estimators": trial.suggest_int("n_estimators", 50, 300),
        "max_depth": trial.suggest_int("max_depth", 3, 15),
        "learning_rate": trial.suggest_float("learning_rate", 0.01, 0.3, log=True),
        "subsample": trial.suggest_float("subsample", 0.6, 1.0),
        "colsample_bytree": trial.suggest_float("colsample_bytree", 0.6, 1.0),
        "min_child_weight": trial.suggest_int("min_child_weight", 1, 10),
        "reg_alpha": trial.suggest_float("reg_alpha", 0, 1),
        "reg_lambda": trial.suggest_float("reg_lambda", 0, 1),
    }
    model = XGBClassifier(**params, random_state=42, eval_metric="logloss")
    score = cross_val_score(model, X_train, y_train, cv=5, scoring="accuracy").mean()
    return score

# Study 생성 & 최적화 실행
study = optuna.create_study(direction="maximize")
study.optimize(objective, n_trials=50, show_progress_bar=True)

print(f"최고 점수: {study.best_value:.4f}")
print(f"최적 파라미터: {study.best_params}")

# 최적 모델로 학습
best_model = XGBClassifier(**study.best_params, random_state=42)
best_model.fit(X_train, y_train)

SHAP — 모델 해석 (Explainable AI) ⭐

모델이 "왜" 그런 예측을 했는지 설명합니다. 의료 AI에서 필수!

import shap

# TreeExplainer (Tree 기반 모델에 가장 빠름)
explainer = shap.TreeExplainer(best_model)
shap_values = explainer.shap_values(X_test)

# 1. 전역: Summary Plot (가장 중요한 시각화!)
shap.summary_plot(shap_values, X_test, feature_names=feature_names)
# → 특성 중요도 + 영향 방향을 동시에 확인

# 2. 전역: Bar Plot (평균 절대 SHAP 값)
shap.summary_plot(shap_values, X_test, plot_type="bar")

# 3. 지역: 개별 환자 예측 설명 (Waterfall Plot)
shap.waterfall_plot(shap.Explanation(
    values=shap_values[0],
    base_values=explainer.expected_value,
    data=X_test.iloc[0].values,
    feature_names=feature_names
))
# → "이 환자가 당뇨로 예측된 이유: 혈당 ↑, BMI ↑, 나이 ↑"

# 4. 특성 간 상호작용
shap.dependence_plot("BMI", shap_values, X_test, interaction_index="Age")

모델 비교 코드 패턴

from sklearn.linear_model import LogisticRegression
from sklearn.tree import DecisionTreeClassifier
from sklearn.ensemble import RandomForestClassifier
from xgboost import XGBClassifier

models = {
    "Logistic": LogisticRegression(max_iter=1000),
    "DecisionTree": DecisionTreeClassifier(max_depth=5),
    "RandomForest": RandomForestClassifier(n_estimators=100),
    "XGBoost": XGBClassifier(n_estimators=200, eval_metric="logloss")
}

results = {}
for name, model in models.items():
    scores = cross_val_score(model, X_scaled, y, cv=5, scoring="f1")
    results[name] = scores
    print(f"{name:15s}: F1 = {scores.mean():.4f} ± {scores.std():.4f}")

# 모델 비교 시각화
plt.figure(figsize=(10, 5))
plt.boxplot(results.values(), labels=results.keys())
plt.ylabel("F1 Score")
plt.title("모델별 교차검증 성능 비교")
plt.grid(axis="y", alpha=0.3)
plt.tight_layout()
plt.show()

6-1. 의료 데이터를 읽는 눈

진단코드 (ICD/KCD)

의사가 진료 후 기록하는 진단명은 표준화된 코드로 저장됩니다.

구조화 데이터 vs 비구조화 데이터

표준 의료용어 체계

6-2. 환자 여정 (Patient Journey)

AI 서비스는 환자 여정의 특정 단계에 개입합니다.

6-3. 일상질환 × AI 서비스

A. 정신건강

• 정신건강 데이터의 특수성: 객관적 바이오마커 부족, 핵심 데이터가 비구조화(면담 내용), 높은 맥락 의존성
• PHQ-9: 우울증 선별 (9문항, 0~27점, 10점 이상 → 유의미)
• GAD-7: 불안장애 선별 (7문항, 0~21점)
• 사례: Woebot(CBT 기반 챗봇), 마인드카페(상담 매칭), 트로스트(자가관리)

B. 금연/중독

• 행동 데이터 중심: 시간, 장소, 동반활동, 감정, 사회적 맥락
• JITAI (Just-In-Time Adaptive Intervention): 가장 취약한 순간에 맞춤 개입
• 금연 시도자 ~65%가 첫 1주 내 재흡연, 1년 후 유지율 5~10%

C. 생활습관 (당뇨·고혈압)

• CGM(연속혈당측정): 5분마다 자동 → 하루 288개 데이터
• TIR(Time in Range): 목표 혈당 범위 내 시간 비율
• 가정혈압: 백의고혈압/가면고혈압 문제 해결

D. 수면

• PSG(다원수면검사) = Gold Standard > 웨어러블 > 자가보고
• CBT-I (불면증 인지행동치료): 수면제보다 장기 효과 우수, 1차 치료 권장
• DTx 사례: Somryst(FDA 승인), 솜즈(한국 식약처 허가)

6-4. 규제환경 — 세 가지 카테고리

6-5. 핵심 교훈 (Key Takeaways)

1. 의학을 몰라도 데이터 구조(ICD, 정형/비정형, FHIR)를 이해하면 기획 가능
2. "대단한 질병"이 아닌 일상 건강문제에서 AI 서비스 기회가 더 큼
3. 환자 여정의 어느 단계에 개입하는지가 데이터·규제·사업모델을 결정
4. 규제·개인정보·임상검증을 모르면 아무리 좋은 기술도 시장 진출 불가
5. 의료진과의 효과적 협업이 성패를 좌우

7-1. 퍼셉트론 / MLP

퍼셉트론은 신경망의 가장 작은 단위입니다. 입력에 가중치를 곱하고 합산한 후 활성화 함수를 적용합니다.

논리 게이트 진리표 — 왜 XOR이 문제인가

• 단일 퍼셉트론 = 선형 문제만 해결 (AND, OR는 되지만 XOR은 불가!)
• MLP (다층 퍼셉트론) = 은닉층을 추가하면 비선형 문제(XOR) 해결 가능

import torch
import torch.nn as nn

# MLP 모델 정의
class MLP(nn.Module):
    def __init__(self, input_size, hidden_size, output_size):
        super().__init__()
        self.fc1 = nn.Linear(input_size, hidden_size)    # 입력→은닉1
        self.fc2 = nn.Linear(hidden_size, hidden_size)   # 은닉1→은닉2
        self.fc3 = nn.Linear(hidden_size, output_size)   # 은닉2→출력
        self.relu = nn.ReLU()
        self.dropout = nn.Dropout(0.3)   # 과적합 방지

    def forward(self, x):
        x = self.relu(self.fc1(x))       # 은닉층 1
        x = self.dropout(x)
        x = self.relu(self.fc2(x))       # 은닉층 2
        x = self.fc3(x)                  # 출력층 (활성화 X → 손실함수에서 처리)
        return x

# 모델 생성
model = MLP(input_size=10, hidden_size=64, output_size=2)

실습: 순방향 전파로 집 가격 예측 (Stage 1)

파이썬만으로 순방향 전파를 구현하여 신경망의 기본 작동 원리를 체감합니다.

# Stage 1: 순방향 전파 (순수 Python, PyTorch 없이)
input_data = [30, 2]  # 집 크기 30평, 방 2개

weights = {
    'node_0': [50, 10],   # 은닉층 첫 번째 노드 가중치
    'node_1': [-1, 0],    # 은닉층 두 번째 노드 가중치
    'output': [0.5, 25]   # 출력층 가중치
}

# 은닉층 계산
node_0_value = input_data[0] * weights['node_0'][0] + input_data[1] * weights['node_0'][1]  # 1520
node_1_value = input_data[0] * weights['node_1'][0] + input_data[1] * weights['node_1'][1]  # -30

hidden_layer_values = [node_0_value, node_1_value]

# 출력층 계산
output = hidden_layer_values[0] * weights['output'][0] + hidden_layer_values[1] * weights['output'][1]
print(f"예측된 집의 가격: {int(output)}억")  # → 10억

다층 신경망 + ReLU 활성화 함수

은닉층을 2개로 늘리고 ReLU를 적용하면 음수 값이 0으로 처리되어 비선형 학습이 가능해집니다.

# Stage 1: 2층 은닉층 + ReLU
def relu(x):
    return max(0, x)

input_data = [30.0, 2.0]
weights = {
    'hidden1_node0': [50.0, 10.0], 'hidden1_node1': [-1.0, 0.0],
    'hidden2_node0': [0.1, 10.0],  'hidden2_node1': [-0.1, 0.0],
    'output': [0.1, 25.0]
}

# 첫 번째 은닉층 + ReLU
h1_0 = relu(input_data[0]*weights['hidden1_node0'][0] + input_data[1]*weights['hidden1_node0'][1])  # 1520
h1_1 = relu(input_data[0]*weights['hidden1_node1'][0] + input_data[1]*weights['hidden1_node1'][1])  # 0 (ReLU: -30→0)

# 두 번째 은닉층 + ReLU
h2_0 = relu(h1_0*weights['hidden2_node0'][0] + h1_1*weights['hidden2_node0'][1])  # 152
h2_1 = relu(h1_0*weights['hidden2_node1'][0] + h1_1*weights['hidden2_node1'][1])  # 0 (ReLU: -152→0)

# 출력
output = h2_0*weights['output'][0] + h2_1*weights['output'][1]
print(f"예측 가격: {int(output)}억")  # → 15억

7-2. 손실함수 & 활성화함수

손실함수 — 예측이 얼마나 틀렸는지 측정

손실함수의 값이 낮을수록 모델의 예측이 실제값과 더 잘 일치한다는 뜻. 모델 학습의 목표 = 손실 최소화.

왜 활성화함수가 필요한가? (Non-Linearity)

활성화함수 — 비선형성을 부여

7-3. 역전파 & 옵티마이저

순전파 → 역전파 흐름

순전파(Forward Propagation): 입력 데이터가 신경망의 각 층을 지나면서, 노드의 가중치와 결합되고 활성화 함수를 통과하며, 마지막에는 예측값(ŷ)을 출력하는 과정.

역전파(Backpropagation): 손실을 각 가중치로 미분하여 기울기를 계산하는 과정 (Chain Rule). 신경망의 끝에서 시작해서, 실제 값과의 차이를 각 층을 거슬러 올라가며 노드의 가중치에 전달하고, 가중치를 조금씩 조정합니다.

순전파 가중치 계산 예시

경사하강법 (Gradient Descent)

w_new = w_old - η × d(기울기)
# η(eta) = learning rate (학습률)
# 보통 0.001, 0.01 등 작은 값 사용

옵티마이저 비교

옵티마이저는 손실 함수의 값을 최소화하기 위해 모델 파라미터(W)를 조정하는 방법을 정의합니다.

딥러닝 모델 학습 순서

모델 학습 과정은 데이터 준비, 모델 정의, 손실함수와 옵티마이저 선택이 완료된 이후 실행됩니다.

1. model.train() — 학습 모드 설정 (Dropout, BatchNorm이 학습용으로 동작)
2. 순전파 — 입력 → 모델 → 예측값
3. 손실 계산 — 예측값과 정답의 차이
4. 기울기 초기화 — zero_grad() (필수!)
5. 역전파 — loss.backward() (자동 미분)
6. 가중치 업데이트 — optimizer.step()

PyTorch 학습 루프 (핵심! 외우세요)

for epoch in range(epochs):
    model.train()                # 학습 모드 (Dropout 활성화)
    pred = model(X)              # 1. 순전파
    loss = criterion(pred, y)    # 2. 손실 계산
    optimizer.zero_grad()        # 3. 기울기 초기화 (필수!)
    loss.backward()              # 4. 역전파 (자동 미분)
    optimizer.step()             # 5. 가중치 업데이트

실습: 역전파로 가중치 업데이트 (Stage 2)

파이썬만으로 역전파를 구현합니다. 신경망이 예측 오류에서 배워 가중치를 조정하는 과정입니다.

# Stage 2: 역전파 (순수 Python)
def relu_derivative(output):
    """ReLU 미분: 양수→1, 음수→0"""
    return 1 if output > 0 else 0

# 1. 오차 계산
predicted_value = 15    # 순전파로 얻은 예측값
target_value = 18       # 실제 목표값
error = predicted_value - target_value  # -3

# 2. 오류의 기울기 (MSE 미분)
error_gradient = error * 2  # -6

# 3. 가중치에 대한 손실 함수의 기울기
hidden_node_output = 152  # ReLU 통과 후 은닉노드 출력값
hidden_node_gradient = relu_derivative(hidden_node_output)  # 1 (양수이므로)
weight_gradient = hidden_node_output * error_gradient * hidden_node_gradient  # -912

# 4. 가중치 업데이트: 새 가중치 = 이전 - 학습률 × 기울기
learning_rate = 0.01
current_weight = 0.1
next_weight = current_weight - learning_rate * weight_gradient  # 9.22
print(f"업데이트된 가중치: {next_weight}")  # → 9.22

7-4. PyTorch 기초

CPU vs GPU — 왜 딥러닝에 GPU가 필요한가

Tensor = 다차원 배열 (NumPy 배열의 GPU 버전)

텐서 데이터 타입 — 텐서가 저장하는 데이터의 종류와 메모리 크기/값 범위를 결정합니다.

import torch

# 텐서 생성
x = torch.tensor([1.0, 2.0, 3.0])
x = torch.zeros(3, 4)           # 3×4 영행렬 (가중치 초기화, 임시 텐서 생성용)
x = torch.ones(3, 4)            # 3×4 모두 1 (기본값 설정, 마스킹 연산용)
x = torch.full((3, 4), 7)       # 3×4 모두 7 (특정 값으로 초기화)
x = torch.rand(3, 4)            # 0~1 균일분포 (가중치 초기화에 사용)
x = torch.randn(3, 4)           # 정규분포 랜덤
x = torch.from_numpy(np_array)  # NumPy → Tensor

# GPU 이동
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
x = x.to(device)                # GPU로 이동 (학습 속도 향상)

# 자동 미분 (역전파의 핵심!)
# 미분이란? 함수의 순간적인 변화율. x가 아주 조금 변할 때 f(x)가 얼마나 변하는지.
# f'(x) = lim(Δx→0) [f(x+Δx) - f(x)] / Δx
x = torch.tensor([2.0], requires_grad=True)
y = x**2 + 3*x + 1       # y = x² + 3x + 1
y.backward()              # 미분 실행
print(x.grad)             # tensor([7.]) → dy/dx = 2x+3 = 2(2)+3 = 7

DataLoader (배치 학습)

from torch.utils.data import TensorDataset, DataLoader

# 데이터셋 → DataLoader
dataset = TensorDataset(X_tensor, y_tensor)
train_loader = DataLoader(dataset, batch_size=32, shuffle=True)

# 배치 순회
for batch_X, batch_y in train_loader:
    pred = model(batch_X)
    loss = criterion(pred, batch_y)
    ...

전체 학습 코드 (템플릿) ⭐⭐

import torch
import torch.nn as nn
from torch.utils.data import DataLoader, TensorDataset

# 1. 데이터 준비
X_train_t = torch.FloatTensor(X_train_scaled)
y_train_t = torch.LongTensor(y_train.values)
train_dataset = TensorDataset(X_train_t, y_train_t)
train_loader = DataLoader(train_dataset, batch_size=32, shuffle=True)

# 2. 모델 정의
model = MLP(input_size=10, hidden_size=64, output_size=2)
model = model.to(device)

# 3. 손실함수 & 옵티마이저
criterion = nn.CrossEntropyLoss()
optimizer = torch.optim.Adam(model.parameters(), lr=0.001)

# 4. 학습 루프
for epoch in range(100):
    model.train()          # 학습 모드 (Dropout 활성화)
    total_loss = 0
    for batch_X, batch_y in train_loader:
        batch_X, batch_y = batch_X.to(device), batch_y.to(device)

        pred = model(batch_X)            # 순전파
        loss = criterion(pred, batch_y)  # 손실 계산
        optimizer.zero_grad()            # 기울기 초기화 (필수!)
        loss.backward()                  # 역전파
        optimizer.step()                 # 가중치 업데이트
        total_loss += loss.item()

    if (epoch+1) % 10 == 0:
        print(f"Epoch {epoch+1}: Loss = {total_loss/len(train_loader):.4f}")

# 5. 평가
model.eval()               # 평가 모드 (Dropout 비활성화)
with torch.no_grad():      # 기울기 계산 안 함 (메모리 절약)
    X_test_t = torch.FloatTensor(X_test_scaled).to(device)
    pred = model(X_test_t)
    predicted = pred.argmax(dim=1)
    accuracy = (predicted.cpu() == torch.LongTensor(y_test.values)).float().mean()
    print(f"정확도: {accuracy:.4f}")

nn.Module로 MLP 정의 (가장 기본적인 모델)

class MLP(nn.Module):
    def __init__(self, input_size, hidden_size, output_size):
        super().__init__()
        self.layers = nn.Sequential(
            nn.Linear(input_size, hidden_size),    # 입력 → 은닉
            nn.BatchNorm1d(hidden_size),           # 배치 정규화
            nn.ReLU(),                              # 활성화 함수
            nn.Dropout(0.3),                        # 과적합 방지
            nn.Linear(hidden_size, hidden_size//2), # 은닉 → 은닉
            nn.ReLU(),
            nn.Dropout(0.3),
            nn.Linear(hidden_size//2, output_size)  # 은닉 → 출력
        )

    def forward(self, x):
        return self.layers(x)

# 사용
model = MLP(input_size=10, hidden_size=64, output_size=2)
print(model)  # 구조 확인

# 파라미터 수 확인
total_params = sum(p.numel() for p in model.parameters())
print(f"총 파라미터: {total_params:,}개")

MNIST 실습 — 손글씨 숫자 분류 (0~9)

딥러닝의 "Hello World". 28×28 흑백 이미지를 10개 클래스로 분류합니다.

import torch
import torch.nn as nn
from torchvision import datasets, transforms
from torch.utils.data import DataLoader

# 1. 데이터 준비 (28×28 → 784 벡터로 펼치기)
transform = transforms.Compose([
    transforms.ToTensor(),           # [0,255] → [0,1] + Tensor 변환
    transforms.Normalize((0.5,), (0.5,))  # 정규화: (x-0.5)/0.5 → [-1,1]
])

train_data = datasets.MNIST("./data", train=True, download=True, transform=transform)
test_data  = datasets.MNIST("./data", train=False, transform=transform)

train_loader = DataLoader(train_data, batch_size=64, shuffle=True)
test_loader  = DataLoader(test_data, batch_size=64)

# 2. MLP 모델 (이미지를 1차원으로 펼쳐서 입력)
class MNIST_MLP(nn.Module):
    def __init__(self):
        super().__init__()
        self.flatten = nn.Flatten()           # 28×28 → 784
        self.layers = nn.Sequential(
            nn.Linear(784, 256),
            nn.BatchNorm1d(256),
            nn.ReLU(),
            nn.Dropout(0.3),
            nn.Linear(256, 128),
            nn.ReLU(),
            nn.Linear(128, 10)                # 10개 클래스 출력
        )

    def forward(self, x):
        x = self.flatten(x)
        return self.layers(x)

# 3. 학습
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model = MNIST_MLP().to(device)
criterion = nn.CrossEntropyLoss()
optimizer = torch.optim.Adam(model.parameters(), lr=0.001)

for epoch in range(10):
    model.train()
    for images, labels in train_loader:
        images, labels = images.to(device), labels.to(device)
        pred = model(images)
        loss = criterion(pred, labels)
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()

# 4. 평가
model.eval()
correct = 0
with torch.no_grad():
    for images, labels in test_loader:
        images, labels = images.to(device), labels.to(device)
        pred = model(images).argmax(dim=1)
        correct += (pred == labels).sum().item()

print(f"MNIST 정확도: {correct / len(test_data):.4f}")  # ~98%

Config(설정) 관리 — 재현성의 핵심

모델의 아키텍처, 학습률, 배치 크기, 에포크 수, 데이터셋 경로 등을 하나로 관리하면 재현성과 코드 유연성이 높아집니다.

# Config 예시 (dict 또는 dataclass)
config = {
    "seed": 42,
    "batch_size": 64,
    "lr": 0.001,
    "epochs": 100,
    "hidden_size": 256,
    "dropout": 0.3,
    "device": "cuda" if torch.cuda.is_available() else "cpu",
}
# → 다른 연구자가 같은 설정으로 실험을 재현할 수 있음

7-4-1. 실습: PyTorch로 단순 선형 회귀 구현 (Stage 3) ⭐

PyTorch 텐서 연산을 활용하여 편향(bias)을 포함한 단순 선형 회귀를 직접 구현합니다. 목표: 가장 잘 맞는 직선을 찾는 것 = W(가중치)와 b(편향)을 최적화하는 것!

① 선형 변환과 편향(bias)의 역할

선형 변환의 수학적 표현: Y = XW + b

② 데이터 준비 & 파라미터 초기화

import torch
import torch.optim as optim

# 훈련 데이터: x=[1,2,3] → y=[3,6,9] (y=3x 관계)
x_train = torch.tensor([[1], [2], [3]], dtype=torch.float)
y_train = torch.tensor([[3], [6], [9]], dtype=torch.float)

# 가중치와 편향을 0으로 초기화
# requires_grad=True → 학습 가능한 파라미터 (자동 미분 대상)
W = torch.zeros(1, requires_grad=True)
b = torch.zeros(1, requires_grad=True)

print(f"초기 가중치: {W}")  # tensor([0.], requires_grad=True)
print(f"초기 편향: {b}")    # tensor([0.], requires_grad=True)

③ 학습 과정 (순전파 → 오차 → 역전파 → 업데이트)

# 각 단계 개별 이해
pred = x_train * W + b                   # ① 순전파
loss = torch.mean((pred - y_train) ** 2)  # ② MSE 오차
loss.backward()                           # ③ 역전파 (기울기 계산)
optimizer.step()                          # ④ 가중치 업데이트

④ SGD vs GD — 왜 확률적 경사 하강법을 쓰는가

# SGD 옵티마이저 설정
# [W, b] = 최적화할 파라미터 목록, lr = 학습률
optimizer = optim.SGD([W, b], lr=0.01)

⑤ 전체 학습 코드 (1000 에포크) ⭐⭐

import torch
import torch.optim as optim

# 데이터 준비
x_train = torch.tensor([[1], [2], [3]], dtype=torch.float)
y_train = torch.tensor([[3], [6], [9]], dtype=torch.float)

# 파라미터 초기화
W = torch.zeros(1, requires_grad=True)
b = torch.zeros(1, requires_grad=True)

# 옵티마이저 설정
optimizer = optim.SGD([W, b], lr=0.01)

# 학습 루프
epochs = 1000
for epoch in range(epochs):
    pred = x_train * W + b                   # 순전파: 예측값 계산
    loss = torch.mean((pred - y_train) ** 2)  # MSE 손실 계산

    optimizer.zero_grad()  # 기울기 초기화 (필수!)
    loss.backward()        # 역전파: 기울기 계산
    optimizer.step()       # 가중치 업데이트

    if (epoch+1) % 100 == 0:
        print(f'Epoch [{epoch+1}/{epochs}], W: {W.item():.3f}, b: {b.item():.3f}, Loss: {loss.item():.4f}')

# 출력 예시:
# Epoch [100/1000],  W: 2.618, b: 0.869, Loss: 0.1089
# Epoch [500/1000],  W: 2.854, b: 0.332, Loss: 0.0159
# Epoch [1000/1000], W: 2.956, b: 0.100, Loss: 0.0014
# → W≈3, b≈0으로 수렴 (정답: y=3x)

⑥ 학습된 모델로 예측하기

# 예측 시에는 torch.no_grad()로 자동 미분을 비활성화
# → 메모리 절약 + 불필요한 기울기 계산 방지
test_x = torch.tensor([[10]], dtype=torch.float)

with torch.no_grad():
    pred_y = test_x * W + b
    print(f"입력 10일 때 예측값: {pred_y.item():.2f}")
    # → 약 29.66 (이론값 30에 근접)

7-4-2. nn.Linear & nn.Module로 모델 구현 (Stage 4)

Stage 3에서 텐서를 직접 다뤘다면, 이제 PyTorch의 nn.Linear와 nn.Module로 같은 모델을 더 체계적으로 구현합니다.

nn.Linear — 가중치·편향 자동 관리

import torch
import torch.nn as nn
import torch.optim as optim
import torch.nn.functional as F

x_train = torch.tensor([[1], [2], [3]], dtype=torch.float)
y_train = torch.tensor([[3], [6], [9]], dtype=torch.float)

# nn.Linear: in_features=1 → out_features=1 (단순 선형)
model = nn.Linear(in_features=1, out_features=1)

# 자동 초기화된 파라미터 확인
for param in model.parameters():
    print(param.data)  # 가중치, 편향이 무작위 값으로 자동 설정됨

# 학습 (F.mse_loss로 손실 계산 간소화)
optimizer = optim.SGD(model.parameters(), lr=0.01)

for epoch in range(1000):
    pred = model(x_train)              # 순전파 (model에 직접 입력!)
    loss = F.mse_loss(pred, y_train)   # MSE 손실 (수동 계산 불필요)
    optimizer.zero_grad()
    loss.backward()
    optimizer.step()

# 예측
with torch.no_grad():
    pred_y = model(torch.tensor([[10]], dtype=torch.float))
    print(f"예측값: {pred_y.item():.2f}")  # → 약 29.5 (≈30)

nn.Module — 사용자 정의 모델 클래스

모델 구조를 클래스로 캡슐화하면 재사용성과 확장성이 크게 향상됩니다.

# nn.Module 상속으로 사용자 정의 모델 만들기
class MyLinearModel(nn.Module):
    def __init__(self):
        super(MyLinearModel, self).__init__()
        self.linear = nn.Linear(1, 1)  # 선형 레이어 정의

    def forward(self, x):
        return self.linear(x)  # 입력 x에 선형 변환 수행

model = MyLinearModel()
optimizer = optim.SGD(model.parameters(), lr=0.01)

for epoch in range(1000):
    pred = model(x_train)
    cost = F.mse_loss(pred, y_train)
    optimizer.zero_grad()
    cost.backward()
    optimizer.step()

# model.linear.weight, model.linear.bias로 학습된 파라미터 접근
print(f"W: {model.linear.weight.item():.3f}, b: {model.linear.bias.item():.3f}")

7-4-3. 다중 선형 회귀 (Stage 5)

독립 변수가 여러 개인 경우: Y = X₁W₁ + X₂W₂ + b. 집의 크기와 방의 개수로 월세를 예측합니다.

기본 텐서 연산으로 구현

import torch
import torch.optim as optim
torch.manual_seed(42)

# 특성: [방 크기(평), 방 개수]  →  타겟: 월세(만원)
x_train = torch.tensor([[10, 1],
                         [15, 2]], dtype=torch.float)
y_train = torch.tensor([[40],
                         [70]], dtype=torch.float)

# 가중치 2개 (특성 2개) + 편향 1개
W = torch.zeros(2, requires_grad=True)  # ← 단순 회귀는 1개, 다중은 특성 수만큼!
b = torch.zeros(1, requires_grad=True)

optimizer = optim.SGD([W, b], lr=0.001)

for epoch in range(10000):
    pred = x_train.mm(W.unsqueeze(1)) + b  # 행렬곱: (2×2)×(2×1) + (1) = (2×1)
    loss = torch.mean((pred - y_train) ** 2)
    optimizer.zero_grad()
    loss.backward()
    optimizer.step()

# 새 집 예측: 20평, 방 3개
with torch.no_grad():
    test_x = torch.tensor([[20, 3]], dtype=torch.float)
    pred_y = test_x.mm(W.unsqueeze(1)) + b
    print(f"예측 월세: {pred_y.item():.1f}만원")  # → 약 97.7만원

nn.Module로 구현

import torch.nn as nn
import torch.nn.functional as F

class LinearRegressionModel(nn.Module):
    def __init__(self):
        super().__init__()
        self.linear = nn.Linear(2, 1)  # in_features=2 (크기,방수), out_features=1 (가격)

    def forward(self, x):
        return self.linear(x)

model = LinearRegressionModel()
optimizer = optim.SGD(model.parameters(), lr=0.001)

for epoch in range(10000):
    pred = model(x_train)
    loss = F.mse_loss(pred, y_train)
    optimizer.zero_grad()
    loss.backward()
    optimizer.step()

7-4-4. MLP 회귀와 DataLoader (Stage 6)

선형 회귀를 넘어 은닉층이 있는 MLP(다층 퍼셉트론)로 회귀 문제를 풀고, DataLoader로 미니 배치 학습을 구현합니다.

import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader, TensorDataset

torch.manual_seed(0)

# 데이터: 4개 특성 → 1개 타겟
input_tensor = torch.tensor([[3, 0.31, 22.6, 11.7],
                              [1, 2.48, 13.5, 7.52],
                              [3, 1.52, 18.9, 17.1]], dtype=torch.float32)
target = torch.tensor([[307], [110], [369]], dtype=torch.float32)

# DataLoader: 미니배치로 데이터 관리
dataset = TensorDataset(input_tensor, target)
data_loader = DataLoader(dataset, batch_size=1, shuffle=True)  # 배치 크기=1, 셔플 활성화

# MLP 모델: 입력(4) → 은닉(5, ReLU) → 출력(1)
class SimpleMLP(nn.Module):
    def __init__(self, input_size):
        super().__init__()
        self.hidden = nn.Linear(input_size, 5)   # 은닉층
        self.relu = nn.ReLU()                     # 활성화 함수
        self.output = nn.Linear(5, 1)             # 출력층

    def forward(self, x):
        x = self.hidden(x)     # 은닉층 통과
        x = self.relu(x)       # ReLU 적용
        x = self.output(x)     # 출력층 통과
        return x

model = SimpleMLP(input_size=4)
criterion = nn.MSELoss()
optimizer = optim.SGD(model.parameters(), lr=0.001)

# 미니 배치 학습 루프
loss_history = []
for epoch in range(1000):
    for inputs, targets in data_loader:       # DataLoader가 배치 단위로 제공
        outputs = model(inputs)
        loss = criterion(outputs, targets)
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()
        loss_history.append(loss.item())

    if (epoch+1) % 200 == 0:
        print(f'Epoch [{epoch+1}/1000], Loss: {loss.item():.4f}')

7-5. 과적합 방지

과적합 = 학습 데이터는 잘 맞추지만 새 데이터에선 실패 (같은 문제집만 10만번 풀면 수능 망하는 것과 같음)

# Early Stopping 구현
best_loss = float("inf")
patience = 10
counter = 0

for epoch in range(300):
    model.train()
    # ... 학습 코드 ...

    # 검증
    model.eval()
    with torch.no_grad():
        val_loss = criterion(model(X_val_t), y_val_t).item()

    if val_loss < best_loss:
        best_loss = val_loss
        counter = 0
        torch.save(model.state_dict(), "best_model.pt")  # 최고 모델 저장
    else:
        counter += 1
        if counter >= patience:
            print(f"Early Stopping at epoch {epoch}")
            break

# Data Augmentation (이미지 전처리)
from torchvision import transforms

train_transform = transforms.Compose([
    transforms.RandomHorizontalFlip(p=0.5),    # 좌우 반전
    transforms.RandomRotation(15),              # ±15도 회전
    transforms.RandomResizedCrop(224, scale=(0.8, 1.0)),
    transforms.ColorJitter(brightness=0.2, contrast=0.2),
    transforms.ToTensor(),
    transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225])
])

test_transform = transforms.Compose([       # 테스트는 증강 X!
    transforms.Resize(256),
    transforms.CenterCrop(224),
    transforms.ToTensor(),
    transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225])
])

7-6. CNN (합성곱 신경망)

이미지 처리에 특화된 신경망입니다. 필터(커널)가 이미지 위를 슬라이딩하며 특징을 추출합니다.

CNN 구조

출력 크기 공식

Output = (Input - Kernel + 2×Padding) / Stride + 1

class SimpleCNN(nn.Module):
    def __init__(self):
        super().__init__()
        self.features = nn.Sequential(
            nn.Conv2d(3, 32, kernel_size=3, padding=1),   # 3채널→32채널
            nn.BatchNorm2d(32), nn.ReLU(), nn.MaxPool2d(2),
            nn.Conv2d(32, 64, kernel_size=3, padding=1),  # 32→64채널
            nn.BatchNorm2d(64), nn.ReLU(), nn.MaxPool2d(2),
        )
        self.classifier = nn.Sequential(
            nn.Flatten(),
            nn.Linear(64*8*8, 256), nn.ReLU(), nn.Dropout(0.5),
            nn.Linear(256, 10)
        )

    def forward(self, x):
        x = self.features(x)       # 특징 추출
        x = self.classifier(x)     # 분류
        return x

# 사용
model = SimpleCNN()
input_img = torch.randn(1, 3, 32, 32)  # (batch, 채널, 높이, 너비)
output = model(input_img)               # (1, 10) → 10개 클래스 확률

Pooling (풀링)

• Max Pooling: 영역 내 최대값 → 가장 강한 특징 유지
• Average Pooling: 영역 내 평균 → 전체적 특징 보존
• Global Average Pooling: 각 채널을 숫자 1개로 압축

7-7. 전이학습 (Transfer Learning)

이미 학습된 모델을 가져다가 내 데이터에 맞게 조금만 추가 학습하는 방법입니다.

from torchvision import models

# 사전학습 모델 로드
model = models.resnet18(weights=models.ResNet18_Weights.IMAGENET1K_V1)

# 모든 파라미터 동결 (기존 학습 유지)
for param in model.parameters():
    param.requires_grad = False

# 마지막 분류층만 교체 (1000클래스 → 5클래스)
model.fc = nn.Linear(model.fc.in_features, 5)

# 전이학습 Fine-tuning 전체 코드
from torchvision import models, datasets, transforms

# 모델 준비
model = models.resnet18(weights=models.ResNet18_Weights.IMAGENET1K_V1)
for param in model.parameters():
    param.requires_grad = False          # 전체 동결

model.fc = nn.Sequential(               # 마지막 층 교체
    nn.Linear(512, 128),
    nn.ReLU(),
    nn.Dropout(0.3),
    nn.Linear(128, 2)                    # 2클래스 (정상/비정상)
)

# 학습할 파라미터만 옵티마이저에 전달
optimizer = torch.optim.Adam(
    filter(lambda p: p.requires_grad, model.parameters()),
    lr=0.001
)

# 데이터 로드 (폴더별 클래스 자동 인식)
# data/train/normal/, data/train/abnormal/
train_data = datasets.ImageFolder("data/train", transform=train_transform)
train_loader = DataLoader(train_data, batch_size=32, shuffle=True)

전략 선택 가이드

7-8. RNN (순환 신경망)

시퀀스 데이터(순서가 중요한 데이터)를 처리합니다: 텍스트, 시계열, 심전도(ECG)

• 은닉 상태(Hidden State): 이전 정보를 다음 시점으로 전달하는 "기억"
• 문제: 기울기 소실 — 긴 시퀀스에서 오래된 정보가 사라짐
• 해결: LSTM(3개 게이트 + 셀 상태로 장기 기억 유지), GRU(LSTM 단순화, 2개 게이트)

# LSTM 분류 모델 (시계열/텍스트용)
class LSTMClassifier(nn.Module):
    def __init__(self, input_size, hidden_size, num_layers, num_classes):
        super().__init__()
        self.lstm = nn.LSTM(input_size, hidden_size, num_layers,
                            batch_first=True, bidirectional=True)
        self.fc = nn.Linear(hidden_size * 2, num_classes)  # *2 (양방향)

    def forward(self, x):
        # x shape: (batch, seq_len, input_size)
        out, (h_n, c_n) = self.lstm(x)
        # out[:, -1, :] = 마지막 시점의 출력
        out = self.fc(out[:, -1, :])
        return out

# 사용 예: 7일간 활력징후 → 위험도 예측
model = LSTMClassifier(input_size=5, hidden_size=64,
                       num_layers=2, num_classes=3)
# input: (batch, 7, 5) → output: (batch, 3)

RNN vs LSTM vs GRU 비교

실전: ECG 분류기 (CNN + LSTM 하이브리드)

1D CNN으로 특징 추출 → LSTM으로 시퀀스 처리. 의료 시계열 분석의 실전 패턴입니다.

class ECGClassifier(nn.Module):
    def __init__(self, num_classes=5):
        super().__init__()
        # 1D CNN: 파형의 지역적 특징 추출
        self.cnn = nn.Sequential(
            nn.Conv1d(1, 32, kernel_size=5, padding=2),
            nn.BatchNorm1d(32), nn.ReLU(), nn.MaxPool1d(2),
            nn.Conv1d(32, 64, kernel_size=5, padding=2),
            nn.BatchNorm1d(64), nn.ReLU(), nn.MaxPool1d(2),
        )
        # LSTM: 시간적 패턴 파악
        self.lstm = nn.LSTM(input_size=64, hidden_size=128,
                            num_layers=2, batch_first=True,
                            bidirectional=True, dropout=0.3)
        self.fc = nn.Sequential(
            nn.Linear(256, 64), nn.ReLU(), nn.Dropout(0.5),
            nn.Linear(64, num_classes)
        )

    def forward(self, x):              # x: (batch, 1, seq_len)
        x = self.cnn(x)                 # (batch, 64, seq_len//4)
        x = x.permute(0, 2, 1)          # (batch, seq_len//4, 64)
        out, _ = self.lstm(x)           # (batch, seq_len//4, 256)
        return self.fc(out[:, -1, :])   # 마지막 시점만 사용

7-9. Transformer & LLM

RNN의 한계(순차처리, 장기의존성)를 Self-Attention으로 해결합니다.

Self-Attention 핵심

Scaled Dot-Product Attention 구현

import math
import torch.nn.functional as F

class ScaledDotProductAttention(nn.Module):
    def __init__(self, d_k):
        super().__init__()
        self.scale = math.sqrt(d_k)

    def forward(self, Q, K, V, mask=None):
        # Q·K^T / √d_k
        scores = torch.matmul(Q, K.transpose(-2, -1)) / self.scale
        if mask is not None:
            scores = scores.masked_fill(mask == 0, -1e9)
        attention_weights = F.softmax(scores, dim=-1)
        output = torch.matmul(attention_weights, V)
        return output, attention_weights

# Multi-Head Attention
class MultiHeadAttention(nn.Module):
    def __init__(self, d_model, num_heads):
        super().__init__()
        self.d_k = d_model // num_heads
        self.num_heads = num_heads
        self.W_q = nn.Linear(d_model, d_model)
        self.W_k = nn.Linear(d_model, d_model)
        self.W_v = nn.Linear(d_model, d_model)
        self.W_o = nn.Linear(d_model, d_model)
        self.attention = ScaledDotProductAttention(self.d_k)

    def forward(self, Q, K, V, mask=None):
        B = Q.size(0)
        # (batch, seq, d_model) → (batch, heads, seq, d_k)
        Q = self.W_q(Q).view(B, -1, self.num_heads, self.d_k).transpose(1, 2)
        K = self.W_k(K).view(B, -1, self.num_heads, self.d_k).transpose(1, 2)
        V = self.W_v(V).view(B, -1, self.num_heads, self.d_k).transpose(1, 2)
        out, weights = self.attention(Q, K, V, mask)
        out = out.transpose(1, 2).contiguous().view(B, -1, self.num_heads * self.d_k)
        return self.W_o(out), weights

# Positional Encoding (위치 정보 주입)
class PositionalEncoding(nn.Module):
    def __init__(self, d_model, max_len=5000):
        super().__init__()
        pe = torch.zeros(max_len, d_model)
        position = torch.arange(0, max_len).unsqueeze(1).float()
        div_term = torch.exp(
            torch.arange(0, d_model, 2).float() * -(math.log(10000.0) / d_model)
        )
        pe[:, 0::2] = torch.sin(position * div_term)  # 짝수: sin
        pe[:, 1::2] = torch.cos(position * div_term)  # 홀수: cos
        self.register_buffer("pe", pe.unsqueeze(0))

    def forward(self, x):
        return x + self.pe[:, :x.size(1)]

BERT vs GPT

7-10. 심화 응용분야

컴퓨터 비전 태스크

U-Net (의료 영상 분할 핵심)

인코더(축소) + 디코더(복원) + Skip Connection으로 세밀한 분할

# U-Net 구현 (의료 영상 분할의 표준 모델)
class DoubleConv(nn.Module):
    """Conv → BN → ReLU × 2"""
    def __init__(self, in_ch, out_ch):
        super().__init__()
        self.conv = nn.Sequential(
            nn.Conv2d(in_ch, out_ch, 3, padding=1),
            nn.BatchNorm2d(out_ch), nn.ReLU(inplace=True),
            nn.Conv2d(out_ch, out_ch, 3, padding=1),
            nn.BatchNorm2d(out_ch), nn.ReLU(inplace=True)
        )
    def forward(self, x): return self.conv(x)

class UNet(nn.Module):
    def __init__(self, in_ch=1, out_ch=1):
        super().__init__()
        # Encoder (축소 경로)
        self.enc1 = DoubleConv(in_ch, 64)
        self.enc2 = DoubleConv(64, 128)
        self.enc3 = DoubleConv(128, 256)
        self.enc4 = DoubleConv(256, 512)
        self.pool = nn.MaxPool2d(2)

        # Bottleneck (병목)
        self.bottleneck = DoubleConv(512, 1024)

        # Decoder (복원 경로) + Skip Connection
        self.up4 = nn.ConvTranspose2d(1024, 512, 2, stride=2)
        self.dec4 = DoubleConv(1024, 512)   # 512(up) + 512(skip) = 1024
        self.up3 = nn.ConvTranspose2d(512, 256, 2, stride=2)
        self.dec3 = DoubleConv(512, 256)
        self.up2 = nn.ConvTranspose2d(256, 128, 2, stride=2)
        self.dec2 = DoubleConv(256, 128)
        self.up1 = nn.ConvTranspose2d(128, 64, 2, stride=2)
        self.dec1 = DoubleConv(128, 64)

        self.out = nn.Conv2d(64, out_ch, 1)  # 1×1 conv

    def forward(self, x):
        e1 = self.enc1(x)                     # Skip 1
        e2 = self.enc2(self.pool(e1))          # Skip 2
        e3 = self.enc3(self.pool(e2))          # Skip 3
        e4 = self.enc4(self.pool(e3))          # Skip 4
        b  = self.bottleneck(self.pool(e4))    # 병목

        d4 = self.dec4(torch.cat([self.up4(b), e4], dim=1))  # concat!
        d3 = self.dec3(torch.cat([self.up3(d4), e3], dim=1))
        d2 = self.dec2(torch.cat([self.up2(d3), e2], dim=1))
        d1 = self.dec1(torch.cat([self.up1(d2), e1], dim=1))
        return self.out(d1)

# Dice Loss (분할 평가에 필수)
class DiceLoss(nn.Module):
    def forward(self, pred, target, smooth=1e-5):
        pred = torch.sigmoid(pred).view(-1)
        target = target.view(-1)
        intersection = (pred * target).sum()
        return 1 - (2 * intersection + smooth) / (pred.sum() + target.sum() + smooth)

GAN & VAE (생성 모델)

# GAN (Generative Adversarial Network)
class Generator(nn.Module):
    def __init__(self, latent_dim=100, img_size=28):
        super().__init__()
        self.model = nn.Sequential(
            nn.Linear(latent_dim, 256),  nn.LeakyReLU(0.2), nn.BatchNorm1d(256),
            nn.Linear(256, 512),         nn.LeakyReLU(0.2), nn.BatchNorm1d(512),
            nn.Linear(512, img_size * img_size), nn.Tanh()   # [-1, 1]
        )
    def forward(self, z):
        return self.model(z).view(-1, 1, 28, 28)

class Discriminator(nn.Module):
    def __init__(self, img_size=28):
        super().__init__()
        self.model = nn.Sequential(
            nn.Flatten(),
            nn.Linear(img_size * img_size, 512), nn.LeakyReLU(0.2), nn.Dropout(0.3),
            nn.Linear(512, 256),                 nn.LeakyReLU(0.2), nn.Dropout(0.3),
            nn.Linear(256, 1),                   nn.Sigmoid()
        )
    def forward(self, img):
        return self.model(img)

# GAN 학습: D(진짜→1, 가짜→0) + G(가짜→1로 속이기)
# VAE (Variational AutoEncoder)
class VAE(nn.Module):
    def __init__(self, input_dim=784, latent_dim=20):
        super().__init__()
        self.encoder = nn.Sequential(nn.Linear(input_dim, 400), nn.ReLU())
        self.fc_mu  = nn.Linear(400, latent_dim)      # 평균
        self.fc_var = nn.Linear(400, latent_dim)       # 분산
        self.decoder = nn.Sequential(
            nn.Linear(latent_dim, 400), nn.ReLU(),
            nn.Linear(400, input_dim),  nn.Sigmoid()
        )

    def reparameterize(self, mu, log_var):
        std = torch.exp(0.5 * log_var)
        return mu + torch.randn_like(std) * std        # 재매개변수화 트릭

    def forward(self, x):
        h = self.encoder(x.view(-1, 784))
        mu, log_var = self.fc_mu(h), self.fc_var(h)
        z = self.reparameterize(mu, log_var)
        return self.decoder(z), mu, log_var

# VAE Loss = 복원 오차 + KL Divergence
def vae_loss(recon_x, x, mu, log_var):
    recon = F.binary_cross_entropy(recon_x, x.view(-1, 784), reduction="sum")
    kl = -0.5 * torch.sum(1 + log_var - mu.pow(2) - log_var.exp())
    return recon + kl

핵심 인사이트: 모델별 데이터 인식 방식

Day 1 — 2D 의료 이미지 분류 (흉부 X-ray)

의료 이미지 분석 시 주의사항

왜 Accuracy만 보면 위험한가

불균형 데이터에서 "다 정상"이라고 찍어도 높게 나오는 것이 Accuracy입니다.

과적합(Overfitting) 이해하기

• 원인: 데이터에만 맞춤 (외우기)
• 대응: 데이터 증강 / Dropout / 정규화(Regularization) / 교차검증

환경 설정 & 시드 고정

재현성을 위해 Python, NumPy, PyTorch 시드를 모두 고정합니다.

import torch, random, numpy as np

SEED = 42

random.seed(SEED)            # 파이썬 기본 랜덤 시드 고정
np.random.seed(SEED)         # NumPy 랜덤 시드 고정
torch.manual_seed(SEED)      # PyTorch CPU 시드 고정

if torch.cuda.is_available():  # 토치에서 GPU를 사용할 수 있는지 확인 → YES or NO
    torch.cuda.manual_seed_all(SEED)  # 사용 가능한 모든 GPU의 시드를 42로 고정

print("torch:", torch.__version__)         # __(언더바 2개) → torch 버전 확인
print("cuda available:", torch.cuda.is_available())  # GPU 사용 가능 여부
if torch.cuda.is_available():
    print("gpu:", torch.cuda.get_device_name(0))  # 0번째 GPU 장치 이름 출력

데이터셋 탐색 (Chest X-ray)

폐렴(PNEUMONIA) vs 정상(NORMAL) 이진 분류 데이터셋입니다.

import os                    # 파일/폴더 경로를 파이썬 명령어로 다루기 위한 라이브러리
from glob import glob        # 특정 폴더 안의 파일들을 한번에 찾기

ROOT = "/content/drive/MyDrive/.../chest_xray/chest_xray"
# 학습데이터의 위치를 지정 / 폴더 옆 점3개 → 경로 복사하는게 가장 안전

splits = ["train", "val", "test"]  # List → ""로 감싸진 데이터 = 문자형
classes = ["NORMAL", "PNEUMONIA"]  # 폴더에 있는 이름을 찾기 위한 코드

for sp in splits:                  # splits 리스트를 하나씩 반복
    print(f"\n[{sp}]")             # 1번째 반복: "train"
    for c in classes:              # classes 리스트를 하나씩 반복
        n = len(glob(os.path.join(ROOT, sp, c, '*')))  # '*' = 모두, 몽땅, all
        print(f" {c} : {n}")       # 해당 클래스의 파일 개수 출력

이미지 시각화

import matplotlib.pyplot as plt  # 별칭은 plt로 줄여서 사용
from PIL import Image            # 이미지 파일을 열고 다룰 수 있는 PIL 라이브러리 (첨부파일 열기)

# show_samples 함수 만들기 : x-ray 사진을 2행4열로 보여주기 위한 기능 구현
def show_samples(splits="train", cls="PNEUMONIA", n=8):  # 함수 실행시 사용될 기본 데이터 설정
    path = glob(os.path.join(ROOT, splits, cls, '*'))[:n]  # 인덱스는 0부터 시작 → n-1까지
    plt.figure(figsize=(12, 6))    # 가로 12 / 세로 6 크기의 화면 만들기
    for i, p in enumerate(path):   # i에 숫자를 반복마다 하나씩 증가, path가 p에 들어감
        img = Image.open(p).convert('RGB')  # 흑백은 알아서 흑백, 컬러가 있다면 컬러로 표시
        plt.subplot(2, 4, i+1)     # 2행 4열, 인덱스를 1부터 (i+1)
        plt.imshow(img)
        plt.axis("off")            # 좌표 눈금이나 선을 제거 → 이미지만 깔끔하게 표시
    plt.suptitle(f"{splits} / {cls}")  # 그래프 제목
    plt.show()                     # 위에 설정한대로 보여줘

show_samples("train", "PNEUMONIA")  # train에 있는 폐렴 환자 데이터를 2행 4열로 시각화
show_samples("train", "NORMAL")

Day 3 — 3D 의료영상 분할 (MONAI + CT)

Medical Imaging AI란?

X-ray, CT, MRI, 초음파처럼 의료 목적으로 촬영된 영상을 인공지능이 분석하는 분야입니다. 일반 이미지 AI와 달리 해부학적 구조, 병변, 조직 변화 등 훨씬 정밀한 대상을 다루며, "어디에 있는지", "얼마나 큰지"까지 함께 분석합니다.

Classification vs Segmentation

왜 의료영상에 MONAI를 쓰는가?

MONAI (Medical Open Network for AI)는 PyTorch 기반의 의료영상 전용 딥러닝 프레임워크입니다.

• 일반 라이브러리와 달리 NIfTI 형식, voxel spacing, orientation, affine 등 의료영상 특유 정보를 직접 처리
• 단순 픽셀 배열이 아닌, 공간적 정보와 구조까지 고려하는 Metadata-aware Processing
• segmentation, detection, classification 등 의료영상 문제에 최적화된 transforms, 네트워크, metric 제공

의료영상 파일 형식 — DICOM vs NIfTI

주요 라이브러리

!pip install -U monai nibabel matplotlib tqdm
# -U : 최신 버전으로 업그레이드해서 설치해줘

NIfTI 파일 구조 이해

CT/MRI 데이터는 일반 이미지(PNG, JPEG)와 다릅니다. .nii.gz 형식의 3D 볼륨 데이터입니다.

import nibabel as nib

img = nib.load(train_images[0])
# png, jpeg와 달리 컴퓨터가 기본 제공하는 형식이 아니라 그냥 열리지 않음
# nib.load(...)는 NIfTI 파일을 열어주는 함수
# train_images[0]는 목록 중 첫 번째 파일
# → 첫 번째 CT파일 하나를 열어서 img 변수에 담는다는 뜻

arr = img.get_fdata()
# img는 아직 의료영상 파일 객체
# get_fdata()는 그 안에 들어 있는 실제 영상 숫자 데이터를 꺼내오는 함수

print("원본 shape:", arr.shape)
# 결과: (512, 512, 55) → 높이, 너비, 깊이(슬라이스 수)
# 512×512 크기의 이미지가 55장 쌓여있다는 뜻

print("affine:\n", img.affine)
# 의료영상의 좌표변환 정보. 보통 4×4 행렬로 출력
# 배열 안의 위치가 실제 공간에서 어디를 의미하는지 알려줌
# CT라도 방향이 다르거나 위치 해석이 다를 수 있어서 affine 정보가 필수
# → 이 숫자 배열이 실제 몸 안의 어느 방향과 위치를 의미하는지 알려주는 표

print("voxel size (pixdim):", img.header["pixdim"])
# img.header는 의료영상 파일의 설명서 같은 정보
# ["pixdim"]은 voxel 간격 정보
# 각 축에서 voxel 하나가 실제로 얼마나 떨어져 있는지 (mm 단위)
# CT 데이터는 배열 크기만 같아도 실제 몸에서의 물리적 크기는 다를 수 있음
# 예: 어떤 데이터는 1mm 간격, 어떤 데이터는 5mm 간격
# → 영상 안에 있는 한 칸(voxel)이 실제로 얼마나 큰지 알려주는 정보

3D 핵심 개념

1. Voxel (Volumetric Pixel)

Voxel은 3차원 데이터 안의 최소 단위입니다. 2D 이미지의 pixel이 평면 위 한 점을 나타낸다면, voxel은 공간 속 작은 부피 단위를 나타냅니다. CT나 MRI에서는 이 Voxel 하나하나에 조직 밀도, 신호 강도 같은 값이 들어갑니다. 따라서 3D 의료영상은 Voxel의 집합으로 구성된 입체 구조라고 볼 수 있습니다.

2. Affine Matrix (아핀 행렬)

회전, 이동, 스케일 변화 등을 포함한 선형 변환 행렬입니다. 의료영상에서는 배열 인덱스를 실제 좌표계와 연결하는 역할을 합니다. 즉, 단순히 "몇 번째 Voxel인가"가 아니라 "실제 몸 안에서 어느 위치인가"를 알려주는 수학적 도구입니다. 서로 다른 영상이나 라벨을 정렬할 때 affine 정보는 매우 중요합니다.

3. Header Metadata

영상 데이터에 대한 구조적 정보를 담고 있는 메타데이터입니다. 차원수, 각 축 길이, voxel 크기, 좌표계 관련 정보 등이 들어갑니다. 시각화, resampling, orientation 정렬, spacing 보정 같은 작업의 기준이 됩니다. 따라서 header metadata는 단순 부가 정보가 아니라 의료영상 분석의 기준점 역할을 합니다.

CT 슬라이스 시각화

z_mid = arr.shape[2] // 2
# arr.shape[2]는 세 번째 축의 길이 = 슬라이스 총 개수
# // 2는 2로 나눈 몫 = 가운데 인덱스

plt.figure(figsize=(5, 5))
# 새로운 그림판을 만듭니다. 가로 5, 세로 5 크기

plt.imshow(arr[:,:,z_mid].T, cmap="gray", origin="lower")
# arr[:,:,z_mid] : 첫번째(:)=높이, 두번째(:)=너비, z_mid=가운데 슬라이스
# .T : 전치(transpose) → 행과 열을 바꿔줌 → 이미지를 보기 좋게 돌리는 역할
#       의료영상은 축 방향 때문에 그냥 출력하면 눕거나 뒤집혀 보여서 .T를 자주 사용
# cmap="gray" : CT는 보통 흑백이므로 회색조로 표시
# origin="lower" : 이미지의 원점을 아래쪽으로 → 아래에서 위로 보는 느낌

plt.title("Original Axial slice")
plt.axis("off")  # x축, y축 눈금과 테두리 숫자를 없앰
plt.show()

3방향 시점 (Axial / Coronal / Sagittal)

3D 의료영상은 한 번에 볼 수 없으므로, 세 가지 방향으로 잘라서 봅니다.

Slice는 3D volume에서 특정 위치를 잘라낸 단면 한 장을 뜻합니다. 3D 전체를 한 번에 보기 어렵기 때문에 slice를 연속적으로 넘기며 구조를 파악합니다. 실습에서는 보통 가운데 slice나 label이 많이 포함된 slice를 선택해 시각화합니다.

MONAI 전처리 파이프라인

원본 CT를 모델에 넣기 전, 방향·크기·밝기를 정리하는 과정입니다.

# 이 코드는 의료영상 데이터를 모델이 사용하기 좋은 형태로 바꾸는 전처리 파이프라인입니다.
# 원본 CT 파일을 바로 모델에 넣지 않고, 방향·크기·밝기 등을 정리해서
# 학습하기 좋은 상태로 바꾸는 과정입니다.

from monai.transforms import (
    Compose,                  # 여러 전처리 단계를 순서대로 묶는 도구
                              # "1번 하고 → 2번 하고 → 3번 하기" 처럼 연결
    LoadImaged,               # 이미지와 라벨 파일을 읽어오는 기능
    EnsureChannelFirstd,      # 채널 축을 맨 앞으로 맞춰주는 기능
                              # 딥러닝 모델이 기대하는 입력 형태로 변환
    Orientationd,             # 영상 방향을 일정한 기준으로 맞춰주는 기능
    Spacingd,                 # voxel 간격을 일정하게 맞춰주는 기능
    ScaleIntensityRanged,     # 영상 밝기값 범위를 일정하게 조정하는 기능
    CropForegroundd,          # 빈 배경을 잘라내고, 실제 몸이 있는 부분만 남기는 기능
    RandCropByPosNegLabeld,   # 장기 있는 patch와 없는 patch를 적절히 섞어서 잘라주는 기능
    # batch : 모델이 한 번에 처리하는 샘플 묶음
    # patch : 큰 이미지나 3D 볼륨에서 잘라낸 작은 조각
    # X-ray 이미지 한 장 = 샘플 1개, CT는 너무 커서 Patch로 잘라 씀
    EnsureTyped,              # 데이터를 MONAI/PyTorch가 다룰 수 있는 형식으로 변환
    DivisiblePadd,            # 이미지 크기를 특정 숫자로 나누어 떨어지게 padding
                              # → 모델에서 shape 오류가 안 나도록 도움
)

# base_transforms 전처리 파이프라인 변수 선언
# Compose([....]) 안에 들어 있는 작업들을 순서대로 실행
base_transforms = Compose([

    LoadImaged(keys=["image", "label"]),
    # image와 label 파일을 불러오는 단계

    EnsureChannelFirstd(keys=["image", "label"]),
    # 채널 축을 맨 앞으로 오게 맞춤
    # 딥러닝 모델은 보통 (채널, 높이, 너비) 또는 (채널, 높이, 너비, 깊이) 형태를 기대
    # 의료영상은 원래 채널 축이 없거나 뒤에 있는 경우가 있어서 위치를 맞춰줌

    Orientationd(keys=["image", "label"], axcodes="RAS"),
    # 영상의 방향을 일정한 기준으로 맞추는 단계
    # axcodes="RAS" → R=Right, A=Anterior, S=Superior
    # 오른쪽, 앞쪽, 위쪽 방향을 기준으로 축 정렬

    Spacingd(keys=["image", "label"],
             pixdim=(1.5, 1.5, 2.0),       # x축, y축, z축 간격 (mm)
             mode=("bilinear", "nearest")), # image는 bilinear, label은 nearest

    ScaleIntensityRanged(keys=["image"],
                         a_min=-57, a_max=164,   # CT 원본 밝기 범위
                         b_min=0.0, b_max=1.0,   # 변환 후 범위
                         clip=True),
    # CT 밝기값을 정리하는 단계
    # CT는 밝기값 범위가 매우 넓지만 비장 같은 복부 장기에는 모든 범위가 필요 없음
    # 너무 낮은 값 = 공기/바깥 배경, 너무 높은 값 = 뼈/강한 밀도 구조
    # → 비장을 보기 좋은 CT 값 범위를 정해서 모델이 그 범위만 보도록 함

    CropForegroundd(keys=["image", "label"], source_key="image"),
    # 실제 의미 있는 부분만 남기고 바깥의 빈 배경을 잘라내는 작업
    # source_key="image"를 기준으로 foreground를 찾음

    EnsureTyped(keys=["image", "label"]),
    # image와 label을 MONAI/PyTorch가 다룰 수 있는 형식으로 변환
    # 보통 텐서 형태나 MONAI typed object로 정리
])

# 전체 흐름 요약:
# 파일 읽고 → 채널 축 맞추고 → 방향 맞추고 → spacing 맞추고
# → 밝기값 정리하고 → 배경 자르고 → 최종 형식 맞추기

의료영상 전처리가 특히 중요한 이유

의료영상은 병원과 장비에 따라 저장 방식이 조금씩 다릅니다. 같은 장기라도:

• 촬영 방향이 다를 수 있음 → Orientationd로 RAS 기준 정렬
• voxel 간격이 다를 수 있음 → Spacingd로 통일 (예: 1.5mm × 1.5mm × 2.0mm)
• intensity 값 범위가 다를 수 있음 → ScaleIntensityRanged로 정규화

Patch Transform (학습용 랜덤 크롭)

base_transforms는 검증/추론용이고, 학습에는 랜덤 패치 크롭을 추가한 patch_transforms를 사용합니다.

# patch_transforms: base_transforms + 랜덤 크롭
# 학습 시 3D 볼륨 전체를 넣으면 메모리 부족 → 작은 patch로 잘라서 학습

patch_transforms = Compose([
    # --- base_transforms와 동일한 부분 ---
    LoadImaged(keys=["image", "label"]),
    EnsureChannelFirstd(keys=["image", "label"]),
    Orientationd(keys=["image", "label"], axcodes="RAS"),
    Spacingd(keys=["image", "label"],
             pixdim=(1.5, 1.5, 2.0),
             mode=("bilinear", "nearest")),
    ScaleIntensityRanged(keys=["image"],
                         a_min=-57, a_max=164,
                         b_min=0.0, b_max=1.0, clip=True),
    CropForegroundd(keys=["image", "label"], source_key="image"),

    # --- 여기서부터 학습 전용 추가 ---
    RandCropByPosNegLabeld(
        keys=["image", "label"],
        label_key="label",           # 정답 라벨 기준으로 자름
        spatial_size=(96, 96, 96),   # 96×96×96 크기의 patch
        pos=1,                       # 장기가 있는 patch 비율
        neg=1,                       # 장기가 없는 patch 비율 (1:1로 섞음)
        num_samples=4,               # 한 볼륨에서 4개의 patch 추출
        image_key="image",
        image_threshold=0,
    ),
    EnsureTyped(keys=["image", "label"]),
])

# patch용 CacheDataset & DataLoader
patch_ds = CacheDataset(data=train_files,
                        transform=patch_transforms,
                        cache_rate=1.0, num_workers=0)

patch_loader = DataLoader(patch_ds, batch_size=1,
                          shuffle=True, num_workers=0)

# Day 4 학습에서 실제 사용할 train_loader
train_loader = patch_loader

sample_patch = next(iter(patch_loader))
print("patch image shape:", sample_patch["image"].shape)
print("patch label shape:", sample_patch["label"].shape)
# → [1, 1, 96, 96, 96] : 96×96×96 크기의 patch

CacheDataset & DataLoader

MONAI의 CacheDataset은 전처리 결과를 메모리에 캐싱하여 학습 속도를 높입니다.

# 이미지와 라벨 파일을 쌍으로 묶기
train_images = sorted(glob.glob(os.path.join(data_dir, "imagesTr", "*.nii.gz")))
# imagesTr 폴더 안에 있는 .nii.gz 파일들을 전부 찾음
# glob.glob(...) → 조건에 맞는 파일 목록을 가져옴
# sorted(...) → 파일 이름 순서대로 정렬
train_labels = sorted(glob.glob(os.path.join(data_dir, "labelsTr", "*.nii.gz")))
# label은 각 CT 이미지에 대응되는 정답 라벨 파일 목록

data_dicts = [
    {"image": img, "label": lbl}
    for img, lbl in zip(train_images, train_labels)
]
# [ ] → 리스트 만드는 코드 / { } → 딕셔너리 만드는 코드
# zip(train_images, train_labels) : 이미지와 라벨을 한 쌍씩 묶어주는 코드
# 결과 예시:
# [
#   {"image": "spleen_1.nii.gz", "label": "spleen_1.nii.gz"},
#   {"image": "spleen_2.nii.gz", "label": "spleen_2.nii.gz"},
#   ...
# ]

# 학습용 / 검증용 데이터 나누기
train_files = data_dicts[:-9]  # 뒤에서 9개를 제외한 나머지 → 학습용
val_files   = data_dicts[-9:]  # 마지막 9개 → 검증용

device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
# GPU가 있으면 GPU를 쓰고, 없으면 CPU를 쓰겠다는 뜻

# --- CacheDataset & DataLoader ---
from monai.data import CacheDataset, DataLoader

# 전처리된 데이터를 실제로 모델이 사용할 수 있는 형태로 바꾸는 단계
train_ds = CacheDataset(data=train_files,
                        transform=base_transforms,
                        cache_rate=1.0,   # 전처리 결과를 100% 메모리에 캐싱
                        num_workers=0)

val_ds = CacheDataset(data=val_files,
                      transform=val_transforms,  # DivisiblePadd(k=16) 추가된 버전
                      cache_rate=1.0, num_workers=0)
# 검증용에는 DivisiblePadd 추가 → Day4 UNet validation/inference 연결을 위해

train_loader = DataLoader(train_ds, batch_size=1, shuffle=True, num_workers=0)
val_loader   = DataLoader(val_ds,   batch_size=1, shuffle=False, num_workers=0)

sample = next(iter(train_loader))
print("image shape:", sample["image"].shape)
# 결과: torch.Size([1, 1, 239, 207, 98])
# → [batch, channel, height, width, depth]

실습 데이터셋: Task09_Spleen (비장 분할)

2D 분류 vs 3D 분할 비교

Dice Loss — 의료 영상 분할의 핵심 손실함수 ⭐

왜 BCE/CE만으로는 부족한가?

의료 영상은 클래스 불균형이 극심합니다. 예를 들어:

Dice의 핵심 아이디어: "겹치는 정도"를 목표로 학습

"예측한 병변 영역"과 "정답 병변 영역"이 얼마나 많이 겹치냐를 측정합니다.

• 분자: 겹친 픽셀 (진짜로 맞춘 병변)
• 분모: 예측 병변 크기 + 정답 병변 크기
• → 병변이 작아도 "겹침"을 직접 키우는 방향으로 학습됨

Dice Score vs Dice Loss

Dice Score는 클수록 좋은 지표(1.0 = 완벽 일치)입니다. 그런데 손실함수는 작을수록 좋아야 하므로:

Dice Score = 2 × |A ∩ B| / (|A| + |B|)    # 0~1, 클수록 좋음

Dice Loss  = 1 - Dice Score                # 0~1, 작을수록 좋음

Dice Loss는 미분이 가능한가?

집합 연산처럼 보이지만, 실제로는 전부 덧셈/곱셈/나눗셈이라 미분 가능합니다:

# pred는 sigmoid를 통과한 확률값 (0~1)
# target(y)은 정답 마스크 (0 또는 1)

intersection = Σ (p_i × y_i)     # 교집합: 예측 확률 × 정답을 곱해서 합산
sum_p        = Σ p_i             # 예측 영역 크기
sum_y        = Σ y_i             # 정답 영역 크기 (상수)

Dice  = 2 × intersection / (sum_p + sum_y + smooth)
# smooth = 1e-5 등 작은 값 → 0 나누기 방지

Loss  = 1 - Dice                # ← return 값

Dice Score 계산 예시

정답 비장 픽셀 100개, 모델이 60개만 비장이라고 예측한 경우:

Dice Score = 2 × intersection / (pre_sum + label_sum)
           = 2 × 60 / (60 + 100)
           = 120 / 160
           = 0.75

BCE와 Dice Loss 비교

Day 3 (2) — 3D U-Net 학습 파이프라인

전처리가 끝난 데이터를 실제로 3D U-Net 모델에 넣어 학습하고, 검증하고, 결과를 시각화하는 전체 과정입니다.

3D U-Net 모델 구성

MONAI가 제공하는 UNet 클래스로 3D segmentation 모델을 만듭니다.

from monai.networks.nets import UNet
from monai.networks.layers import Norm

model = UNet(
    spatial_dims=3,           # 3D 데이터 처리
    in_channels=1,            # CT 흑백 = 채널 1개
    out_channels=2,           # 배경(0) + 비장(1) = 2클래스
    channels=(16, 32, 64, 128, 256),  # 각 단계 feature map 수
    strides=(2, 2, 2, 2),    # 매 단계 크기 절반 축소
    num_res_units=2,          # Residual 연결 (원래 정보를 뒤에 다시 더함)
    norm=Norm.BATCH           # Batch Normalization
).to(device)

# 입출력 shape 확인
sample = next(iter(patch_loader))
x = sample["image"].to(device)

with torch.no_grad():
    y = model(x)

print("입력 shape:", x.shape)  # [4, 1, 96, 96, 96]
print("출력 shape:", y.shape)  # [4, 2, 96, 96, 96]
# 4개 patch / 클래스 채널 2개 (배경, 비장) / 공간 크기 96×96×96

손실함수 · 옵티마이저 · 평가 지표

from monai.losses import DiceLoss
from monai.metrics import DiceMetric
from monai.transforms import AsDiscrete

# 손실함수: 겹침이 좋으면 loss 감소, 나쁘면 loss 증가
loss_function = DiceLoss(to_onehot_y=True, softmax=True)
# to_onehot_y=True → 정답(label)을 one-hot 형식으로 변환
# softmax=True → 모델 출력을 확률로 바꿈 (배경:0.8, 비장:0.2 식으로)

# 옵티마이저: 모델 가중치를 수정하는 도구
optimizer = torch.optim.Adam(model.parameters(), lr=1e-4)

# 평가 지표: Dice 점수로 검증 성능 측정
dice_metric = DiceMetric(include_background=False, reduction="mean")
# include_background=False → 배경은 평가에서 제외

# 후처리: 예측 결과를 최종 비교용 형태로 변환
post_pred  = AsDiscrete(argmax=True, to_onehot=2)
# argmax=True → 가장 큰 점수의 클래스 선택
# to_onehot=2 → 2채널 one-hot 형태로 변환
post_label = AsDiscrete(to_onehot=2)

학습 함수 — train_one_epoch()

모델을 학습 데이터 한 바퀴(1 epoch)만큼 학습시키는 함수입니다.

def train_one_epoch(model, loader, loss_function, optimizer, device):
    model.train()          # 학습 모드로 전환
    epoch_loss = 0.0       # 이번 epoch의 loss를 누적할 변수
    step = 0               # batch 처리 횟수 카운터

    for batch_data in tqdm(loader, desc="Training"):
        step += 1
        inputs = batch_data["image"].to(device)
        labels = batch_data["label"].to(device)

        optimizer.zero_grad()    # 이전 batch의 gradient 초기화
        outputs = model(inputs)  # 모델 예측
        loss = loss_function(outputs, labels)  # 손실 계산
        loss.backward()          # 오차 역전파 (gradient 계산)
        optimizer.step()         # 가중치 실제 수정

        epoch_loss += loss.item()

    epoch_loss /= max(step, 1)   # 평균 loss 반환
    return epoch_loss

Sliding Window Inference (슬라이딩 윈도우 추론)

3D 의료영상은 전체 볼륨이 너무 커서 한 번에 모델에 넣을 수 없습니다. 작은 3D 창(window)으로 나누어 예측한 뒤 다시 합칩니다.

from monai.inferers import sliding_window_inference

roi_size = (160, 160, 160)   # 한 번에 볼 3D 창 크기
sw_batch_size = 1            # 창을 한 번에 1개씩 모델에 넣음 (메모리 절약)

def inferer(input, model):
    return sliding_window_inference(
        inputs=input,
        roi_size=roi_size,
        sw_batch_size=sw_batch_size,
        predictor=model,
    )

검증 함수 — validate_one_epoch()

학습은 하지 않고, 검증용 데이터를 넣어서 Dice 점수로 모델 성능을 평가합니다.

from monai.data import decollate_batch

def validate_one_epoch(model, loader, device, inferer):
    model.eval()           # 평가 모드 전환
    dice_metric.reset()    # 이전 결과 초기화

    with torch.no_grad():  # gradient 계산 안 함 (학습 X)
        for val_data in tqdm(loader, desc="Validation"):
            val_inputs  = val_data["image"].to(device)
            val_labels  = val_data["label"].to(device)

            val_outputs = inferer(val_inputs, model)
            # 큰 볼륨을 sliding window로 나누어 예측

            # 예측 결과를 최종 비교용 형태로 변환
            val_outputs_list = [post_pred(i) for i in decollate_batch(val_outputs)]
            val_labels_list  = [post_label(i) for i in decollate_batch(val_labels)]
            # decollate_batch: batch를 개별 샘플로 분리
            # post_pred: argmax → one-hot 변환
            # post_label: one-hot 변환

            dice_metric(y_pred=val_outputs_list, y=val_labels_list)

    metric = dice_metric.aggregate().item()
    # 누적된 결과를 합쳐서 최종 Dice 점수 산출
    return metric

전체 학습 실행

train_one_epoch + validate_one_epoch를 반복하며, 최고 성능 모델을 저장합니다.

best_metric = -1
best_metric_epoch = -1
best_model_path = os.path.join(root_dir, "best_metric_model.pth")

epoch_loss_values = []   # loss 변화 기록
metric_values = []       # Dice 점수 변화 기록
max_epochs = 5           # 수업용 5 epoch

for epoch in range(max_epochs):
    print(f"\nEpoch {epoch + 1}/{max_epochs}")

    # 1) 학습
    train_loss = train_one_epoch(model, train_loader,
                                 loss_function, optimizer, device)
    epoch_loss_values.append(train_loss)

    # 2) 검증
    val_dice = validate_one_epoch(model, val_loader, device, inferer)
    metric_values.append(val_dice)

    print(f"loss={train_loss:.4f}, val_dice={val_dice:.4f}")

    # 3) 최고 성능이면 모델 저장
    if val_dice > best_metric:
        best_metric = val_dice
        best_metric_epoch = epoch + 1
        torch.save(model.state_dict(), best_model_path)
        # state_dict() = 모델의 학습된 가중치를 딕셔너리로 저장
        print("최고 성능 모델 저장 완료!")

5 Epoch 학습 결과 예시 (Task09_Spleen):

결과 시각화

학습이 끝나면 최고 성능 모델을 로드하고, 검증 데이터에 대한 예측 결과를 시각화합니다.

# 최고 성능 모델 로드
model.load_state_dict(torch.load(best_model_path, map_location=device))
model.eval()

with torch.no_grad():
    val_example = next(iter(val_loader))
    val_inputs  = val_example["image"].to(device)
    val_labels  = val_example["label"].to(device)
    val_outputs = inferer(val_inputs, model)
    pred_mask   = torch.argmax(val_outputs, dim=1)  # 가장 높은 점수 클래스 선택

# numpy로 변환
image_np = val_inputs[0, 0].cpu().numpy()
label_np = val_labels[0, 0].cpu().numpy()
pred_np  = pred_mask[0].cpu().numpy()

# 비장이 가장 많이 보이는 slice 찾기
slice_sums = [label_np[:, :, i].sum() for i in range(label_np.shape[2])]
best_slice = int(np.argmax(slice_sums))

# 3장 비교: 원본 / 정답 / 예측
plt.figure(figsize=(15, 5))
for i, (data, title) in enumerate([
    (image_np[:,:,best_slice], "Image"),
    (label_np[:,:,best_slice], "Label"),
    (pred_np[:,:,best_slice],  "Prediction"),
]):
    plt.subplot(1, 3, i+1)
    plt.imshow(data, cmap="gray")
    plt.title(title)
    plt.axis("off")
plt.show()

# 오버레이: 원본 위에 예측 마스크 겹치기
plt.figure(figsize=(6, 6))
plt.imshow(image_np[:,:,best_slice], cmap="gray")
plt.imshow(pred_np[:,:,best_slice], cmap="Reds", alpha=0.35)
plt.title("Overlay")
plt.axis("off")
plt.show()

학습 곡선 시각화

plt.figure(figsize=(12, 5))

plt.subplot(1, 2, 1)
plt.title("Epoch Average Loss")
plt.plot(range(1, len(epoch_loss_values)+1), epoch_loss_values, marker="o")
plt.xlabel("Epoch")
plt.ylabel("Loss")

plt.subplot(1, 2, 2)
plt.title("Validation Dice")
plt.plot(range(1, len(metric_values)+1), metric_values, marker="o")
plt.xlabel("Epoch")
plt.ylabel("Dice")

plt.tight_layout()
plt.show()

전체 파이프라인 요약

1. 데이터 준비      → NIfTI 파일 로드, 학습/검증 분리
2. 전처리           → Orientation, Spacing, Intensity, Crop, Patch
3. 모델 생성        → 3D UNet (MONAI)
4. 손실함수/옵티마이저 → DiceLoss + Adam
5. 학습 (train)     → forward → loss → backward → step
6. 검증 (validate)  → sliding window inference → Dice Score
7. 모델 저장        → best Dice 기준으로 .pth 파일 저장
8. 시각화           → Image / Label / Prediction 비교

Day 5 — 오류 분석 · 후처리 · 모델 개선

학습된 3D U-Net 모델의 예측 결과를 분석하고, 후처리(post-processing)를 통해 성능을 개선하는 과정입니다.

오류 분석의 3단계

Dice Score 기반 케이스 분석

Dice Score는 예측 마스크와 정답 마스크의 겹침 정도를 0~1로 나타낸 값입니다. 1에 가까울수록 잘 맞고, 0에 가까울수록 거의 안 맞습니다.

import numpy as np

def binary_dice(pred_np, label_np, target_class=1):
    pred_bin = (pred_np == target_class)
    label_bin = (label_np == target_class)

    intersection = np.sum(pred_bin & label_bin)  # 둘 다 비장인 위치
    pred_sum = np.sum(pred_bin)                   # 예측에서 비장 영역 크기
    label_sum = np.sum(label_bin)                 # 정답에서 비장 영역 크기

    if pred_sum + label_sum == 0:
        return 1.0  # 둘 다 없으면 완전 일치
    return 2.0 * intersection / (pred_sum + label_sum)
# Dice 공식: 2 × 겹치는 영역 / (예측 크기 + 정답 크기)

Validation 전체 샘플 Dice 비교

검증 데이터 전체를 순회하며 각 샘플의 Dice를 계산하고, worst/best case를 찾습니다.

case_results = []
model.eval()

with torch.no_grad():
    for idx, val_data in enumerate(val_loader):
        val_inputs = val_data["image"].to(device)
        val_labels = val_data["label"].to(device)

        val_outputs = inferer(val_inputs, model)
        pred_mask = torch.argmax(val_outputs, dim=1)

        image_np = val_inputs[0, 0].cpu().numpy()
        label_np = val_labels[0, 0].cpu().numpy()
        pred_np = pred_mask[0].cpu().numpy()

        dice = binary_dice(pred_np, label_np, target_class=1)
        case_results.append({"idx": idx, "dice": dice, "image": image_np,
                             "label": label_np, "pred": pred_np})

# Dice 기준 정렬 → worst/best 추출
case_results = sorted(case_results, key=lambda x: x["dice"])
worst_case = case_results[0]   # 가장 낮은 Dice
best_case = case_results[-1]   # 가장 높은 Dice

Patch 단위 Dice 분석

학습에 사용된 patch_loader의 모든 patch를 하나씩 검사해서 Dice 분포를 확인합니다.

patch_batch_results = []
model.eval()

with torch.no_grad():
    for batch_idx, batch_data in enumerate(patch_loader):
        inputs = batch_data["image"].to(device)
        labels = batch_data["label"].to(device)
        outputs = model(inputs)
        pred_mask = torch.argmax(outputs, dim=1)

        for patch_idx in range(inputs.shape[0]):
            image_np = inputs[patch_idx, 0].cpu().numpy()
            label_np = labels[patch_idx, 0].cpu().numpy()
            pred_np = pred_mask[patch_idx].cpu().numpy()
            dice = binary_dice(pred_np, label_np, target_class=1)
            patch_batch_results.append({
                "batch_idx": batch_idx, "patch_idx": patch_idx,
                "dice": dice, "image": image_np,
                "label": label_np, "pred": pred_np,
            })

patch_batch_results = sorted(patch_batch_results, key=lambda x: x["dice"])

후처리: 최대 연결 영역 추출 (Largest Connected Component)

모델 예측에는 작은 노이즈 덩어리가 섞일 수 있습니다. 가장 큰 연결 영역만 남기면 Dice가 크게 향상됩니다.

from scipy import ndimage

def keep_largest_component(binary_mask):
    labeled, num_features = ndimage.label(binary_mask)
    # labeled: 각 덩어리에 번호를 붙인 배열
    # num_features: 덩어리 총 개수

    if num_features == 0:
        return binary_mask  # 덩어리가 없으면 그대로 반환

    sizes = ndimage.sum(binary_mask, labeled, range(1, num_features + 1))
    # 각 덩어리 크기 계산 (리스트 형태, 예: [12, 245, 16])
    largest_label = np.argmax(sizes) + 1  # 가장 큰 덩어리 번호 (+1: 번호가 1부터 시작)
    largest_component = (labeled == largest_label).astype(np.uint8)
    return largest_component

# 후처리 적용 예시 (worst case)
worst_pred_bin = (worst_case["pred"] == 1).astype(np.uint8)
worst_pred_post = keep_largest_component(worst_pred_bin)

dice_before = binary_dice(worst_case["pred"], worst_case["label"], target_class=1)
dice_after = binary_dice(worst_pred_post, worst_case["label"], target_class=1)
# 예: 후처리 전 0.014 → 후처리 후 0.611 (대폭 개선)

시각화: 후처리 전후 비교

비장 영역이 가장 많은 slice를 찾아서 Original / Label / Prediction (전후)를 비교합니다.

def get_best_slice_from_label(label_np):
    slice_sums = [label_np[:, :, i].sum() for i in range(label_np.shape[2])]
    return int(np.argmax(slice_sums))

best_slice = get_best_slice_from_label(worst_case["label"])

plt.figure(figsize=(16, 8))
plt.subplot(2, 3, 1); plt.title("Original")
plt.imshow(worst_case["image"][:, :, best_slice], cmap="gray"); plt.axis("off")
plt.subplot(2, 3, 2); plt.title("Label")
plt.imshow(worst_case["label"][:, :, best_slice], cmap="gray"); plt.axis("off")
plt.subplot(2, 3, 3); plt.title(f"Before Post\nDice={dice_before:.4f}")
plt.imshow(worst_case["pred"][:, :, best_slice], cmap="gray"); plt.axis("off")
plt.subplot(2, 3, 5); plt.title(f"After Post\nDice={dice_after:.4f}")
plt.imshow(worst_pred_post[:, :, best_slice], cmap="gray"); plt.axis("off")
plt.subplot(2, 3, 6); plt.title("Overlay After")
plt.imshow(worst_case["image"][:, :, best_slice], cmap="gray")
plt.imshow(worst_pred_post[:, :, best_slice], cmap="Reds", alpha=0.35); plt.axis("off")
plt.tight_layout(); plt.show()

Dice 관련 변형 지표

Dice Score의 2를 직접 바꾸는 것은 권장하지 않습니다. 대신 아래 변형 지표를 활용합니다.

Day 5 전체 흐름 요약

1. 모델 로드        → best_metric_model.pth 불러오기
2. Validation 추론  → sliding_window_inference로 전체 볼륨 예측
3. Dice 계산        → 샘플별 binary_dice 계산
4. 오류 분석        → worst/best case 식별 및 시각화
5. Patch 분석       → patch 단위 Dice 분포 확인
6. 후처리           → keep_largest_component로 노이즈 제거
7. 성능 비교        → 후처리 전후 Dice 비교 (대폭 개선 확인)

11-1. 코드 네이밍 규칙

프로그래밍에서 이름을 짓는 방식에는 정해진 관례가 있습니다. 팀 프로젝트든 개인 프로젝트든 이 규칙을 따르면 코드가 읽기 쉬워지고, 역할이 이름만 봐도 드러납니다.

1. 전부 소문자 — lowercase

쓰는 곳: 파일명, 폴더명, URL, CSS 클래스

pages/about.js
components/header.js
.container { }
/api/user-list

2. 앞글자만 대문자 — PascalCase (단어마다)

쓰는 곳: React 컴포넌트, 클래스, 타입

// React 컴포넌트
function UserProfile() { }
function LoginButton() { }

// 클래스
class DatabaseConnection { }

3. 중간만 대문자 — camelCase

쓰는 곳: 변수, 함수, 메서드

// 변수
const userName = "홍길동"
const isLoggedIn = true

// 함수
function getUserData() { }
function handleClick() { }

4. 전부 대문자 — UPPER_SNAKE_CASE

쓰는 곳: 절대 안 바뀌는 상수값

const MAX_COUNT = 100
const API_KEY = "abc123"
const BASE_URL = "https://..."

5. 복수형 s 붙이기

쓰는 곳: 여러 개를 담는 배열, 컬렉션

// 여러 개 → 복수
const users = [user1, user2]
const products = []

// 하나 → 단수
const user = { name: "홍길동" }
const product = { id: 1 }

한눈에 요약

Python과의 차이

스터디 그룹에서 발표하거나 조사한 내용을 정리한 노트입니다.

S-1. 중입자선 치료와 AI Segmentation (2026-03-10)

치료 파이프라인에서의 AI 역할

중입자선 치료는 CT/MRI 촬영부터 실제 조사까지 여러 단계를 거치며, CNN이 자동 Segmentation 구간에서 핵심 역할을 합니다.

CT/MRI 촬영 → AI(CNN) 자동 Segmentation → 의사 검토·수정
→ TPS(치료계획시스템) → 물리학자 검증 → 가속기 정밀 조사

역할 분담

중입자선 치료에서 Segmentation이 특히 중요한 이유

어떤 Segmentation 모델이 쓰이나?

실제 활용 Task 분류

Medical Image Segmentation
├── Semantic Segmentation    → 종양 영역 전체 픽셀 분류
├── Instance Segmentation    → 개별 종양 덩어리 구분
└── Organ-at-Risk (OAR)      → 주변 정상 장기 보호 경계 설정

S-2. 의료 영상 AI — Classification · Detection · Segmentation (2026-03-10)

세 가지 Task 비교

왜 의료영상에서 Segmentation을 특히 많이 연구할까?

1. 위치 + 크기 = 치료에 직결 — 단순히 "있다"가 아니라 정확한 위치·경계·부피를 알 수 있어서 수술 계획, 방사선 치료 planning, 항암 반응 평가까지 연결
2. 정량화 가능 — "종양 부피 12.4cc, 지난 검사 대비 18% 감소" 같은 숫자 기반 평가 가능
3. 추적 관찰 — 3개월마다 CT를 찍을 때 병변이 줄었는지·커졌는지 정량 비교 가능
4. 하지만 어려움도 많음 — 라벨링이 매우 힘들고, 경계가 애매하고, 전문가 간 합의도 100%가 아님

3D 의료영상에서 CNN(U-Net)이 여전히 강력한 이유

U-Net — 의료 영상 Segmentation의 표준 베이스라인

• Encoder-Decoder 구조로 위치 정보 + 경계 정보 모두 포착
• Skip connection으로 해상도 유지 — 세밀한 경계 복원에 핵심
• 비교적 구조가 단순하고 학습이 안정적
• 학습 데이터가 적어도 잘 작동 (의료 데이터 특성에 적합)

주요 적용 대상: 비장(spleen), 간(liver), 신장(kidney), 폐(lung), 뇌 구조, 종양 영역

S-3. 딥러닝 기초 복습 (2026-03-11)

핵심 개념 4가지

퍼셉트론과 딥러닝의 관계

• 퍼셉트론 = 활성화 함수로 계단 함수(step function)를 사용하는 인공 뉴런
• 신경망 = 인공 뉴런이 여러 층으로 쌓인 것 (입력층 → 은닉층 → 출력층)
• "딥(Deep)" = 층이 깊다는 의미. 층이 많을수록 추상적인 특징 학습 가능

퍼셉트론: z = w1·x1 + w2·x2 + w3·x3 + b → 활성화 → 출력

XOR 문제 → 단층으로 불가 → 다층 퍼셉트론(MLP) + 역전파 등장 → 딥러닝의 시작

의료영상에서의 층별 학습 비유

1층 (저수준 특징): "X-ray에서 밝은/어두운 부분 인식"
2층 (중간 특징): "폐의 윤곽, 갈비뼈 구조 파악"
3층 (고수준 특징): "폐에 비정상적 흰 영역 발견"
4층 (최종 판단): "폐렴 가능성 높음"

주요 아키텍처

S-4. 바이브 코딩 — 반려견 유전체 분석 프로젝트 (2026-03-11)

프로젝트 개요

마이크로어레이 SNP 데이터(약 70만 유전 변이)를 활용한 반려견 유전체 분석 프로젝트입니다.

3단계 파이프라인

3단계 챌린지 상세

1. 믹스견 혈통 분석 — 순종 데이터를 레퍼런스 패널로 사용, PLINK-ADMIXTURE로 어떤 순종과 몇 % 유사한지 확인
2. 유전적 교배 위험 계산 — 두 개체의 유전체를 비교, ROH(Runs of Homozygosity) 개념으로 근친교배 위험도 예측
3. 외모 예측 모델링 — 털 색깔, 곱슬거림, 몸집 크기 등을 다유전자 위험 점수(PRS)로 추정

S-5. 의료 AI 이미지 Segmentation 평가 기준 (2026-03-11)

실제 의료 AI 평가 순서

제품 개발 단계에서는 Sensitivity 기준 먼저 통과 → FP 허용 범위 설정 → Dice 개선 순으로 진행합니다.

FP per scan — 왜 "덩어리" 기준인가?

• voxel 단위 FP는 의미 거의 없음. 의사는 "이상 의심 부위가 몇 개 뜨는가?" 즉 region 단위로 봄
• 예시: 100개 CT에서 총 FP 300개 → FP per scan = 300 / 100 = 3
• 이상적 기준: Case-level sensitivity 95% 이상 + FP per scan 1~3 이하

"Dice는 높은데 Precision은 낮다" — 모델 스타일 해석

직관적 예시:
진짜 병변 = 100 voxel
모델: 95 voxel 맞춤 (TP=95) + 40 voxel 과검출 (FP=40) + 5 voxel 놓침 (FN=5)
→ Dice 꽤 높음, 하지만 Precision 낮음

임상 해석:
→ 놓치는 건 적다 (Triage용 가능)
→ FP 많아서 정밀 자동 판독용은 위험

Voxel vs Case 레벨 비교

전체 커리큘럼의 흐름을 한눈에 볼 수 있는 구조도입니다. 각 과목이 어떻게 연결되는지 파악하면 학습 방향을 잡기 훨씬 쉬워집니다.

프로그래밍 기초
└─ Python
   ├─ 1. 변수 & 자료형
   │  ├─ 변수 (Variable) ─ 값 저장, 동적 타입 언어
   │  ├─ 기본 자료형 (Scalar Type)
   │  │   ├─ int (정수)  ├─ float (실수)
   │  │   ├─ str (문자열) └─ bool (참/거짓)
   │  └─ 타입 확인 & 변환: type(), int(), float(), str(), bool()
   │
   ├─ 2. 컬렉션 (자료구조)
   │  ├─ list ─ 순서 O, 변경 가능 (mutable), 인덱싱/슬라이싱
   │  ├─ tuple ─ 순서 O, 변경 불가 (immutable)
   │  ├─ dict ─ key:value 구조, 빠른 조회, JSON과 유사
   │  └─ set ─ 중복 제거, 순서 없음, 집합 연산
   │
   ├─ 3. 내장 함수
   │  ├─ 기본: len(), sum(), max(), min()
   │  ├─ 타입: type(), isinstance(), list(), dict(), set()
   │  └─ 반복 보조: range(), enumerate(), zip()
   │
   ├─ 4. 제어 흐름
   │  ├─ 조건문: if / elif / else
   │  ├─ 반복문: for (컬렉션 순회) / while (조건 기반)
   │  └─ 흐름 제어: break, continue, pass
   │
   └─ 5. 함수 (Function)
      ├─ def, 매개변수, return
      ├─ 목적: 코드 재사용, 로직 분리, 가독성
      └─ 확장: 기본값 매개변수, *args/**kwargs, lambda

NumPy ─ 빠른 수치 계산 엔진, Pandas/Scikit-learn 내부 기반
 ├─ ndarray
 │  ├─ shape / ndim / size
 │  ├─ dtype (int, float, bool)
 │  ├─ indexing / slicing / fancy indexing
 │  └─ broadcasting
 │
 ├─ 벡터화 연산
 │  ├─ for-loop 제거, element-wise 연산
 │  ├─ ufunc (np.add, np.log 등)
 │  └─ 연산 속도 향상
 │
 ├─ axis 개념
 │  ├─ axis=0 (열 기준) / axis=1 (행 기준)
 │  └─ 다차원 축 이해
 │
 └─ 통계 연산
    ├─ mean / sum / min / max
    ├─ std / var
    └─ argmin / argmax

Pandas
 ├─ 데이터 구조: Series / DataFrame
 ├─ 데이터 타입: 수치형 (int,float) / 범주형 (Nominal, Ordinal) ⭐
 ├─ 로딩: read_csv, read_excel, read_sql
 ├─ 구조 파악 (EDA): head, info, shape, describe, value_counts
 ├─ 결측치: isnull, MCAR/MAR/MNAR, dropna, fillna
 ├─ 데이터 선택: loc/iloc, boolean indexing, isin/query
 └─ 기초 통계: mean, median, mode, std, quantile

Matplotlib (기본 시각화)
 ├─ Figure / Axes 구조 ⭐⭐
 ├─ 기본: plot, scatter, bar, hist, boxplot
 └─ 꾸미기: title, legend, color, xlabel/ylabel, grid

Seaborn (통계 시각화)
 ├─ 범주형 ⭐: countplot, barplot, boxplot, violinplot
 ├─ 수치형: scatterplot, lineplot, heatmap
 ├─ 다변량: pairplot
 └─ 범주 분리: hue (색상), style (마커), size (크기)

Database
 ├─ RDBMS (관계형 DB)
 │  ├─ Table, 관계 (1:1, 1:N, N:M)
 │  ├─ Key (PK, FK)
 │  └─ SQL (SELECT, JOIN, GROUP BY)
 │
 └─ NoSQL
    └─ MongoDB (Database → Collection → Document)

👉 실무: DB → Pandas로 가져와서 분석 시작

전처리
 ├─ 스케일링
 │  ├─ StandardScaler (표준화): (x-mean)/std → 평균 0, 표준편차 1
 │  ├─ MinMaxScaler (정규화): (x-min)/(max-min) → 0~1 범위
 │  └─ RobustScaler: 중앙값/사분위 기반 → 이상치에 강건
 │
 ├─ 이상치 제거
 │  ├─ Z-score 방식 (임계값 보통 3)
 │  └─ IQR 방식 (Boxplot 기준)
 │
 ├─ 범주형 인코딩 ⭐
 │  ├─ Label Encoding ─ 범주→정수 (Tree 모델용)
 │  ├─ One-Hot Encoding ─ 범주→이진벡터 (선형/거리 모델용)
 │  └─ Ordinal Encoding ─ 순서 있는 범주용
 │
 └─ 파생 변수: binning, feature engineering

머신러닝 (Machine Learning)
│
├─ 학습 패러다임
│  ├─ 지도학습 (Supervised)
│  │  ├─ 분류: Logistic Regression, KNN, SVM, Decision Tree, Ensemble
│  │  └─ 회귀: Linear Regression, Ridge, Lasso, SVR, Tree 기반
│  │
│  ├─ 비지도학습 (Unsupervised)
│  │  ├─ 군집화: K-Means, DBSCAN, GMM, Hierarchical
│  │  ├─ 차원 축소: PCA, t-SNE, UMAP
│  │  └─ 이상치 탐지: Isolation Forest, LOF
│  │
│  └─ 강화학습: Agent/Environment, State/Action/Reward
│
├─ 핵심 개념
│  ├─ Bias-Variance Tradeoff
│  ├─ Underfitting (모델 너무 단순) vs Overfitting (과적합)
│  └─ Generalization (일반화) ─ 실무에서 가장 중요한 목표
│
├─ 데이터 분할 ⭐⭐
│  ├─ Train Set (학습) / Validation Set (튜닝) / Test Set (최종 평가)
│  ├─ K-Fold CV / Stratified K-Fold (분류) / TimeSeriesSplit
│  └─ ⚠ Data Leakage 주의 ─ 전처리는 Train 기준으로만!
│
├─ 평가 지표
│  ├─ 분류: Accuracy, Precision/Recall, F1, ROC-AUC, Confusion Matrix
│  └─ 회귀: MSE/RMSE, MAE, R²
│
└─ Feature Engineering ⭐ (성능에 가장 큰 영향)
   ├─ 스케일링 → 인코딩 → 선택 → Pipeline
   ├─ Filter / Wrapper / Embedded 방법
   ├─ Feature Importance: Gain, Permutation, SHAP
   └─ "FE 없으면 HPO 의미 없음"

앙상블 (Ensemble Learning)
│
├─ Bagging 계열 ─ 병렬 학습, 분산(Variance) 감소
│  ├─ Random Forest ─ 여러 Decision Tree + Feature Randomness
│  └─ Extra Trees ─ split 완전 랜덤, 더 강한 분산 감소
│
├─ Boosting 계열 ─ 순차 학습, 편향(Bias) 감소
│  ├─ 공통: learning_rate, n_estimators, max_depth, 과적합 위험
│  ├─ XGBoost ─ L1/L2 정규화, 결측치 자동, 병렬 처리
│  ├─ LightGBM ─ leaf-wise 성장, 대용량 강점, 매우 빠름
│  └─ CatBoost ─ 범주형 강점, target leakage 방지
│
└─ Stacking ─ 모델 위의 모델
   ├─ Base Models: 서로 다른 성향 조합 (선형 + 트리 + 부스팅)
   ├─ Meta Model: Base 예측을 입력으로 재학습
   └─ ⚠ CV 기반 예측값만 사용, 데이터 누수 방지

모델 개선
├─ 데이터 관점: 데이터 수 증가, Feature Engineering, 노이즈 제거
├─ 모델 관점: 모델 변경, 앙상블, 복잡도 조절
├─ 하이퍼파라미터 최적화 (HPO)
│  ├─ GridSearch / RandomSearch
│  ├─ Bayesian Optimization
│  └─ Optuna / Hyperopt
│
└─ 핵심 주의사항
   ├─ Tree 모델은 스케일링 거의 불필요
   ├─ Label Encoding 함부로 쓰면 위험
   ├─ One-Hot은 안전하지만 차원 증가
   ├─ Feature Selection은 Train 기준
   └─ FE 없으면 HPO 의미 없음

모르는 용어가 있으면 알려주세요. 여기에 추가해 드립니다!