MCP Market Statistics Server

"""심리 지표 도구 테스트""" import pytest import random import math from datetime import datetime, timedelta from unittest.mock import AsyncMock from src.tools.sentiment_tools import MarketSentimentTool from src.exceptions import DataValidationError, DatabaseConnectionError class TestMarketSentimentTool: """시장 심리 지표 도구 테스트""" @pytest.fixture def mock_db_manager(self): """Mock 데이터베이스 매니저""" return AsyncMock() @pytest.fixture def mock_cache_manager(self): """Mock 캐시 매니저""" return AsyncMock() @pytest.fixture def sentiment_tool(self, mock_db_manager, mock_cache_manager): """심리 지표 도구 인스턴스""" return MarketSentimentTool(mock_db_manager, mock_cache_manager) @pytest.fixture def sample_sentiment_data(self): """샘플 심리 지표 데이터""" base_date = datetime.now().date() return { "put_call_ratio": [ { "date": base_date - timedelta(days=i), "put_volume": 85000000 + random.randint(-10000000, 10000000), "call_volume": 120000000 + random.randint(-15000000, 15000000), "put_call_ratio": 0.7 + random.uniform(-0.2, 0.2), "put_oi": 950000 + random.randint(-100000, 100000), "call_oi": 1200000 + random.randint(-150000, 150000) } for i in range(30) ], "vkospi_data": [ { "date": base_date - timedelta(days=i), "vkospi_value": 22.5 + random.uniform(-5.0, 5.0), "change": random.uniform(-3.0, 3.0), "change_rate": random.uniform(-15.0, 15.0), "high": 25.0 + random.uniform(-3.0, 3.0), "low": 20.0 + random.uniform(-3.0, 3.0) } for i in range(30) ], "market_indicators": [ { "date": base_date - timedelta(days=i), "kospi_close": 2650 + random.uniform(-100, 100), "kospi_change_rate": random.uniform(-3.0, 3.0), "volume_ratio": 0.8 + random.uniform(-0.3, 0.3), "advancing_issues": random.randint(600, 900), "declining_issues": random.randint(500, 800), "new_highs": random.randint(5, 50), "new_lows": random.randint(5, 45) } for i in range(30) ], "news_sentiment": [ { "date": base_date - timedelta(days=i), "positive_count": random.randint(20, 80), "negative_count": random.randint(15, 75), "neutral_count": random.randint(50, 150), "sentiment_score": random.uniform(-1.0, 1.0), "buzz_intensity": random.uniform(0.1, 1.0) } for i in range(30) ] } def test_tool_initialization(self, sentiment_tool, mock_db_manager, mock_cache_manager): """도구 초기화 테스트""" assert sentiment_tool.name == "get_market_sentiment" assert sentiment_tool.description is not None assert "심리" in sentiment_tool.description or "감성" in sentiment_tool.description assert sentiment_tool.db_manager == mock_db_manager assert sentiment_tool.cache_manager == mock_cache_manager def test_tool_definition(self, sentiment_tool): """도구 정의 테스트""" definition = sentiment_tool.get_tool_definition() assert definition.name == "get_market_sentiment" 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 "indicators" in properties assert "period" in properties assert "include_fear_greed_index" in properties # indicators 파라미터 검증 indicators_prop = properties["indicators"] assert indicators_prop["type"] == "array" assert "items" in indicators_prop @pytest.mark.asyncio async def test_execute_comprehensive_sentiment(self, sentiment_tool, sample_sentiment_data): """종합 심리 지표 분석 테스트""" # 캐시 미스 sentiment_tool.cache_manager.get.return_value = None # 데이터베이스 응답 설정 (순서대로 쿼리될 것으로 예상) sentiment_tool.db_manager.fetch_all.side_effect = [ sample_sentiment_data["put_call_ratio"], sample_sentiment_data["vkospi_data"], sample_sentiment_data["market_indicators"], sample_sentiment_data["news_sentiment"] ] # 실행 result = await sentiment_tool.execute({ "market": "KOSPI", "indicators": ["put_call_ratio", "vkospi", "market_breadth", "news_sentiment"], "period": "30d", "include_fear_greed_index": True }) # 결과 검증 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 "period" in data assert "sentiment_indicators" in data assert "fear_greed_index" in data assert "overall_sentiment" in data # Put/Call Ratio 검증 sentiment_indicators = data["sentiment_indicators"] assert "put_call_ratio" in sentiment_indicators pcr = sentiment_indicators["put_call_ratio"] assert "current_ratio" in pcr assert "interpretation" in pcr assert "trend" in pcr # VKOSPI 검증 assert "vkospi" in sentiment_indicators vkospi = sentiment_indicators["vkospi"] assert "current_value" in vkospi assert "volatility_regime" in vkospi # Fear & Greed Index 검증 fgi = data["fear_greed_index"] assert "score" in fgi assert "category" in fgi assert 0 <= fgi["score"] <= 100 assert fgi["category"] in ["극도 공포", "공포", "중립", "탐욕", "극도 탐욕"] @pytest.mark.asyncio async def test_execute_put_call_ratio_only(self, sentiment_tool, sample_sentiment_data): """Put/Call Ratio만 분석 테스트""" sentiment_tool.cache_manager.get.return_value = None sentiment_tool.db_manager.fetch_all.return_value = sample_sentiment_data["put_call_ratio"] # 실행 result = await sentiment_tool.execute({ "market": "KOSPI", "indicators": ["put_call_ratio"], "period": "30d" }) # 결과 검증 content = result[0] import json data = json.loads(content.text) assert "sentiment_indicators" in data sentiment_indicators = data["sentiment_indicators"] assert "put_call_ratio" in sentiment_indicators assert len(sentiment_indicators) == 1 # 하나의 지표만 @pytest.mark.asyncio async def test_execute_vkospi_analysis(self, sentiment_tool, sample_sentiment_data): """VKOSPI 분석 테스트""" sentiment_tool.cache_manager.get.return_value = None sentiment_tool.db_manager.fetch_all.return_value = sample_sentiment_data["vkospi_data"] # 실행 result = await sentiment_tool.execute({ "market": "KOSPI", "indicators": ["vkospi"], "period": "30d" }) # 결과 검증 content = result[0] import json data = json.loads(content.text) vkospi = data["sentiment_indicators"]["vkospi"] assert "current_value" in vkospi assert "volatility_regime" in vkospi assert "trend_analysis" in vkospi # 변동성 체제 확인 regime = vkospi["volatility_regime"] assert regime in ["낮음", "보통", "높음", "매우 높음"] @pytest.mark.asyncio async def test_market_breadth_sentiment(self, sentiment_tool, sample_sentiment_data): """시장 폭 기반 심리 분석 테스트""" sentiment_tool.cache_manager.get.return_value = None sentiment_tool.db_manager.fetch_all.return_value = sample_sentiment_data["market_indicators"] # 실행 result = await sentiment_tool.execute({ "market": "KOSPI", "indicators": ["market_breadth"], "period": "30d" }) # 결과 검증 content = result[0] import json data = json.loads(content.text) breadth = data["sentiment_indicators"]["market_breadth"] assert "advance_decline_sentiment" in breadth assert "volume_sentiment" in breadth assert "new_highs_lows_ratio" in breadth @pytest.mark.asyncio async def test_news_sentiment_analysis(self, sentiment_tool, sample_sentiment_data): """뉴스 심리 분석 테스트""" sentiment_tool.cache_manager.get.return_value = None sentiment_tool.db_manager.fetch_all.return_value = sample_sentiment_data["news_sentiment"] # 실행 result = await sentiment_tool.execute({ "market": "KOSPI", "indicators": ["news_sentiment"], "period": "7d", "include_buzz_analysis": True }) # 결과 검증 content = result[0] import json data = json.loads(content.text) news = data["sentiment_indicators"]["news_sentiment"] assert "sentiment_score" in news assert "sentiment_trend" in news assert "buzz_analysis" in news # 버즈 분석 확인 buzz = news["buzz_analysis"] assert "intensity_level" in buzz assert "trending_topics" in buzz def test_put_call_ratio_interpretation(self, sentiment_tool): """Put/Call Ratio 해석 테스트""" # 극도 공포 (높은 Put/Call Ratio) interpretation = sentiment_tool._interpret_put_call_ratio(1.2) assert "공포" in interpretation # 중립 interpretation = sentiment_tool._interpret_put_call_ratio(0.8) assert "중립" in interpretation or "보통" in interpretation # 탐욕 (낮은 Put/Call Ratio) interpretation = sentiment_tool._interpret_put_call_ratio(0.4) assert "탐욕" in interpretation def test_vkospi_volatility_regime(self, sentiment_tool): """VKOSPI 변동성 체제 분류 테스트""" # 낮은 변동성 regime = sentiment_tool._classify_volatility_regime(15.0) assert regime == "낮음" # 보통 변동성 regime = sentiment_tool._classify_volatility_regime(22.0) assert regime == "보통" # 높은 변동성 regime = sentiment_tool._classify_volatility_regime(30.0) assert regime == "높음" # 매우 높은 변동성 regime = sentiment_tool._classify_volatility_regime(40.0) assert regime == "매우 높음" def test_fear_greed_index_calculation(self, sentiment_tool): """Fear & Greed Index 계산 테스트""" # 샘플 지표들 indicators = { "put_call_ratio": {"score": 30, "weight": 0.25}, # 공포쪽 "vkospi": {"score": 25, "weight": 0.20}, # 공포쪽 "market_breadth": {"score": 70, "weight": 0.25}, # 탐욕쪽 "news_sentiment": {"score": 60, "weight": 0.15}, # 약간 탐욕 "volume": {"score": 55, "weight": 0.15} # 중립 } fgi_score = sentiment_tool._calculate_fear_greed_index(indicators) assert 0 <= fgi_score <= 100 assert isinstance(fgi_score, (int, float)) def test_fear_greed_category(self, sentiment_tool): """Fear & Greed 카테고리 분류 테스트""" # 극도 공포 category = sentiment_tool._get_fear_greed_category(15) assert category == "극도 공포" # 공포 category = sentiment_tool._get_fear_greed_category(35) assert category == "공포" # 중립 category = sentiment_tool._get_fear_greed_category(50) assert category == "중립" # 탐욕 category = sentiment_tool._get_fear_greed_category(70) assert category == "탐욕" # 극도 탐욕 category = sentiment_tool._get_fear_greed_category(90) assert category == "극도 탐욕" def test_sentiment_trend_analysis(self, sentiment_tool): """심리 트렌드 분석 테스트""" # 상승 트렌드 values = [20, 25, 30, 35, 40] trend = sentiment_tool._analyze_sentiment_trend(values) assert trend["direction"] == "상승" # 하락 트렌드 values = [60, 55, 50, 45, 40] trend = sentiment_tool._analyze_sentiment_trend(values) assert trend["direction"] == "하락" # 보합 values = [50, 48, 52, 49, 51] trend = sentiment_tool._analyze_sentiment_trend(values) assert trend["direction"] == "보합" @pytest.mark.asyncio async def test_social_media_sentiment(self, sentiment_tool): """소셜 미디어 심리 분석 테스트""" # 소셜 미디어 데이터 social_data = [ { "date": datetime.now().date() - timedelta(days=i), "platform": "twitter", "positive_mentions": random.randint(100, 500), "negative_mentions": random.randint(50, 300), "neutral_mentions": random.randint(200, 800), "sentiment_score": random.uniform(-1.0, 1.0), "engagement_rate": random.uniform(0.01, 0.15), "trending_keywords": ["KOSPI", "주식", "투자", "증시"] } for i in range(7) ] sentiment_tool.cache_manager.get.return_value = None sentiment_tool.db_manager.fetch_all.return_value = social_data # 실행 result = await sentiment_tool.execute({ "market": "KOSPI", "indicators": ["social_media"], "period": "7d" }) # 결과 검증 content = result[0] import json data = json.loads(content.text) social = data["sentiment_indicators"]["social_media"] assert "sentiment_score" in social assert "engagement_trend" in social assert "trending_topics" in social @pytest.mark.asyncio async def test_cache_functionality(self, sentiment_tool): """캐시 기능 테스트""" # 캐시 히트 시나리오 cached_data = { "timestamp": datetime.now().isoformat(), "market": "KOSPI", "sentiment_indicators": { "put_call_ratio": {"current_ratio": 0.75, "interpretation": "중립"} }, "overall_sentiment": "중립" } sentiment_tool.cache_manager.get.return_value = cached_data # 실행 result = await sentiment_tool.execute({ "market": "KOSPI", "indicators": ["put_call_ratio"], "period": "30d" }) # 캐시에서 데이터 반환 확인 content = result[0] import json data = json.loads(content.text) assert data == cached_data # 데이터베이스 호출 없음 확인 sentiment_tool.db_manager.fetch_all.assert_not_called() @pytest.mark.asyncio async def test_error_handling(self, sentiment_tool): """에러 처리 테스트""" sentiment_tool.cache_manager.get.return_value = None sentiment_tool.db_manager.fetch_all.side_effect = DatabaseConnectionError("DB 연결 실패") with pytest.raises(DatabaseConnectionError): await sentiment_tool.execute({ "market": "KOSPI", "indicators": ["put_call_ratio"], "period": "30d" }) @pytest.mark.asyncio async def test_invalid_parameters(self, sentiment_tool): """잘못된 파라미터 테스트""" # 잘못된 시장 with pytest.raises(ValueError, match="Invalid market"): await sentiment_tool.execute({ "market": "INVALID", "indicators": ["put_call_ratio"], "period": "30d" }) # 빈 지표 목록 with pytest.raises(ValueError, match="At least one indicator"): await sentiment_tool.execute({ "market": "KOSPI", "indicators": [], "period": "30d" }) # 잘못된 기간 with pytest.raises(ValueError, match="Invalid period"): await sentiment_tool.execute({ "market": "KOSPI", "indicators": ["put_call_ratio"], "period": "invalid" }) @pytest.mark.asyncio async def test_insufficient_data_handling(self, sentiment_tool): """데이터 부족 처리 테스트""" # 데이터가 부족한 경우 insufficient_data = [ { "date": datetime.now().date(), "put_volume": 85000000, "call_volume": 120000000, "put_call_ratio": 0.7 } ] sentiment_tool.cache_manager.get.return_value = None sentiment_tool.db_manager.fetch_all.return_value = insufficient_data result = await sentiment_tool.execute({ "market": "KOSPI", "indicators": ["put_call_ratio"], "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_sentiment_aggregation(self, sentiment_tool): """심리 지표 종합 테스트""" # 다양한 지표들 indicators = { "put_call_ratio": 0.85, # 약간 공포 "vkospi": 28.5, # 높은 변동성 (공포) "advance_decline": 0.65, # 약간 부정적 "news_sentiment": 0.2 # 약간 긍정적 } overall = sentiment_tool._aggregate_sentiment_indicators(indicators) assert "score" in overall assert "sentiment" in overall assert "confidence" in overall assert 0 <= overall["score"] <= 100 assert overall["sentiment"] in ["매우 부정적", "부정적", "중립", "긍정적", "매우 긍정적"] @pytest.mark.asyncio async def test_historical_comparison(self, sentiment_tool, sample_sentiment_data): """과거 대비 심리 변화 분석 테스트""" sentiment_tool.cache_manager.get.return_value = None sentiment_tool.db_manager.fetch_all.return_value = sample_sentiment_data["put_call_ratio"] # 실행 (과거 비교 포함) result = await sentiment_tool.execute({ "market": "KOSPI", "indicators": ["put_call_ratio"], "period": "30d", "include_historical_comparison": True }) # 결과 검증 content = result[0] import json data = json.loads(content.text) assert "historical_comparison" in data historical = data["historical_comparison"] assert "vs_1week_ago" in historical assert "vs_1month_ago" in historical assert "percentile_rank" in historical