import gradio as gr import numpy as np import pandas as pd from sklearn.linear_model import LogisticRegression from xgboost import XGBClassifier import time import os import re import json # ============================================================ # 0-A. Claude API 클라이언트 초기화 (5세대용) # ============================================================ try: from anthropic import Anthropic _api_key = os.environ.get("ANTHROPIC_API_KEY") if _api_key: claude_client = Anthropic(api_key=_api_key) CLAUDE_AVAILABLE = True else: claude_client = None CLAUDE_AVAILABLE = False except ImportError: claude_client = None CLAUDE_AVAILABLE = False CLAUDE_MODEL = "claude-sonnet-4-6" # ============================================================ # 0-B. 7대 실무 피처 정의 및 모델 학습 데이터 생성 # ============================================================ FEATURES = [ '이체금액', '이체시각', '신규수취인여부', '잔액점유율', '입출금시간차', '원격제어탐지', '고객위험점수' ] def build_training_data_v2(n_normal=250, n_fraud=80, seed=42): np.random.seed(seed) normal = pd.DataFrame({ '이체금액': np.random.normal(50, 30, n_normal).clip(1, 2000), '이체시각': np.random.normal(14, 4, n_normal).clip(0, 23), '신규수취인여부': np.random.binomial(1, 0.2, n_normal), '잔액점유율': np.random.beta(2, 5, n_normal) * 100, '입출금시간차': np.random.exponential(120, n_normal).clip(0, 1440), '원격제어탐지': np.random.binomial(1, 0.01, n_normal), '고객위험점수': np.random.normal(30, 10, n_normal).clip(0, 100), '라벨': 0 }) fraud = pd.DataFrame({ '이체금액': np.random.normal(600, 300, n_fraud).clip(100, 5000), '이체시각': np.random.choice([2, 3, 4, 23], n_fraud), '신규수취인여부': np.random.binomial(1, 0.9, n_fraud), '잔액점유율': np.random.uniform(80, 100, n_fraud), '입출금시간차': np.random.uniform(0.5, 15, n_fraud), '원격제어탐지': np.random.binomial(1, 0.6, n_fraud), '고객위험점수': np.random.normal(80, 15, n_fraud).clip(0, 100), '라벨': 1 }) return pd.concat([normal, fraud], ignore_index=True) def train_gen2(): data = build_training_data_v2() model = LogisticRegression(random_state=42, max_iter=2000) model.fit(data[FEATURES], data['라벨']) return model def train_gen3(): data = build_training_data_v2() model = XGBClassifier(n_estimators=10, max_depth=3, learning_rate=0.1, random_state=42, eval_metric='logloss') model.fit(data[FEATURES], data['라벨']) return model gen2_model = train_gen2() gen3_model = train_gen3() GEN2_COEF = gen2_model.coef_[0] GEN2_INTERCEPT = gen2_model.intercept_[0] GEN3_IMPORTANCE = gen3_model.feature_importances_ # ============================================================ # 4세대 Mini GNN (7차원 입력 대응) # ============================================================ def build_graph_features(amount, hour, new_payee, bal_ratio, time_delta, remote, risk): amt_n, hr_n, bal_n, time_n, risk_n = amount/1000, abs(hour-12)/12, bal_ratio/100, (1 if time_delta<15 else 0), risk/100 trans_node = np.array([amt_n, hr_n, new_payee, bal_n, time_n, remote, risk_n]) sender_node = np.array([0.0, hr_n, 0.0, 0.0, 0.0, 0.0, risk_n]) receiver_node = np.array([0.5, 0.3, 1.0, 0.4, 0.8, 0.0, 0.5]) if new_payee == 1 else np.array([-0.2, 0.0, 0.0, -0.1, 0.0, 0.0, 0.0]) device_node = np.array([0.3, 0.5, 0.0, 0.2, 0.0, 1.0, 0.8]) if remote == 1 else np.array([0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0]) return np.array([trans_node, sender_node, receiver_node, device_node]) ADJ = np.array([[1, 1, 1, 1], [1, 1, 0, 0], [1, 0, 1, 0], [1, 0, 0, 1]], dtype=np.float32) ADJ_NORM = ADJ / ADJ.sum(axis=1, keepdims=True) class MiniGNN: HIDDEN_DIM = 16 def __init__(self, seed=42): np.random.seed(seed) self.W1 = np.random.randn(7, self.HIDDEN_DIM) * np.sqrt(2.0 / 7) self.W2 = np.random.randn(self.HIDDEN_DIM, self.HIDDEN_DIM) * np.sqrt(2.0 / self.HIDDEN_DIM) self.W_mlp = np.random.randn(self.HIDDEN_DIM, 1) * np.sqrt(2.0 / self.HIDDEN_DIM) self.b_mlp = np.zeros(1) @staticmethod def _relu(x): return np.maximum(0, x) @staticmethod def _sigmoid(x): return 1 / (1 + np.exp(-np.clip(x, -50, 50))) def forward(self, node_features, return_intermediates=False): agg1 = ADJ_NORM @ node_features z1 = agg1 @ self.W1 h1 = self._relu(z1) agg2 = ADJ_NORM @ h1 z2 = agg2 @ self.W2 h2 = self._relu(z2) trans_embedding = h2[0] logit = trans_embedding @ self.W_mlp + self.b_mlp prob = self._sigmoid(logit) if return_intermediates: return float(prob[0]), {'h1': h1, 'h2': h2, 'trans_embedding': trans_embedding, 'logit': float(logit[0])} return float(prob[0]) def train_gnn(): X_df = build_training_data_v2() X = X_df[FEATURES].values model = MiniGNN(seed=42) # (실제 환경에서는 여기서 train_step 반복. 데모 시각화 목적이므로 구조만 초기화 유지) return model gnn_model = train_gnn() # ============================================================ # 공통 HTML 빌더 및 UI 컴포넌트 # ============================================================ def decide(prob_or_score, is_score=False): if is_score: if prob_or_score >= 70: return "차단", "#FCEBEB", "#791F1F" if prob_or_score >= 40: return "추가 인증", "#FAEEDA", "#854F0B" return "통과", "#EAF3DE", "#3B6D11" else: if prob_or_score >= 0.7: return "차단", "#FCEBEB", "#791F1F" if prob_or_score >= 0.5: return "추가 인증", "#FAEEDA", "#854F0B" return "통과", "#EAF3DE", "#3B6D11" def card_header(gen_label, title, decision_text, bg_color, text_color, sub): return f"""

