경보 시스템과 실시간 알림

효과적인 경보 시스템은 RPM의 핵심입니다. 적절한 타이밍에 적절한 사람에게 적절한 알림을 보내는 것은 생명을 구할 수 있습니다. 이 장에서는 지능형 경보 시스템 설계, 임계값 설정, 경보 피로 방지, 그리고 다중 채널 알림 전략을 상세히 다룹니다.

경보 피로 문제

경보 피로(alert fatigue)는 과도한 알림으로 인해 의료진이 중요한 경보를 무시하거나 느리게 반응하는 현상입니다. 연구에 따르면 중환자실에서 발생하는 경보의 85-99%는 임상적으로 의미 없거나 거짓 양성입니다.

경보 피로의 위험성:

• 진짜 응급 상황을 놓칠 수 있음
• 의료진의 스트레스 증가와 번아웃
• 환자 안전 위협
• 워크플로우 중단으로 인한 생산성 저하
• 경보 무음 처리 또는 비활성화 경향

지능형 경보 시스템 설계

계층화된 경보 시스템

모든 경보를 동일하게 취급하지 말고, 중요도에 따라 분류합니다:

심각도	설명	대응 시간	알림 방법
🔴 위급 (Critical)	즉각적 생명 위협	즉시	전화, SMS, 앱 푸시 (모두), 소리 알림
🟠 긴급 (Urgent)	빠른 개입 필요	15분 이내	SMS, 앱 푸시, 이메일
🟡 경고 (Warning)	주의 필요, 모니터링	1-4시간	앱 푸시, 이메일
🟢 정보 (Info)	참고용 정보	일일 검토	대시보드 표시만

스마트 임계값 설정

고정된 임계값 대신 개인화되고 맥락을 고려한 임계값을 사용합니다:

class SmartThresholdEngine:
    def __init__(self, patient_id):
        self.patient_id = patient_id
        self.baseline = self.calculate_baseline()

    def calculate_baseline(self):
        """
        환자의 최근 30일 데이터로 개인별 기준선 계산
        """
        historical_data = get_patient_data(
            patient_id=self.patient_id,
            days=30
        )

        return {
            'heart_rate': {
                'mean': np.mean(historical_data['heart_rate']),
                'std': np.std(historical_data['heart_rate']),
                'percentile_10': np.percentile(historical_data['heart_rate'], 10),
                'percentile_90': np.percentile(historical_data['heart_rate'], 90)
            },
            'blood_pressure': {
                'systolic_mean': np.mean(historical_data['systolic']),
                'diastolic_mean': np.mean(historical_data['diastolic'])
            }
        }

    def check_heart_rate(self, current_hr, context):
        """
        맥락을 고려한 심박수 이상 감지
        """
        baseline_hr = self.baseline['heart_rate']['mean']
        std = self.baseline['heart_rate']['std']

        # 활동 중이면 더 높은 임계값 허용
        if context['activity_level'] == 'high':
            threshold_multiplier = 1.5
        elif context['activity_level'] == 'moderate':
            threshold_multiplier = 1.2
        else:  # resting
            threshold_multiplier = 1.0

        # 개인별 기준선에서 2 표준편차 이상 벗어나면 경고
        upper_threshold = baseline_hr + (2 * std * threshold_multiplier)
        lower_threshold = baseline_hr - (2 * std)

        if current_hr > upper_threshold:
            return {
                'alert': True,
                'severity': 'warning',
                'message': f'심박수가 개인 기준선({baseline_hr:.0f})보다 {current_hr - baseline_hr:.0f} bpm 높습니다'
            }
        elif current_hr < lower_threshold:
            return {
                'alert': True,
                'severity': 'warning',
                'message': f'심박수가 개인 기준선({baseline_hr:.0f})보다 {baseline_hr - current_hr:.0f} bpm 낮습니다'
            }

        return {'alert': False}

    def detect_trend_alert(self, recent_values, metric_name):
        """
        추세 기반 조기 경보
        """
        # 최근 24시간 데이터로 추세 분석
        if len(recent_values) < 10:
            return None

        # 선형 회귀로 추세 계산
        x = np.arange(len(recent_values))
        slope, intercept = np.polyfit(x, recent_values, 1)

        # 급격한 상승 또는 하강 감지
        if metric_name == 'blood_pressure_systolic':
            if slope > 5:  # 시간당 5 mmHg 이상 증가
                return {
                    'alert': True,
                    'severity': 'warning',
                    'message': '혈압이 지속적으로 상승하고 있습니다',
                    'trend': 'increasing',
                    'slope': slope
                }
        elif metric_name == 'oxygen_saturation':
            if slope < -0.5:  # 시간당 0.5% 이상 감소
                return {
                    'alert': True,
                    'severity': 'urgent',
                    'message': '산소포화도가 지속적으로 감소하고 있습니다',
                    'trend': 'decreasing',
                    'slope': slope
                }

        return None

