web-page-backend/stock/app/screener/ai_news/analyzer.py

"""Claude Haiku 기반 종목 뉴스 호재/악재 분석."""

from __future__ import annotations

import datetime as dt
import json
import logging
import os
from typing import Any, Dict, List

log = logging.getLogger(__name__)

DEFAULT_MODEL = os.getenv("AI_NEWS_MODEL", "claude-haiku-4-5-20251001")

PROMPT_TEMPLATE = """다음은 종목 {name}({ticker})에 대한 최근 뉴스 {n}개의 헤드라인입니다.

{news_block}

이 뉴스들이 종목 주가에 호재인지 악재인지 종합 평가하세요.

규칙:
- score: -10(매우 강한 악재) ~ +10(매우 강한 호재) 사이의 실수. 명확한 방향성이 없으면 0(중립).
- 뉴스가 호재·악재로 섞여 있으면 주가에 더 우세한 쪽을 기준으로 부호를 정하세요.
- reason은 반드시 score 부호와 같은 방향의 근거만 쓰세요.
  · score가 양수(호재)면 호재 근거만, 음수(악재)면 악재 근거만 적습니다.
  · 호재 평가에 악재 내용을, 악재 평가에 호재 내용을 섞지 마세요.
- reason: 30자 이내 한 줄.

JSON으로만 응답하세요. 다른 텍스트 금지:
{{"score": <float>, "reason": "<string>"}}"""


def _clamp(x: float, lo: float = -10.0, hi: float = 10.0) -> float:
    return max(lo, min(hi, x))


def _format_news_block(news: List[Dict[str, Any]]) -> str:
    """news dict 리스트 → prompt 에 들어가는 텍스트 블록.

    summary 가 있으면 title 다음 줄에 indent 해서 포함 (최대 200자).
    pub_date 가 있으면 title 앞에 표시.
    """
    lines: List[str] = []
    for n in news:
        date = (n.get("pub_date") or "").strip()
        title = (n.get("title") or "").strip()
        summary = (n.get("summary") or "").strip()
        prefix = f"[{date}] " if date else ""
        if summary:
            lines.append(f"- {prefix}{title}\n  {summary[:200]}")
        else:
            lines.append(f"- {prefix}{title}")
    return "\n".join(lines)


async def score_sentiment(
    llm,
    ticker: str,
    news: List[Dict[str, Any]],
    *,
    name: str | None = None,
    model: str = DEFAULT_MODEL,
    asof: dt.date | None = None,
) -> Dict[str, Any]:
    """Returns {ticker, score_raw, reason, news_count, tokens_input, tokens_output, model}.

    asof(현재 KST 일자)를 주면 prompt 맨 앞에 오늘 날짜를 명시해 LLM이 현재 시점 기준으로 판단한다.
    """
    news_block = _format_news_block(news)
    prompt = PROMPT_TEMPLATE.format(
        name=name or ticker, ticker=ticker,
        n=len(news), news_block=news_block,
    )
    if asof is not None:
        prompt = f"오늘 날짜: {asof.isoformat()} (이 시점 기준으로 뉴스를 평가하세요)\n\n" + prompt
    resp = await llm.messages.create(
        model=model,
        max_tokens=200,
        temperature=0,
        system="너는 한국 주식 뉴스 감성 분석가다. JSON 객체 하나만 반환한다.",
        messages=[
            {"role": "user", "content": prompt},
            # Assistant prefill — 첫 토큰을 강제로 '{' 로 시작해 JSON 응답을 보장
            {"role": "assistant", "content": "{"},
        ],
    )
    raw = resp.content[0].text if resp.content else ""
    # prefill '{' 이 응답에 포함되지 않으므로 다시 붙임
    text = "{" + raw if not raw.lstrip().startswith("{") else raw
    in_tokens = int(getattr(resp.usage, "input_tokens", 0) or 0)
    out_tokens = int(getattr(resp.usage, "output_tokens", 0) or 0)

    try:
        data = json.loads(text)
        score = _clamp(float(data["score"]))
        reason = str(data["reason"])[:200]
        return {
            "ticker": ticker,
            "score_raw": score,
            "reason": reason,
            "news_count": len(news),
            "tokens_input": in_tokens,
            "tokens_output": out_tokens,
            "model": model,
        }
    except (json.JSONDecodeError, KeyError, TypeError, ValueError) as e:
        log.warning("ai_news parse fail for %s: %s (raw=%r)", ticker, e, text[:100])
        return {
            "ticker": ticker,
            "score_raw": 0.0,
            "reason": f"parse fail: {e!s}"[:200],
            "news_count": len(news),
            "tokens_input": in_tokens,
            "tokens_output": out_tokens,
            "model": model,
        }