453 lines
18 KiB
Python
453 lines
18 KiB
Python
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
|
|
|
|
# Optional: KIS API dependency (not currently available in MCP server directly)
|
|
korea_investment_api = None
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class MCPTools:
|
|
"""MCP Tools Registry"""
|
|
|
|
|
|
def __init__(self):
|
|
self.tools: Dict[str, Callable] = {}
|
|
self._register_tools()
|
|
# Initialize KIS API if available
|
|
self.korea_investment_api = None
|
|
|
|
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 (기본 NASD 가정)
|
|
exchange = "NASD"
|
|
# TODO: 거래소 판별 로직 고도화 필요 (현재는 NASD 고정)
|
|
|
|
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]:
|
|
"""주문 실행 (실제로는 backend API 호출)"""
|
|
if not korea_investment_api:
|
|
return {"error": "Korea Investment API not available"}
|
|
|
|
try:
|
|
# 국내/해외 구분
|
|
is_domestic = stock_code.isdigit() and len(stock_code) == 6
|
|
|
|
if is_domestic:
|
|
result = await korea_investment_api.place_order(
|
|
stock_code=stock_code,
|
|
order_type=order_type,
|
|
quantity=quantity,
|
|
price=int(price) if price else None
|
|
)
|
|
else:
|
|
# 해외 주문 (지정가 필수)
|
|
if price is None:
|
|
return {"error": "Overseas orders require a limit price."}
|
|
|
|
result = await korea_investment_api.place_overseas_order(
|
|
symbol=stock_code,
|
|
order_type=order_type,
|
|
quantity=quantity,
|
|
price=price,
|
|
exchange_code="NASD" # 기본값
|
|
)
|
|
|
|
return result
|
|
except Exception as e:
|
|
logger.error(f"Failed to place order: {e}")
|
|
raise
|
|
|
|
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)
|
|
link = "https://finance.naver.com" + title_tag['href']
|
|
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()
|