feat(stock-lab): telegram.py 메시지 빌더 (Top10 + 아이콘 + 페이지 링크)

This commit is contained in:
2026-05-12 09:32:34 +09:00
parent 7db9869722
commit 01ebd2e7d9
2 changed files with 123 additions and 0 deletions

View File

@@ -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,
}

View File

@@ -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"]