수정본
This commit is contained in:
188
mcp-server/gemini_client.py
Normal file
188
mcp-server/gemini_client.py
Normal file
@@ -0,0 +1,188 @@
|
||||
from google import genai
|
||||
from google.genai import types
|
||||
from config import settings
|
||||
from utils import _dumps
|
||||
from typing import Dict, Any, List
|
||||
import json
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class GeminiClient:
|
||||
"""Gemini AI 클라이언트 (ClaudeClient와 동일한 인터페이스)"""
|
||||
|
||||
def __init__(self):
|
||||
self.client = genai.Client(api_key=settings.GEMINI_API_KEY)
|
||||
self.model_name = settings.GEMINI_MODEL
|
||||
self.generate_config = types.GenerateContentConfig(
|
||||
max_output_tokens=settings.MAX_TOKENS,
|
||||
temperature=settings.TEMPERATURE,
|
||||
)
|
||||
|
||||
async def analyze(
|
||||
self,
|
||||
system_prompt: str,
|
||||
user_message: str,
|
||||
tools: List[Dict[str, Any]] = None
|
||||
) -> Dict[str, Any]:
|
||||
"""Gemini를 통한 분석 실행"""
|
||||
try:
|
||||
response = await self.client.aio.models.generate_content(
|
||||
model=self.model_name,
|
||||
contents=f"{system_prompt}\n\n{user_message}",
|
||||
config=self.generate_config,
|
||||
)
|
||||
|
||||
text = response.text if response.text else ""
|
||||
return {
|
||||
"content": [{"type": "text", "text": text}],
|
||||
"stop_reason": "end_turn",
|
||||
"usage": {
|
||||
"input_tokens": response.usage_metadata.prompt_token_count if response.usage_metadata else 0,
|
||||
"output_tokens": response.usage_metadata.candidates_token_count if response.usage_metadata else 0,
|
||||
}
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Gemini analysis error: {e}")
|
||||
raise
|
||||
|
||||
def _parse_json_response(self, response: Dict[str, Any], fallback: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""응답 텍스트에서 JSON 파싱 (공통 헬퍼)"""
|
||||
for content in response["content"]:
|
||||
if content["type"] == "text":
|
||||
try:
|
||||
text = content["text"]
|
||||
json_start = text.find('{')
|
||||
json_end = text.rfind('}') + 1
|
||||
if json_start != -1 and json_end > json_start:
|
||||
return json.loads(text[json_start:json_end])
|
||||
except (json.JSONDecodeError, Exception):
|
||||
return {**fallback, "analysis": content["text"]}
|
||||
return fallback
|
||||
|
||||
async def validate_trading_rule(
|
||||
self,
|
||||
rule: Dict[str, Any],
|
||||
market_data: Dict[str, Any],
|
||||
news_data: List[Dict[str, Any]] = None
|
||||
) -> Dict[str, Any]:
|
||||
"""트레이딩 규칙 검증"""
|
||||
system_prompt = """주식 트레이딩 규칙 검증 전문가. 규칙/시장/뉴스를 분석하여 JSON으로만 응답.
|
||||
필수 필드: decision(BUY/SELL/HOLD), confidence(0-1), risk_level, recommendation.
|
||||
악재(유상증자,감자,소송,실적악화) 시 confidence 낮추고 경고 포함. JSON 외 텍스트 금지."""
|
||||
|
||||
user_message = f"규칙:{_dumps(rule)}\n시장:{_dumps(market_data)}"
|
||||
if news_data:
|
||||
user_message += f"\n뉴스:{_dumps(news_data)}"
|
||||
|
||||
response = await self.analyze(system_prompt, user_message)
|
||||
return self._parse_json_response(
|
||||
response,
|
||||
{"decision": "HOLD", "confidence": 0.0, "analysis": "분석 실패"}
|
||||
)
|
||||
|
||||
async def analyze_market(
|
||||
self,
|
||||
stock_code: str,
|
||||
market_data: Dict[str, Any],
|
||||
news: List[Dict[str, Any]] = None
|
||||
) -> Dict[str, Any]:
|
||||
"""시장 분석"""
|
||||
system_prompt = """주식 시장 분석 전문가. JSON으로만 응답.
|
||||
필수 필드: trend(UP/DOWN/SIDEWAYS), sentiment(POSITIVE/NEGATIVE/NEUTRAL), recommendation(BUY/SELL/HOLD), confidence(0-1), analysis(간략설명). JSON 외 텍스트 금지."""
|
||||
|
||||
news_limit = settings.NEWS_LIMIT if settings.TOKEN_SAVE_MODE else 5
|
||||
news_text = ""
|
||||
if news:
|
||||
news_text = "\n뉴스:" + ",".join([item.get('title', '') for item in news[:news_limit]])
|
||||
|
||||
user_message = f"종목:{stock_code}\n시장:{_dumps(market_data)}{news_text}"
|
||||
|
||||
response = await self.analyze(system_prompt, user_message)
|
||||
return self._parse_json_response(
|
||||
response,
|
||||
{"trend": "UNKNOWN", "sentiment": "NEUTRAL", "recommendation": "HOLD", "confidence": 0.0}
|
||||
)
|
||||
|
||||
async def calculate_fair_price(
|
||||
self,
|
||||
stock_code: str,
|
||||
market_data: Dict[str, Any],
|
||||
financial_data: Dict[str, Any] = None
|
||||
) -> Dict[str, Any]:
|
||||
"""AI 기반 적정 매수/매도 가격 산출"""
|
||||
system_prompt = """주식 가격 전략 전문가. JSON으로만 응답.
|
||||
필수 필드: buy_prices(aggressive,neutral,conservative), sell_prices(target1,target2,target3), stop_loss, strategy, confidence(0-1). JSON 외 텍스트 금지."""
|
||||
|
||||
user_message = (
|
||||
f"종목:{stock_code} 현재가:{market_data.get('current_price')} "
|
||||
f"고가:{market_data.get('high_price')} 저가:{market_data.get('low_price')} "
|
||||
f"거래량:{market_data.get('volume')} 지표:{_dumps(market_data.get('indicators', {}))}"
|
||||
)
|
||||
|
||||
response = await self.analyze(system_prompt, user_message)
|
||||
return self._parse_json_response(
|
||||
response,
|
||||
{"error": "분석 실패", "confidence": 0.0}
|
||||
)
|
||||
|
||||
async def assess_risk(
|
||||
self,
|
||||
portfolio: Dict[str, Any],
|
||||
trade_plan: Dict[str, Any]
|
||||
) -> Dict[str, Any]:
|
||||
"""리스크 평가"""
|
||||
system_prompt = """투자 리스크 관리 전문가. JSON으로만 응답.
|
||||
필수 필드: risk_level(LOW/MEDIUM/HIGH/CRITICAL), risk_factors, approval(APPROVED/REJECTED), recommendation. JSON 외 텍스트 금지."""
|
||||
|
||||
user_message = f"포트폴리오:{_dumps(portfolio)}\n거래계획:{_dumps(trade_plan)}"
|
||||
|
||||
response = await self.analyze(system_prompt, user_message)
|
||||
return self._parse_json_response(
|
||||
response,
|
||||
{"error": "Failed to assess risk"}
|
||||
)
|
||||
|
||||
async def optimize_trading_rules(
|
||||
self,
|
||||
stock_code: str,
|
||||
market_data: Dict[str, Any],
|
||||
news_data: List[Dict[str, Any]] = None
|
||||
) -> Dict[str, Any]:
|
||||
"""시장 상황에 맞는 트레이딩 규칙 최적화"""
|
||||
system_prompt = """알고리즘 트레이딩 전략가. JSON으로만 응답.
|
||||
필수 필드: strategy_name, market_condition, parameters(buy_conditions,sell_conditions(take_profit_rate,stop_loss_rate,trailing_stop)), risk_assessment, confidence_score(0-100).
|
||||
악재시 보수적, 호재시 적극적 설정. JSON 외 텍스트 금지."""
|
||||
|
||||
user_message = f"종목:{stock_code}\n시장:{_dumps(market_data)}"
|
||||
if news_data:
|
||||
user_message += f"\n뉴스:{_dumps(news_data)}"
|
||||
|
||||
response = await self.analyze(system_prompt, user_message)
|
||||
return self._parse_json_response(
|
||||
response,
|
||||
{"error": "Failed to generate optimized rules"}
|
||||
)
|
||||
|
||||
async def generate_trade_journal(
|
||||
self,
|
||||
trade_info: Dict[str, Any],
|
||||
market_data: Dict[str, Any]
|
||||
) -> Dict[str, Any]:
|
||||
"""매매일지 자동 작성"""
|
||||
system_prompt = """매매일지 작성 전문가. JSON으로만 응답.
|
||||
필수 필드: title, content(간략 Markdown), mood(평정심/흥분/공포/아쉬움/만족), tags(배열), score(0-100).
|
||||
매매근거, 잘한점/아쉬운점, 개선사항 포함. JSON 외 텍스트 금지."""
|
||||
|
||||
user_message = f"거래:{_dumps(trade_info)}\n시장:{_dumps(market_data)}"
|
||||
|
||||
response = await self.analyze(system_prompt, user_message)
|
||||
return self._parse_json_response(
|
||||
response,
|
||||
{"error": "Failed to generate journal"}
|
||||
)
|
||||
|
||||
|
||||
gemini_client = GeminiClient()
|
||||
Reference in New Issue
Block a user