MCP Market Statistics Server

"""투자자 동향 분석 도구""" import json import logging 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 InvestorFlowsTool(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_investor_flows" @property def description(self) -> str: return "투자자별(개인, 기관, 외국인) 매매 동향과 자금 흐름을 분석합니다." 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개월)" }, "investor_type": { "type": "string", "enum": ["개인", "기관", "외국인", "ALL"], "default": "ALL", "description": "조회할 투자자 유형" }, "include_market_impact": { "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") investor_type = arguments.get("investor_type", "ALL") include_market_impact = arguments.get("include_market_impact", False) self._validate_parameters(market, period, investor_type) # 캐시 확인 cache_key = self._generate_cache_key(market, period, investor_type, include_market_impact) 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_investor_data(market, period, investor_type, include_market_impact) # 캐시 저장 await self.cache_manager.set(cache_key, data, ttl=self.cache_ttl) self.logger.info(f"Investor flows 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 investor flows tool: {e}") raise def _validate_parameters(self, market: str, period: str, investor_type: str): """파라미터 검증""" valid_markets = ["KOSPI", "KOSDAQ", "ALL"] valid_periods = ["1d", "1w", "1m", "3m"] valid_investor_types = ["개인", "기관", "외국인", "ALL"] if market not in valid_markets: raise ValueError(f"Invalid market: {market}") if period not in valid_periods: raise ValueError(f"Invalid period: {period}") if investor_type not in valid_investor_types: raise ValueError(f"Invalid investor_type: {investor_type}") def _generate_cache_key(self, market: str, period: str, investor_type: str, include_impact: bool) -> str: """캐시 키 생성""" return f"investor_flows:{market}:{period}:{investor_type}:{include_impact}" 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_investor_data(self, market: str, period: str, investor_type: str, include_impact: bool) -> Dict[str, Any]: """데이터베이스에서 투자자 동향 데이터 조회""" try: # 기간 설정 days = self._get_period_days(period) # 쿼리 구성 query = """ SELECT date, investor_type, buy_amount, sell_amount, net_amount, market FROM investor_flows WHERE date >= CURRENT_DATE - INTERVAL '%s days' """ params = [days] # 시장 필터 if market != "ALL": query += " AND market = %s" params.append(market) # 투자자 유형 필터 if investor_type != "ALL": query += " AND investor_type = %s" params.append(investor_type) query += " ORDER BY date DESC, investor_type" # 데이터 조회 flows_data = await self.db_manager.fetch_all(query, *params) # 결과 구성 result = { "timestamp": datetime.now().isoformat(), "period": period, "market": market, "investor_type": investor_type, "flows": [self._format_flow_data(flow) for flow in flows_data], "summary": self._calculate_summary(flows_data) } # 빈 데이터 처리 if not flows_data: result["message"] = "요청한 기간의 투자자 동향 데이터가 없습니다" result["flows"] = [] # 누적 플로우 계산 (주간/월간 기간의 경우) if period in ["1w", "1m", "3m"] and flows_data: result["cumulative_flows"] = self._calculate_cumulative_flows(flows_data) # 시장 영향도 분석 포함 if include_impact and flows_data: impact_data = await self._analyze_market_impact(market, period, flows_data) result["market_impact"] = impact_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 investor 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_flow_data(self, flow: Dict[str, Any]) -> Dict[str, Any]: """플로우 데이터 포맷팅""" net_amount = flow.get("net_amount", 0) return { "date": flow.get("date").isoformat() if flow.get("date") else None, "investor_type": flow.get("investor_type"), "buy_amount": int(flow.get("buy_amount", 0)), "sell_amount": int(flow.get("sell_amount", 0)), "net_amount": int(net_amount), "net_amount_formatted": self._format_amount(net_amount), "market": flow.get("market") } def _format_amount(self, amount: int) -> str: """금액 포맷팅 (조, 억 단위)""" if amount == 0: return "0원" abs_amount = abs(amount) sign = "-" if amount < 0 else "" if abs_amount >= 1000000000000: # 1조 이상 trillion = abs_amount / 1000000000000 if trillion == int(trillion): return f"{sign}{int(trillion)}조" else: return f"{sign}{trillion:.1f}조" elif abs_amount >= 100000000: # 1억 이상 hundred_million = abs_amount / 100000000 if hundred_million >= 1000: # 1000억 이상은 콤마 표시 return f"{sign}{int(hundred_million):,}억" else: return f"{sign}{int(hundred_million)}억" else: return f"{sign}{int(abs_amount):,}원" def _calculate_summary(self, flows_data: List[Dict[str, Any]]) -> Dict[str, Any]: """투자자별 순매매 요약 계산""" summary = { "net_individual": 0, "net_foreign": 0, "net_institution": 0, "total_trading_value": 0 } for flow in flows_data: investor_type = flow.get("investor_type", "") net_amount = flow.get("net_amount", 0) buy_amount = flow.get("buy_amount", 0) sell_amount = flow.get("sell_amount", 0) if investor_type == "개인": summary["net_individual"] += net_amount elif investor_type == "외국인": summary["net_foreign"] += net_amount elif investor_type == "기관": summary["net_institution"] += net_amount summary["total_trading_value"] += buy_amount + sell_amount # 포맷팅된 버전도 추가 summary["net_individual_formatted"] = self._format_amount(summary["net_individual"]) summary["net_foreign_formatted"] = self._format_amount(summary["net_foreign"]) summary["net_institution_formatted"] = self._format_amount(summary["net_institution"]) return summary def _calculate_cumulative_flows(self, flows_data: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """누적 플로우 계산""" # 날짜별로 그룹화 daily_flows = {} for flow in flows_data: date = flow.get("date") if date: date_str = date.isoformat() if hasattr(date, 'isoformat') else str(date) if date_str not in daily_flows: daily_flows[date_str] = 0 daily_flows[date_str] += flow.get("net_amount", 0) # 날짜 순으로 정렬하고 누적 계산 sorted_dates = sorted(daily_flows.keys()) cumulative = 0 cumulative_flows = [] for date_str in sorted_dates: daily_net = daily_flows[date_str] cumulative += daily_net cumulative_flows.append({ "date": date_str, "daily_net": daily_net, "cumulative_net": cumulative, "daily_net_formatted": self._format_amount(daily_net), "cumulative_net_formatted": self._format_amount(cumulative) }) return cumulative_flows async def _analyze_market_impact(self, market: str, period: str, flows_data: List[Dict[str, Any]]) -> Dict[str, Any]: """시장 영향도 분석""" try: # 동일 기간 시장 지수 변화 조회 days = self._get_period_days(period) index_data = await self.db_manager.fetch_one( """ SELECT (SELECT current_value FROM market_indices WHERE index_code = %s ORDER BY timestamp DESC LIMIT 1) - (SELECT current_value FROM market_indices WHERE index_code = %s AND timestamp <= CURRENT_TIMESTAMP - INTERVAL '%s days' ORDER BY timestamp DESC LIMIT 1) as index_change, ((SELECT current_value FROM market_indices WHERE index_code = %s ORDER BY timestamp DESC LIMIT 1) / (SELECT current_value FROM market_indices WHERE index_code = %s AND timestamp <= CURRENT_TIMESTAMP - INTERVAL '%s days' ORDER BY timestamp DESC LIMIT 1) - 1) * 100 as index_change_rate """, market if market != "ALL" else "KOSPI", market if market != "ALL" else "KOSPI", days, market if market != "ALL" else "KOSPI", market if market != "ALL" else "KOSPI", days ) if not index_data: return {"error": "시장 지수 데이터를 찾을 수 없습니다"} # 투자자별 순매매와 지수 변화 상관관계 분석 summary = self._calculate_summary(flows_data) index_change_rate = index_data.get("index_change_rate", 0) # 단순 상관관계 계산 (실제로는 더 복잡한 통계적 분석이 필요) foreign_correlation = -1 if (summary["net_foreign"] < 0 and index_change_rate > 0) or \ (summary["net_foreign"] > 0 and index_change_rate < 0) else 1 individual_correlation = 1 if (summary["net_individual"] > 0 and index_change_rate > 0) or \ (summary["net_individual"] < 0 and index_change_rate < 0) else -1 return { "index_change": float(index_data.get("index_change", 0)), "index_change_rate": float(index_change_rate), "correlation_analysis": { "foreign_vs_index": foreign_correlation, "individual_vs_index": individual_correlation, "interpretation": self._interpret_correlation(foreign_correlation, individual_correlation) } } except Exception as e: self.logger.warning(f"Market impact analysis failed: {e}") return {"error": f"시장 영향도 분석 중 오류 발생: {e}"} def _interpret_correlation(self, foreign_corr: float, individual_corr: float) -> str: """상관관계 해석""" interpretations = [] if foreign_corr < 0: interpretations.append("외국인 자금이 지수와 반대 방향으로 움직임") else: interpretations.append("외국인 자금이 지수와 같은 방향으로 움직임") if individual_corr > 0: interpretations.append("개인 투자자가 지수와 같은 방향으로 움직임") else: interpretations.append("개인 투자자가 지수와 반대 방향으로 움직임") return "; ".join(interpretations)