from typing import Dict, Any, Callable import pandas as pd import numpy as np from ta.trend import SMAIndicator, EMAIndicator, MACD from ta.momentum import RSIIndicator, StochasticOscillator from ta.volatility import BollingerBands import logging logger = logging.getLogger(__name__) class MCPTools: """MCP Tools Registry""" def __init__(self): self.tools: Dict[str, Callable] = {} self._register_tools() def _register_tools(self): """도구 등록""" self.tools = { "get_market_data": self.get_market_data, "calculate_indicators": self.calculate_indicators, "check_portfolio": self.check_portfolio, "validate_rules": self.validate_rules, "execute_order": self.execute_order, "get_news": self.get_news } def get_tool_definitions(self): """도구 정의 반환 (Claude API 용)""" return [ { "name": "get_market_data", "description": "주식의 시장 데이터를 조회합니다.", "input_schema": { "type": "object", "properties": { "stock_code": { "type": "string", "description": "종목 코드 (예: 005930)" }, "period": { "type": "string", "description": "조회 기간 (D: 일봉, W: 주봉, M: 월봉)", "enum": ["D", "W", "M"] } }, "required": ["stock_code"] } }, { "name": "calculate_indicators", "description": "기술적 지표를 계산합니다.", "input_schema": { "type": "object", "properties": { "price_data": { "type": "array", "description": "가격 데이터 배열" }, "indicators": { "type": "array", "description": "계산할 지표 목록 (RSI, MACD, BB, SMA, EMA)", "items": {"type": "string"} } }, "required": ["price_data", "indicators"] } }, { "name": "check_portfolio", "description": "현재 포트폴리오 상태를 확인합니다.", "input_schema": { "type": "object", "properties": { "stock_code": { "type": "string", "description": "특정 종목 필터 (선택사항)" } } } }, { "name": "validate_rules", "description": "트레이딩 규칙을 검증합니다.", "input_schema": { "type": "object", "properties": { "rule": { "type": "object", "description": "검증할 규칙" }, "market_context": { "type": "object", "description": "시장 상황" } }, "required": ["rule", "market_context"] } }, { "name": "execute_order", "description": "주문을 실행합니다.", "input_schema": { "type": "object", "properties": { "stock_code": {"type": "string"}, "order_type": { "type": "string", "enum": ["BUY", "SELL"] }, "quantity": {"type": "integer"}, "price": {"type": "number", "description": "지정가 (시장가는 null)"} }, "required": ["stock_code", "order_type", "quantity"] } }, { "name": "get_news", "description": "주식 관련 뉴스를 조회합니다.", "input_schema": { "type": "object", "properties": { "stock_code": { "type": "string", "description": "종목 코드" }, "limit": { "type": "integer", "description": "조회할 뉴스 개수", "default": 10 } } } } ] async def call_tool(self, tool_name: str, parameters: Dict[str, Any]) -> Dict[str, Any]: """도구 호출""" if tool_name not in self.tools: raise ValueError(f"Unknown tool: {tool_name}") try: result = await self.tools[tool_name](**parameters) return { "status": "success", "result": result } except Exception as e: logger.error(f"Tool {tool_name} execution error: {e}") return { "status": "error", "error": str(e) } async def get_market_data(self, stock_code: str, period: str = "D") -> Dict[str, Any]: """시장 데이터 조회 (실제로는 Korea Investment API 호출)""" # 여기서는 샘플 데이터 반환 # 실제로는 backend API를 호출해야 함 return { "stock_code": stock_code, "period": period, "current_price": 70000, "change_rate": 1.5, "volume": 1000000, "data": [] } async def calculate_indicators( self, price_data: list, indicators: list ) -> Dict[str, Any]: """기술적 지표 계산""" try: df = pd.DataFrame(price_data) if 'close' not in df.columns: raise ValueError("Price data must include 'close' column") results = {} # RSI 계산 if "RSI" in indicators: rsi = RSIIndicator(close=df['close'], window=14) results["RSI"] = float(rsi.rsi().iloc[-1]) # MACD 계산 if "MACD" in indicators: macd = MACD(close=df['close']) results["MACD"] = { "macd": float(macd.macd().iloc[-1]), "signal": float(macd.macd_signal().iloc[-1]), "histogram": float(macd.macd_diff().iloc[-1]) } # Bollinger Bands 계산 if "BB" in indicators: bb = BollingerBands(close=df['close']) results["BollingerBands"] = { "upper": float(bb.bollinger_hband().iloc[-1]), "middle": float(bb.bollinger_mavg().iloc[-1]), "lower": float(bb.bollinger_lband().iloc[-1]) } # SMA 계산 if "SMA" in indicators: sma_20 = SMAIndicator(close=df['close'], window=20) sma_60 = SMAIndicator(close=df['close'], window=60) results["SMA"] = { "sma_20": float(sma_20.sma_indicator().iloc[-1]), "sma_60": float(sma_60.sma_indicator().iloc[-1]) } # EMA 계산 if "EMA" in indicators: ema_12 = EMAIndicator(close=df['close'], window=12) ema_26 = EMAIndicator(close=df['close'], window=26) results["EMA"] = { "ema_12": float(ema_12.ema_indicator().iloc[-1]), "ema_26": float(ema_26.ema_indicator().iloc[-1]) } return results except Exception as e: logger.error(f"Error calculating indicators: {e}") raise async def check_portfolio(self, stock_code: str = None) -> Dict[str, Any]: """포트폴리오 확인""" # 실제로는 backend API 호출 return { "total_value": 10000000, "holdings": [ { "stock_code": "005930", "quantity": 10, "average_price": 70000, "current_price": 71000 } ] } async def validate_rules( self, rule: Dict[str, Any], market_context: Dict[str, Any] ) -> Dict[str, Any]: """규칙 검증""" # 간단한 규칙 검증 로직 conditions = rule.get("conditions", {}) # RSI 조건 확인 if "rsi_below" in conditions: market_rsi = market_context.get("RSI", 50) if market_rsi < conditions["rsi_below"]: return { "valid": True, "reason": f"RSI {market_rsi} is below {conditions['rsi_below']}" } return { "valid": False, "reason": "Conditions not met" } async def execute_order( self, stock_code: str, order_type: str, quantity: int, price: float = None ) -> Dict[str, Any]: """주문 실행 (실제로는 backend API 호출)""" return { "order_id": "ORD20241104001", "status": "PENDING", "stock_code": stock_code, "order_type": order_type, "quantity": quantity, "price": price } async def get_news(self, stock_code: str = None, limit: int = 10) -> Dict[str, Any]: """뉴스 조회""" # 실제로는 뉴스 API 호출 return { "stock_code": stock_code, "news": [] } mcp_tools = MCPTools()