"""MCP 도구 레지스트리 모듈 (mcp_tools.py) MCP 서버가 제공하는 도구(Tool)들을 등록하고 실행하는 모듈. Claude/Gemini AI가 function calling으로 이 도구들을 호출할 수 있다. 등록된 도구: - get_market_data: 국내/해외 주식 시장 데이터 조회 (KIS API) - calculate_indicators: 기술적 지표 계산 (RSI, MACD, BB, SMA, EMA) - check_portfolio: 포트폴리오 보유 현황 조회 - validate_rules: 트레이딩 규칙 검증 (RSI 조건 등) - execute_order: 주문 실행 (차단됨 - 백엔드 API 경유 필수) - get_news: 네이버 금융 뉴스 크롤링 아키텍처 주의사항: - korea_investment_api는 None (MCP 서버에서 KIS API 직접 호출 불가) - execute_order()는 리스크 검증 우회 방지를 위해 항상 차단 응답 반환 - 주문은 반드시 백엔드 /api/trades 엔드포인트를 통해 실행해야 함 """ from typing import Dict, Any, Callable import logging 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 httpx from bs4 import BeautifulSoup from config import settings # KIS API 인스턴스 — MCP 서버에서는 직접 사용 불가 (None) # 주문은 백엔드 API를 통해서만 실행해야 한다 (리스크 검증 파이프라인 우회 방지) korea_investment_api = None logger = logging.getLogger(__name__) # Major US stock exchange mapping _EXCHANGE_MAP = { # NYSE stocks "JPM": "NYSE", "V": "NYSE", "JNJ": "NYSE", "WMT": "NYSE", "PG": "NYSE", "UNH": "NYSE", "HD": "NYSE", "BAC": "NYSE", "XOM": "NYSE", "CVX": "NYSE", "KO": "NYSE", "DIS": "NYSE", "MCD": "NYSE", "NKE": "NYSE", "BA": "NYSE", # AMEX (mostly ETFs) "SPY": "AMEX", "QQQ": "AMEX", "IWM": "AMEX", "XLF": "AMEX", } def _detect_exchange(ticker: str) -> str: return _EXCHANGE_MAP.get(ticker.upper(), "NAS") # Default to NAS 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]: """시장 데이터 조회 (국내/해외 통합)""" if not korea_investment_api: return {"error": "Korea Investment API not available"} try: # 국내 주식 여부 판단 (6자리 숫자) is_domestic = stock_code.isdigit() and len(stock_code) == 6 if is_domestic: # 현재가 조회 current_price_data = await korea_investment_api.get_current_price(stock_code) # 차트 데이터 조회 chart_data = await korea_investment_api.get_daily_chart(stock_code, period, limit=settings.CHART_DAYS) output2 = chart_data.get('output2', []) formatted_data = [] for item in output2: formatted_data.append({ "date": item.get('stck_bsop_date'), "open": int(item.get('stck_oprc')) if item.get('stck_oprc') else 0, "high": int(item.get('stck_hgpr')) if item.get('stck_hgpr') else 0, "low": int(item.get('stck_lwpr')) if item.get('stck_lwpr') else 0, "close": int(item.get('stck_clpr')) if item.get('stck_clpr') else 0, "volume": int(item.get('acml_vol')) if item.get('acml_vol') else 0 }) current_output = current_price_data.get('output', {}) return { "stock_code": stock_code, "period": period, "current_price": int(current_output.get('stck_prpr', 0)) if current_output else 0, "change_rate": float(current_output.get('prdy_ctrt', 0)) if current_output else 0, "volume": int(current_output.get('acml_vol', 0)) if current_output else 0, "data": formatted_data } else: # 해외 주식 logic (거래소 자동 판별) exchange = _detect_exchange(stock_code) current_price_data = await korea_investment_api.get_overseas_price(stock_code, exchange) chart_data = await korea_investment_api.get_overseas_daily_chart(stock_code, exchange, period) output2 = chart_data.get('output2', []) formatted_data = [] for item in output2: # 해외주식 데이터 키 매핑 (Key names depend on actual API response, adjusted to common KIS overseas response) formatted_data.append({ "date": item.get('xymd'), # 일자 "open": float(item.get('open')) if item.get('open') else 0.0, "high": float(item.get('high')) if item.get('high') else 0.0, "low": float(item.get('low')) if item.get('low') else 0.0, "close": float(item.get('clos')) if item.get('clos') else 0.0, "volume": float(item.get('tvol')) if item.get('tvol') else 0.0 }) curr = current_price_data.get('output', {}) return { "stock_code": stock_code, "period": period, "current_price": float(curr.get('last', 0)) if curr else 0.0, "change_rate": float(curr.get('rate', 0)) if curr else 0.0, "volume": float(curr.get('tvol', 0)) if curr else 0.0, "data": formatted_data } except Exception as e: logger.error(f"Failed to get market data for {stock_code}: {e}") raise 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]: """포트폴리오 확인""" if not korea_investment_api: return {"error": "Korea Investment API not available"} try: balance = await korea_investment_api.get_balance() output1 = balance.get('output1', []) output2 = balance.get('output2', []) # Total asset summary holdings = [] for item in output1: # Filter if stock_code provided if stock_code and item.get('pdno') != stock_code: continue holdings.append({ "stock_code": item.get('pdno'), "stock_name": item.get('prdt_name'), "quantity": int(item.get('hldg_qty')), "average_price": float(item.get('pchs_avg_pric')), "current_price": float(item.get('prpr')), "profit_loss": float(item.get('evlu_pfls_amt')), "profit_rate": float(item.get('evlu_pfls_rt')) }) total_eval = 0 if output2: total_eval = float(output2[0].get('tot_evlu_mony', 0)) return { "total_value": total_eval, "holdings": holdings } except Exception as e: logger.error(f"Failed to get portfolio: {e}") raise 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]: """주문 실행. [수정 3] MCP 서버에서 KIS API를 직접 호출하는 것은 리스크 검증 파이프라인을 우회하므로 허용하지 않는다. 주문은 반드시 백엔드 API(/api/trades)를 통해 실행해야 한다. MCP 서버 내부에서 korea_investment_api 인스턴스가 None인지 여부와 무관하게 이 함수는 항상 에러를 반환한다. 실제 주문은 auto_trade_claude.py의 auto_trade_from_claude() 함수(리스크 검증 포함)를 사용하거나 백엔드 /api/trades 엔드포인트를 호출해야 한다. """ logger.warning( "[execute_order] MCP 서버에서 직접 주문 호출 차단 — " "백엔드 /api/trades 엔드포인트를 통해 주문하십시오. " f"요청 정보: stock_code={stock_code}, order_type={order_type}, qty={quantity}, price={price}" ) return { "error": ( "MCP 서버에서 KIS API 직접 주문은 리스크 검증 우회 위험으로 차단되었습니다. " "백엔드 API(/api/trades)를 통해 주문하십시오." ), "blocked": True, } async def get_news(self, stock_code: str = None, limit: int = 5) -> Dict[str, Any]: """네이버 금융 뉴스 크롤링""" if not stock_code: return {"error": "stock_code is required"} try: url = f"https://finance.naver.com/item/news_news.naver?code={stock_code}&page=1" headers = { "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36" } async with httpx.AsyncClient() as client: response = await client.get(url, headers=headers) if response.status_code != 200: return {"error": f"Failed to fetch news: {response.status_code}"} soup = BeautifulSoup(response.text, 'html.parser') news_items = [] # 네이버 금융 뉴스 구조에 맞춰 파싱 for row in soup.select('table.type5 tbody tr'): title_tag = row.select_one('.title a') if not title_tag: continue title = title_tag.get_text(strip=True) # [수정 6] href 속성이 없는 경우 KeyError 방지 href = title_tag.get('href', '') link = ("https://finance.naver.com" + href) if href else '' date_tag = row.select_one('.date') date = date_tag.get_text(strip=True) if date_tag else "" source_tag = row.select_one('.info') source = source_tag.get_text(strip=True) if source_tag else "" # 관련뉴스/공시 제외하고 싶으면 필터링 가능 item = {"title": title, "date": date, "source": source} if not settings.TOKEN_SAVE_MODE: item["link"] = link news_items.append(item) if len(news_items) >= limit: break return { "stock_code": stock_code, "news": news_items } except Exception as e: logger.error(f"Error fetching news for {stock_code}: {e}") return {"error": str(e), "news": []} mcp_tools = MCPTools()