auto-stock/mcp-server/claude_client.py

"""Claude AI 클라이언트 모듈 (claude_client.py)

Anthropic Claude API를 사용하여 주식 분석, 리스크 평가, 매매일지 생성 등을 수행.

주요 기능:
  - analyze(): 범용 Claude 분석 (캐싱 컨텍스트 지원)
  - validate_trading_rule(): 트레이딩 규칙 검증
  - analyze_market(): 시장 분석 (추세/심리/추천)
  - calculate_fair_price(): AI 기반 적정가 산출
  - assess_risk(): 리스크 평가 (토큰 최소화)
  - optimize_trading_rules(): 전략 파라미터 최적화
  - generate_trade_journal(): 매매일지 자동 생성

캐싱 전략:
  - _CACHED_CONTEXT를 system 프롬프트의 첫 블록으로 전송 (cache_control: ephemeral)
  - 반복 호출 시 Anthropic 서버 측에서 캐시 히트로 입력 토큰 절약
"""

from anthropic import AsyncAnthropic
from config import settings
from utils import _dumps
from typing import Dict, Any, List
import json
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 클라이언트 (하이브리드 Haiku/Sonnet 지원)"""

    def __init__(self):
        self.client = AsyncAnthropic(api_key=settings.ANTHROPIC_API_KEY)
        self.model = settings.CLAUDE_MODEL
        self.haiku_model = settings.CLAUDE_HAIKU_MODEL

    async def analyze(
        self,
        system_prompt: str,
        user_message: str,
        tools: List[Dict[str, Any]] = None,
        max_tokens: int = None,
        model_override: str = None,
    ) -> Dict[str, Any]:
        """Claude를 통한 분석 실행.

        시스템 프롬프트는 두 블록으로 구성:
          1) _CACHED_CONTEXT — cache_control 적용, 반복 호출 시 캐시에서 재사용.
          2) 메서드별 role 지시문 — 캐싱 없이 매번 전송.

        Args:
            model_override: 특정 모델 강제 지정 (하이브리드 모드에서 Haiku/Sonnet 분기용)
        """
        try:
            use_model = model_override or self.model
            messages = [
                {"role": "user", "content": user_message}
            ]

            params = {
                "model": use_model,
                "max_tokens": max_tokens or settings.MAX_TOKENS,
                # anthropic SDK 1.x: temperature 등 샘플링 파라미터 제거됨 (전달 시 TypeError)
                "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,
                "model_used": use_model,
                "usage": {
                    "input_tokens": usage.input_tokens,
                    "output_tokens": usage.output_tokens,
                    "cache_read_tokens": cache_read,
                    "cache_write_tokens": cache_write,
                },
            }
            logger.debug(
                f"[Claude] model={use_model} in={usage.input_tokens} out={usage.output_tokens} "
                f"cache_hit={cache_read}"
            )
            
            for block in response.content:
                if block.type == "text":
                    result["content"].append({
                        "type": "text",
                        "text": block.text
                    })
                elif block.type == "tool_use":
                    result["content"].append({
                        "type": "tool_use",
                        "id": block.id,
                        "name": block.name,
                        "input": block.input
                    })
            
            return result
        
        except Exception as e:
            logger.error(f"Claude analysis error: {e}")
            raise
    
    def _parse_json_response(self, response: Dict[str, Any], fallback: Dict[str, Any]) -> Dict[str, Any]:
        """응답 텍스트에서 JSON 파싱 (공통 헬퍼).

        GeminiClient._parse_json_response()와 동일한 로직.
        1) 응답 content에서 text 블록을 찾아 JSON 파싱 시도
        2) 순수 json.loads 실패 시 텍스트 내 첫 '{' ~ 마지막 '}' 범위를 추출하여 재시도
        3) 모두 실패하면 fallback에 analysis 필드를 추가하여 반환
        """
        for content in response["content"]:
            if content["type"] == "text":
                text = content["text"]
                try:
                    return json.loads(text)
                except json.JSONDecodeError:
                    pass
                # JSON이 마크다운 코드블록 등으로 감싸진 경우 추출 시도
                try:
                    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):
                    pass
                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 낮추고 경고 포함."""

        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(간략설명)."""

        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)."""

        user_message = f"종목:{stock_code} 현재가:{market_data.get('current_price')} 고가:{market_data.get('high_price')} 저가:{market_data.get('low_price')} 거래량:{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]:
        """리스크 평가 (토큰 최소화 — max_tokens=RISK_MAX_TOKENS).

        하이브리드 모드: Haiku 사용 (단순 승인/거부 판단)
        """
        system_prompt = """리스크 평가. JSON만 출력.
필수: risk_level(LOW/MEDIUM/HIGH), approval(APPROVED/REJECTED), reason(한줄)."""

        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)}"

        # 하이브리드: 단순 판단은 Haiku로 비용 절감
        _model = self.haiku_model if settings.HYBRID_AI_ENABLED else None
        response = await self.analyze(system_prompt, user_message, max_tokens=settings.RISK_MAX_TOKENS, model_override=_model)
        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).
악재시 보수적, 호재시 적극적 설정."""

        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]:
        """매매일지 자동 작성.

        하이브리드 모드: Haiku 사용 (템플릿 기반 정리 작업)
        """
        system_prompt = """매매일지 작성 전문가. JSON으로 응답.
