web-page-backend/stock-lab/app/ai_analyst.py

"""ai_analyst.py — Gemini Pro 기반 주식 포트폴리오 AI 분석

환경변수:
  GEMINI_API_KEY  — Google AI Studio API 키 (필수)
  GEMINI_MODEL    — 사용할 모델 (기본: gemini-1.5-pro-latest)
"""
from __future__ import annotations

import os
import time
import logging
from datetime import datetime, date
from typing import Any

logger = logging.getLogger(__name__)

# ── 캐시 (메모리, 5분 TTL) ────────────────────────────────────────────────────
_cache: dict[str, Any] = {}
_CACHE_TTL = 300  # 5분


# ── Gemini 모델 로드 ──────────────────────────────────────────────────────────

def _get_model():
    api_key = os.getenv("GEMINI_API_KEY", "").strip()
    if not api_key:
        raise ValueError("GEMINI_API_KEY 환경변수가 설정되지 않았습니다.")

    import google.generativeai as genai  # lazy import (설치 안 된 경우 대비)

    genai.configure(api_key=api_key)
    model_name = os.getenv("GEMINI_MODEL", "gemini-1.5-flash")
    return genai.GenerativeModel(
        model_name,
        generation_config={
            "temperature": 0.65,
            "max_output_tokens": 4096,
        },
    )


# ── 프롬프트 빌더 ─────────────────────────────────────────────────────────────

def _build_prompt(holdings: list[dict], news: list[dict]) -> str:
    today = date.today().strftime("%Y년 %m월 %d일")

    # 보유 종목 섹션
    if holdings:
        holdings_lines = []
        for h in holdings:
            cp = h.get("current_price")
            rate = h.get("profit_rate")
            profit = h.get("profit_amount")
            rate_str = f"{rate:+.2f}%" if rate is not None else "시세 조회 불가"
            profit_str = f"({profit:+,.0f}원)" if profit is not None else ""
            cp_str = f"{cp:,}원" if cp else "조회 불가"
            holdings_lines.append(
                f"- **{h['name']}** ({h['ticker']}) | 계좌: {h.get('broker', '-')}\n"
                f"  수량 {h['quantity']}주 | 평균매입 {h['avg_price']:,}원 | "
                f"현재가 {cp_str} | 손익 {rate_str} {profit_str}"
            )
        holdings_text = "\n".join(holdings_lines)
    else:
        holdings_text = "보유 종목 없음"

    # 뉴스 섹션 (최근 8개)
    news_lines = []
    for i, n in enumerate(news[:8], 1):
        title = n.get("title", "").strip()
        category = n.get("category", "")
        if title:
            news_lines.append(f"{i}. [{category}] {title}")
    news_text = "\n".join(news_lines) if news_lines else "뉴스 없음"

    return f"""당신은 15년 이상 경력의 한국 주식시장 전문 애널리스트입니다.
오늘은 {today}입니다. 아래 포트폴리오와 뉴스를 바탕으로 전문가 분석을 제공해주세요.

---

## 현재 보유 포트폴리오

{holdings_text}

---

## 오늘의 주요 뉴스

{news_text}

---

## 분석 요청

다음 형식으로 명확하게 작성해주세요:

### 📈 오늘의 시장 환경
뉴스를 바탕으로 오늘 한국 주식시장의 전반적인 분위기와 주요 이슈를 2-3문장으로 요약하세요.

### 🔍 종목별 분석 및 행동 지침
각 보유 종목에 대해 아래 형식으로 작성하세요:

**[종목명 (티커)]**
- 현황: 현재 손익 상태와 포지션 평가
- 분석: 업황·섹터 동향, 관련 뉴스 영향, 주요 리스크/기회
- 🎯 행동 지침: **[매도 / 보유 / 추가매수 / 분할매도]** — 구체적 이유와 목표 참고 가격대

### 💼 포트폴리오 종합 의견
전체 포트폴리오의 섹터 편중, 리밸런싱 필요 여부, 현금 비중 조언을 작성하세요.

### ⚠️ 오늘 주의해야 할 리스크
매크로·섹터·개별 종목 측면에서 오늘 특히 주의할 리스크를 2-3가지 나열하세요.

---
분석은 반드시 한국어로, 구체적인 수치와 근거를 들어 전문적으로 작성해주세요.
투자 결정은 최종적으로 투자자 본인이 판단함을 명시하세요.
"""


# ── 메인 분석 함수 ────────────────────────────────────────────────────────────

def analyze_portfolio(
    holdings: list[dict],
    news: list[dict],
    force: bool = False,
) -> dict:
    """포트폴리오 AI 분석 (5분 캐시).

    Returns:
        {
          "analysis": str,           # 마크다운 분석 텍스트
          "generated_at": str,       # ISO timestamp
          "cached": bool,
          "holdings_count": int,
        }
    """
    now = time.time()
    today_str = date.today().isoformat()
    cached = _cache.get("portfolio")

    if not force and cached and cached.get("date") == today_str and (now - cached["ts"]) < _CACHE_TTL:
        logger.info("[AI] cache hit (%.0fs ago)", now - cached["ts"])
        return {**cached["result"], "cached": True}

    if not holdings:
        return {
            "analysis": "포트폴리오에 보유 종목이 없습니다. 종목을 추가한 후 다시 분석해주세요.",
            "generated_at": datetime.now().isoformat(),
            "cached": False,
            "holdings_count": 0,
        }

    try:
        model = _get_model()
        prompt = _build_prompt(holdings, news)
        logger.info("[AI] Calling Gemini API, holdings=%d, news=%d", len(holdings), len(news))
        t0 = time.time()
        response = model.generate_content(prompt)
        analysis_text = response.text
        logger.info("[AI] Gemini response in %.1fs", time.time() - t0)
    except Exception as exc:
        logger.exception("[AI] Gemini API error")
        raise RuntimeError(str(exc)) from exc

    result = {
        "analysis": analysis_text,
        "generated_at": datetime.now().isoformat(),
        "cached": False,
        "holdings_count": len(holdings),
    }

    _cache["portfolio"] = {"result": result, "ts": now, "date": today_str}
    return result