MCP Market Statistics Server

"""이상 징후 탐지 도구 테스트""" import pytest import random import math from datetime import datetime, timedelta from unittest.mock import AsyncMock, MagicMock from src.tools.anomaly_tools import AnomalyDetectionTool from src.exceptions import DataValidationError, DatabaseConnectionError class TestAnomalyDetectionTool: """이상 징후 탐지 도구 테스트""" @pytest.fixture def mock_db_manager(self): """Mock 데이터베이스 매니저""" return AsyncMock() @pytest.fixture def mock_cache_manager(self): """Mock 캐시 매니저""" return AsyncMock() @pytest.fixture def anomaly_tool(self, mock_db_manager, mock_cache_manager): """이상 징후 탐지 도구 인스턴스""" return AnomalyDetectionTool(mock_db_manager, mock_cache_manager) @pytest.fixture def sample_market_data(self): """샘플 시장 데이터 (정상 패턴 + 이상치 포함)""" base_date = datetime.now().date() data = [] for i in range(100): # 100일 데이터 date = base_date - timedelta(days=i) # 대부분 정상 데이터 if i < 90: price = 2650 + random.gauss(0, 30) # 정상 변동 volume = 450000000 + random.gauss(0, 50000000) # 정상 거래량 volatility = 0.015 + random.uniform(-0.005, 0.005) # 정상 변동성 else: # 이상치 데이터 (마지막 10일) price = 2650 + random.gauss(0, 150) # 큰 변동 volume = 450000000 + random.gauss(0, 200000000) # 거래량 급증 volatility = 0.035 + random.uniform(-0.015, 0.015) # 높은 변동성 data.append({ "date": date, "market": "KOSPI", "close_price": max(price, 100), # 음수 방지 "volume": max(int(volume), 1000000), # 최소값 보장 "daily_return": random.gauss(0.001, 0.02), "volatility": max(volatility, 0.001), # 양수 보장 "vix": 20 + random.gauss(0, 5), "put_call_ratio": 0.8 + random.uniform(-0.3, 0.3) }) return data def test_tool_initialization(self, anomaly_tool, mock_db_manager, mock_cache_manager): """도구 초기화 테스트""" assert anomaly_tool.name == "detect_market_anomalies" assert anomaly_tool.description is not None assert "이상 징후" in anomaly_tool.description or "anomaly" in anomaly_tool.description.lower() assert anomaly_tool.db_manager == mock_db_manager assert anomaly_tool.cache_manager == mock_cache_manager def test_tool_definition(self, anomaly_tool): """도구 정의 테스트""" definition = anomaly_tool.get_tool_definition() assert definition.name == "detect_market_anomalies" assert definition.description is not None assert definition.inputSchema is not None # 입력 스키마 검증 schema = definition.inputSchema assert schema["type"] == "object" assert "properties" in schema properties = schema["properties"] assert "market" in properties assert "detection_methods" in properties assert "lookback_period" in properties assert "sensitivity" in properties assert "include_realtime_alerts" in properties # detection_methods 파라미터 검증 methods_prop = properties["detection_methods"] assert methods_prop["type"] == "array" assert "statistical" in str(methods_prop) assert "isolation_forest" in str(methods_prop) assert "timeseries" in str(methods_prop) @pytest.mark.asyncio async def test_execute_statistical_anomaly_detection(self, anomaly_tool, sample_market_data): """통계적 이상 탐지 테스트""" # 캐시 미스 anomaly_tool.cache_manager.get.return_value = None # 데이터베이스 응답 설정 anomaly_tool.db_manager.fetch_all.return_value = sample_market_data # 실행 result = await anomaly_tool.execute({ "market": "KOSPI", "detection_methods": ["statistical"], "lookback_period": "30d", "sensitivity": 2.0 }) # 결과 검증 assert len(result) == 1 content = result[0] assert content.type == "text" # JSON 파싱하여 내용 확인 import json data = json.loads(content.text) assert "timestamp" in data assert "market" in data assert "anomaly_detection_results" in data # 통계적 이상 탐지 결과 검증 results = data["anomaly_detection_results"] assert "statistical" in results stat_results = results["statistical"] assert "z_score_anomalies" in stat_results assert "iqr_anomalies" in stat_results assert "anomaly_count" in stat_results assert "anomaly_percentage" in stat_results @pytest.mark.asyncio async def test_execute_isolation_forest_detection(self, anomaly_tool, sample_market_data): """Isolation Forest 이상 탐지 테스트""" anomaly_tool.cache_manager.get.return_value = None anomaly_tool.db_manager.fetch_all.return_value = sample_market_data # 실행 result = await anomaly_tool.execute({ "market": "KOSPI", "detection_methods": ["isolation_forest"], "lookback_period": "60d", "contamination": 0.1 }) # 결과 검증 content = result[0] import json data = json.loads(content.text) results = data["anomaly_detection_results"] assert "isolation_forest" in results if_results = results["isolation_forest"] assert "anomaly_scores" in if_results assert "anomalies_detected" in if_results assert "feature_importance" in if_results assert "model_stats" in if_results @pytest.mark.asyncio async def test_execute_timeseries_anomaly_detection(self, anomaly_tool, sample_market_data): """시계열 이상 패턴 탐지 테스트""" anomaly_tool.cache_manager.get.return_value = None anomaly_tool.db_manager.fetch_all.return_value = sample_market_data # 실행 result = await anomaly_tool.execute({ "market": "KOSPI", "detection_methods": ["timeseries"], "lookback_period": "90d" }) # 결과 검증 content = result[0] import json data = json.loads(content.text) results = data["anomaly_detection_results"] assert "timeseries" in results ts_results = results["timeseries"] assert "seasonal_anomalies" in ts_results assert "trend_breaks" in ts_results assert "volatility_regimes" in ts_results assert "structural_breaks" in ts_results @pytest.mark.asyncio async def test_comprehensive_anomaly_detection(self, anomaly_tool, sample_market_data): """종합 이상 탐지 테스트 (모든 방법)""" anomaly_tool.cache_manager.get.return_value = None anomaly_tool.db_manager.fetch_all.return_value = sample_market_data # 실행 result = await anomaly_tool.execute({ "market": "KOSPI", "detection_methods": ["statistical", "isolation_forest", "timeseries"], "lookback_period": "60d", "sensitivity": 2.5, "contamination": 0.05, "include_feature_analysis": True, "include_risk_assessment": True }) # 결과 검증 content = result[0] import json data = json.loads(content.text) assert "anomaly_detection_results" in data assert "anomaly_summary" in data assert "risk_assessment" in data assert "feature_analysis" in data # 모든 탐지 방법 결과 확인 results = data["anomaly_detection_results"] assert "statistical" in results assert "isolation_forest" in results assert "timeseries" in results # 종합 요약 확인 summary = data["anomaly_summary"] assert "total_anomalies_detected" in summary assert "severity_distribution" in summary assert "confidence_scores" in summary def test_z_score_calculation(self, anomaly_tool): """Z-score 이상 탐지 계산 테스트""" # 정상 데이터 + 이상치 data = [1, 2, 3, 2, 1, 2, 3, 50, 2, 1] # 50이 이상치 anomalies = anomaly_tool._detect_z_score_anomalies(data, threshold=2.0) assert len(anomalies) > 0 assert any(item["value"] == 50 for item in anomalies) assert all(-1 <= item["z_score"] or item["z_score"] >= 1 for item in anomalies) def test_iqr_anomaly_detection(self, anomaly_tool): """IQR 이상 탐지 테스트""" # 이상치가 포함된 데이터 data = list(range(10, 20)) + [100, 105] # 100, 105가 이상치 anomalies = anomaly_tool._detect_iqr_anomalies(data) assert len(anomalies) >= 2 assert any(item["value"] == 100 for item in anomalies) assert any(item["value"] == 105 for item in anomalies) def test_isolation_forest_model(self, anomaly_tool): """Isolation Forest 모델 테스트""" # 다차원 피처 데이터 features = [ [1, 2, 3], [2, 3, 4], [1, 1, 2], [2, 2, 3], [10, 15, 20], # 이상치 [1, 2, 2], [2, 3, 3] ] anomalies, scores, model_stats = anomaly_tool._detect_isolation_forest_anomalies( features, contamination=0.2 ) assert len(anomalies) > 0 assert len(scores) == len(features) assert "n_estimators" in model_stats assert "contamination" in model_stats def test_volatility_regime_detection(self, anomaly_tool): """변동성 체제 탐지 테스트""" # 변동성 데이터 (낮음 -> 높음 -> 낮음) volatility_data = ([0.01] * 30 + [0.05] * 20 + [0.01] * 30) regimes = anomaly_tool._detect_volatility_regimes(volatility_data) assert len(regimes) >= 2 # 최소 2개의 체제 변화 assert all("start_index" in regime for regime in regimes) assert all("regime_type" in regime for regime in regimes) assert all("volatility_level" in regime for regime in regimes) def test_structural_break_detection(self, anomaly_tool): """구조적 변화점 탐지 테스트""" # 구조적 변화가 있는 시계열 (평균 변화) data = [10] * 50 + [20] * 50 # 50일 후 평균 변화 breaks = anomaly_tool._detect_structural_breaks(data) assert len(breaks) > 0 # 변화점이 대략 50 근처에서 발견되어야 함 assert any(40 < break_info["breakpoint"] < 60 for break_info in breaks) @pytest.mark.asyncio async def test_realtime_alert_system(self, anomaly_tool): """실시간 알림 시스템 테스트""" # 심각한 이상 징후 데이터 severe_anomaly_data = [{ "date": datetime.now().date(), "market": "KOSPI", "close_price": 1000, # 급락 "volume": 1500000000, # 거래량 급증 "volatility": 0.08, # 높은 변동성 "daily_return": -0.15, # 큰 하락 "vix": 45, # 공포지수 상승 "put_call_ratio": 2.0 # Put 급증 }] anomaly_tool.cache_manager.get.return_value = None anomaly_tool.db_manager.fetch_all.return_value = severe_anomaly_data # 실행 (실시간 알림 포함) result = await anomaly_tool.execute({ "market": "KOSPI", "detection_methods": ["statistical"], "include_realtime_alerts": True, "alert_threshold": "high" }) content = result[0] import json data = json.loads(content.text) # 실시간 알림 정보 확인 if "realtime_alerts" in data: alerts = data["realtime_alerts"] assert "critical_anomalies" in alerts assert "alert_level" in alerts assert "recommended_actions" in alerts @pytest.mark.asyncio async def test_cache_functionality(self, anomaly_tool): """캐시 기능 테스트""" # 캐시 히트 시나리오 cached_data = { "timestamp": datetime.now().isoformat(), "market": "KOSPI", "anomaly_detection_results": { "statistical": {"anomaly_count": 5} } } anomaly_tool.cache_manager.get.return_value = cached_data # 실행 result = await anomaly_tool.execute({ "market": "KOSPI", "detection_methods": ["statistical"], "lookback_period": "30d" }) # 캐시에서 데이터 반환 확인 content = result[0] import json data = json.loads(content.text) assert data == cached_data # 데이터베이스 호출 없음 확인 anomaly_tool.db_manager.fetch_all.assert_not_called() @pytest.mark.asyncio async def test_error_handling(self, anomaly_tool): """에러 처리 테스트""" anomaly_tool.cache_manager.get.return_value = None anomaly_tool.db_manager.fetch_all.side_effect = DatabaseConnectionError("DB 연결 실패") with pytest.raises(DatabaseConnectionError): await anomaly_tool.execute({ "market": "KOSPI", "detection_methods": ["statistical"] }) @pytest.mark.asyncio async def test_invalid_parameters(self, anomaly_tool): """잘못된 파라미터 테스트""" # 잘못된 시장 with pytest.raises(ValueError, match="Invalid market"): await anomaly_tool.execute({ "market": "INVALID", "detection_methods": ["statistical"] }) # 빈 탐지 방법 목록 with pytest.raises(ValueError, match="At least one detection method"): await anomaly_tool.execute({ "market": "KOSPI", "detection_methods": [] }) # 잘못된 민감도 값 with pytest.raises(ValueError, match="Invalid sensitivity"): await anomaly_tool.execute({ "market": "KOSPI", "detection_methods": ["statistical"], "sensitivity": -1.0 }) @pytest.mark.asyncio async def test_insufficient_data_handling(self, anomaly_tool): """데이터 부족 처리 테스트""" # 데이터가 부족한 경우 (5일 데이터) insufficient_data = [ { "date": datetime.now().date() - timedelta(days=i), "market": "KOSPI", "close_price": 2650, "volume": 450000000, "volatility": 0.015 } for i in range(5) ] anomaly_tool.cache_manager.get.return_value = None anomaly_tool.db_manager.fetch_all.return_value = insufficient_data result = await anomaly_tool.execute({ "market": "KOSPI", "detection_methods": ["statistical"], "lookback_period": "30d" }) content = result[0] import json data = json.loads(content.text) assert "warning" in data or "insufficient data" in str(data).lower() def test_feature_extraction(self, anomaly_tool, sample_market_data): """피처 추출 테스트""" features = anomaly_tool._extract_features(sample_market_data[:30]) assert len(features) == 30 assert len(features[0]) > 5 # 최소 5개 피처 assert all(isinstance(row, list) for row in features) assert all(isinstance(val, (int, float)) for row in features for val in row) def test_risk_scoring(self, anomaly_tool): """리스크 스코어링 테스트""" # 다양한 이상 탐지 결과 detection_results = { "statistical": { "anomaly_count": 10, "severe_anomalies": 3 }, "isolation_forest": { "anomalies_detected": 8, "avg_anomaly_score": -0.15 } } risk_assessment = anomaly_tool._assess_risk_level(detection_results) assert "overall_risk_level" in risk_assessment assert "risk_factors" in risk_assessment assert "confidence_score" in risk_assessment assert risk_assessment["overall_risk_level"] in ["low", "medium", "high", "critical"]