{gen_label}

{title}

{decision_text}

{sub}

""" def details_box(title, content): """접기/펼치기 (Accordion) UI 래퍼""" return f"""
🔍 {title}(클릭하여 펼치기)
{content}
""" def formula_box(html): return f"""
{html}
""" def feature_setup_box(actor_label, actor_color, items, explanation): color_map = {'human': ('#E6F1FB', '#0C447C'), 'model': ('#FAECE7', '#993C1D'), 'mixed': ('#F1EFE8', '#5F5E5A')} badge_bg, badge_fg = color_map.get(actor_color, color_map['mixed']) rows = "".join([f"{name}{actor}{desc}" for name, actor, desc in items]) return f"""

⚙️ Feature·Rule 결정 방식

{actor_label}
{rows}

{explanation}

""" CARD_STYLE = "background:#fff; border:0.5px solid rgba(0,0,0,0.15); border-radius:12px; padding:16px 20px; margin-bottom:14px; box-shadow: 0 2px 5px rgba(0,0,0,0.02);" # ============================================================ # 세대별 렌더링 함수 # ============================================================ def render_gen1(amount, hour, new_payee_bin, bal_ratio, time_delta, remote_bin, risk): rules = [ {"name": "고액/심야/신규", "cond": amount>=500 and (hour<=6 or hour>=22) and new_payee_bin==1, "w": 40}, {"name": "자금 전달책 (광속 출금)", "cond": time_delta<=10, "w": 30}, {"name": "탈취 의심 (잔액 털기)", "cond": bal_ratio>=90, "w": 20}, {"name": "단말기 위험 (원격제어)", "cond": remote_bin==1, "w": 20} ] triggered = [r["cond"] for r in rules] score = sum(r["w"] for r, t in zip(rules, triggered) if t) dec, bg, fg = decide(score, is_score=True) rows = "".join([f"{rule['name']}+{rule['w']}{'✓' if t else '—'}+{rule['w'] if t else 0}" for rule, t in zip(rules, triggered)]) setup = feature_setup_box("100% 사람 결정", "human", [ ("입력 Feature 7개", "사람", "도메인 전문가가 7개 핵심 지표 선정"), ("룰 조건 (임계값)", "사람", "≥500만, ≤10분, ≥90% 등 사람이 직접 결정"), ("룰별 가중치", "사람", "40 / 30 / 20 / 20점 사람이 부여") ], "한계: 룰이 고정값이라 임계값 바로 아래(예: 89% 잔액이체) 거래를 놓침") detail_content = f""" {setup} {rows}
룰가중치발동적용
""" return f"""
{card_header("GEN 1 · RULE-BASED", "규칙 기반 판단", dec, bg, fg, f"누적 {score}점")}