복합 지표 기반 경보

단일 바이오마커가 아닌 여러 지표를 조합하여 더 정확한 경보를 생성합니다:

class CompositeAlertSystem:
    def check_sepsis_risk(self, vitals):
        """
        패혈증 조기 경보 (qSOFA 점수 기반)
        """
        qsofa_score = 0

        # 1. 호흡수 >= 22/분
        if vitals['respiratory_rate'] >= 22:
            qsofa_score += 1

        # 2. 의식 변화 (GCS < 15 또는 활동량 급감)
        if vitals.get('gcs_score', 15) < 15 or \
           vitals.get('activity_decrease', 0) > 50:
            qsofa_score += 1

        # 3. 수축기 혈압 <= 100 mmHg
        if vitals['systolic_bp'] <= 100:
            qsofa_score += 1

        if qsofa_score >= 2:
            return {
                'alert': True,
                'severity': 'urgent',
                'condition': '패혈증 위험',
                'qsofa_score': qsofa_score,
                'recommendation': '즉시 의료진 평가 필요'
            }

        return {'alert': False}

    def check_heart_failure_decompensation(self, data):
        """
        심부전 악화 조기 감지
        """
        risk_factors = []

        # 1. 체중 급증 (3일 이내 2kg 이상)
        if data['weight_change_3days'] >= 2.0:
            risk_factors.append('급격한 체중 증가')

        # 2. 호흡곤란 증가
        if data['dyspnea_score'] > data['baseline_dyspnea'] + 2:
            risk_factors.append('호흡곤란 악화')

        # 3. 하지 부종 증가
        if data['edema_score'] > data['baseline_edema']:
            risk_factors.append('부종 증가')

        # 4. 산소포화도 감소
        if data['spo2'] < 92:
            risk_factors.append('산소포화도 저하')

        # 5. 안정시 심박수 증가
        if data['resting_hr'] > data['baseline_resting_hr'] + 15:
            risk_factors.append('안정시 심박수 증가')

        # 2개 이상 위험 요소 = 경보
        if len(risk_factors) >= 2:
            return {
                'alert': True,
                'severity': 'urgent',
                'condition': '심부전 악화 가능성',
                'risk_factors': risk_factors,
                'recommendation': '이뇨제 조정 및 의사 연락 고려'
            }

        return {'alert': False}

경보 피로 방지 전략

1. 지능형 필터링

• 신호 품질 확인: 노이즈가 많거나 신뢰할 수 없는 데이터로 경보 생성 방지
• 지속성 검증: 이상 값이 일정 시간(예: 5분) 지속될 때만 경보
• 재확인 메커니즘: 중요 경보는 다른 센서로 교차 확인

def validate_alert(alert_data, sensor_quality, duration):
    """
    경보 생성 전 검증
    """
    # 센서 품질 확인
    if sensor_quality['signal_strength'] < 0.7:
        return False  # 신호 품질 불량, 경보 억제

    # 지속성 확인
    if duration < 300:  # 5분
        return False  # 너무 짧은 이벤트, 경보 억제

    # 모션 아티팩트 확인
    if sensor_quality['motion_detected']:
        return False  # 움직임으로 인한 오류 가능성

    return True  # 경보 생성 허용

