gahusb
ace0339d33
- git mv stock-lab/ → stock/ - docker-compose.yml: 서비스 키 + container_name + build.context + frontend.depends_on + agent-office STOCK_LAB_URL → STOCK_URL - agent-office/app: config.py, service_proxy.py, agents/stock.py, tests/ STOCK_LAB_URL → STOCK_URL - nginx/default.conf: proxy_pass http://stock-lab → http://stock (3 lines) - CLAUDE.md / README.md / STATUS.md / scripts/ 문구 갱신 - stock/ 내부 자기 참조 갱신 lab 네이밍 정책 (feedback_lab_naming.md) graduation. API URL / Python import / DB 파일명 변경 없음.
83 lines
2.7 KiB
Python
83 lines
2.7 KiB
Python
"""Telegram payload builder. Caller (agent-office) handles actual delivery."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import datetime as dt
|
|
|
|
# 노드별 풀 라벨 (아이콘 대신 사용 — 사용자가 명확한 이름 선호)
|
|
NODE_LABELS = {
|
|
"foreign_buy": "외국인",
|
|
"volume_surge": "거래량급증",
|
|
"momentum": "20일모멘텀",
|
|
"high52w": "52주신고가",
|
|
"rs_rating": "RS레이팅",
|
|
"ma_alignment": "이평선정배열",
|
|
"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:
|
|
"""1,234,567원 형태 (None 시 '-')."""
|
|
if n is None:
|
|
return "\\-"
|
|
return f"{int(n):,}원"
|
|
|
|
|
|
def _format_active_nodes(scores: dict, threshold: int = 70) -> str:
|
|
"""70점 이상 노드를 '라벨 점수' 형태로 나열, 콤마 구분."""
|
|
active = []
|
|
for name, sc in scores.items():
|
|
label = NODE_LABELS.get(name)
|
|
if label is None or sc < threshold:
|
|
continue
|
|
active.append(f"{_escape_md(label)} {int(sc)}")
|
|
return " · ".join(active) if active else "\\(70점 이상 노드 없음\\)"
|
|
|
|
|
|
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]:
|
|
nodes_str = _format_active_nodes(r.get("scores", {}))
|
|
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" {nodes_str}\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,
|
|
}
|