feat(screener): ai_news telegram message builder (MarkdownV2 + cost line)

stock-lab/app/screener/ai_news/telegram.py

@@ -0,0 +1,61 @@
+"""ai_news Top 5/5 텔레그램 메시지 빌더 (MarkdownV2)."""
+
+from __future__ import annotations
+
+from typing import Any, Dict, List
+
+
+_MD_SPECIAL = r"_*[]()~`>#+-=|{}.!\\"
+
+
+def _escape(text: str) -> str:
+    return "".join("\\" + c if c in _MD_SPECIAL else c for c in str(text))
+
+
+def _cost_won(tokens_input: int, tokens_output: int) -> int:
+    """Claude Haiku 가격 환산 (대략): in $1/M × ₩1300, out $5/M × ₩1300."""
+    return int(tokens_input * 0.0013 + tokens_output * 0.0065)
+
+
+def _row_line(idx: int, r: Dict[str, Any]) -> str:
+    score = r["score_raw"]
+    sign = "+" if score >= 0 else ""
+    return (
+        f"{idx}\\. {_escape(r['ticker'])} \\({sign}{score:.1f}\\) — "
+        f"{_escape(r['reason'])}"
+    )
+
+
+def build_message(
+    *,
+    asof: str,
+    top_pos: List[Dict[str, Any]],
+    top_neg: List[Dict[str, Any]],
+    tokens_input: int,
+    tokens_output: int,
+) -> str:
+    lines: List[str] = [
+        f"🌅 *AI 뉴스 분석* \\({_escape(asof)} 08:00\\)",
+        "",
+        "📈 *호재 Top 5*",
+    ]
+    if top_pos:
+        for i, r in enumerate(top_pos, 1):
+            lines.append(_row_line(i, r))
+    else:
+        lines.append(_escape("- (없음)"))
+
+    lines += ["", "📉 *악재 Top 5*"]
+    if top_neg:
+        for i, r in enumerate(top_neg, 1):
+            lines.append(_row_line(i, r))
+    else:
+        lines.append(_escape("- (없음)"))
+
+    cost = _cost_won(tokens_input, tokens_output)
+    lines += [
+        "",
+        f"_분석: 시총 상위 100종목 · 토큰 {tokens_input:,} in / {tokens_output:,} out · "
+        f"약 ₩{cost:,}_",
+    ]
+    return "\n".join(lines)

stock-lab/tests/test_ai_news_telegram.py

@@ -0,0 +1,54 @@
+from app.screener.ai_news import telegram as tg
+
+
+def _row(ticker, score, reason="r"):
+    return {"ticker": ticker, "score_raw": score, "reason": reason,
+            "news_count": 5, "tokens_input": 100, "tokens_output": 20,
+            "model": "m"}
+
+
+def test_build_message_includes_top_sections():
+    msg = tg.build_message(
+        asof="2026-05-13",
+        top_pos=[_row("005930", 8.5, "HBM 호재")],
+        top_neg=[_row("373220", -6.3, "수주 지연")],
+        tokens_input=10000, tokens_output=2000,
+    )
+    assert "AI 뉴스 분석" in msg
+    assert "호재 Top" in msg
+    assert "악재 Top" in msg
+    assert "005930" in msg
+    assert "8.5" in msg
+    assert "HBM" in msg
+    assert "373220" in msg
+
+
+def test_build_message_escapes_markdownv2_specials():
+    msg = tg.build_message(
+        asof="2026-05-13",
+        top_pos=[_row("005930", 3.0, "테스트(괄호) [대괄호]")],
+        top_neg=[],
+        tokens_input=100, tokens_output=20,
+    )
+    # MarkdownV2 특수문자 ( ) [ ] 이 escape 되어야 함
+    assert r"\(" in msg or r"\)" in msg
+    assert r"\[" in msg or r"\]" in msg
+
+
+def test_build_message_cost_won_line():
+    msg = tg.build_message(
+        asof="2026-05-13", top_pos=[], top_neg=[],
+        tokens_input=10000, tokens_output=2000,
+    )
+    # tokens_input × 0.0013 + tokens_output × 0.0065 = 13 + 13 = ₩26
+    assert "₩26" in msg or "₩ 26" in msg or "₩" in msg
+
+
+def test_build_message_empty_lists():
+    msg = tg.build_message(
+        asof="2026-05-13", top_pos=[], top_neg=[],
+        tokens_input=0, tokens_output=0,
+    )
+    # 빈 리스트라도 헤더는 있어야 함
+    assert "호재 Top" in msg
+    assert "악재 Top" in msg