MCP Market Statistics Server

"""시장 이상 탐지기 테스트""" import pytest import asyncio import time from datetime import datetime, timedelta from unittest.mock import AsyncMock, MagicMock, patch from typing import Dict, List, Any from src.ai.market_anomaly_detector import MarketAnomalyDetector from src.exceptions import ModelNotTrainedError, InsufficientDataError, PredictionError class TestMarketAnomalyDetector: """시장 이상 탐지기 테스트""" @pytest.fixture def detector_config(self): """탐지기 설정""" return { "anomaly_threshold": 2.5, "window_size": 20, "min_anomaly_duration": 3, "algorithms": ["isolation_forest", "statistical", "lstm_autoencoder"], "sensitivity": 0.8, "ensemble_method": "majority_vote", "feature_weights": { "price": 0.3, "volume": 0.25, "volatility": 0.2, "sentiment": 0.15, "technical_indicators": 0.1 } } @pytest.fixture def anomaly_detector(self, detector_config): """시장 이상 탐지기 인스턴스""" return MarketAnomalyDetector(detector_config) @pytest.fixture def normal_market_data(self): """정상 시장 데이터""" return [ { "timestamp": "2024-01-15T09:00:00", "symbol": "005930", "price": 75000, "volume": 1000000, "volatility": 0.02, "sentiment_score": 0.1, "rsi": 55 }, { "timestamp": "2024-01-15T09:30:00", "symbol": "005930", "price": 75100, "volume": 1050000, "volatility": 0.021, "sentiment_score": 0.12, "rsi": 56 }, { "timestamp": "2024-01-15T10:00:00", "symbol": "005930", "price": 75200, "volume": 980000, "volatility": 0.019, "sentiment_score": 0.08, "rsi": 57 } ] @pytest.fixture def anomalous_market_data(self): """이상 시장 데이터""" return [ { "timestamp": "2024-01-15T11:00:00", "symbol": "005930", "price": 78000, # 급등 "volume": 5000000, # 거래량 급증 "volatility": 0.08, # 변동성 급증 "sentiment_score": 0.7, # 강한 긍정 감정 "rsi": 85 # 과매수 }, { "timestamp": "2024-01-15T11:30:00", "symbol": "005930", "price": 79500, "volume": 6000000, "volatility": 0.09, "sentiment_score": 0.8, "rsi": 90 } ] def test_detector_initialization(self, anomaly_detector, detector_config): """탐지기 초기화 테스트""" assert anomaly_detector.anomaly_threshold == detector_config["anomaly_threshold"] assert anomaly_detector.window_size == detector_config["window_size"] assert anomaly_detector.min_anomaly_duration == detector_config["min_anomaly_duration"] assert anomaly_detector.algorithms == detector_config["algorithms"] assert anomaly_detector.sensitivity == detector_config["sensitivity"] assert anomaly_detector.ensemble_method == detector_config["ensemble_method"] assert anomaly_detector.feature_weights == detector_config["feature_weights"] assert anomaly_detector.is_trained == False @pytest.mark.asyncio async def test_model_training(self, anomaly_detector, normal_market_data): """모델 훈련 테스트""" # 훈련 전 상태 assert anomaly_detector.is_trained == False # 충분한 훈련 데이터 생성 (정상 데이터) training_data = normal_market_data * 30 # 90개 데이터포인트 # 모델 훈련 training_result = await anomaly_detector.train(training_data) # 훈련 후 상태 assert anomaly_detector.is_trained == True assert "models" in training_result assert "training_metrics" in training_result assert "validation_score" in training_result @pytest.mark.asyncio async def test_single_point_anomaly_detection(self, anomaly_detector, normal_market_data, anomalous_market_data): """단일 데이터포인트 이상 탐지 테스트""" # 모델 훈련 training_data = normal_market_data * 30 await anomaly_detector.train(training_data) # 정상 데이터 테스트 normal_result = await anomaly_detector.detect_anomaly(normal_market_data[0]) assert "is_anomaly" in normal_result assert "anomaly_score" in normal_result assert "anomaly_type" in normal_result assert "confidence" in normal_result assert normal_result["is_anomaly"] == False # 이상 데이터 테스트 anomalous_result = await anomaly_detector.detect_anomaly(anomalous_market_data[0]) assert anomalous_result["is_anomaly"] == True assert anomalous_result["anomaly_score"] > normal_result["anomaly_score"] @pytest.mark.asyncio async def test_batch_anomaly_detection(self, anomaly_detector, normal_market_data, anomalous_market_data): """배치 이상 탐지 테스트""" # 모델 훈련 training_data = normal_market_data * 30 await anomaly_detector.train(training_data) # 혼합 데이터 배치 mixed_data = normal_market_data + anomalous_market_data batch_results = await anomaly_detector.detect_batch_anomalies(mixed_data) # 결과 검증 assert len(batch_results) == len(mixed_data) # 정상 데이터는 이상으로 탐지되지 않아야 함 for i in range(len(normal_market_data)): assert batch_results[i]["is_anomaly"] == False # 이상 데이터는 이상으로 탐지되어야 함 for i in range(len(normal_market_data), len(mixed_data)): assert batch_results[i]["is_anomaly"] == True @pytest.mark.asyncio async def test_time_series_anomaly_detection(self, anomaly_detector, normal_market_data, anomalous_market_data): """시계열 이상 탐지 테스트""" # 모델 훈련 training_data = normal_market_data * 30 await anomaly_detector.train(training_data) # 시계열 데이터 time_series_data = normal_market_data + anomalous_market_data + normal_market_data time_series_results = await anomaly_detector.detect_time_series_anomalies( time_series_data, window_size=3 ) # 결과 검증 assert "anomaly_periods" in time_series_results assert "anomaly_summary" in time_series_results assert "trend_analysis" in time_series_results # 이상 기간이 감지되어야 함 anomaly_periods = time_series_results["anomaly_periods"] assert len(anomaly_periods) > 0 # 각 이상 기간에 대한 정보 확인 for period in anomaly_periods: assert "start_time" in period assert "end_time" in period assert "severity" in period assert "anomaly_type" in period @pytest.mark.asyncio async def test_anomaly_type_classification(self, anomaly_detector, normal_market_data): """이상 유형 분류 테스트""" # 모델 훈련 training_data = normal_market_data * 30 await anomaly_detector.train(training_data) # 다양한 유형의 이상 데이터 price_anomaly = { "timestamp": "2024-01-15T12:00:00", "symbol": "005930", "price": 85000, # 급등 "volume": 1000000, "volatility": 0.02, "sentiment_score": 0.1, "rsi": 55 } volume_anomaly = { "timestamp": "2024-01-15T12:30:00", "symbol": "005930", "price": 75000, "volume": 10000000, # 거래량 급증 "volatility": 0.02, "sentiment_score": 0.1, "rsi": 55 } volatility_anomaly = { "timestamp": "2024-01-15T13:00:00", "symbol": "005930", "price": 75000, "volume": 1000000, "volatility": 0.15, # 변동성 급증 "sentiment_score": 0.1, "rsi": 55 } # 각 유형별 탐지 price_result = await anomaly_detector.detect_anomaly(price_anomaly) volume_result = await anomaly_detector.detect_anomaly(volume_anomaly) volatility_result = await anomaly_detector.detect_anomaly(volatility_anomaly) # 이상 유형이 올바르게 분류되는지 확인 assert price_result["anomaly_type"] in ["price_spike", "multi_feature"] assert volume_result["anomaly_type"] in ["volume_spike", "multi_feature"] assert volatility_result["anomaly_type"] in ["volatility_spike", "multi_feature"] @pytest.mark.asyncio async def test_anomaly_severity_scoring(self, anomaly_detector, normal_market_data): """이상 심각도 점수 테스트""" # 모델 훈련 training_data = normal_market_data * 30 await anomaly_detector.train(training_data) # 다양한 심각도의 이상 데이터 mild_anomaly = { "timestamp": "2024-01-15T14:00:00", "symbol": "005930", "price": 76000, # 약간 상승 "volume": 1200000, # 약간 증가 "volatility": 0.025, # 약간 증가 "sentiment_score": 0.2, "rsi": 60 } severe_anomaly = { "timestamp": "2024-01-15T14:30:00", "symbol": "005930", "price": 85000, # 급등 "volume": 8000000, # 급증 "volatility": 0.12, # 급증 "sentiment_score": 0.8, "rsi": 95 } mild_result = await anomaly_detector.detect_anomaly(mild_anomaly) severe_result = await anomaly_detector.detect_anomaly(severe_anomaly) # 심각도 점수 비교 assert severe_result["anomaly_score"] > mild_result["anomaly_score"] assert severe_result["confidence"] >= mild_result["confidence"] @pytest.mark.asyncio async def test_ensemble_method_voting(self, anomaly_detector, normal_market_data, anomalous_market_data): """앙상블 방법 투표 테스트""" # 모델 훈련 training_data = normal_market_data * 30 await anomaly_detector.train(training_data) # 앙상블 결과 확인 result = await anomaly_detector.detect_anomaly(anomalous_market_data[0]) assert "ensemble_scores" in result assert "algorithm_votes" in result # 각 알고리즘의 결과가 포함되어야 함 ensemble_scores = result["ensemble_scores"] for algorithm in anomaly_detector.algorithms: assert algorithm in ensemble_scores @pytest.mark.asyncio async def test_feature_importance_analysis(self, anomaly_detector, normal_market_data, anomalous_market_data): """피처 중요도 분석 테스트""" # 모델 훈련 training_data = normal_market_data * 30 await anomaly_detector.train(training_data) # 피처 중요도 분석 importance_result = await anomaly_detector.analyze_feature_importance(anomalous_market_data[0]) assert "feature_contributions" in importance_result assert "top_contributing_features" in importance_result assert "anomaly_explanation" in importance_result # 피처 기여도 확인 feature_contributions = importance_result["feature_contributions"] assert all(feature in feature_contributions for feature in ["price", "volume", "volatility"]) @pytest.mark.asyncio async def test_anomaly_threshold_adjustment(self, anomaly_detector, normal_market_data): """이상 임계값 조정 테스트""" # 모델 훈련 training_data = normal_market_data * 30 await anomaly_detector.train(training_data) # 임계값 변경 전 설정 original_threshold = anomaly_detector.anomaly_threshold # 임계값 조정 new_threshold = 1.5 await anomaly_detector.adjust_threshold(new_threshold) assert anomaly_detector.anomaly_threshold == new_threshold # 조정된 임계값으로 검증 test_data = { "timestamp": "2024-01-15T15:00:00", "symbol": "005930", "price": 76500, "volume": 1500000, "volatility": 0.03, "sentiment_score": 0.3, "rsi": 65 } result = await anomaly_detector.detect_anomaly(test_data) assert "threshold_used" in result assert result["threshold_used"] == new_threshold @pytest.mark.asyncio async def test_anomaly_pattern_learning(self, anomaly_detector, normal_market_data, anomalous_market_data): """이상 패턴 학습 테스트""" # 모델 훈련 training_data = normal_market_data * 30 await anomaly_detector.train(training_data) # 새로운 이상 패턴 학습 new_anomaly_patterns = anomalous_market_data * 5 learning_result = await anomaly_detector.learn_new_patterns(new_anomaly_patterns) assert "patterns_learned" in learning_result assert "model_updated" in learning_result assert learning_result["model_updated"] == True # 학습 후 탐지 성능 확인 post_learning_result = await anomaly_detector.detect_anomaly(anomalous_market_data[0]) assert post_learning_result["confidence"] > 0.7 @pytest.mark.asyncio async def test_real_time_anomaly_detection(self, anomaly_detector, normal_market_data): """실시간 이상 탐지 테스트""" # 모델 훈련 training_data = normal_market_data * 30 await anomaly_detector.train(training_data) # 실시간 스트리밍 시뮬레이션 streaming_data = [] for i in range(10): data_point = { "timestamp": f"2024-01-15T{16+i}:00:00", "symbol": "005930", "price": 75000 + i * 100, "volume": 1000000 + i * 50000, "volatility": 0.02 + i * 0.001, "sentiment_score": 0.1 + i * 0.05, "rsi": 55 + i } streaming_data.append(data_point) # 실시간 처리 real_time_results = [] for data_point in streaming_data: result = await anomaly_detector.process_real_time_data(data_point) real_time_results.append(result) # 결과 검증 assert len(real_time_results) == len(streaming_data) for result in real_time_results: assert "is_anomaly" in result assert "processing_time" in result assert result["processing_time"] < 1.0 # 1초 이내 처리 @pytest.mark.asyncio async def test_anomaly_alert_system(self, anomaly_detector, normal_market_data, anomalous_market_data): """이상 알림 시스템 테스트""" # 모델 훈련 training_data = normal_market_data * 30 await anomaly_detector.train(training_data) # 알림 설정 alert_config = { "severity_threshold": "medium", "notification_channels": ["email", "webhook"], "cooldown_period": 300 # 5분 } await anomaly_detector.configure_alerts(alert_config) # 이상 탐지 및 알림 생성 alert_result = await anomaly_detector.detect_with_alerts(anomalous_market_data[0]) assert "alert_triggered" in alert_result assert "alert_details" in alert_result if alert_result["alert_triggered"]: alert_details = alert_result["alert_details"] assert "severity" in alert_details assert "message" in alert_details assert "timestamp" in alert_details @pytest.mark.asyncio async def test_error_handling(self, anomaly_detector): """오류 처리 테스트""" # 훈련되지 않은 모델로 탐지 시도 with pytest.raises(ModelNotTrainedError): await anomaly_detector.detect_anomaly({"price": 100}) # 불충분한 데이터로 훈련 시도 insufficient_data = [{"price": 100}] # 너무 적은 데이터 with pytest.raises(InsufficientDataError): await anomaly_detector.train(insufficient_data) # 잘못된 데이터 형식 await anomaly_detector.train([{"price": 100, "volume": 1000}] * 50) invalid_data = {"invalid_field": "value"} result = await anomaly_detector.detect_anomaly(invalid_data) assert result["is_anomaly"] == False # 기본값으로 처리 @pytest.mark.asyncio async def test_performance_metrics(self, anomaly_detector, normal_market_data, anomalous_market_data): """성능 메트릭 테스트""" # 모델 훈련 training_data = normal_market_data * 30 await anomaly_detector.train(training_data) # 성능 테스트 데이터 test_data = normal_market_data + anomalous_market_data start_time = time.time() batch_results = await anomaly_detector.detect_batch_anomalies(test_data) end_time = time.time() processing_time = end_time - start_time # 성능 메트릭 확인 metrics = anomaly_detector.get_performance_metrics() assert "total_detections" in metrics assert "average_processing_time" in metrics assert "accuracy_metrics" in metrics # 처리 시간이 합리적인지 확인 assert processing_time < 10.0 # 10초 이내 assert len(batch_results) == len(test_data) def test_statistical_anomaly_detection(self, anomaly_detector): """통계적 이상 탐지 테스트""" # 정상 데이터 (평균 100, 표준편차 10) normal_values = [100, 105, 95, 110, 90, 102, 98, 107, 93, 101] # 이상 데이터 anomalous_value = 200 # z-score > 3 # 통계적 이상 탐지 z_score = anomaly_detector._calculate_z_score(anomalous_value, normal_values) is_statistical_anomaly = anomaly_detector._is_statistical_anomaly(z_score) assert abs(z_score) > 3 # 강한 이상 assert is_statistical_anomaly == True # 정상 값 테스트 normal_z_score = anomaly_detector._calculate_z_score(105, normal_values) is_normal = anomaly_detector._is_statistical_anomaly(normal_z_score) assert abs(normal_z_score) < 2 assert is_normal == False