"""보유종목 인텔리전스 텔레그램 포매터 (advisory).""" import logging from typing import Any, Dict from ..telegram.messaging import send_raw logger = logging.getLogger("agent-office") _ACTION_KR = {"add": "🟢 추가매수", "hold": "⚪ 보유", "trim": "🟡 축소", "sell": "🔴 매도"} _SEV = {"high": "🔴", "med": "🟠", "low": "🟡"} def format_holdings_brief(payload: Dict[str, Any]) -> str: date = payload.get("date") or "?" lines = [f"📊 보유종목 인텔리전스 ({date})", ""] ph = payload.get("portfolio_health") or {} if ph: lines.append(f"포트 손익 {ph.get('total_pnl_rate',0):+.1f}% · " f"종목 {ph.get('positions',0)} · 최대비중 {ph.get('max_weight',0)*100:.0f}% · " f"현금 {ph.get('cash_ratio',0)*100:.0f}%") lines.append("") for h in payload.get("holdings", []): act = _ACTION_KR.get(h.get("action"), h.get("action", "?")) pnl = h.get("pnl_rate") pnl_txt = f"{pnl:+.1f}%" if pnl is not None else "—" line = f"{act} {h.get('name') or h.get('ticker')} ({pnl_txt})" if h.get("reasons"): line += f" — {h['reasons']}" lines.append(line) for iss in (h.get("issues") or [])[:3]: lines.append(f" {_SEV.get(iss.get('severity'),'•')} {iss.get('summary','')}") lines.append("") lines.append("ℹ️ 투자 판단 보조용 제안입니다(자동매매 아님).") return "\n".join(lines) async def send_holdings_brief(payload: Dict[str, Any]) -> None: text = format_holdings_brief(payload) try: await send_raw(text) except Exception as e: logger.warning(f"[telegram_stock] holdings brief send failed: {e}")