Spaces:

ableman82
/

FDS_test2

Runtime error

Update app.py

1bd6b4d verified 6 months ago

4.03 kB

import sys
import subprocess
import pandas as pd

# [필수 라이브러리 체크 및 자동 설치]
for package in ["transformers", "torch", "shap", "pandas"]:
    try:
        __import__(package)
    except ModuleNotFoundError:
        subprocess.check_call([sys.executable, "-m", "pip", "install", package])

from transformers import pipeline
import shap

# ==========================================
# 1. 고도화된 5세대 FDS용 LLM 모델 로드
# ==========================================
print("🔄 FDS 비정형 맥락 분석을 위한 LLM 파이프라인을 초기화 중입니다...")
classifier = pipeline("sentiment-analysis", model="distilbert-base-uncased-finetuned-sst-2-english", top_k=None)

# ==========================================
# 2. SHAP 기반의 XAI 예측 함수 정의 (오류 수정 완료)
# ==========================================
def fds_llm_predict(texts):
    """
    SHAP이 내부적으로 텍스트를 마스킹하여 numpy.ndarray 형태로 전송하므로,
    Hugging Face 파이프라인이 인식할 수 있도록 강제로 파이썬 list 타입으로 캐스팅합니다.
    """
    # 🌟 [CRITICAL FIX] numpy array 등을 순수 파이썬 리스트로 변환
    if not isinstance(texts, list):
        texts = list(texts)
        
    results = classifier(texts)
    fraud_probabilities = []
    for res in results:
        # 'NEGATIVE'(의심 징후) 레이블의 확률 스코어를 추출
        risk_score = next(item['score'] for item in res if item['label'] == 'NEGATIVE')
        fraud_probabilities.append(risk_score)
    return fraud_probabilities

# 텍스트 데이터의 단어(Word) 단위를 마스킹하며 추적하는 SHAP 익스플레이너 생성
explainer = shap.Explainer(fds_llm_predict, shap.maskers.Text(tokenizer=r"\W+"))

# ==========================================
# 3. 실무 시연용 의심 트랜잭션 콘텍스트 정의
# ==========================================
suspicious_transaction_context = (
    "The elderly customer requested an urgent transfer of $45,000 to an unknown account. "
    "She appears extremely nervous, continually checking her smartphone, and mentioned that "
    "a stranger instructed her via an unverified remote control app to complete this transaction immediately."
)

print("\n" + "="*60)
print("📥 [수집된 비정형 데이터 분석 대상]")
print(suspicious_transaction_context)
print("="*60)

# ==========================================
# 4. LLM 추론 및 SHAP 가중치 분석 실행
# ==========================================
print("\n🤖 LLM 분석 및 SHAP 가치 계산 가동 중...")

# 1) LLM 최종 판정
final_risk_score = fds_llm_predict([suspicious_transaction_context])[0]

# 2) SHAP 가치 계산 (이제 에러 없이 정상 작동합니다)
shap_values = explainer([suspicious_transaction_context])

# ==========================================
# 5. 실무자 보고용 결과 데이터 정제
# ==========================================
words = shap_values.data[0]       
contributions = shap_values.values[0] 

fds_report = pd.DataFrame({
    'Detected_Word': words,
    'Risk_Contribution': contributions
})

fds_report = fds_report[fds_report['Detected_Word'].str.strip() != ""]
fds_report_sorted = fds_report.sort_values(by='Risk_Contribution', ascending=False)

# ==========================================
# 6. 관제 시스템 출력 시뮬레이션
# ==========================================
print("\n🚨 [FDS 5세대 관제 시스템 알림]")
if final_risk_score > 0.85:
    print(f"▶ 최종 조치: ❌ [즉시 차단] 금융사기 의심 문맥 포착")
elif final_risk_score > 0.50:
    print(f"▶ 최종 조치: ⚠️ [추가 인증] 의심 징후 탐지")
else:
    print(f"▶ 최종 조치: ✅ [정상 승인]")

print(f"▶ 종합 리스크 스코어: {round(final_risk_score * 100, 2)}%\n")

