73 lines
2.2 KiB
Python
73 lines
2.2 KiB
Python
"""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,
|
|
}
|