diff --git a/stock-lab/app/screener/ai_news/analyzer.py b/stock-lab/app/screener/ai_news/analyzer.py new file mode 100644 index 0000000..bccbd52 --- /dev/null +++ b/stock-lab/app/screener/ai_news/analyzer.py @@ -0,0 +1,76 @@ +"""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": , "reason": ""}}""" + + +def _clamp(x: float, lo: float = -10.0, hi: float = 10.0) -> float: + return max(lo, min(hi, x)) + + +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 = "\n".join(f"- {n['title']}" for n in 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, + messages=[{"role": "user", "content": prompt}], + ) + text = resp.content[0].text if resp.content else "" + 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, + } diff --git a/stock-lab/tests/test_ai_news_analyzer.py b/stock-lab/tests/test_ai_news_analyzer.py new file mode 100644 index 0000000..311e8e9 --- /dev/null +++ b/stock-lab/tests/test_ai_news_analyzer.py @@ -0,0 +1,55 @@ +import json +import pytest +from unittest.mock import AsyncMock, MagicMock + +from app.screener.ai_news import analyzer + + +def _mk_llm(content_text: str, in_tokens: int = 100, out_tokens: int = 20): + llm = AsyncMock() + resp = MagicMock() + block = MagicMock() + block.text = content_text + resp.content = [block] + resp.usage = MagicMock(input_tokens=in_tokens, output_tokens=out_tokens) + llm.messages = MagicMock() + llm.messages.create = AsyncMock(return_value=resp) + return llm + + +NEWS = [{"title": "삼성전자, HBM 양산"}, {"title": "메모리 가격 반등"}] + + +@pytest.mark.asyncio +async def test_score_sentiment_success_parses_json(): + llm = _mk_llm(json.dumps({"score": 7.5, "reason": "HBM 호재"})) + out = await analyzer.score_sentiment(llm, "005930", NEWS, name="삼성전자") + assert out["ticker"] == "005930" + assert out["score_raw"] == 7.5 + assert out["reason"] == "HBM 호재" + assert out["news_count"] == 2 + assert out["tokens_input"] == 100 + assert out["tokens_output"] == 20 + + +@pytest.mark.asyncio +async def test_score_sentiment_json_parse_fail_returns_zero(): + llm = _mk_llm("not valid json") + out = await analyzer.score_sentiment(llm, "005930", NEWS) + assert out["score_raw"] == 0.0 + assert "parse fail" in out["reason"] + assert out["tokens_input"] == 100 # 호출은 발생했음 + + +@pytest.mark.asyncio +async def test_score_sentiment_clamps_out_of_range(): + llm = _mk_llm(json.dumps({"score": 15.0, "reason": "초강세"})) + out = await analyzer.score_sentiment(llm, "005930", NEWS) + assert out["score_raw"] == 10.0 # +10 클램프 + + +@pytest.mark.asyncio +async def test_score_sentiment_clamps_negative_out_of_range(): + llm = _mk_llm(json.dumps({"score": -42.0, "reason": "초악재"})) + out = await analyzer.score_sentiment(llm, "005930", NEWS) + assert out["score_raw"] == -10.0