2. 경보 통합 및 그룹화

동일한 근본 원인에서 발생하는 여러 경보를 하나로 통합합니다:

class AlertAggregator:
    def __init__(self):
        self.pending_alerts = []
        self.aggregation_window = 300  # 5분

    def aggregate_alerts(self, new_alert):
        """
        관련 경보들을 그룹화
        """
        # 최근 5분 내 관련 경보 찾기
        related_alerts = [
            a for a in self.pending_alerts
            if self.are_related(a, new_alert) and
               (time.time() - a['timestamp']) < self.aggregation_window
        ]

        if related_alerts:
            # 통합 경보 생성
            return self.create_composite_alert(related_alerts + [new_alert])
        else:
            # 새 경보로 추가
            self.pending_alerts.append(new_alert)
            return new_alert

    def are_related(self, alert1, alert2):
        """
        두 경보가 관련되어 있는지 확인
        """
        # 예: 저혈압, 빈맥, 저체온은 쇼크와 관련될 수 있음
        shock_indicators = ['low_bp', 'tachycardia', 'hypothermia']

        if alert1['type'] in shock_indicators and \
           alert2['type'] in shock_indicators:
            return True

        return False

    def create_composite_alert(self, alerts):
        """
        여러 경보를 하나의 종합 경보로 생성
        """
        return {
            'type': 'composite',
            'severity': max(a['severity'] for a in alerts),
            'message': f'{len(alerts)}개 관련 이상 징후 감지',
            'sub_alerts': alerts,
            'possible_conditions': self.infer_conditions(alerts)
        }

3. 적응형 임계값

머신러닝을 사용하여 각 환자의 정상 범위를 학습하고, 시간이 지남에 따라 임계값을 자동으로 조정합니다:

• 환자의 활동 패턴, 약물 변화, 계절적 변화 고려
• 거짓 양성 경보가 많으면 임계값을 완화
• 놓친 이벤트가 있으면 임계값을 엄격하게 조정

에스컬레이션 프로토콜

계층적 알림 체계

경보의 심각도와 대응 시간에 따라 적절한 사람에게 순차적으로 알립니다:

class EscalationProtocol:
    def __init__(self, patient_id):
        self.patient_id = patient_id
        self.care_team = self.get_care_team()

    def get_care_team(self):
        """
        환자의 케어 팀 정보 가져오기
        """
        return {
            'primary_nurse': {
                'name': 'Kim RN',
                'phone': '+82-10-1234-5678',
                'email': 'kim.rn@hospital.com',
                'priority': 1
            },
            'physician': {
                'name': 'Dr. Lee',
                'phone': '+82-10-9876-5432',
                'email': 'dr.lee@hospital.com',
                'priority': 2
            },
            'specialist': {
                'name': 'Dr. Park (Cardiology)',
                'phone': '+82-10-5555-1234',
                'email': 'dr.park@hospital.com',
                'priority': 3
            },
            'emergency': {
                'name': '119',
                'phone': '119',
                'priority': 4
            }
        }

    def escalate_alert(self, alert):
        """
        경보 에스컬레이션 실행
        """
        if alert['severity'] == 'critical':
            # 위급: 모든 팀원에게 동시 알림
            self.notify_all(alert)

        elif alert['severity'] == 'urgent':
            # 긴급: 단계별 에스컬레이션
            self.staged_escalation(alert, timeout=15)  # 15분

        elif alert['severity'] == 'warning':
            # 경고: 1차 담당자에게만 알림
            self.notify_primary(alert)

    def staged_escalation(self, alert, timeout):
        """
        단계별 에스컬레이션 (응답 없으면 상위 레벨로)
        """
        escalation_chain = [
            'primary_nurse',
            'physician',
            'specialist',
            'emergency'
        ]

        for level in escalation_chain:
            contact = self.care_team[level]

            # 알림 발송
            self.send_notification(contact, alert)

            # 확인 대기
            if self.wait_for_acknowledgment(alert['id'], timeout):
                # 확인됨, 에스컬레이션 중지
                log_alert_handled(alert['id'], contact['name'])
                return

            # 응답 없음, 다음 레벨로
            log_escalation(alert['id'], level)

        # 모든 레벨 무응답 시 최종 조치
        self.trigger_emergency_protocol(alert)

