From da5cef192594236739f6a97028f7c4f9dc009e2d Mon Sep 17 00:00:00 2001 From: shuim Date: Fri, 6 Mar 2026 14:40:31 +0900 Subject: [PATCH] =?UTF-8?q?MCP=20=EC=88=98=EC=A0=95=EC=A4=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/api/endpoints/mcp.py | 3 + backend/app/api/endpoints/recommendations.py | 96 ++++++++------------ backend/app/services/trade_executor.py | 56 +----------- mcp-server/claude_client.py | 59 ++++++++++-- 4 files changed, 91 insertions(+), 123 deletions(-) diff --git a/backend/app/api/endpoints/mcp.py b/backend/app/api/endpoints/mcp.py index 171921d..bd23885 100644 --- a/backend/app/api/endpoints/mcp.py +++ b/backend/app/api/endpoints/mcp.py @@ -72,6 +72,9 @@ async def request_mcp_risk_assessment( - approve=False → 주문 스킵 - MCP 장애 / 타임아웃 → fail-open (approve=True, 매매 중단 방지) """ + if not settings.MCP_VALIDATION_ENABLED: + return True, "mcp_disabled" + try: client = get_http_client() except RuntimeError: diff --git a/backend/app/api/endpoints/recommendations.py b/backend/app/api/endpoints/recommendations.py index 538c190..5ff7b32 100644 --- a/backend/app/api/endpoints/recommendations.py +++ b/backend/app/api/endpoints/recommendations.py @@ -3,7 +3,6 @@ from typing import List, Optional from datetime import datetime, date, timedelta import asyncio import logging -import httpx from collections import namedtuple from app.core.database import get_db, AsyncSessionLocal @@ -662,72 +661,49 @@ async def execute_auto_trading(db: AsyncSession = Depends(get_db)): @router.get("/ai/{stock_code}") async def get_ai_recommendation(stock_code: str): - from app.services.korea_investment_api import korea_investment_api + """종목 기술적 분석 — AI 미사용, 전략 엔진(RSI/BB/MA/ATR/ADX) 직접 계산.""" + from app.services.market_analyzer import analyze_stock_for_batch try: - price_data = {} - try: - price_data = await korea_investment_api.get_current_price(stock_code) - except Exception as e: - logger.warning(f"Failed to fetch price data locally: {e}") + rec = await analyze_stock_for_batch(stock_code, stock_code) + if rec is None: + raise HTTPException(status_code=404, detail="분석 데이터를 가져올 수 없습니다.") - news_data = [] - try: - async with httpx.AsyncClient() as client: - mcp_url = f"{settings.MCP_SERVER_URL}/tools/call" - payload = { - "tool_name": "get_news", - "parameters": {"stock_code": stock_code, "limit": 3} - } - response = await client.post(mcp_url, json=payload, timeout=10.0) - if response.status_code == 200: - result = response.json() - if result.get("status") == "success": - data = result.get("result", {}) - if "news" in data: - news_data = data["news"] - else: - news_data = data - except Exception as e: - logger.warning(f"Failed to fetch news via MCP: {e}") + signals = rec.signals or [] + score = rec.score or 0 - analysis_result = "AI 분석을 수행할 수 없습니다." - try: - async with httpx.AsyncClient() as client: - analyze_url = f"{settings.MCP_SERVER_URL}/analyze" - market_summary = {} - if "output" in price_data: - output = price_data["output"] - market_summary = { - "current_price": output.get("stck_prpr"), - "change_rate": output.get("prdy_ctrt"), - "volume": output.get("acml_vol"), - "high": output.get("stck_hgpr"), - "low": output.get("stck_lwpr") - } + # 전략 신호 기반 요약 문자열 생성 + if score >= 3: + summary = f"매수 신호 감지 (점수: {score}). 주요 신호: {', '.join(signals[:3])}" + elif score <= -2: + summary = f"매도/경고 신호 (점수: {score}). 주요 신호: {', '.join(signals[:3])}" + else: + summary = f"보유/관망 (점수: {score}). 신호: {', '.join(signals[:3]) if signals else '없음'}" - payload = { - "analysis_type": "MARKET_ANALYSIS", - "data": { - "stock_code": stock_code, - "market_data": market_summary, - "news": news_data - } - } - response = await client.post(analyze_url, json=payload, timeout=45.0) - if response.status_code == 200: - analysis_res = response.json() - if "result" in analysis_res: - analysis_result = analysis_res["result"] - except Exception as e: - logger.error(f"Failed to fetch AI analysis: {e}") - return { "stock_code": stock_code, - "analysis": analysis_result, - "news": news_data, - "market_data": price_data.get("output", {}) + "analysis": summary, + "signals": signals, + "score": score, + "market_data": { + "current_price": rec.current_price, + "change_rate": rec.change_rate, + "volume": rec.volume, + "rsi": rec.rsi, + "ma5": rec.ma5, + "ma20": rec.ma20, + "ma60": rec.ma60, + "bollinger_upper": rec.bollinger_upper, + "bollinger_lower": rec.bollinger_lower, + "bollinger_middle": rec.bollinger_middle, + "atr_14": rec.atr_14, + "adx_14": rec.adx_14, + "stochastic_k": rec.stochastic_k, + "stochastic_d": rec.stochastic_d, + } } + except HTTPException: + raise except Exception as e: - logger.error(f"AI Recommendation error: {e}") + logger.error(f"Strategy analysis error for {stock_code}: {e}") raise HTTPException(status_code=500, detail=str(e)) diff --git a/backend/app/services/trade_executor.py b/backend/app/services/trade_executor.py index 661cd1b..6dd451c 100644 --- a/backend/app/services/trade_executor.py +++ b/backend/app/services/trade_executor.py @@ -1,6 +1,5 @@ import asyncio import logging -import httpx from decimal import Decimal from datetime import timedelta from sqlalchemy.ext.asyncio import AsyncSession @@ -25,58 +24,9 @@ _order_semaphore = asyncio.Semaphore(1) async def _mcp_validate_buy(stock: dict) -> tuple: - if not settings.MCP_VALIDATION_ENABLED: - return True, "mcp_disabled" - - try: - payload = { - "analysis_type": "RULE_VALIDATION", - "data": { - "rule": { - "id": stock.get("rule_id"), - "name": stock.get("rule_name", "auto_trade"), - }, - "market_data": { - "stock_code": stock.get("stock_code"), - "stock_name": stock.get("stock_name"), - "current_price": stock.get("current_price"), - "signals": stock.get("signals", []), - "score": stock.get("score"), - "ml_signal": stock.get("ml_signal"), - "ml_confidence": stock.get("ml_confidence"), - "ma20": stock.get("ma20"), - "ma60": stock.get("ma60"), - } - } - } - - async with httpx.AsyncClient(timeout=10.0) as client: - resp = await client.post( - f"{settings.MCP_SERVER_URL}/analyze", - json=payload - ) - resp.raise_for_status() - data = resp.json() - - result = data.get("result", {}) - decision = result.get("decision", "HOLD") - confidence = float(result.get("confidence", 0.0)) - - if decision == "BUY" and confidence >= settings.MCP_CONFIDENCE_THRESHOLD: - return True, f"mcp_approved(conf={confidence:.2f})" - else: - return False, f"mcp_rejected(decision={decision}, conf={confidence:.2f})" - - except httpx.HTTPStatusError as e: - if e.response.status_code == 500: - logger.warning(f"⚠️ MCP 서버 500 에러 — 룰 기반으로 우회: {e}") - return True, "mcp_500_fallback_to_rules" - logger.warning(f"⚠️ MCP 검증 실패 — 매매 차단: {e}") - return False, "mcp_unavailable" - - except Exception as e: - logger.warning(f"⚠️ MCP 검증 실패 — 매매 차단: {e}") - return False, "mcp_unavailable" + # 전략 엔진(RSI/BB/MA/ATR 등)이 이미 신호를 검증하므로 AI 호출 불필요. + # RISK_ASSESSMENT는 매매 직전 mcp.py에서 별도 수행. + return True, "strategy_validated" async def _execute_buy( diff --git a/mcp-server/claude_client.py b/mcp-server/claude_client.py index 2425093..70ee7c3 100644 --- a/mcp-server/claude_client.py +++ b/mcp-server/claude_client.py @@ -8,46 +8,85 @@ import logging logger = logging.getLogger(__name__) +# 캐싱 공통 컨텍스트 — 전략 규칙을 한 번만 전송하고 이후 호출에서 캐시로 재사용. +# Claude Sonnet 기준 캐싱 최소 임계값: 1024 토큰. +_CACHED_CONTEXT = """[자동 주식 매매 시스템 — 리스크 규칙] +포지션 사이징: 종목당 총자본 3~4%, 최대 보유 20~25개, 레버리지 ETF 최대 20%. +손절(SL): 안정종목 -6%, 변동성종목 -10%, ATR 우선(진입가 - 2.5*ATR). +익절(TP): +2R 도달 시 50% 분할매도, 나머지는 고점-(1.5*ATR) 트레일링스탑. +MDD -10%: 신규진입 중단, 포지션 30% 축소, 단기전략 중단. +MDD -20%: 전 포지션 50% 강제청산 (레버리지 우선). +가격필터: 국내 5,000원 초과, 미국 $4 초과 (ETF·리츠 예외). +상관관계: 섹터당 최대 3종목, 상관계수 0.7 초과 진입 거부. +시장국면: SIDEWAYS(ADX<20) 진입 금지. +전략A(평균회귀): 상승장 RSI<40·BB하단, 하락장 RSI<30·BB하단·반전캔들, 거래량>20MA. +전략B(코스닥스캐너): 총자본 25%, 점수=0.5*RSI+0.3*BB+0.2*Vol 상위 20개. +단기전략(15분): RSI<30·BB하단·5MA돌파·거래량>1.3배, MDD -10% 미만 비활성. +""" + + class ClaudeClient: """Claude AI 클라이언트""" def __init__(self): self.client = AsyncAnthropic(api_key=settings.ANTHROPIC_API_KEY) self.model = settings.CLAUDE_MODEL - + async def analyze( self, system_prompt: str, user_message: str, tools: List[Dict[str, Any]] = None ) -> Dict[str, Any]: - """Claude를 통한 분석 실행""" + """Claude를 통한 분석 실행. + + 시스템 프롬프트는 두 블록으로 구성: + 1) _CACHED_CONTEXT — cache_control 적용, 반복 호출 시 캐시에서 재사용. + 2) 메서드별 role 지시문 — 캐싱 없이 매번 전송. + """ try: messages = [ {"role": "user", "content": user_message} ] - + params = { "model": self.model, "max_tokens": settings.MAX_TOKENS, "temperature": settings.TEMPERATURE, - "system": system_prompt, - "messages": messages + "system": [ + { + "type": "text", + "text": _CACHED_CONTEXT, + "cache_control": {"type": "ephemeral"}, + }, + { + "type": "text", + "text": system_prompt, + }, + ], + "messages": messages, } - + if tools: params["tools"] = tools - + response = await self.client.messages.create(**params) # 응답 파싱 + usage = response.usage + cache_read = getattr(usage, "cache_read_input_tokens", 0) or 0 + cache_write = getattr(usage, "cache_creation_input_tokens", 0) or 0 + if cache_read: + logger.debug(f"[cache] hit={cache_read} write={cache_write} input={usage.input_tokens} output={usage.output_tokens}") result = { "content": [], "stop_reason": response.stop_reason, "usage": { - "input_tokens": response.usage.input_tokens, - "output_tokens": response.usage.output_tokens - } + "input_tokens": usage.input_tokens, + "output_tokens": usage.output_tokens, + "cache_read_tokens": cache_read, + "cache_write_tokens": cache_write, + }, } for block in response.content: