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