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.analytics.advanced_pattern_recognition import AdvancedPatternRecognition from src.exceptions import ModelNotTrainedError, InsufficientDataError, PredictionError class TestAdvancedPatternRecognition: """고급 패턴 인식 테스트""" @pytest.fixture def pattern_config(self): """패턴 인식 설정""" return { "pattern_types": { "chart_patterns": ["head_shoulders", "double_top", "triangle", "flag", "wedge"], "candlestick_patterns": ["doji", "hammer", "engulfing", "harami", "star"], "technical_patterns": ["divergence", "breakout", "support_resistance", "trend_reversal"], "harmonic_patterns": ["gartley", "butterfly", "bat", "crab", "shark"], "elliott_wave": ["impulse", "corrective", "diagonal"] }, "sensitivity_levels": { "high": 0.9, "medium": 0.7, "low": 0.5 }, "lookback_periods": { "short": 20, "medium": 50, "long": 200 }, "confirmation_requirements": { "volume_confirmation": True, "momentum_confirmation": True, "time_confirmation": True }, "pattern_scoring": { "accuracy_weight": 0.4, "profit_potential_weight": 0.3, "risk_reward_weight": 0.3 } } @pytest.fixture def pattern_recognition(self, pattern_config): """패턴 인식 엔진 인스턴스""" return AdvancedPatternRecognition(pattern_config) @pytest.fixture def sample_ohlcv_data(self): """샘플 OHLCV 데이터""" base_price = 50000 data = [] # 헤드앤숄더 패턴 시뮬레이션 pattern_prices = [ 50000, 51000, 52000, 53000, 52000, # 왼쪽 어깨 51000, 54000, 55000, 54000, 51000, # 머리 52000, 53000, 52000, 51000, 50000 # 오른쪽 어깨 ] for i, price in enumerate(pattern_prices): timestamp = datetime.now() - timedelta(days=len(pattern_prices) - i) data.append({ "timestamp": timestamp.isoformat(), "open": price - 100, "high": price + 200, "low": price - 200, "close": price, "volume": 1000000 + i * 10000 }) return data @pytest.fixture def sample_price_series(self): """샘플 가격 시계열 데이터""" prices = [] base = 50000 # 상승 트렌드 -> 하락 트렌드 전환 패턴 for i in range(100): if i < 50: # 상승 트렌드 price = base + i * 100 + (i % 5) * 50 else: # 하락 트렌드 price = base + 5000 - (i - 50) * 80 - ((i - 50) % 5) * 40 prices.append({ "timestamp": (datetime.now() - timedelta(days=100 - i)).isoformat(), "price": price, "volume": 1000000 + (i % 10) * 50000 }) return prices def test_pattern_recognition_initialization(self, pattern_recognition, pattern_config): """패턴 인식 엔진 초기화 테스트""" assert pattern_recognition.pattern_types == pattern_config["pattern_types"] assert pattern_recognition.sensitivity_levels == pattern_config["sensitivity_levels"] assert pattern_recognition.lookback_periods == pattern_config["lookback_periods"] assert pattern_recognition.confirmation_requirements == pattern_config["confirmation_requirements"] assert pattern_recognition.is_trained == False @pytest.mark.asyncio async def test_model_training(self, pattern_recognition, sample_ohlcv_data): """모델 훈련 테스트""" # 훈련 전 상태 assert pattern_recognition.is_trained == False # 훈련 데이터 준비 (더 많은 데이터) extended_data = sample_ohlcv_data * 10 # 150개 데이터포인트 # 모델 훈련 training_result = await pattern_recognition.train(extended_data) # 훈련 후 상태 assert pattern_recognition.is_trained == True assert "patterns_learned" in training_result assert "training_metrics" in training_result assert "model_performance" in training_result @pytest.mark.asyncio async def test_chart_pattern_detection(self, pattern_recognition, sample_ohlcv_data): """차트 패턴 감지 테스트""" # 모델 훈련 extended_data = sample_ohlcv_data * 10 await pattern_recognition.train(extended_data) # 차트 패턴 감지 chart_patterns = await pattern_recognition.detect_chart_patterns( sample_ohlcv_data, pattern_types=["head_shoulders", "double_top"] ) # 결과 검증 assert "detected_patterns" in chart_patterns assert "pattern_details" in chart_patterns assert "confidence_scores" in chart_patterns # 각 감지된 패턴에 대한 상세 정보 확인 for pattern in chart_patterns["detected_patterns"]: assert "pattern_type" in pattern assert "start_index" in pattern assert "end_index" in pattern assert "confidence" in pattern assert "key_points" in pattern assert 0 <= pattern["confidence"] <= 1 @pytest.mark.asyncio async def test_candlestick_pattern_detection(self, pattern_recognition, sample_ohlcv_data): """캔들스틱 패턴 감지 테스트""" # 모델 훈련 extended_data = sample_ohlcv_data * 10 await pattern_recognition.train(extended_data) # 캔들스틱 패턴 감지 candlestick_patterns = await pattern_recognition.detect_candlestick_patterns( sample_ohlcv_data[-5:], # 최근 5개 캔들 sensitivity="medium" ) # 결과 검증 assert "patterns" in candlestick_patterns assert "interpretations" in candlestick_patterns assert "trading_signals" in candlestick_patterns # 각 패턴에 대한 해석 확인 for pattern in candlestick_patterns["patterns"]: assert "pattern_name" in pattern assert "candle_index" in pattern assert "bullish_bearish" in pattern assert "reliability" in pattern assert pattern["bullish_bearish"] in ["bullish", "bearish", "neutral"] @pytest.mark.asyncio async def test_technical_pattern_detection(self, pattern_recognition, sample_price_series): """기술적 패턴 감지 테스트""" # 모델 훈련 await pattern_recognition.train(sample_price_series) # 기술적 패턴 감지 technical_patterns = await pattern_recognition.detect_technical_patterns( sample_price_series, indicators=["rsi", "macd", "bollinger"] ) # 결과 검증 assert "divergences" in technical_patterns assert "breakouts" in technical_patterns assert "support_resistance_levels" in technical_patterns assert "trend_reversals" in technical_patterns # 지지/저항 레벨 검증 sr_levels = technical_patterns["support_resistance_levels"] assert "support_levels" in sr_levels assert "resistance_levels" in sr_levels assert len(sr_levels["support_levels"]) > 0 assert len(sr_levels["resistance_levels"]) > 0 @pytest.mark.asyncio async def test_harmonic_pattern_detection(self, pattern_recognition, sample_ohlcv_data): """하모닉 패턴 감지 테스트""" # 모델 훈련 extended_data = sample_ohlcv_data * 10 await pattern_recognition.train(extended_data) # 하모닉 패턴 감지 harmonic_patterns = await pattern_recognition.detect_harmonic_patterns( sample_ohlcv_data, fibonacci_tolerance=0.05 ) # 결과 검증 assert "detected_patterns" in harmonic_patterns assert "fibonacci_ratios" in harmonic_patterns assert "potential_reversal_zones" in harmonic_patterns # 각 하모닉 패턴의 피보나치 비율 검증 for pattern in harmonic_patterns["detected_patterns"]: assert "pattern_type" in pattern assert "points" in pattern # X, A, B, C, D 포인트 assert "ratios" in pattern assert "completion_zone" in pattern assert "profit_targets" in pattern @pytest.mark.asyncio async def test_elliott_wave_analysis(self, pattern_recognition, sample_price_series): """엘리엇 파동 분석 테스트""" # 모델 훈련 await pattern_recognition.train(sample_price_series) # 엘리엇 파동 분석 elliott_analysis = await pattern_recognition.analyze_elliott_waves( sample_price_series, wave_degree="primary" ) # 결과 검증 assert "wave_count" in elliott_analysis assert "current_wave" in elliott_analysis assert "wave_structure" in elliott_analysis assert "next_wave_projection" in elliott_analysis # 파동 구조 검증 wave_structure = elliott_analysis["wave_structure"] for wave in wave_structure: assert "wave_label" in wave assert "start_point" in wave assert "end_point" in wave assert "wave_type" in wave assert wave["wave_type"] in ["impulse", "corrective", "diagonal"] @pytest.mark.asyncio async def test_pattern_confirmation(self, pattern_recognition, sample_ohlcv_data): """패턴 확인 테스트""" # 모델 훈련 extended_data = sample_ohlcv_data * 10 await pattern_recognition.train(extended_data) # 패턴 감지 detected_patterns = await pattern_recognition.detect_all_patterns(sample_ohlcv_data) # 패턴 확인 confirmed_patterns = await pattern_recognition.confirm_patterns( detected_patterns, sample_ohlcv_data, confirmation_criteria=["volume", "momentum", "time"] ) # 결과 검증 assert "confirmed_patterns" in confirmed_patterns assert "confirmation_scores" in confirmed_patterns assert "failed_confirmations" in confirmed_patterns # 확인된 패턴은 더 높은 신뢰도를 가져야 함 for pattern in confirmed_patterns["confirmed_patterns"]: assert "original_confidence" in pattern assert "confirmed_confidence" in pattern assert pattern["confirmed_confidence"] >= pattern["original_confidence"] @pytest.mark.asyncio async def test_pattern_backtesting(self, pattern_recognition, sample_ohlcv_data): """패턴 백테스팅 테스트""" # 모델 훈련 extended_data = sample_ohlcv_data * 10 await pattern_recognition.train(extended_data) # 패턴 백테스팅 backtest_results = await pattern_recognition.backtest_patterns( extended_data, entry_rules={"confidence_threshold": 0.7}, exit_rules={"profit_target": 0.05, "stop_loss": 0.02} ) # 결과 검증 assert "total_patterns" in backtest_results assert "successful_patterns" in backtest_results assert "win_rate" in backtest_results assert "average_profit" in backtest_results assert "profit_factor" in backtest_results assert "pattern_performance" in backtest_results # 패턴별 성과 확인 for pattern_type, performance in backtest_results["pattern_performance"].items(): assert "count" in performance assert "win_rate" in performance assert "avg_return" in performance assert "best_return" in performance assert "worst_return" in performance @pytest.mark.asyncio async def test_multi_timeframe_pattern_analysis(self, pattern_recognition, sample_ohlcv_data): """멀티 타임프레임 패턴 분석 테스트""" # 모델 훈련 extended_data = sample_ohlcv_data * 10 await pattern_recognition.train(extended_data) # 멀티 타임프레임 분석 mtf_analysis = await pattern_recognition.analyze_multi_timeframe_patterns( sample_ohlcv_data, timeframes=["5m", "15m", "1h", "4h"] ) # 결과 검증 assert "timeframe_patterns" in mtf_analysis assert "confluence_zones" in mtf_analysis assert "dominant_patterns" in mtf_analysis assert "timeframe_alignment" in mtf_analysis # 각 타임프레임별 패턴 확인 for timeframe, patterns in mtf_analysis["timeframe_patterns"].items(): assert isinstance(patterns, list) for pattern in patterns: assert "pattern_type" in pattern assert "strength" in pattern assert "direction" in pattern @pytest.mark.asyncio async def test_pattern_strength_scoring(self, pattern_recognition, sample_ohlcv_data): """패턴 강도 점수 테스트""" # 모델 훈련 extended_data = sample_ohlcv_data * 10 await pattern_recognition.train(extended_data) # 패턴 감지 detected_patterns = await pattern_recognition.detect_all_patterns(sample_ohlcv_data) # 패턴 강도 점수 계산 pattern_scores = await pattern_recognition.calculate_pattern_strength( detected_patterns["all_patterns"], market_context={"trend": "bullish", "volatility": "medium"} ) # 결과 검증 assert "scored_patterns" in pattern_scores assert "strength_distribution" in pattern_scores assert "strongest_patterns" in pattern_scores # 각 패턴의 점수 확인 for pattern in pattern_scores["scored_patterns"]: assert "strength_score" in pattern assert "components" in pattern assert 0 <= pattern["strength_score"] <= 1 # 점수 구성 요소 components = pattern["components"] assert "pattern_clarity" in components assert "volume_support" in components assert "market_alignment" in components @pytest.mark.asyncio async def test_pattern_prediction(self, pattern_recognition, sample_ohlcv_data): """패턴 기반 예측 테스트""" # 모델 훈련 extended_data = sample_ohlcv_data * 10 await pattern_recognition.train(extended_data) # 현재 패턴 감지 current_patterns = await pattern_recognition.detect_all_patterns(sample_ohlcv_data) # 패턴 기반 예측 predictions = await pattern_recognition.predict_from_patterns( current_patterns["all_patterns"], prediction_horizon=5 ) # 결과 검증 assert "price_predictions" in predictions assert "direction_probability" in predictions assert "target_levels" in predictions assert "confidence_intervals" in predictions # 예측 확률 검증 direction_prob = predictions["direction_probability"] assert "bullish" in direction_prob assert "bearish" in direction_prob assert "neutral" in direction_prob assert abs(sum(direction_prob.values()) - 1.0) < 0.01 # 합이 1 @pytest.mark.asyncio async def test_pattern_alerts(self, pattern_recognition, sample_ohlcv_data): """패턴 알림 테스트""" # 모델 훈련 extended_data = sample_ohlcv_data * 10 await pattern_recognition.train(extended_data) # 알림 규칙 설정 alert_rules = { "patterns": ["head_shoulders", "double_top", "breakout"], "min_confidence": 0.8, "min_strength": 0.7 } await pattern_recognition.configure_alerts(alert_rules) # 실시간 패턴 감지 및 알림 alerts = await pattern_recognition.check_pattern_alerts(sample_ohlcv_data) # 결과 검증 assert "triggered_alerts" in alerts assert "alert_summary" in alerts assert "next_check_time" in alerts # 각 알림 확인 for alert in alerts["triggered_alerts"]: assert "pattern_type" in alert assert "confidence" in alert assert "strength" in alert assert "message" in alert assert "severity" in alert assert "timestamp" in alert @pytest.mark.asyncio async def test_pattern_visualization_data(self, pattern_recognition, sample_ohlcv_data): """패턴 시각화 데이터 테스트""" # 모델 훈련 extended_data = sample_ohlcv_data * 10 await pattern_recognition.train(extended_data) # 패턴 감지 patterns = await pattern_recognition.detect_chart_patterns(sample_ohlcv_data) # 시각화 데이터 생성 viz_data = await pattern_recognition.generate_visualization_data( patterns["detected_patterns"], sample_ohlcv_data ) # 결과 검증 assert "chart_annotations" in viz_data assert "pattern_overlays" in viz_data assert "key_levels" in viz_data assert "trading_zones" in viz_data # 차트 주석 확인 for annotation in viz_data["chart_annotations"]: assert "x" in annotation assert "y" in annotation assert "text" in annotation assert "type" in annotation @pytest.mark.asyncio async def test_pattern_statistics(self, pattern_recognition, sample_ohlcv_data): """패턴 통계 테스트""" # 모델 훈련 extended_data = sample_ohlcv_data * 20 # 더 많은 데이터 await pattern_recognition.train(extended_data) # 패턴 통계 수집 statistics = await pattern_recognition.get_pattern_statistics( extended_data, time_period="all" ) # 결과 검증 assert "pattern_frequency" in statistics assert "pattern_success_rates" in statistics assert "average_pattern_duration" in statistics assert "pattern_correlations" in statistics assert "seasonal_patterns" in statistics # 패턴 빈도 확인 pattern_freq = statistics["pattern_frequency"] assert isinstance(pattern_freq, dict) for pattern, freq in pattern_freq.items(): assert freq >= 0 @pytest.mark.asyncio async def test_real_time_pattern_detection(self, pattern_recognition, sample_ohlcv_data): """실시간 패턴 감지 테스트""" # 모델 훈련 extended_data = sample_ohlcv_data * 10 await pattern_recognition.train(extended_data) # 실시간 스트리밍 시뮬레이션 streaming_buffer = [] for candle in sample_ohlcv_data: streaming_buffer.append(candle) # 실시간 패턴 감지 real_time_patterns = await pattern_recognition.detect_real_time_patterns( streaming_buffer, min_data_points=5 ) # 결과 검증 assert "active_patterns" in real_time_patterns assert "completed_patterns" in real_time_patterns assert "emerging_patterns" in real_time_patterns assert "pattern_updates" in real_time_patterns @pytest.mark.asyncio async def test_pattern_combination_analysis(self, pattern_recognition, sample_ohlcv_data): """패턴 조합 분석 테스트""" # 모델 훈련 extended_data = sample_ohlcv_data * 10 await pattern_recognition.train(extended_data) # 모든 패턴 감지 all_patterns = await pattern_recognition.detect_all_patterns(sample_ohlcv_data) # 패턴 조합 분석 combinations = await pattern_recognition.analyze_pattern_combinations( all_patterns["all_patterns"], combination_rules={"max_overlap": 0.5, "min_correlation": 0.3} ) # 결과 검증 assert "pattern_combinations" in combinations assert "synergy_scores" in combinations assert "conflict_analysis" in combinations assert "optimal_combinations" in combinations # 각 조합 확인 for combo in combinations["pattern_combinations"]: assert "patterns" in combo assert "combined_strength" in combo assert "synergy_score" in combo assert "recommendation" in combo @pytest.mark.asyncio async def test_error_handling(self, pattern_recognition): """오류 처리 테스트""" # 훈련되지 않은 모델로 패턴 감지 시도 with pytest.raises(ModelNotTrainedError): await pattern_recognition.detect_chart_patterns([]) # 불충분한 데이터로 훈련 시도 insufficient_data = [{"open": 100, "close": 100}] # 너무 적은 데이터 with pytest.raises(InsufficientDataError): await pattern_recognition.train(insufficient_data) # 잘못된 패턴 타입 await pattern_recognition.train([{"open": 100, "close": 100}] * 100) with pytest.raises(ValueError): await pattern_recognition.detect_chart_patterns([], pattern_types=["invalid_pattern"]) def test_pattern_configuration(self, pattern_recognition): """패턴 설정 테스트""" # 감도 레벨 조정 pattern_recognition.set_sensitivity("high") assert pattern_recognition.current_sensitivity == pattern_recognition.sensitivity_levels["high"] # 패턴 타입 활성화/비활성화 pattern_recognition.enable_pattern_types(["head_shoulders", "double_top"]) assert "head_shoulders" in pattern_recognition.active_pattern_types assert "double_top" in pattern_recognition.active_pattern_types pattern_recognition.disable_pattern_types(["double_top"]) assert "double_top" not in pattern_recognition.active_pattern_types @pytest.mark.asyncio async def test_performance_optimization(self, pattern_recognition, sample_ohlcv_data): """성능 최적화 테스트""" # 모델 훈련 extended_data = sample_ohlcv_data * 50 # 큰 데이터셋 await pattern_recognition.train(extended_data) # 성능 측정 start_time = time.time() # 병렬 패턴 감지 tasks = [ pattern_recognition.detect_chart_patterns(extended_data), pattern_recognition.detect_candlestick_patterns(extended_data[-10:]), pattern_recognition.detect_technical_patterns(extended_data) ] results = await asyncio.gather(*tasks) end_time = time.time() # 성능 확인 processing_time = end_time - start_time assert processing_time < 5.0 # 5초 이내 assert len(results) == 3 # 캐시 활용 확인 metrics = pattern_recognition.get_performance_metrics() assert metrics["cache_hit_rate"] > 0 # 캐시가 활용되었는지 확인