Initial commit
This commit is contained in:
jsahn
+435
@@ -0,0 +1,435 @@
|
||||
"""Rendering: MarketPayload (+ optional LLM Report) -> Korean Markdown, and Markdown -> Telegram HTML chunks.
|
||||
|
||||
Every number in the output comes from the payload. The Report is used for prose only.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import unicodedata
|
||||
from datetime import UTC, datetime
|
||||
from urllib.parse import quote
|
||||
from zoneinfo import ZoneInfo
|
||||
|
||||
from us2kr.models import EarningsItem, EconEvent, MarketPayload, NewsItem, Quote, Report
|
||||
from us2kr.symbols import GROUP_TITLES_KO, POINT_CHANGE_GROUPS, SYMBOL_GROUPS
|
||||
|
||||
KST = ZoneInfo("Asia/Seoul")
|
||||
MAX_NEWS_ITEMS = 12
|
||||
DASH = "—"
|
||||
|
||||
_TIMING_KO = {"amc": "장마감후", "bmo": "장전", "unknown": ""}
|
||||
|
||||
|
||||
# ------------------------------------------------------------------ formatting helpers
|
||||
|
||||
|
||||
def _kst(dt: datetime) -> datetime:
|
||||
if dt.tzinfo is None:
|
||||
dt = dt.replace(tzinfo=UTC)
|
||||
return dt.astimezone(KST)
|
||||
|
||||
|
||||
def _fmt_num(v: float | None, decimals: int = 2) -> str:
|
||||
return DASH if v is None else f"{v:,.{decimals}f}"
|
||||
|
||||
|
||||
def _fmt_pct(v: float | None) -> str:
|
||||
if v is None:
|
||||
return DASH
|
||||
if abs(v) < 0.005: # avoid printing "-0.00%"
|
||||
v = 0.0
|
||||
return f"{v:+.2f}%"
|
||||
|
||||
|
||||
def _fmt_bp(change_abs: float | None) -> str:
|
||||
return DASH if change_abs is None else f"{change_abs * 100:+.1f}bp"
|
||||
|
||||
|
||||
def _fmt_money(v: float | None) -> str:
|
||||
if v is None:
|
||||
return DASH
|
||||
if abs(v) >= 1e9:
|
||||
return f"{v / 1e9:,.2f}B"
|
||||
if abs(v) >= 1e6:
|
||||
return f"{v / 1e6:,.1f}M"
|
||||
return f"{v:,.2f}"
|
||||
|
||||
|
||||
def _cell(text: str) -> str:
|
||||
"""Escape a Markdown table cell (pipes, newlines)."""
|
||||
return text.replace("|", "\\|").replace("\n", " ").strip()
|
||||
|
||||
|
||||
def _text(s: object) -> str:
|
||||
"""Normalise a prose string from the Report: strip, collapse whitespace/newlines."""
|
||||
return re.sub(r"\s*\n\s*", " ", str(s or "")).strip()
|
||||
|
||||
|
||||
def _link_text(s: str) -> str:
|
||||
return _text(s).replace("[", "(").replace("]", ")")
|
||||
|
||||
|
||||
# ------------------------------------------------------------------ Markdown sections
|
||||
|
||||
|
||||
def _header(payload: MarketPayload, report: Report | None) -> list[str]:
|
||||
lines = [
|
||||
f"# 🇺🇸→🇰🇷 미국 마감 브리핑 — {payload.krx_date.isoformat()} (미 {payload.session_date.isoformat()} 세션)"
|
||||
]
|
||||
close_kst = _kst(payload.us_close_time).strftime("%Y-%m-%d %H:%M")
|
||||
gen_kst = _kst(payload.generated_at).strftime("%Y-%m-%d %H:%M")
|
||||
lines.append(f"미국 마감: {close_kst} KST · 생성: {gen_kst} KST")
|
||||
if not payload.krx_open_today:
|
||||
lines.append("⚠️ 오늘 한국 휴장")
|
||||
if report is None:
|
||||
lines.append("🧩 축소 모드(LLM 요약 없음)")
|
||||
return lines
|
||||
|
||||
|
||||
def _bullets(items: list[str]) -> list[str]:
|
||||
return [f"- {_text(it)}" for it in items if _text(it)]
|
||||
|
||||
|
||||
def _section(title: str, body: list[str]) -> list[str]:
|
||||
if not body:
|
||||
return []
|
||||
return [f"## {title}", *body, ""]
|
||||
|
||||
|
||||
def _report_sections(report: Report) -> list[str]:
|
||||
out: list[str] = []
|
||||
if _text(report.headline):
|
||||
out += _section("총평", [f"**{_text(report.headline)}**"])
|
||||
if _text(report.market_overview):
|
||||
out += _section("시장 개요", [_text(report.market_overview)])
|
||||
out += _section("주요 요인", _bullets(report.key_drivers))
|
||||
if _text(report.rates_fx_commodities):
|
||||
out += _section("금리·환율·원자재", [_text(report.rates_fx_commodities)])
|
||||
out += _section("한국 장 시사점", _bullets(report.korea_implications))
|
||||
out += _section("오늘 볼 것", _bullets(report.watchlist_today))
|
||||
out += _section("리스크", _bullets(report.risk_flags))
|
||||
return out
|
||||
|
||||
|
||||
def _ordered_groups(payload: MarketPayload) -> list[tuple[str, str, list[Quote]]]:
|
||||
"""(group key, Korean title, quotes) in GROUP_TITLES_KO order; unknown groups appended."""
|
||||
by_group = payload.quotes_by_group()
|
||||
ordered: list[tuple[str, str, list[Quote]]] = []
|
||||
for g, title in GROUP_TITLES_KO.items():
|
||||
if g in by_group:
|
||||
ordered.append((g, title, _order_quotes(g, by_group[g])))
|
||||
for g, qs in by_group.items():
|
||||
if g not in GROUP_TITLES_KO:
|
||||
ordered.append((g, g, qs))
|
||||
return ordered
|
||||
|
||||
|
||||
def _order_quotes(group: str, quotes: list[Quote]) -> list[Quote]:
|
||||
order = list(SYMBOL_GROUPS.get(group, {}))
|
||||
rank = {t: i for i, t in enumerate(order)}
|
||||
return sorted(quotes, key=lambda q: (rank.get(q.symbol, len(rank)), q.symbol))
|
||||
|
||||
|
||||
def _quote_table(group: str, quotes: list[Quote]) -> list[str]:
|
||||
points = group in POINT_CHANGE_GROUPS
|
||||
has_after = any(q.after_hours_price is not None for q in quotes)
|
||||
header = ["종목", "종가", "변동"] + (["애프터"] if has_after else [])
|
||||
lines = ["| " + " | ".join(header) + " |", "|" + "|".join("---" for _ in header) + "|"]
|
||||
for q in quotes:
|
||||
name = _cell(q.name or q.symbol)
|
||||
if points:
|
||||
close = _fmt_num(q.close, 3)
|
||||
change = _fmt_bp(q.change_abs)
|
||||
else:
|
||||
close = _fmt_num(q.close, 2)
|
||||
change = _fmt_pct(q.change_pct)
|
||||
cells = [name, close, change]
|
||||
if has_after:
|
||||
if q.after_hours_price is None:
|
||||
cells.append(DASH)
|
||||
else:
|
||||
ah = _fmt_num(q.after_hours_price, 3 if points else 2)
|
||||
if q.after_hours_change_pct is not None:
|
||||
ah += f" ({_fmt_pct(q.after_hours_change_pct)})"
|
||||
cells.append(ah)
|
||||
lines.append("| " + " | ".join(cells) + " |")
|
||||
return lines
|
||||
|
||||
|
||||
# Normal URL punctuation and existing %XX escapes stay as-is; '(' ')' and whitespace become %28 %29 %20 so
|
||||
# they cannot terminate the Markdown ``(url)`` destination early (see _LINK_RE).
|
||||
_URL_SAFE = ":/?#[]@!$&'*+,;=%"
|
||||
|
||||
|
||||
def _link_url(url: str) -> str:
|
||||
"""Percent-encode characters that would break a Markdown / Telegram link destination."""
|
||||
return quote(url.strip(), safe=_URL_SAFE)
|
||||
|
||||
|
||||
def _news_lines(news: list[NewsItem]) -> list[str]:
|
||||
lines = []
|
||||
for n in news[:MAX_NEWS_ITEMS]:
|
||||
when = _kst(n.published).strftime("%H:%M")
|
||||
lines.append(f"- [{_link_text(n.title)}]({_link_url(n.link)}) — {_text(n.source)}, {when} KST")
|
||||
return lines
|
||||
|
||||
|
||||
def _event_line(e: EconEvent) -> str:
|
||||
star = "★ " if e.importance == "high" else ""
|
||||
when = f" {_kst(e.time).strftime('%m/%d %H:%M')} KST" if e.time else ""
|
||||
if e.kind == "released":
|
||||
return (
|
||||
f"- [발표{when}] {star}{_text(e.name)}: 실제 {e.actual or DASH} / 예상 {e.forecast or DASH}"
|
||||
f" / 이전 {e.previous or DASH}"
|
||||
)
|
||||
tail = f" (예상 {e.forecast})" if e.forecast else ""
|
||||
return f"- [예정{when}] {star}{_text(e.name)}{tail}"
|
||||
|
||||
|
||||
def _earnings_line(it: EarningsItem) -> str:
|
||||
timing = _TIMING_KO.get(it.timing, "")
|
||||
tag = f"실적·{timing}" if timing else "실적"
|
||||
parts = []
|
||||
if it.eps_actual is not None or it.eps_estimate is not None:
|
||||
eps = f"EPS {_fmt_num(it.eps_actual)}"
|
||||
if it.eps_estimate is not None:
|
||||
eps += f" (예상 {_fmt_num(it.eps_estimate)})"
|
||||
parts.append(eps)
|
||||
if it.revenue_actual is not None or it.revenue_estimate is not None:
|
||||
rev = f"매출 {_fmt_money(it.revenue_actual)}"
|
||||
if it.revenue_estimate is not None:
|
||||
rev += f" (예상 {_fmt_money(it.revenue_estimate)})"
|
||||
parts.append(rev)
|
||||
detail = " · ".join(parts) if parts else "발표 예정"
|
||||
return f"- [{tag}] {_text(it.name)}({it.symbol}) {it.report_date.isoformat()}: {detail}"
|
||||
|
||||
|
||||
def render_markdown(payload: MarketPayload, report: Report | None, model_used: str | None) -> str:
|
||||
"""Render the Korean Markdown briefing. Works with ``report=None`` (degraded: tables only)."""
|
||||
lines: list[str] = _header(payload, report)
|
||||
lines.append("")
|
||||
if report is not None:
|
||||
lines += _report_sections(report)
|
||||
for g, title, qs in _ordered_groups(payload):
|
||||
lines += _section(f"📊 {title}", _quote_table(g, qs))
|
||||
lines += _section("히스토리 노트", _bullets(payload.history_notes))
|
||||
lines += _section("주요 뉴스", _news_lines(payload.news))
|
||||
ev_lines = [_event_line(e) for e in payload.events] + [_earnings_line(e) for e in payload.earnings]
|
||||
lines += _section("경제지표·실적", ev_lines)
|
||||
lines += _section("수집 누락", _bullets(payload.data_gaps))
|
||||
lines += ["---", f"요약 모델: {model_used or '없음'}"]
|
||||
return "\n".join(lines).rstrip() + "\n"
|
||||
|
||||
|
||||
# ------------------------------------------------------------------ Telegram HTML
|
||||
|
||||
_LINK_RE = re.compile(r"\[([^\]]+)\]\(([^)\s]+)\)")
|
||||
_BOLD_RE = re.compile(r"\*\*(.+?)\*\*")
|
||||
_HEADER_RE = re.compile(r"^(#{1,6})\s+(.*)$")
|
||||
_BULLET_RE = re.compile(r"^\s*[-*]\s+")
|
||||
_HR_RE = re.compile(r"^\s*(-{3,}|\*{3,}|_{3,})\s*$")
|
||||
_SEP_CELL_RE = re.compile(r":?-{2,}:?")
|
||||
_PRE_OPEN, _PRE_CLOSE = "<pre>", "</pre>"
|
||||
|
||||
_Unit = tuple[str, list[str]] # ("text", [html_line]) | ("pre", [escaped_lines])
|
||||
|
||||
|
||||
def _esc(s: str) -> str:
|
||||
return s.replace("&", "&").replace("<", "<").replace(">", ">")
|
||||
|
||||
|
||||
def _inline(md: str) -> str:
|
||||
"""Escape HTML and convert inline Markdown (links, bold) to Telegram HTML."""
|
||||
out: list[str] = []
|
||||
pos = 0
|
||||
for m in _LINK_RE.finditer(md):
|
||||
out.append(_esc(md[pos : m.start()]))
|
||||
url = _esc(m.group(2)).replace('"', """)
|
||||
out.append(f'<a href="{url}">{_esc(m.group(1))}</a>')
|
||||
pos = m.end()
|
||||
out.append(_esc(md[pos:]))
|
||||
return _BOLD_RE.sub(r"<b>\1</b>", "".join(out))
|
||||
|
||||
|
||||
def _width(s: str) -> int:
|
||||
"""Approximate display width (CJK / emoji count as 2 columns)."""
|
||||
w = 0
|
||||
for ch in s:
|
||||
if unicodedata.combining(ch):
|
||||
continue
|
||||
if unicodedata.east_asian_width(ch) in ("W", "F") or ord(ch) >= 0x1F000:
|
||||
w += 2
|
||||
else:
|
||||
w += 1
|
||||
return w
|
||||
|
||||
|
||||
def _split_row(line: str) -> list[str]:
|
||||
inner = line.strip().removeprefix("|")
|
||||
if inner.endswith("|") and not inner.endswith("\\|"):
|
||||
inner = inner[:-1]
|
||||
return [c.strip().replace("\\|", "|") for c in re.split(r"(?<!\\)\|", inner)]
|
||||
|
||||
|
||||
def _format_table(md_rows: list[str]) -> list[str]:
|
||||
rows: list[list[str]] = []
|
||||
for r in md_rows:
|
||||
cells = _split_row(r)
|
||||
if cells and all(_SEP_CELL_RE.fullmatch(c) for c in cells if c):
|
||||
continue
|
||||
rows.append(cells)
|
||||
if not rows:
|
||||
return []
|
||||
ncol = max(len(r) for r in rows)
|
||||
rows = [r + [""] * (ncol - len(r)) for r in rows]
|
||||
widths = [max(_width(r[i]) for r in rows) for i in range(ncol)]
|
||||
out = []
|
||||
for r in rows:
|
||||
cells = []
|
||||
for i, c in enumerate(r):
|
||||
pad = " " * (widths[i] - _width(c))
|
||||
cells.append(c + pad if i == 0 else pad + c)
|
||||
out.append(_esc(" ".join(cells).rstrip()))
|
||||
return out
|
||||
|
||||
|
||||
def _to_units(lines: list[str]) -> list[_Unit]:
|
||||
units: list[_Unit] = []
|
||||
i = 0
|
||||
while i < len(lines):
|
||||
line = lines[i]
|
||||
if line.lstrip().startswith("|"):
|
||||
table = []
|
||||
while i < len(lines) and lines[i].lstrip().startswith("|"):
|
||||
table.append(lines[i])
|
||||
i += 1
|
||||
pre = _format_table(table)
|
||||
if pre:
|
||||
units.append(("pre", pre))
|
||||
continue
|
||||
i += 1
|
||||
if _HR_RE.match(line):
|
||||
continue
|
||||
if not line.strip():
|
||||
if units and units[-1] == ("text", [""]):
|
||||
continue # collapse blank runs
|
||||
units.append(("text", [""]))
|
||||
continue
|
||||
m = _HEADER_RE.match(line)
|
||||
if m:
|
||||
units.append(("text", [f"<b>{_inline(m.group(2).strip())}</b>"]))
|
||||
elif _BULLET_RE.match(line):
|
||||
units.append(("text", ["• " + _inline(_BULLET_RE.sub("", line, count=1))]))
|
||||
else:
|
||||
units.append(("text", [_inline(line)]))
|
||||
while units and units[0] == ("text", [""]):
|
||||
units.pop(0)
|
||||
while units and units[-1] == ("text", [""]):
|
||||
units.pop()
|
||||
return units
|
||||
|
||||
|
||||
def _split_sections(markdown: str) -> list[list[str]]:
|
||||
sections: list[list[str]] = []
|
||||
cur: list[str] = []
|
||||
for line in markdown.splitlines():
|
||||
if line.startswith("## ") and cur:
|
||||
sections.append(cur)
|
||||
cur = []
|
||||
cur.append(line)
|
||||
if cur:
|
||||
sections.append(cur)
|
||||
return sections
|
||||
|
||||
|
||||
def _units_html(units: list[_Unit]) -> str:
|
||||
parts: list[str] = []
|
||||
for kind, lines in units:
|
||||
if kind == "pre":
|
||||
parts.append("\n".join([_PRE_OPEN, *lines, _PRE_CLOSE]))
|
||||
else:
|
||||
parts.extend(lines)
|
||||
return "\n".join(parts)
|
||||
|
||||
|
||||
def _joined_len(lines: list[str]) -> int:
|
||||
return sum(len(x) for x in lines) + max(len(lines) - 1, 0)
|
||||
|
||||
|
||||
def _hard_split(line: str, limit: int) -> list[str]:
|
||||
return [line[i : i + limit] for i in range(0, len(line), limit)] or [""]
|
||||
|
||||
|
||||
def _pack_units(units: list[_Unit], limit: int) -> list[str]:
|
||||
"""Pack one oversized section line by line; <pre> blocks are closed/reopened across chunks."""
|
||||
chunks: list[str] = []
|
||||
buf: list[str] = []
|
||||
|
||||
def flush() -> None:
|
||||
while buf and not buf[-1]:
|
||||
buf.pop()
|
||||
if buf:
|
||||
chunks.append("\n".join(buf))
|
||||
buf.clear()
|
||||
|
||||
for kind, lines in units:
|
||||
if kind == "text":
|
||||
for line in lines:
|
||||
if not line and not buf:
|
||||
continue
|
||||
if _joined_len([*buf, line]) > limit:
|
||||
flush()
|
||||
if len(line) > limit:
|
||||
for piece in _hard_split(line, limit):
|
||||
flush()
|
||||
buf.append(piece)
|
||||
continue
|
||||
buf.append(line)
|
||||
continue
|
||||
pre: list[str] = []
|
||||
for line in lines:
|
||||
if _joined_len([*buf, _PRE_OPEN, *pre, line, _PRE_CLOSE]) <= limit:
|
||||
pre.append(line)
|
||||
continue
|
||||
if pre:
|
||||
buf.extend([_PRE_OPEN, *pre, _PRE_CLOSE])
|
||||
pre = []
|
||||
flush()
|
||||
pre.append(line) # may exceed limit only if a single line is longer than limit
|
||||
if pre:
|
||||
buf.extend([_PRE_OPEN, *pre, _PRE_CLOSE])
|
||||
flush()
|
||||
return chunks
|
||||
|
||||
|
||||
def render_telegram_chunks(markdown: str, limit: int = 4000) -> list[str]:
|
||||
"""Convert the briefing Markdown to Telegram HTML and split it into chunks of at most ``limit`` chars.
|
||||
|
||||
Sections (``## `` headers) are packed greedily; an oversized section is split on lines, and
|
||||
``<pre>`` blocks are closed and reopened so every chunk is balanced HTML.
|
||||
"""
|
||||
chunks: list[str] = []
|
||||
cur = ""
|
||||
for section in _split_sections(markdown):
|
||||
units = _to_units(section)
|
||||
if not units:
|
||||
continue
|
||||
html = _units_html(units)
|
||||
candidate = html if not cur else f"{cur}\n\n{html}"
|
||||
if len(candidate) <= limit:
|
||||
cur = candidate
|
||||
continue
|
||||
if cur:
|
||||
chunks.append(cur)
|
||||
cur = ""
|
||||
if len(html) <= limit:
|
||||
cur = html
|
||||
continue
|
||||
parts = _pack_units(units, limit)
|
||||
if parts:
|
||||
chunks.extend(parts[:-1])
|
||||
cur = parts[-1]
|
||||
if cur:
|
||||
chunks.append(cur)
|
||||
return chunks
|
||||
Reference in New Issue
Block a user