print("💡 [XAI 소명 가이드 - 위험 기여도 Top 5 단어]")
print("-" * 50)
print(fds_report_sorted.head(5).to_string(index=False))
print("-" * 50)

	import sys
	import subprocess
	import pandas as pd

	# [필수 라이브러리 체크 및 자동 설치]
	for package in ["transformers", "torch", "shap", "pandas"]:
	try:
	__import__(package)
	except ModuleNotFoundError:
	subprocess.check_call([sys.executable, "-m", "pip", "install", package])

	from transformers import pipeline
	import shap

	# ==========================================
	# 1. 고도화된 5세대 FDS용 LLM 모델 로드
	# ==========================================
	print("🔄 FDS 비정형 맥락 분석을 위한 LLM 파이프라인을 초기화 중입니다...")
	classifier = pipeline("sentiment-analysis", model="distilbert-base-uncased-finetuned-sst-2-english", top_k=None)

	# ==========================================
	# 2. SHAP 기반의 XAI 예측 함수 정의 (오류 수정 완료)
	# ==========================================
	def fds_llm_predict(texts):
	"""
	SHAP이 내부적으로 텍스트를 마스킹하여 numpy.ndarray 형태로 전송하므로,
	Hugging Face 파이프라인이 인식할 수 있도록 강제로 파이썬 list 타입으로 캐스팅합니다.
	"""
	# 🌟 [CRITICAL FIX] numpy array 등을 순수 파이썬 리스트로 변환
	if not isinstance(texts, list):
	texts = list(texts)

	results = classifier(texts)
	fraud_probabilities = []
	for res in results:
	# 'NEGATIVE'(의심 징후) 레이블의 확률 스코어를 추출
	risk_score = next(item['score'] for item in res if item['label'] == 'NEGATIVE')
	fraud_probabilities.append(risk_score)
	return fraud_probabilities

	# 텍스트 데이터의 단어(Word) 단위를 마스킹하며 추적하는 SHAP 익스플레이너 생성
	explainer = shap.Explainer(fds_llm_predict, shap.maskers.Text(tokenizer=r"\W+"))

	# ==========================================
	# 3. 실무 시연용 의심 트랜잭션 콘텍스트 정의
	# ==========================================
	suspicious_transaction_context = (
	"The elderly customer requested an urgent transfer of $45,000 to an unknown account. "
	"She appears extremely nervous, continually checking her smartphone, and mentioned that "
	"a stranger instructed her via an unverified remote control app to complete this transaction immediately."
	)

	print("\n" + "="*60)
	print("📥 [수집된 비정형 데이터 분석 대상]")
	print(suspicious_transaction_context)
	print("="*60)

	# ==========================================
	# 4. LLM 추론 및 SHAP 가중치 분석 실행
	# ==========================================
	print("\n🤖 LLM 분석 및 SHAP 가치 계산 가동 중...")

	# 1) LLM 최종 판정
	final_risk_score = fds_llm_predict([suspicious_transaction_context])[0]

	# 2) SHAP 가치 계산 (이제 에러 없이 정상 작동합니다)
	shap_values = explainer([suspicious_transaction_context])

	# ==========================================
	# 5. 실무자 보고용 결과 데이터 정제
	# ==========================================
	words = shap_values.data[0]
	contributions = shap_values.values[0]

	fds_report = pd.DataFrame({
	'Detected_Word': words,
	'Risk_Contribution': contributions
	})

	fds_report = fds_report[fds_report['Detected_Word'].str.strip() != ""]
	fds_report_sorted = fds_report.sort_values(by='Risk_Contribution', ascending=False)

	# ==========================================
	# 6. 관제 시스템 출력 시뮬레이션
	# ==========================================
	print("\n🚨 [FDS 5세대 관제 시스템 알림]")
	if final_risk_score > 0.85:
	print(f"▶ 최종 조치: ❌ [즉시 차단] 금융사기 의심 문맥 포착")
	elif final_risk_score > 0.50:
	print(f"▶ 최종 조치: ⚠️ [추가 인증] 의심 징후 탐지")
	else:
	print(f"▶ 최종 조치: ✅ [정상 승인]")

	print(f"▶ 종합 리스크 스코어: {round(final_risk_score * 100, 2)}%\n")

	print("💡 [XAI 소명 가이드 - 위험 기여도 Top 5 단어]")
	print("-" * 50)
	print(fds_report_sorted.head(5).to_string(index=False))
	print("-" * 50)