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 파일명 변경 없음.
74 lines
2.1 KiB
Python
74 lines
2.1 KiB
Python
"""ai_news Top 5/5 텔레그램 메시지 빌더 (MarkdownV2)."""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Any, Dict, List
|
||
|
||
|
||
_MD_SPECIAL = r"_*[]()~`>#+-=|{}.!\\"
|
||
|
||
|
||
def _escape(text: str) -> str:
|
||
return "".join("\\" + c if c in _MD_SPECIAL else c for c in str(text))
|
||
|
||
|
||
def _cost_won(tokens_input: int, tokens_output: int) -> int:
|
||
"""Claude Haiku 가격 환산 (대략): in $1/M × ₩1300, out $5/M × ₩1300."""
|
||
return int(tokens_input * 0.0013 + tokens_output * 0.0065)
|
||
|
||
|
||
def _row_line(idx: int, r: Dict[str, Any]) -> str:
|
||
score = r["score_raw"]
|
||
# score 문자열 자체를 _escape 통과 — '+', '-', '.' 모두 MarkdownV2 reserved
|
||
score_str = _escape(f"{score:+.1f}")
|
||
name = r.get("name") or ""
|
||
ticker = r["ticker"]
|
||
label = (
|
||
f"{_escape(name)} \\({_escape(ticker)}\\)"
|
||
if name else _escape(ticker)
|
||
)
|
||
return f"{idx}\\. {label} \\({score_str}\\) — {_escape(r['reason'])}"
|
||
|
||
|
||
def build_message(
|
||
*,
|
||
asof: str,
|
||
top_pos: List[Dict[str, Any]],
|
||
top_neg: List[Dict[str, Any]],
|
||
tokens_input: int,
|
||
tokens_output: int,
|
||
mapping: Dict[str, int] | None = None,
|
||
) -> str:
|
||
lines: List[str] = [
|
||
f"🌅 *AI 뉴스 분석* \\({_escape(asof)} 08:00\\)",
|
||
"",
|
||
"📈 *호재 Top 5*",
|
||
]
|
||
if top_pos:
|
||
for i, r in enumerate(top_pos, 1):
|
||
lines.append(_row_line(i, r))
|
||
else:
|
||
lines.append(_escape("- (없음)"))
|
||
|
||
lines += ["", "📉 *악재 Top 5*"]
|
||
if top_neg:
|
||
for i, r in enumerate(top_neg, 1):
|
||
lines.append(_row_line(i, r))
|
||
else:
|
||
lines.append(_escape("- (없음)"))
|
||
|
||
cost = _cost_won(tokens_input, tokens_output)
|
||
mapping_part = ""
|
||
if mapping:
|
||
mapping_part = (
|
||
f"매핑 {mapping['hit_tickers']}/100 ticker "
|
||
f"\\({mapping['matched_pairs']}쌍 / articles {mapping['total_articles']}건\\) · "
|
||
)
|
||
lines += [
|
||
"",
|
||
f"_분석: 시총 상위 100종목 · {mapping_part}"
|
||
f"토큰 {tokens_input:,} in / {tokens_output:,} out · "
|
||
f"약 ₩{cost:,}_",
|
||
]
|
||
return "\n".join(lines)
|