MCP Market Statistics Server

"""시장 국면 판단 도구 테스트""" import pytest import random import math from datetime import datetime, timedelta from unittest.mock import AsyncMock, MagicMock from src.tools.regime_tools import MarketRegimeTool from src.exceptions import DataValidationError, DatabaseConnectionError class TestMarketRegimeTool: """시장 국면 판단 도구 테스트""" @pytest.fixture def mock_db_manager(self): """Mock 데이터베이스 매니저""" return AsyncMock() @pytest.fixture def mock_cache_manager(self): """Mock 캐시 매니저""" return AsyncMock() @pytest.fixture def regime_tool(self, mock_db_manager, mock_cache_manager): """시장 국면 판단 도구 인스턴스""" return MarketRegimeTool(mock_db_manager, mock_cache_manager) @pytest.fixture def sample_regime_data(self): """샘플 시장 국면 데이터 (다양한 국면 포함)""" base_date = datetime.now().date() data = [] # 120일 데이터 - 여러 국면을 시뮬레이션 for i in range(120): date = base_date - timedelta(days=i) if i < 30: # Bull market (최근 30일) price_trend = 0.002 # 상승 추세 volatility = 0.012 # 낮은 변동성 volume_factor = 1.1 # 약간 높은 거래량 elif i < 60: # Sideways market (30-60일) price_trend = 0.0005 # 약한 상승 volatility = 0.008 # 매우 낮은 변동성 volume_factor = 0.9 # 낮은 거래량 elif i < 90: # Volatile market (60-90일) price_trend = 0.001 # 혼재 volatility = 0.025 # 높은 변동성 volume_factor = 1.3 # 높은 거래량 else: # Bear market (90-120일) price_trend = -0.0015 # 하락 추세 volatility = 0.018 # 중간 변동성 volume_factor = 1.2 # 높은 거래량 # 시장 데이터 생성 base_price = 2650 daily_return = random.gauss(price_trend, volatility) volume = int(450000000 * volume_factor * (1 + random.gauss(0, 0.2))) data.append({ "date": date, "market": "KOSPI", "close_price": base_price, "daily_return": daily_return, "volume": max(volume, 100000000), "volatility": volatility + random.uniform(-0.003, 0.003), "rsi": 50 + random.gauss(0, 15), "macd": random.gauss(0, 2), "bollinger_position": random.uniform(0, 1), "vix": 20 + random.gauss(0, 8), "put_call_ratio": 0.8 + random.uniform(-0.3, 0.3), "advance_decline_ratio": 1.0 + random.gauss(0, 0.3) }) return data def test_tool_initialization(self, regime_tool, mock_db_manager, mock_cache_manager): """도구 초기화 테스트""" assert regime_tool.name == "get_market_regime" assert regime_tool.description is not None assert "국면" in regime_tool.description or "regime" in regime_tool.description.lower() assert regime_tool.db_manager == mock_db_manager assert regime_tool.cache_manager == mock_cache_manager def test_tool_definition(self, regime_tool): """도구 정의 테스트""" definition = regime_tool.get_tool_definition() assert definition.name == "get_market_regime" 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 "analysis_methods" in properties assert "lookback_period" in properties assert "regime_types" in properties assert "include_predictions" in properties # analysis_methods 파라미터 검증 methods_prop = properties["analysis_methods"] assert methods_prop["type"] == "array" assert "hmm" in str(methods_prop) assert "statistical" in str(methods_prop) assert "technical" in str(methods_prop) @pytest.mark.asyncio async def test_execute_hmm_regime_analysis(self, regime_tool, sample_regime_data): """HMM 기반 시장 국면 분석 테스트""" # 캐시 미스 regime_tool.cache_manager.get.return_value = None # 데이터베이스 응답 설정 regime_tool.db_manager.fetch_all.return_value = sample_regime_data # 실행 result = await regime_tool.execute({ "market": "KOSPI", "analysis_methods": ["hmm"], "lookback_period": "120d", "n_regimes": 3 }) # 결과 검증 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 "regime_analysis_results" in data # HMM 결과 검증 results = data["regime_analysis_results"] assert "hmm" in results hmm_results = results["hmm"] assert "current_regime" in hmm_results assert "regime_probabilities" in hmm_results assert "regime_history" in hmm_results assert "transition_matrix" in hmm_results assert "regime_characteristics" in hmm_results @pytest.mark.asyncio async def test_execute_statistical_regime_analysis(self, regime_tool, sample_regime_data): """통계적 시장 국면 분석 테스트""" regime_tool.cache_manager.get.return_value = None regime_tool.db_manager.fetch_all.return_value = sample_regime_data # 실행 result = await regime_tool.execute({ "market": "KOSPI", "analysis_methods": ["statistical"], "lookback_period": "90d", "volatility_threshold": 0.02 }) # 결과 검증 content = result[0] import json data = json.loads(content.text) results = data["regime_analysis_results"] assert "statistical" in results stat_results = results["statistical"] assert "volatility_regime" in stat_results assert "return_regime" in stat_results assert "volume_regime" in stat_results assert "regime_classification" in stat_results @pytest.mark.asyncio async def test_execute_technical_regime_analysis(self, regime_tool, sample_regime_data): """기술적 지표 기반 시장 국면 분석 테스트""" regime_tool.cache_manager.get.return_value = None regime_tool.db_manager.fetch_all.return_value = sample_regime_data # 실행 result = await regime_tool.execute({ "market": "KOSPI", "analysis_methods": ["technical"], "lookback_period": "60d" }) # 결과 검증 content = result[0] import json data = json.loads(content.text) results = data["regime_analysis_results"] assert "technical" in results tech_results = results["technical"] assert "trend_regime" in tech_results assert "momentum_regime" in tech_results assert "overbought_oversold" in tech_results assert "support_resistance_levels" in tech_results @pytest.mark.asyncio async def test_comprehensive_regime_analysis(self, regime_tool, sample_regime_data): """종합 시장 국면 분석 테스트 (모든 방법)""" regime_tool.cache_manager.get.return_value = None regime_tool.db_manager.fetch_all.return_value = sample_regime_data # 실행 result = await regime_tool.execute({ "market": "KOSPI", "analysis_methods": ["hmm", "statistical", "technical"], "lookback_period": "120d", "n_regimes": 4, "include_predictions": True, "include_strategy_mapping": True, "prediction_horizon": 10 }) # 결과 검증 content = result[0] import json data = json.loads(content.text) assert "regime_analysis_results" in data assert "regime_consensus" in data assert "predictions" in data assert "strategy_mapping" in data # 모든 분석 방법 결과 확인 results = data["regime_analysis_results"] assert "hmm" in results assert "statistical" in results assert "technical" in results # 컨센서스 확인 consensus = data["regime_consensus"] assert "overall_regime" in consensus assert "confidence_score" in consensus assert "regime_strength" in consensus def test_hmm_model_training(self, regime_tool): """HMM 모델 훈련 테스트""" # 시계열 데이터 (변동성이 다른 3개 구간) returns = ( [random.gauss(0.001, 0.01) for _ in range(30)] + # 저변동성 [random.gauss(0.002, 0.02) for _ in range(30)] + # 중변동성 [random.gauss(-0.001, 0.015) for _ in range(30)] # 하락+중변동성 ) model = regime_tool._train_hmm_model(returns, n_regimes=3) assert "states" in model assert "transition_matrix" in model assert "emission_params" in model assert len(model["states"]) == len(returns) assert len(model["transition_matrix"]) == 3 assert len(model["transition_matrix"][0]) == 3 def test_volatility_regime_detection(self, regime_tool): """변동성 체제 탐지 테스트""" # 변동하는 변동성 데이터 volatilities = [0.01] * 20 + [0.03] * 20 + [0.015] * 20 regime = regime_tool._detect_volatility_regime(volatilities, threshold=0.02) assert "current_regime" in regime assert "regime_history" in regime assert regime["current_regime"] in ["low", "medium", "high"] def test_return_regime_classification(self, regime_tool): """수익률 체제 분류 테스트""" # Bull, Bear, Sideways 패턴 returns = [0.002] * 20 + [-0.0015] * 20 + [0.0003] * 20 regime = regime_tool._classify_return_regime(returns) assert "current_regime" in regime assert "regime_probability" in regime assert regime["current_regime"] in ["bull", "bear", "sideways"] assert 0 <= regime["regime_probability"] <= 1 def test_technical_indicator_analysis(self, regime_tool): """기술적 지표 분석 테스트""" # 샘플 기술 지표 데이터 technical_data = [ {"rsi": 70, "macd": 2.5, "bollinger_position": 0.9}, # Overbought {"rsi": 30, "macd": -1.8, "bollinger_position": 0.1}, # Oversold {"rsi": 50, "macd": 0.2, "bollinger_position": 0.5} # Neutral ] analysis = regime_tool._analyze_technical_indicators(technical_data) assert "trend_regime" in analysis assert "momentum_regime" in analysis assert "overbought_oversold" in analysis def test_regime_transition_detection(self, regime_tool): """국면 전환 탐지 테스트""" # 국면 전환 시계열 (0 -> 1 -> 2) regime_sequence = [0] * 30 + [1] * 30 + [2] * 30 transitions = regime_tool._detect_regime_transitions(regime_sequence) assert len(transitions) >= 2 # 최소 2번의 전환 assert all("from_regime" in t and "to_regime" in t for t in transitions) assert all("transition_date" in t for t in transitions) def test_regime_persistence_analysis(self, regime_tool): """국면 지속성 분석 테스트""" # 다양한 지속성을 가진 국면 regime_sequence = [0] * 50 + [1] * 20 + [2] * 30 persistence = regime_tool._analyze_regime_persistence(regime_sequence) assert "average_duration" in persistence assert "regime_stability" in persistence assert "transition_frequency" in persistence assert persistence["average_duration"] > 0 def test_market_regime_prediction(self, regime_tool): """시장 국면 예측 테스트""" # HMM 모델 결과 시뮬레이션 hmm_model = { "states": [0, 1, 0, 1, 1, 2], "transition_matrix": [ [0.7, 0.2, 0.1], [0.3, 0.5, 0.2], [0.1, 0.3, 0.6] ] } predictions = regime_tool._predict_regime_changes(hmm_model, horizon=5) assert "predicted_regimes" in predictions assert "probability_forecast" in predictions assert "regime_change_probability" in predictions assert len(predictions["predicted_regimes"]) == 5 def test_investment_strategy_mapping(self, regime_tool): """투자 전략 매핑 테스트""" current_regime = { "overall_regime": "bull_market", "volatility_level": "medium", "momentum": "positive" } strategies = regime_tool._map_investment_strategies(current_regime) assert "recommended_strategies" in strategies assert "asset_allocation" in strategies assert "risk_management" in strategies assert len(strategies["recommended_strategies"]) > 0 def test_regime_confidence_scoring(self, regime_tool): """국면 신뢰도 점수 테스트""" # 여러 방법의 결과 analysis_results = { "hmm": {"current_regime": "bull", "confidence": 0.8}, "statistical": {"current_regime": "bull", "confidence": 0.7}, "technical": {"current_regime": "neutral", "confidence": 0.6} } consensus = regime_tool._calculate_regime_consensus(analysis_results) assert "overall_regime" in consensus assert "confidence_score" in consensus assert "method_agreement" in consensus assert 0 <= consensus["confidence_score"] <= 1 def test_volatility_clustering(self, regime_tool): """변동성 클러스터링 테스트""" # GARCH 효과가 있는 변동성 데이터 volatilities = [] prev_vol = 0.015 for i in range(50): shock = random.gauss(0, 0.002) new_vol = 0.8 * prev_vol + 0.2 * abs(shock) volatilities.append(new_vol) prev_vol = new_vol clusters = regime_tool._detect_volatility_clustering(volatilities) assert "cluster_periods" in clusters assert "garch_effects" in clusters assert len(clusters["cluster_periods"]) > 0 @pytest.mark.asyncio async def test_regime_backtest_framework(self, regime_tool, sample_regime_data): """국면 백테스팅 프레임워크 테스트""" # 백테스팅용 과거 데이터 historical_data = sample_regime_data[-60:] # 최근 60일 제외 regime_tool.cache_manager.get.return_value = None regime_tool.db_manager.fetch_all.return_value = sample_regime_data result = await regime_tool.execute({ "market": "KOSPI", "analysis_methods": ["hmm"], "lookback_period": "120d", "include_backtesting": True, "backtest_period": "60d" }) content = result[0] import json data = json.loads(content.text) if "backtesting_results" in data: backtest = data["backtesting_results"] assert "accuracy_metrics" in backtest assert "strategy_performance" in backtest assert "regime_prediction_accuracy" in backtest["accuracy_metrics"] @pytest.mark.asyncio async def test_cache_functionality(self, regime_tool): """캐시 기능 테스트""" # 캐시 히트 시나리오 cached_data = { "timestamp": datetime.now().isoformat(), "market": "KOSPI", "regime_analysis_results": { "hmm": {"current_regime": "bull_market"} } } regime_tool.cache_manager.get.return_value = cached_data # 실행 result = await regime_tool.execute({ "market": "KOSPI", "analysis_methods": ["hmm"], "lookback_period": "60d" }) # 캐시에서 데이터 반환 확인 content = result[0] import json data = json.loads(content.text) assert data == cached_data # 데이터베이스 호출 없음 확인 regime_tool.db_manager.fetch_all.assert_not_called() @pytest.mark.asyncio async def test_error_handling(self, regime_tool): """에러 처리 테스트""" regime_tool.cache_manager.get.return_value = None regime_tool.db_manager.fetch_all.side_effect = DatabaseConnectionError("DB 연결 실패") with pytest.raises(DatabaseConnectionError): await regime_tool.execute({ "market": "KOSPI", "analysis_methods": ["hmm"] }) @pytest.mark.asyncio async def test_invalid_parameters(self, regime_tool): """잘못된 파라미터 테스트""" # 잘못된 시장 with pytest.raises(ValueError, match="Invalid market"): await regime_tool.execute({ "market": "INVALID", "analysis_methods": ["hmm"] }) # 빈 분석 방법 목록 with pytest.raises(ValueError, match="At least one analysis method"): await regime_tool.execute({ "market": "KOSPI", "analysis_methods": [] }) # 잘못된 국면 수 with pytest.raises(ValueError, match="Invalid number of regimes"): await regime_tool.execute({ "market": "KOSPI", "analysis_methods": ["hmm"], "n_regimes": 10 }) @pytest.mark.asyncio async def test_insufficient_data_handling(self, regime_tool): """데이터 부족 처리 테스트""" # 데이터가 부족한 경우 (10일 데이터) insufficient_data = [ { "date": datetime.now().date() - timedelta(days=i), "market": "KOSPI", "close_price": 2650, "daily_return": 0.001, "volume": 450000000, "volatility": 0.015 } for i in range(10) ] regime_tool.cache_manager.get.return_value = None regime_tool.db_manager.fetch_all.return_value = insufficient_data result = await regime_tool.execute({ "market": "KOSPI", "analysis_methods": ["hmm"], "lookback_period": "60d" }) content = result[0] import json data = json.loads(content.text) assert "warning" in data or "insufficient data" in str(data).lower() def test_feature_engineering(self, regime_tool, sample_regime_data): """국면 분석을 위한 피처 엔지니어링 테스트""" features = regime_tool._engineer_regime_features(sample_regime_data[:30]) assert len(features) == 30 assert "volatility_ma" in features[0] assert "return_ma" in features[0] assert "volume_ratio" in features[0] assert "momentum_score" in features[0] def test_regime_characterization(self, regime_tool): """국면 특성화 테스트""" # 3개 국면의 특성 데이터 regime_data = { 0: {"returns": [0.002, 0.003, 0.001], "volatilities": [0.01, 0.012, 0.009]}, 1: {"returns": [-0.001, -0.002, 0.001], "volatilities": [0.025, 0.030, 0.022]}, 2: {"returns": [0.0005, -0.0003, 0.0008], "volatilities": [0.008, 0.007, 0.009]} } characteristics = regime_tool._characterize_regimes(regime_data) assert len(characteristics) == 3 for regime_id, char in characteristics.items(): assert "mean_return" in char assert "mean_volatility" in char assert "regime_label" in char