MCP Market Statistics Server

"""시장 폭 지표 도구""" import json import logging import statistics from datetime import datetime, timedelta from typing import Any, Dict, List from src.tools.base import BaseTool, ToolSchema, TextContent from src.exceptions import DatabaseConnectionError, DataValidationError class MarketBreadthTool(BaseTool): """시장 폭 지표 분석 도구""" def __init__(self, db_manager, cache_manager): super().__init__(db_manager, cache_manager) self.logger = logging.getLogger(__name__) self.cache_ttl = 300 # 5분 @property def name(self) -> str: return "get_market_breadth" @property def description(self) -> str: return "시장 폭 지표(상승/하락 종목 비율, A/D Line 등)를 분석하여 시장의 강도와 방향성을 제공합니다." def get_tool_definition(self) -> ToolSchema: """도구 정의 반환""" return ToolSchema( name=self.name, description=self.description, inputSchema={ "type": "object", "properties": { "market": { "type": "string", "enum": ["KOSPI", "KOSDAQ", "ALL"], "default": "KOSPI", "description": "조회할 시장" }, "period": { "type": "string", "enum": ["1d", "1w", "1m", "3m"], "default": "1d", "description": "조회 기간 (1d: 1일, 1w: 1주, 1m: 1개월, 3m: 3개월)" }, "include_volume_analysis": { "type": "boolean", "default": False, "description": "거래량 분석 포함 여부" } }, "required": [] } ) async def execute(self, arguments: Dict[str, Any]) -> List[TextContent]: """시장 폭 지표 데이터 조회 및 분석""" try: # 파라미터 추출 및 검증 market = arguments.get("market", "KOSPI") period = arguments.get("period", "1d") include_volume_analysis = arguments.get("include_volume_analysis", False) self._validate_parameters(market, period) # 캐시 확인 cache_key = self._generate_cache_key(market, period, include_volume_analysis) cached_data = await self.cache_manager.get(cache_key) if cached_data and self._is_data_fresh(cached_data): self.logger.info(f"Cache hit for {cache_key}") return [TextContent(text=json.dumps(cached_data, ensure_ascii=False))] # 데이터베이스에서 데이터 조회 data = await self._fetch_breadth_data(market, period, include_volume_analysis) # 캐시 저장 await self.cache_manager.set(cache_key, data, ttl=self.cache_ttl) self.logger.info(f"Market breadth data fetched for {market}/{period}") return [TextContent(text=json.dumps(data, ensure_ascii=False, indent=2))] except Exception as e: self.logger.error(f"Error in market breadth tool: {e}") raise def _validate_parameters(self, market: str, period: str): """파라미터 검증""" valid_markets = ["KOSPI", "KOSDAQ", "ALL"] valid_periods = ["1d", "1w", "1m", "3m"] if market not in valid_markets: raise ValueError(f"Invalid market: {market}") if period not in valid_periods: raise ValueError(f"Invalid period: {period}") def _generate_cache_key(self, market: str, period: str, include_volume: bool) -> str: """캐시 키 생성""" return f"market_breadth:{market}:{period}:{include_volume}" def _is_data_fresh(self, data: Dict[str, Any]) -> bool: """데이터 신선도 확인""" if "timestamp" not in data: return False try: timestamp = datetime.fromisoformat(data["timestamp"]) return datetime.now() - timestamp < timedelta(minutes=5) except (ValueError, TypeError): return False async def _fetch_breadth_data(self, market: str, period: str, include_volume: bool) -> Dict[str, Any]: """데이터베이스에서 시장 폭 지표 데이터 조회""" try: # 기간 설정 days = self._get_period_days(period) # 쿼리 구성 query = """ SELECT date, market, advancing, declining, unchanged, total_issues, advance_decline_ratio, advance_volume, decline_volume, timestamp FROM market_breadth WHERE date >= CURRENT_DATE - INTERVAL '%s days' """ params = [days] # 시장 필터 if market != "ALL": query += " AND market = %s" params.append(market) query += " ORDER BY date DESC, market" # 데이터 조회 breadth_data = await self.db_manager.fetch_all(query, *params) # 결과 구성 result = { "timestamp": datetime.now().isoformat(), "period": period, "market": market, "breadth_data": [self._format_breadth_data(data) for data in breadth_data], "summary": self._calculate_summary(breadth_data) } # 빈 데이터 처리 if not breadth_data: result["message"] = "요청한 기간의 시장 폭 데이터가 없습니다" result["breadth_data"] = [] # 트렌드 분석 (주간/월간 기간의 경우) if period in ["1w", "1m", "3m"] and breadth_data: result["trend_analysis"] = self._analyze_trend(breadth_data) # 거래량 분석 포함 if include_volume and breadth_data: result["volume_analysis"] = self._analyze_volume(breadth_data) return result except Exception as e: self.logger.error(f"Database query failed: {e}") if isinstance(e, DatabaseConnectionError): raise raise DatabaseConnectionError(f"Failed to fetch market breadth data: {e}") def _get_period_days(self, period: str) -> int: """기간을 일수로 변환""" period_map = { "1d": 1, "1w": 7, "1m": 30, "3m": 90 } return period_map.get(period, 1) def _format_breadth_data(self, data: Dict[str, Any]) -> Dict[str, Any]: """시장 폭 데이터 포맷팅""" advancing = data.get("advancing", 0) declining = data.get("declining", 0) return { "date": data.get("date").isoformat() if data.get("date") else None, "market": data.get("market"), "advancing": int(advancing), "declining": int(declining), "unchanged": int(data.get("unchanged", 0)), "total_issues": int(data.get("total_issues", 0)), "advance_decline_ratio": float(data.get("advance_decline_ratio", 0)), "advance_volume": int(data.get("advance_volume", 0)), "decline_volume": int(data.get("decline_volume", 0)), "timestamp": data.get("timestamp").isoformat() if data.get("timestamp") else None } def _calculate_ad_ratio(self, advancing: int, declining: int) -> float: """상승하락비율 계산""" if declining == 0: return float('inf') if advancing > 0 else 0.0 return advancing / declining def _interpret_market_sentiment(self, ratio: float) -> str: """시장 심리 해석""" if ratio >= 2.0: return "강한 상승세" elif ratio >= 1.2: return "상승세" elif ratio >= 0.8: return "보합" elif ratio >= 0.5: return "하락세" else: return "강한 하락세" def _calculate_summary(self, breadth_data: List[Dict[str, Any]]) -> Dict[str, Any]: """시장 폭 요약 계산""" if not breadth_data: return {} # 평균 상승하락비율 계산 ratios = [data.get("advance_decline_ratio", 0) for data in breadth_data if data.get("advance_decline_ratio") is not None] avg_ratio = statistics.mean(ratios) if ratios else 0.0 # 시장 심리 판단 market_sentiment = self._interpret_market_sentiment(avg_ratio) # 최신 데이터 기준 추가 정보 latest = breadth_data[0] if breadth_data else {} return { "avg_advance_decline_ratio": round(avg_ratio, 2), "market_sentiment": market_sentiment, "latest_advancing": latest.get("advancing", 0), "latest_declining": latest.get("declining", 0), "latest_unchanged": latest.get("unchanged", 0), "data_points": len(breadth_data) } def _analyze_trend(self, breadth_data: List[Dict[str, Any]]) -> Dict[str, Any]: """트렌드 분석""" if len(breadth_data) < 2: return {"error": "트렌드 분석을 위한 충분한 데이터가 없습니다"} # 시간순 정렬 (오래된 것부터) sorted_data = sorted(breadth_data, key=lambda x: x.get("date", datetime.min)) # 상승하락비율의 변화 계산 ratios = [data.get("advance_decline_ratio", 0) for data in sorted_data] # 선형 회귀를 통한 트렌드 방향 계산 (단순화) if len(ratios) >= 2: first_half_avg = statistics.mean(ratios[:len(ratios)//2]) second_half_avg = statistics.mean(ratios[len(ratios)//2:]) trend_change = second_half_avg - first_half_avg if trend_change > 0.1: direction = "상승" strength = "강함" if trend_change > 0.2 else "약함" elif trend_change < -0.1: direction = "하락" strength = "강함" if trend_change < -0.2 else "약함" else: direction = "보합" strength = "보통" else: direction = "불명" strength = "불명" return { "direction": direction, "strength": strength, "trend_change": round(trend_change if 'trend_change' in locals() else 0.0, 3), "days_analyzed": len(sorted_data) } def _analyze_volume(self, breadth_data: List[Dict[str, Any]]) -> Dict[str, Any]: """거래량 분석""" if not breadth_data: return {} # 최신 데이터 latest = breadth_data[0] advance_volume = latest.get("advance_volume", 0) decline_volume = latest.get("decline_volume", 0) # 거래량 비율 계산 total_volume = advance_volume + decline_volume volume_ratio = advance_volume / decline_volume if decline_volume > 0 else float('inf') # 거래량 트렌드 판단 if volume_ratio >= 1.5: volume_trend = "상승 거래량 우세" elif volume_ratio >= 0.67: volume_trend = "균형" else: volume_trend = "하락 거래량 우세" return { "advance_volume": advance_volume, "decline_volume": decline_volume, "total_volume": total_volume, "volume_ratio": round(volume_ratio, 2) if volume_ratio != float('inf') else "무한대", "volume_trend": volume_trend, "advance_volume_pct": round((advance_volume / total_volume * 100), 1) if total_volume > 0 else 0.0 }