104 lines
3.5 KiB
Python
104 lines
3.5 KiB
Python
"""Claude Haiku 기반 종목 뉴스 호재/악재 분석."""
|
|
|
|
from __future__ import annotations
|
|
|
|
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: 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,
|
|
) -> Dict[str, Any]:
|
|
"""Returns {ticker, score_raw, reason, news_count, tokens_input, tokens_output, model}."""
|
|
news_block = _format_news_block(news)
|
|
prompt = PROMPT_TEMPLATE.format(
|
|
name=name or ticker, ticker=ticker,
|
|
n=len(news), news_block=news_block,
|
|
)
|
|
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,
|
|
}
|