auto-stock/mcp-server/gemini_client.py

"""Gemini AI 클라이언트 모듈 (gemini_client.py)

Google Gemini API를 사용하여 주식 분석, 리스크 평가, 매매일지 생성 등을 수행.
Claude 클라이언트와 동일한 인터페이스를 제공하여 AI_PROVIDER 설정으로 전환 가능.

주요 기능:
  - analyze(): 범용 Gemini 분석
  - validate_trading_rule(): 트레이딩 규칙 검증
  - analyze_market(): 시장 분석
  - calculate_fair_price(): 적정가 산출
  - assess_risk(): 리스크 평가
  - optimize_trading_rules(): 전략 최적화
  - generate_trade_journal(): 매매일지 생성

공통 헬퍼:
  - _parse_json_response(): AI 응답에서 JSON 추출 (Claude 클라이언트에도 동일 로직 존재)
"""

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 클라이언트 (최신 google-genai SDK 버전)"""

    def __init__(self):
        if not settings.GEMINI_API_KEY:
            raise ValueError("GEMINI_API_KEY is not set")
        
        # 최신 SDK에서는 Client 객체를 생성하여 사용합니다.
        self.client = genai.Client(api_key=settings.GEMINI_API_KEY)
        self.model_id = settings.GEMINI_MODEL  # 예: 'gemini-2.0-flash'
        
        # GenerationConfig 설정
        self.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,
        max_tokens: int = None,
    ) -> Dict[str, Any]:
        """Gemini를 통한 분석 실행"""
        try:
            config_with_system = types.GenerateContentConfig(
                system_instruction=system_prompt,
                max_output_tokens=max_tokens or self.config.max_output_tokens,
                temperature=self.config.temperature,
            )

            # 비동기 호출
            response = await self.client.models.generate_content(
                model=self.model_id,
                contents=user_message,
                config=config_with_system
            )

            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), approval(APPROVED/REJECTED), reason(한줄). JSON 외 금지."""

        slim_portfolio = {k: portfolio[k] for k in ("strategy", "position_count", "mdd") if k in portfolio}
        user_message = f"포트:{_dumps(slim_portfolio)} 계획:{_dumps(trade_plan)}"

        response = await self.analyze(system_prompt, user_message, max_tokens=settings.RISK_MAX_TOKENS)
        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()