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
        }