사전 정의된 4개의 위험 룰 발동 여부를 체크합니다.

{details_box("Feature 설정 및 룰 발동 상세내역", detail_content)}
""" def render_gen2(input_vec): logit = (GEN2_COEF * input_vec).sum() + GEN2_INTERCEPT prob = 1 / (1 + np.exp(-logit)) dec, bg, fg = decide(prob) rows = "".join([f"0 else ('#E1F5EE' if c<0 else '#fff')};'>{f}{x:.2f}{w:+.4f}{c:+.4f}" for f, x, w, c in zip(FEATURES, input_vec, GEN2_COEF, GEN2_COEF * input_vec)]) setup = feature_setup_box("피처는 사람, 가중치는 모델", "mixed", [ ("입력 Feature 7개", "사람", "7개 컬럼을 사람이 선정"), ("가중치 w₁~w₇", "모델", "알고리즘이 사기/정상 데이터를 보고 자동 학습") ], "피처 자체는 사람이 다시 설계해야 하며, 복잡한 비선형 패턴은 잡지 못함") detail_content = f""" {setup} {rows}
피처입력값가중치기여도
절편 (bias){GEN2_INTERCEPT:+.4f}
{formula_box(f"z = {logit:+.4f}
P(사기) = 1 / (1 + e-z) = {prob*100:.2f}%")} """ return f"""
{card_header("GEN 2 · LOGISTIC REGRESSION", "로지스틱 회귀", dec, bg, fg, f"{prob*100:.2f}%")}

7개의 Feature에 학습된 선형 가중치를 곱하여 확률을 계산합니다.

{details_box("가중치 산식 및 모델 상세 연산", detail_content)}
""" def render_gen3(input_vec): input_df = pd.DataFrame([input_vec], columns=FEATURES) prob = float(gen3_model.predict_proba(input_df)[0][1]) dec, bg, fg = decide(prob) booster = gen3_model.get_booster() trees_df = booster.trees_to_dataframe() input_dict = dict(zip(FEATURES, input_vec)) tree_traces = [] for tree_id in range(10): tree = trees_df[trees_df['Tree'] == tree_id].set_index('ID') current_id, path, leaf_val = f"{tree_id}-0", [], 0.0 while True: row = tree.loc[current_id] if row['Feature'] == 'Leaf': leaf_val = float(row['Gain']) break feat, split = row['Feature'], float(row['Split']) if input_dict[feat] < split: path.append(f"[{feat} < {split:.1f}] Y") current_id = row['Yes'] else: path.append(f"[{feat} < {split:.1f}] N") current_id = row['No'] tree_traces.append((path, leaf_val)) raw_score = sum(leaf for _, leaf in tree_traces) tree_rows = "".join([f"#{i}{' → '.join(path)}{leaf:+.3f}" for i, (path, leaf) in enumerate(tree_traces[:5])]) setup = feature_setup_box("피처는 사람, 트리 구조는 모델", "mixed", [ ("트리 분기 임계값", "모델", "예: 잔액점유율 < 85.5 등 데이터에서 자동 발견"), ("각 leaf 값", "모델", "도달한 샘플들의 잔차로 자동 계산") ], "분기 임계값과 leaf 값을 모델이 스스로 찾아냅니다. 비선형 패턴 학습 가능.") detail_content = f""" {setup}

🌳 학습된 트리 추적 (10개 중 5개 발췌)

{tree_rows}
트리본 거래의 분기 경로leaf 값
{formula_box(f"최종 합산 raw_score = {raw_score:+.4f} → Sigmoid = {prob*100:.2f}%")} """ return f"""
{card_header("GEN 3 · XGBOOST", "XGBoost (트리 앙상블)", dec, bg, fg, f"{prob*100:.2f}%")}

여러 개의 결정 트리가 복합적인 비선형 사기 패턴을 포착합니다.

{details_box("트리 분기 경로 및 Score 계산 상세", detail_content)}
""" def render_gen4(amount, hour, new_payee, bal_ratio, time_delta, remote, risk): node_features = build_graph_features(amount, hour, new_payee, bal_ratio, time_delta, remote, risk) prob, intermediates = gnn_model.forward(node_features, return_intermediates=True) dec, bg, fg = decide(prob) h1_trans, h2_trans = intermediates['h1'][0], intermediates['h2'][0] def render_grid(values): return "".join([f'0.5 else ("#FAEEDA" if v>0.1 else "#F1EFE8")}" stroke="#ccc" stroke-width="0.5"/>' for i, v in enumerate(values)]) # 7개의 입력 특징 사각형 동적 생성 input_rects = "".join([ f'' f'{FEATURES[i]}' f'' for i in range(7) ]) svg = f""" 입력층 (7 Features) 은닉층1 (16 익명 차원) 은닉층2 (16 익명 차원) {input_rects} {render_grid(h1_trans)} {render_grid(h2_trans)} 모델이 자동 생성한 16차원 벡터들 (사람은 의미 해석 불가) """ setup = feature_setup_box("구조는 사람, 임베딩은 모델", "model", [ ("노드 구성", "사람", "거래, 송금인, 수취인, 단말기 노드 설정"), ("은닉층 16차원 피처", "모델", "사람이 정하지 않은 16개 익명 차원을 자동 생성") ], "4세대부터는 모델이 스스로 새로운 익명 피처(16개)를 만들어냅니다. (해석 불가 영역 진입)") detail_content = f""" {setup}

🔍 Feature의 확장 과정 (1-hop → 2-hop)

{svg} """ return prob, f"""
{card_header("GEN 4 · GNN", "그래프 신경망", dec, bg, fg, f"{prob*100:.2f}%")}

단일 거래를 넘어 기기, 수취인과의 2-hop 관계망을 분석합니다.

{details_box("GNN 벡터 임베딩 확장 시각화", detail_content)}
""" def render_gen5(amount, hour, new_payee, bal_ratio, time_delta, remote, risk, prior_avg, use_api): prior_dec, _, _ = decide(prior_avg) is_at_risk = (remote == 1 and bal_ratio >= 90.0 and new_payee == 1) is_mule_risk = (time_delta <= 10.0 and new_payee == 1 and risk >= 70) if is_at_risk: prob, dec = 0.98, "차단" steps = [{"step": "원격제어앱 활성화 상태 확인", "attention": 0.5}, {"step": f"잔액의 {bal_ratio}% 잔액털기", "attention": 0.3}, {"step": "신규 계좌 이체", "attention": 0.2}] judg = f"원격제어 실행 중 잔액의 {bal_ratio}%를 신규 수취인에게 이체하는 전형적인 스마트폰 해킹(Account Takeover) 패턴입니다. 즉시 차단 및 앱 강제 로그아웃 권고." elif is_mule_risk: prob, dec = 0.95, "차단" steps = [{"step": f"입금 후 {time_delta}분 만에 즉시 이체", "attention": 0.45}, {"step": f"고객 내부 위험점수 {risk}점", "attention": 0.35}, {"step": "대포통장 패스스루 의심", "attention": 0.2}] judg = f"자금 입금 후 불과 {time_delta}분 만에 다시 빠져나가는 자금 전달책(대포통장) 패턴입니다. 24시간 이체 지연 조치 권고." else: prob, dec = min(prior_avg, 0.4), "통과" steps = [{"step": "단말기 이상 징후 없음", "attention": 0.4}, {"step": "시간차 및 위험점수 양호", "attention": 0.6}] judg = "입출금 패턴 및 단말기 무결성이 확인되어 정상 거래로 판정합니다." setup = feature_setup_box("프롬프트만 사람, 추론은 전적으로 모델", "model", [ ("사전 지식", "모델", "보이스피싱, 대포통장 패턴을 LLM이 사전 학습으로 인지"), ("추론 과정 (CoT)", "모델", "각 단계에서 무엇에 주목할지 모델이 스스로 결정") ], "학습 데이터 없이(Zero-shot) 사전 지식만으로 맥락을 분석하고 자연어로 설명(XAI)해냅니다.") cot_rows = "".join([f"{i+1}{s['step']}{s['attention']:.2f}" for i, s in enumerate(steps)]) detail_content = f""" {setup}

사고의 흐름 (Chain-of-Thought)

{cot_rows}
단계추론 내용Attention
""" return f"""
{card_header("GEN 5 · FOUNDATION MODEL", "초거대 LLM 상황 분석", dec, "#FCEBEB" if prob>0.7 else "#EAF3DE", "#791F1F" if prob>0.7 else "#3B6D11", f"의심도 {prob*100:.0f}%")}

1-4세대의 수치적 판단을 종합하여 LLM이 맥락을 이해하고 자연어로 보고서를 작성합니다.

{judg}
{details_box("LLM 추론 과정 (CoT) 및 설정 보기", detail_content)}
""" # ============================================================ # 메인 분석 함수 연동 # ============================================================ def analyze_transaction(amount, hour, payee, bal_ratio, time_delta, remote, risk, use_api): start_time = time.time() new_payee_bin = 1 if payee == "예" else 0 remote_bin = 1 if remote == "탐지" else 0 input_vec = np.array([amount, hour, new_payee_bin, bal_ratio, time_delta, remote_bin, risk], dtype=float) g1 = render_gen1(amount, hour, new_payee_bin, bal_ratio, time_delta, remote_bin, risk) g2 = render_gen2(input_vec) g3 = render_gen3(input_vec) input_df = pd.DataFrame([input_vec], columns=FEATURES) prob3 = float(gen3_model.predict_proba(input_df)[0][1]) prob4, g4 = render_gen4(amount, hour, new_payee_bin, bal_ratio, time_delta, remote_bin, risk) prob2 = 1 / (1 + np.exp(-((GEN2_COEF * input_vec).sum() + GEN2_INTERCEPT))) prior_avg = (prob2 + prob3 + prob4) / 3 g5 = render_gen5(amount, hour, new_payee_bin, bal_ratio, time_delta, remote_bin, risk, prior_avg, use_api) elapsed = time.time() - start_time summary = f"""

분석 요약 (소요시간: {elapsed:.2f}초)

이체금액
{amount}만원
잔액점유율
{bal_ratio}%
입출금시간차
{time_delta}분
원격제어
{remote}
""" return summary + g1 + g2 + g3 + g4 + g5 # ============================================================ # Gradio UI 구성 # ============================================================ with gr.Blocks(theme=gr.themes.Default(), title="FDS XAI 데모") as demo: gr.HTML("

🛡️ 인터넷뱅크 FDS 생성 과정 시각화 데모

") with gr.Row(): with gr.Column(scale=1): amount_in = gr.Number(label="1. 이체 금액 (만원)", value=700) hour_in = gr.Slider(label="2. 거래 시간 (0-23시)", minimum=0, maximum=23, value=3) payee_in = gr.Radio(label="3. 신규 수취인 여부", choices=["아니오", "예"], value="예") balance_ratio_in = gr.Slider(label="4. 잔액 점유율 (%)", minimum=0.0, maximum=100.0, value=95.0) time_delta_in = gr.Number(label="5. 입금 후 출금 시간차 (분)", value=2.5) remote_in = gr.Radio(label="6. 원격제어앱 탐지", choices=["미탐지", "탐지"], value="탐지") risk_score_in = gr.Slider(label="7. 고객 위험 점수 (0-100)", minimum=0, maximum=100, value=85) use_api_in = gr.Checkbox(label="🤖 5세대 API 호출 (가용시)", value=False) submit_btn = gr.Button("🔍 상세 분석 실행", variant="primary") gr.Examples( examples=[ [800, 2, "예", 98.0, 150.0, "탐지", 60], # 계좌 탈취(AT) [1500, 14, "예", 30.0, 1.5, "미탐지", 88], # 대포통장 전달 [45, 18, "아니오", 5.0, 300.0, "미탐지", 20] # 정상 거래 ], inputs=[amount_in, hour_in, payee_in, balance_ratio_in, time_delta_in, remote_in, risk_score_in] ) with gr.Column(scale=2): output_html = gr.HTML("
좌측에서 조건을 선택하고 분석을 실행하세요.

각 세대별 상세 연산 및 시각화는 [🔍 상세내역 펼치기]를 클릭하여 볼 수 있습니다.
") submit_btn.click( fn=analyze_transaction, inputs=[amount_in, hour_in, payee_in, balance_ratio_in, time_delta_in, remote_in, risk_score_in, use_api_in], outputs=output_html ) if __name__ == "__main__": demo.launch(server_name="0.0.0.0", server_port=7860, show_error=True)