diff --git a/stock-lab/app/screener/telegram.py b/stock-lab/app/screener/telegram.py new file mode 100644 index 0000000..842bb69 --- /dev/null +++ b/stock-lab/app/screener/telegram.py @@ -0,0 +1,72 @@ +"""Telegram payload builder. Caller (agent-office) handles actual delivery.""" + +from __future__ import annotations + +import datetime as dt + +NODE_ICONS = { + "foreign_buy": "👤외", + "volume_surge": "⚡거", + "momentum": "🚀모", + "high52w": "🆙고", + "rs_rating": "💪RS", + "ma_alignment": "📈MA", + "vcp_lite": "🌀VCP", +} + +PAGE_BASE = "https://gahusb.synology.me/stock/screener" + + +def _escape_md(s: str) -> str: + """Minimal MarkdownV2 escape — extend if formatting breaks.""" + for ch in r"\_*[]()~`>#+-=|{}.!": + s = s.replace(ch, "\\" + ch) + return s + + +def _format_won(n) -> str: + if n is None: + return "-" + return f"{int(n):,}" + + +def build_telegram_payload(asof: dt.date, mode: str, survivors_count: int, + top_n: int, rows: list, run_id) -> dict: + title = "*KRX 강세주 스크리너*" + header = ( + f"🎯 {title} — {_escape_md(asof.isoformat())} \\({_escape_md(mode)}\\)\n" + f"통과 {survivors_count}종 / Top {top_n} / 본문 1\\-10" + ) + + lines = [] + for r in rows[:10]: + icons = " ".join( + NODE_ICONS[name] for name, sc in r["scores"].items() + if sc >= 70 and name in NODE_ICONS + ) + score_str = f"{r['total_score']:.1f}" + r_pct = r.get("r_pct") + r_pct_str = f"{r_pct:.1f}" if r_pct is not None else "-" + lines.append( + f"{r['rank']}\\. *{_escape_md(r['name'])}* `{r['ticker']}` " + f"⭐ {_escape_md(score_str)}\n" + f" {icons}\n" + f" 진입 {_format_won(r.get('entry_price'))} " + f"손절 {_format_won(r.get('stop_price'))} " + f"익절 {_format_won(r.get('target_price'))} " + f"\\(R {_escape_md(r_pct_str)}%\\)" + ) + + # URL은 inline link로 감싸 URL 내부 . - ? = 이스케이프 회피 + link = ( + f"🔗 [전체 결과·11\\~20위]({PAGE_BASE}?run_id={run_id})" + if run_id else "" + ) + + text = header + "\n\n" + "\n\n".join(lines) + ("\n\n" + link if link else "") + + return { + "chat_target": "default", + "parse_mode": "MarkdownV2", + "text": text, + } diff --git a/stock-lab/app/test_screener_telegram.py b/stock-lab/app/test_screener_telegram.py new file mode 100644 index 0000000..606b459 --- /dev/null +++ b/stock-lab/app/test_screener_telegram.py @@ -0,0 +1,51 @@ +import datetime as dt +from app.screener.telegram import build_telegram_payload + + +def test_build_payload_includes_top10_and_link(): + rows = [ + { + "rank": i, "ticker": f"00{i:04}", "name": f"종목{i}", + "total_score": 90 - i, + "scores": {"foreign_buy": 80 + i, "volume_surge": 60, "momentum": 70, + "high52w": 75, "rs_rating": 85, "ma_alignment": 80, "vcp_lite": 30}, + "close": 50000, "entry_price": 50250, "stop_price": 48500, + "target_price": 53750, "r_pct": 3.5, + } + for i in range(1, 21) + ] + p = build_telegram_payload( + asof=dt.date(2026, 5, 12), + mode="auto", + survivors_count=612, + top_n=20, + rows=rows, + run_id=42, + ) + assert p["parse_mode"] == "MarkdownV2" + text = p["text"] + assert "2026" in text and "05" in text and "12" in text + assert "종목1" in text + assert "종목10" in text + assert "종목11" not in text # 본문 1-10만 + assert "42" in text # run_id 링크 + + +def test_score_threshold_filters_icons(): + rows = [{ + "rank": 1, "ticker": "A", "name": "A주", + "total_score": 80, + "scores": {"foreign_buy": 90, "volume_surge": 50, "momentum": 70, + "high52w": 30, "rs_rating": 80, "ma_alignment": 80, "vcp_lite": 60}, + "close": 50000, "entry_price": 50250, "stop_price": 48500, + "target_price": 53750, "r_pct": 3.5, + }] + p = build_telegram_payload(dt.date(2026, 5, 12), "auto", 100, 1, rows, run_id=1) + # foreign_buy(90), momentum(70), rs_rating(80), ma_alignment(80) 만 표시 (≥70) + assert "👤외" in p["text"] + assert "🚀모" in p["text"] + assert "💪RS" in p["text"] + assert "📈MA" in p["text"] + assert "⚡거" not in p["text"] + assert "🆙고" not in p["text"] + assert "🌀VCP" not in p["text"]