다중 채널 알림

알림 채널별 특성

채널	도달 시간	확인률	적합 용도
전화 (음성 통화)	즉시	높음 (90%+)	위급 상황
SMS	즉시	높음 (80-90%)	긴급 및 위급
앱 푸시 알림	즉시	중간 (60-70%)	긴급 및 경고
이메일	분 단위	낮음 (30-50%)	경고 및 정보
인앱 메시지	앱 열 때	낮음	정보성 알림

알림 내용 최적화

효과적인 알림 메시지는 명확하고, 실행 가능하며, 맥락을 제공합니다:

def create_alert_message(alert_data):
    """
    효과적인 경보 메시지 생성
    """
    # 나쁜 예: "이상 감지"
    # 좋은 예:

    message = {
        'title': f"🔴 {alert_data['patient_name']} - {alert_data['condition']}",
        'body': f"""
환자: {alert_data['patient_name']} (나이: {alert_data['age']}, 성별: {alert_data['gender']})
질환: {alert_data['primary_diagnosis']}

이상 징후:
- {alert_data['primary_issue']}
- 현재 값: {alert_data['current_value']}
- 정상 범위: {alert_data['normal_range']}
- 기준선 대비: {alert_data['deviation_from_baseline']}

권장 조치:
{alert_data['recommended_action']}

추가 정보:
- 최근 바이탈: HR {alert_data['hr']}, BP {alert_data['bp']}, SpO2 {alert_data['spo2']}%
- 최근 증상 보고: {alert_data['recent_symptoms']}
- 마지막 약물 복용: {alert_data['last_medication']}

환자 연락처: {alert_data['patient_phone']}
        """,
        'actions': [
            {'label': '확인', 'action': 'acknowledge'},
            {'label': '환자에게 전화', 'action': 'call_patient'},
            {'label': '상세 보기', 'action': 'view_details'}
        ]
    }

    return message

환자 대상 알림

환자 자가 관리 알림

환자에게도 적절한 알림을 보내어 자가 관리를 돕습니다:

• 약물 복용 알림: 시간에 맞춰 약 복용 리마인더
• 측정 리마인더: 혈압, 혈당, 체중 측정 시간 알림
• 생활습관 코칭: 운동, 수분 섭취, 식사 권장사항
• 이상 징후 교육: 주의해야 할 증상 안내

def create_patient_alert(alert_type, data):
    """
    환자 친화적 알림 메시지 생성
    """
    if alert_type == 'high_blood_pressure':
        return {
            'title': '혈압이 높습니다',
            'message': f"""
혈압이 {data['systolic']}/{data['diastolic']} mmHg로 측정되었습니다.

다음 조치를 취해 주세요:
1. 편안하게 앉아서 5분간 휴식하세요
2. 15분 후 혈압을 다시 측정하세요
3. 여전히 높으면 (140/90 이상) 담당 의사에게 연락하세요

도움이 필요하면 언제든지 연락하세요: {data['nurse_phone']}
            """,
            'urgency': 'medium',
            'icon': '🩺'
        }

    elif alert_type == 'medication_reminder':
        return {
            'title': '💊 약 복용 시간입니다',
            'message': f"""
{data['medication_name']} {data['dose']} 복용 시간입니다.

복용 후 앱에서 확인 버튼을 눌러주세요.
            """,
            'urgency': 'low',
            'actions': ['복용 완료', '나중에 알림', '건너뛰기']
        }

가족/보호자 알림

환자의 가족이나 보호자에게도 중요한 정보를 공유합니다:

• 환자의 동의 하에 선택적 정보 공유
• 긴급 상황 시 자동 알림
• 일일/주간 건강 요약 리포트
• 약물 변경, 진료 예약 등 중요 이벤트 알림