필수 필드: title, content(간략 Markdown), mood(평정심/흥분/공포/아쉬움/만족), tags(배열), score(0-100).
매매근거, 잘한점/아쉬운점, 개선사항 포함."""

        user_message = f"거래:{_dumps(trade_info)}\n시장:{_dumps(market_data)}"

        _model = self.haiku_model if settings.HYBRID_AI_ENABLED else None
        response = await self.analyze(system_prompt, user_message, model_override=_model)
        return self._parse_json_response(
            response,
            {"error": "Failed to generate journal"}
        )

    async def analyze_batch_stocks(
        self,
        stocks: List[Dict[str, Any]],
    ) -> Dict[str, Any]:
        """장 전 배치 분석 — 10종목을 한 번에 받아 각각 매수적합/부적합 판정.

        Sonnet 사용 (기본 self.model). max_tokens=2048 (10종목 응답이므로 넉넉하게).
        입력 stocks: [{stock_code, stock_name, current_price, rsi, bb_position,
                       volume_ratio, score, news_titles}, ...]
        """
        system_prompt = """주식 스윙 트레이딩 분석가. 종목 리스트를 분석하여 JSON 배열로 응답.
각 종목: stock_code, suitable(true/false), confidence(0-1), reason(한줄), risk_factors(배열).
JSON 외 텍스트 금지."""

        user_message = f"종목리스트:{_dumps(stocks)}"

        response = await self.analyze(system_prompt, user_message, max_tokens=2048)
        # 배열 응답 파싱: '{' 대신 '[' ~ ']' 범위도 처리
        return self._parse_batch_response(response)

    def _parse_batch_response(self, response: Dict[str, Any]) -> Dict[str, Any]:
        """배치 분석 응답 파싱 — JSON 배열 또는 객체 모두 처리.

        1) 응답 텍스트에서 JSON 배열([...]) 파싱 시도
        2) JSON 객체({...}) 파싱 시도 (AI가 {stocks: [...]} 형태로 응답할 수 있음)
        3) 모두 실패하면 빈 결과 반환
        """
        for content in response["content"]:
            if content["type"] == "text":
                text = content["text"]
                # 배열 형태 시도
                try:
                    arr_start = text.find('[')
                    arr_end = text.rfind(']') + 1
                    if arr_start != -1 and arr_end > arr_start:
                        parsed = json.loads(text[arr_start:arr_end])
                        if isinstance(parsed, list):
                            return {"stocks": parsed}
                except (json.JSONDecodeError, Exception):
                    pass
                # 객체 형태 시도
                try:
                    obj_start = text.find('{')
                    obj_end = text.rfind('}') + 1
                    if obj_start != -1 and obj_end > obj_start:
                        return json.loads(text[obj_start:obj_end])
                except (json.JSONDecodeError, Exception):
                    pass
                return {"stocks": [], "analysis": text}
        return {"stocks": []}

    async def check_news_sentiment(
        self,
        stock_code: str,
        stock_name: str,
        news_titles: List[str],
    ) -> Dict[str, Any]:
        """장중 뉴스 감성 분석.

        하이브리드 모드: Haiku 사용 (단순 감성 판단). max_tokens=256 (짧은 응답).
        """
        system_prompt = """주식 뉴스 감성 분석. JSON만 출력.
필수: sentiment(POSITIVE/NEGATIVE/NEUTRAL), severity(LOW/MEDIUM/HIGH), action(HOLD/SELL/WATCH), reason(한줄).
JSON 외 텍스트 금지."""

        user_message = f"종목:{stock_code}({stock_name})\n뉴스제목:{_dumps(news_titles)}"

        _model = self.haiku_model if settings.HYBRID_AI_ENABLED else None
        response = await self.analyze(system_prompt, user_message, max_tokens=256, model_override=_model)
        return self._parse_json_response(
            response,
            {"sentiment": "NEUTRAL", "severity": "LOW", "action": "HOLD", "reason": "분석 실패"}
        )

    async def review_daily_trades(
        self,
        trades_summary: Dict[str, Any],
        market_summary: Dict[str, Any],
    ) -> Dict[str, Any]:
        """장 후 당일 매매 복기.

        Sonnet 사용 (기본 self.model). max_tokens=1024.
        trades_summary: {total_trades, wins, losses, pnl, strategies_used, notable_trades}
        market_summary: {kospi_change, regime, top_movers}
        """
        system_prompt = """주식 매매 복기 전문가. 당일 매매를 분석하여 JSON 응답.
필수: overall_grade(A/B/C/D/F), good_points(배열), bad_points(배열), improvements(배열), strategy_drift(bool), drift_detail(있으면), tomorrow_plan(한줄).
JSON 외 텍스트 금지."""

        user_message = f"매매요약:{_dumps(trades_summary)}\n시장요약:{_dumps(market_summary)}"

        response = await self.analyze(system_prompt, user_message, max_tokens=1024)
        return self._parse_json_response(
            response,
            {
                "overall_grade": "C",
                "good_points": [],
                "bad_points": [],
                "improvements": [],
                "strategy_drift": False,
                "tomorrow_plan": "분석 실패",
            }
        )


claude_client = ClaudeClient()