215 lines
7.4 KiB
Python
215 lines
7.4 KiB
Python
from fastapi import FastAPI, HTTPException
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from pydantic import BaseModel
|
|
from typing import Dict, Any, List
|
|
import logging
|
|
|
|
from config import settings
|
|
from claude_client import claude_client
|
|
from mcp_tools import mcp_tools
|
|
|
|
# AI_PROVIDER에 따라 클라이언트 동적 선택
|
|
if settings.AI_PROVIDER == "gemini":
|
|
from gemini_client import gemini_client
|
|
ai_client = gemini_client
|
|
else:
|
|
ai_client = claude_client
|
|
|
|
# 로깅 설정
|
|
logging.basicConfig(
|
|
level=getattr(logging, settings.LOG_LEVEL.upper(), logging.INFO),
|
|
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
app = FastAPI(
|
|
title="Auto Stock Trading MCP Server",
|
|
description="Model Context Protocol Server for Stock Trading with Claude AI",
|
|
version="1.0.0"
|
|
)
|
|
|
|
# CORS 설정 (이슈 5)
|
|
# 문제: allow_origins=["*"] + allow_credentials=True 조합은 CORS 스펙 위반 (브라우저 차단)
|
|
# 해결: 환경변수 MCP_CORS_ORIGINS로 origin 제한 + allow_credentials=False (서버-서버 통신 불필요)
|
|
_raw_cors = settings.MCP_CORS_ORIGINS
|
|
_allowed_origins = (
|
|
[o.strip() for o in _raw_cors.split(",") if o.strip()]
|
|
if _raw_cors
|
|
else ["http://localhost:8000", "http://127.0.0.1:8000", "http://backend:8000"]
|
|
)
|
|
logger.info(f"MCP CORS allowed origins: {_allowed_origins}")
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=_allowed_origins,
|
|
allow_credentials=False, # 서버-서버 통신; 브라우저 세션 쿠키 불필요
|
|
allow_methods=["GET", "POST"],
|
|
allow_headers=["Content-Type", "Authorization"],
|
|
)
|
|
|
|
|
|
class ToolCallRequest(BaseModel):
|
|
tool_name: str
|
|
parameters: Dict[str, Any]
|
|
|
|
|
|
class AnalysisRequest(BaseModel):
|
|
analysis_type: str # RULE_VALIDATION, MARKET_ANALYSIS, RISK_ASSESSMENT
|
|
data: Dict[str, Any]
|
|
|
|
|
|
@app.get("/")
|
|
async def root():
|
|
return {
|
|
"message": "MCP Server for Auto Stock Trading",
|
|
"version": "1.0.0",
|
|
"status": "running"
|
|
}
|
|
|
|
|
|
@app.get("/health")
|
|
async def health_check():
|
|
return {
|
|
"status": "healthy",
|
|
"ai_provider": settings.AI_PROVIDER,
|
|
"claude_configured": bool(settings.ANTHROPIC_API_KEY),
|
|
"gemini_configured": bool(settings.GEMINI_API_KEY),
|
|
}
|
|
|
|
|
|
@app.get("/tools")
|
|
async def list_tools():
|
|
"""사용 가능한 도구 목록"""
|
|
return {
|
|
"tools": mcp_tools.get_tool_definitions()
|
|
}
|
|
|
|
|
|
@app.post("/tools/call")
|
|
async def call_tool(request: ToolCallRequest):
|
|
"""MCP 도구 호출"""
|
|
try:
|
|
result = await mcp_tools.call_tool(request.tool_name, request.parameters)
|
|
return result
|
|
except Exception as e:
|
|
logger.error(f"Tool call error: {e}")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
@app.post("/analyze")
|
|
async def analyze(request: AnalysisRequest):
|
|
"""Claude를 통한 분석"""
|
|
try:
|
|
analysis_type = request.analysis_type
|
|
data = request.data
|
|
|
|
if analysis_type == "RULE_VALIDATION":
|
|
rule = data.get("rule")
|
|
market_data = data.get("market_data")
|
|
if not rule or not market_data:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="rule and market_data are required"
|
|
)
|
|
stock_code = market_data.get("stock_code")
|
|
news_data = []
|
|
if stock_code:
|
|
try:
|
|
news_result = await mcp_tools.get_news(stock_code, limit=settings.NEWS_LIMIT)
|
|
if "news" in news_result:
|
|
news_data = news_result["news"]
|
|
except Exception as e:
|
|
logger.warning(f"Failed to fetch news for validation: {e}")
|
|
result = await ai_client.validate_trading_rule(rule, market_data, news_data)
|
|
return {"analysis_type": analysis_type, "result": result}
|
|
|
|
elif analysis_type == "MARKET_ANALYSIS":
|
|
stock_code = data.get("stock_code")
|
|
market_data = data.get("market_data")
|
|
news = data.get("news", [])
|
|
if not stock_code or not market_data:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="stock_code and market_data are required"
|
|
)
|
|
result = await ai_client.analyze_market(stock_code, market_data, news)
|
|
return {"analysis_type": analysis_type, "result": result}
|
|
|
|
elif analysis_type == "RISK_ASSESSMENT":
|
|
# 이슈 4: request_mcp_risk_assessment()가 trade_plan/portfolio/signals 전달
|
|
portfolio = data.get("portfolio") or {}
|
|
trade_plan = data.get("trade_plan")
|
|
signals = data.get("signals") or {}
|
|
if not trade_plan:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="trade_plan is required"
|
|
)
|
|
result = await ai_client.assess_risk(portfolio, trade_plan)
|
|
return {"analysis_type": analysis_type, "result": result}
|
|
|
|
elif analysis_type == "FAIR_PRICE":
|
|
stock_code = data.get("stock_code")
|
|
market_data = data.get("market_data")
|
|
financial_data = data.get("financial_data")
|
|
if not stock_code or not market_data:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="stock_code and market_data are required"
|
|
)
|
|
result = await ai_client.calculate_fair_price(
|
|
stock_code, market_data, financial_data
|
|
)
|
|
return {"analysis_type": analysis_type, "result": result}
|
|
|
|
elif analysis_type == "RULE_OPTIMIZATION":
|
|
stock_code = data.get("stock_code")
|
|
market_data = data.get("market_data")
|
|
if not stock_code or not market_data:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="stock_code and market_data are required"
|
|
)
|
|
news_data = []
|
|
try:
|
|
news_result = await mcp_tools.get_news(stock_code, limit=settings.NEWS_LIMIT)
|
|
if "news" in news_result:
|
|
news_data = news_result["news"]
|
|
except Exception as e:
|
|
logger.warning(f"Failed to fetch news for optimization: {e}")
|
|
result = await ai_client.optimize_trading_rules(stock_code, market_data, news_data)
|
|
return {"analysis_type": analysis_type, "result": result}
|
|
|
|
elif analysis_type == "GENERATE_JOURNAL":
|
|
trade_info = data.get("trade_info")
|
|
market_data = data.get("market_data")
|
|
if not trade_info or not market_data:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="trade_info and market_data are required"
|
|
)
|
|
result = await ai_client.generate_trade_journal(trade_info, market_data)
|
|
return {"analysis_type": analysis_type, "result": result}
|
|
|
|
else:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"Unknown analysis type: {analysis_type}"
|
|
)
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"Analysis error: {e}")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
uvicorn.run(
|
|
"main:app",
|
|
host=settings.MCP_SERVER_HOST,
|
|
port=settings.MCP_SERVER_PORT,
|
|
reload=True
|
|
)
|