""" 텔레그램 카드 메시지 포맷터. 매수/매도, 계좌, 종목명, 가격 4개를 ▶ 마커 + *bold* (텔레그램 Markdown)로 하이라이트. 시장가일 때 호가창 + 평균체결가 + 슬리피지% 표시. PIN 메시지는 PIN 만 단독 — 메타 정보 일체 없음. 가희 계좌는 카드 헤더에 🔐 마커 추가. """ from __future__ import annotations import json from pathlib import Path from typing import Optional LIMITS_FILE = Path(__file__).resolve().parent / 'limits.json' ACCOUNT_DISPLAY = { '일반': '본인 일반', 'ISA': '본인 ISA', '가희_일반': '가희 일반', '가희_ISA': '가희 ISA', } def _limits() -> dict: return json.loads(LIMITS_FILE.read_text(encoding='utf-8')) def _is_spouse(account: str) -> bool: return account in _limits()['spouse_accounts'] def _account_display(account: str) -> str: return ACCOUNT_DISPLAY.get(account, account) def _md_bold(s: str) -> str: return f'*{s}*' def _money(n) -> str: return f'{int(n):,}원' def _pct(p: float, digits: int = 2) -> str: if abs(p) < 10 ** -digits: p = 0.0 sign = '+' if p > 0 else '' return f'{sign}{p:.{digits}f}%' def _num(p) -> str: """계단 비율 표기 — 10.0 은 '10', 10.5 는 '10.5' 로. 불필요한 .0 을 지운다.""" return f'{float(p or 0):g}' def format_card(request: dict, market_data: dict, card_id: str, estimate: Optional[dict] = None, budget_conversion: Optional[dict] = None, state_warning: Optional[str] = None, amended: bool = False) -> str: cfg = _limits()['card'] side = request['side'] side_word = '매수' if side == 'BUY' else '매도' side_emoji = cfg['buy_emoji'] if side == 'BUY' else cfg['sell_emoji'] marker = cfg['highlight_marker'] spouse_marker = ' 🔐 가희 계좌' if _is_spouse(request['account']) else '' type_marker = '' if request['order_type'] == 'MARKET': type_marker = f' {cfg["warning_emoji"]} 시장가' elif request['order_type'] == 'AGGRESSIVE_LIMIT': type_marker = f' {cfg["warning_emoji"]} 공격적 지정가' elif request['order_type'] == 'STOP_LIMIT': type_marker = f' {cfg["warning_emoji"]} 스톱지정가(예약)' elif request['order_type'] == 'TRAILING_STOP': type_marker = f' {cfg["warning_emoji"]} 트레일링 스톱(예약)' amend_marker = ' ✏️ 수정됨' if amended else '' lines = [f'{side_emoji} {_md_bold(side_word + " 미리보기")} [#{card_id}]{type_marker}{spouse_marker}{amend_marker}', ''] lines.append(f'{marker} {_md_bold(side_word)}') lines.append(f'{marker} 계좌: {_md_bold(_account_display(request["account"]))}') symbol_name = request.get('symbol_name', request['symbol']) lines.append(f'{marker} 종목: {_md_bold(symbol_name)} ({request["symbol"]})') if state_warning: lines.append(state_warning) if request['order_type'] == 'LIMIT': price_part = _md_bold(_money(request['price'])) if market_data.get('prev_close'): ratio = (request['price'] - market_data['prev_close']) / market_data['prev_close'] * 100 price_part += f' (전일종가 {_pct(ratio)})' lines.append(f'{marker} 가격: {price_part}') elif request['order_type'] == 'STOP_LIMIT': stop_p = request.get('stop_price') or 0 cur = market_data.get('current_price') cur_part = f' (현재가 {_money(cur)})' if cur else '' lines.append(f'{marker} 조건: {_md_bold(_money(stop_p) + " 도달 시")}{cur_part}') lines.append(f'{marker} 매도가격: {_md_bold(_money(request["price"]))} (지정가)') elif request['order_type'] == 'TRAILING_STOP': # 승인 대상은 "계단 정의" — 각 계단의 조건단가는 고점이 오르면 제 하락률을 유지하며 # 함께 따라 올라간다. 계단이 1개면 종전의 단일 트레일링과 같다. cur = market_data.get('current_price') cur_part = f' (현재가 {_money(cur)})' if cur else '' # 초기 고점은 항상 등록 시점 현재가 — 과거 고점 기준 옵션은 2026-07-30 제거됨. peak_v = request.get('trail_peak') or 0 steps = request.get('trail_steps') or [] lines.append(f'{marker} 시작 고점: {_md_bold(_money(peak_v))} (등록 시점){cur_part}') min_sell = request.get('min_sell_price') if min_sell: # 최저 매도가가 손절선을 결정했는지(=트레일보다 높았는지) 명시 — 둘 중 어느 게 # 지배했는지 안 보이면 관리자님이 값을 잘못 읽는다. tied = sum(1 for s in steps if s.get('floor_applied')) who = f'{tied}개 계단 적용' if tied else '미적용 (트레일이 더 높음)' lines.append(f'{marker} 최저 매도가: {_md_bold(_money(min_sell))} ({who})') lines.append('') lines.append(f'{marker} 매도 계단 {len(steps)}단') for s in steps: cum = s.get('cum', 0) cum_part = '전량' if cum >= 100 else f'누적 {_num(cum)}%' lines.append(f' {s["n"]}단계 −{_num(s["pct"])}% ' f'{_md_bold(_money(s["cond_uv"]) + " 도달")} → ' f'{s["qty"]:,}주 ({cum_part})') rest = request['qty'] - sum(s['qty'] for s in steps) if rest > 0: lines.append(f' ↳ 나머지 {rest:,}주는 스톱 없음 (계속 보유)') lines.append('') lines.append(f'{marker} 고점이 오르면 계단이 통째로 따라 올라갑니다 (내려가지 않음)') elif request['order_type'] == 'MARKET': lines.append(f'{marker} 가격: {_md_bold("시장가")}') else: if estimate and 'aggressive_price' in estimate: tail = '최우선호가+1틱' if estimate.get('source') == 'fallback': now_dt = market_data.get('now') tm = now_dt.strftime('%H:%M') if now_dt else '' tm_part = f' {tm}' if tm else '' tail = f'호가창 비어 현재가{tm_part} +1틱' lines.append(f'{marker} 가격: {_md_bold(_money(estimate["aggressive_price"]))} ({tail})') else: lines.append(f'{marker} 가격: {_md_bold("최우선호가+1틱")}') lines.append('') qty_line = f'수량: {request["qty"]:,}주' if budget_conversion: if budget_conversion.get('source') == 'fallback': now_dt = market_data.get('now') tm = now_dt.strftime('%H:%M') if now_dt else '' tm_part = f' {tm}' if tm else '' ref_label = f'호가창 비어 현재가{tm_part}' else: ref_label = '매도1호가' if side == 'BUY' else '매수1호가' rem = budget_conversion['remainder'] if budget_conversion.get('bumped'): rem_part = f'1주 추가 매수, 초과액 {_money(abs(rem))}' else: rem_part = f'잔액 {_money(rem)}' qty_line += (f' (예산 {_money(budget_conversion["budget"])} → ' f'{ref_label} {_money(budget_conversion["ref_price"])} 기준 환산, ' f'{rem_part})') lines.append(qty_line) if request['order_type'] == 'MARKET': lines.append(f'현재가: {_money(market_data["current_price"])}') ob = market_data.get('orderbook') or {} levels = (ob.get('asks' if side == 'BUY' else 'bids') or [])[:cfg['orderbook_depth']] label = '매도' if side == 'BUY' else '매수' for i, lvl in enumerate(levels, 1): lines.append(f'{label}{i}호가: {_money(lvl["price"])} (잔량 {lvl["qty"]:,}주)') if estimate: lines.append('') money_label = '예상 금액' if side == 'BUY' else '예상 회수' lines.append(f'예상 평균체결가: {_money(estimate["avg_fill"])}') lines.append(f'{money_label}: 약 {_money(estimate["total_won"])}') lines.append(f'예상 슬리피지: {_pct(estimate["slippage_pct"], 3)}') else: price = request.get('price') or (estimate and estimate.get('aggressive_price')) if price: money_label = '예상 금액' if side == 'BUY' else '예상 회수' lines.append(f'{money_label}: {_money(price * request["qty"])}') if cfg['show_balance_ratio']: if side == 'BUY' and market_data.get('balance_d2') is not None: balance = market_data['balance_d2'] need = (estimate['total_won'] if estimate and request['order_type'] == 'MARKET' else (request.get('price') or (estimate and estimate.get('aggressive_price')) or 0) * request['qty']) if balance and need: ratio = need / balance * 100 lines.append(f'매수가능금액: {_money(balance)} (잔고대비 {ratio:.1f}%)') else: lines.append(f'매수가능금액: {_money(balance)}') if side == 'SELL' and market_data.get('position_qty') is not None: lines.append(f'보유수량: {market_data["position_qty"]:,}주') lines.append(f'{_limits()["pin"]["expiry_seconds"]}초 후 만료') return '\n'.join(lines) def format_card_multi(legs: list, side: str, symbol: str, symbol_name: str, order_type: str, price: Optional[int], card_id: str, state_warning: Optional[str] = None) -> str: """복수 계좌 동시 주문 카드 — 레그별 계좌·수량·예상회수 나열. legs: [{'account': str, 'qty': int, 'position_qty': int?}, ...] """ cfg = _limits()['card'] side_word = '매수' if side == 'BUY' else '매도' side_emoji = cfg['buy_emoji'] if side == 'BUY' else cfg['sell_emoji'] marker = cfg['highlight_marker'] spouse_marker = ' 🔐 가희 계좌' if any(_is_spouse(l['account']) for l in legs) else '' type_marker = f' {cfg["warning_emoji"]} 시장가' if order_type == 'MARKET' else '' lines = [f'{side_emoji} {_md_bold(side_word + " 미리보기")} [#{card_id}]{type_marker}{spouse_marker}' f' · 계좌 {len(legs)}개 동시', ''] lines.append(f'{marker} {_md_bold(side_word)}') lines.append(f'{marker} 종목: {_md_bold(symbol_name)} ({symbol})') if state_warning: lines.append(state_warning) if order_type == 'LIMIT': lines.append(f'{marker} 가격: {_md_bold(_money(price))}') else: lines.append(f'{marker} 가격: {_md_bold("시장가")}') lines.append('') total = 0 for l in legs: part = f'{marker} {_md_bold(_account_display(l["account"]))}: {l["qty"]:,}주' if l.get('position_qty'): part += f' (보유 {l["position_qty"]:,}주)' if order_type == 'LIMIT' and price: part += f' → {_money(price * l["qty"])}' total += price * l['qty'] lines.append(part) if order_type == 'LIMIT' and price: money_label = '예상 금액 합계' if side == 'BUY' else '예상 회수 합계' lines.append(f'{money_label}: {_money(total)}') lines.append(f'{_limits()["pin"]["expiry_seconds"]}초 후 만료') return '\n'.join(lines) def format_pin_message(pin: str) -> str: return pin def format_filled(card_id: str, side: str, symbol_name: str, qty: int, fill_price: int, ord_no: str, remaining_balance: Optional[int] = None) -> str: side_word = '매수' if side == 'BUY' else '매도' out = [f'✅ [#{card_id}] 체결: {symbol_name} {qty:,}주 @ {_money(fill_price)} ({side_word})', f'주문번호: {ord_no}'] if remaining_balance is not None: out.append(f'잔여 예수금: {_money(remaining_balance)}') return '\n'.join(out) def format_submitted(card_id: str, side: str, symbol_name: str, qty: int, price: Optional[int], ord_no: str) -> str: side_word = '매수' if side == 'BUY' else '매도' price_str = _money(price) if price else '시장가' return f'📨 [#{card_id}] {side_word} 접수: {symbol_name} {qty:,}주 @ {price_str}\n주문번호: {ord_no}' def format_trailing_submitted(info: dict) -> str: """트레일링 계단 접수 결과 — 몇 단이 걸렸고 감시가 붙었는지를 한 화면에 담는다. 계단 일부만 접수될 수 있어(레그 독립 제출) 성공·실패를 나란히 적는다. 실패분을 요약해 버리면 방어선이 몇 개인지 모르는 채로 남게 된다. """ placed = info.get('placed') or [] failed = info.get('failed') or [] acct = _account_display(info.get('account', '')) head = '🧪 [모의] 트레일링' if info.get('dry_run') else '✅ 트레일링' lines = [f'{head} {len(placed)}단 등록: {info["symbol_name"]}', f'[{acct}] 총 {info["total_qty"]:,}주 · 시작 고점 {_money(info["peak"])}'] if info.get('min_sell_price'): lines.append(f'최저 매도가 {_money(info["min_sell_price"])}') for s in placed: cum = s.get('cum', 0) cum_part = '전량' if cum >= 100 else f'누적 {_num(cum)}%' no = f' · 주문 {s["ord_no"]}' if s.get('ord_no') else '' lines.append(f'{s["n"]}단계 −{_num(s["pct"])}% {_money(s["cond_uv"])} 도달 → ' f'{s["qty"]:,}주 ({cum_part}){no}') rest = info['total_qty'] - sum(s['qty'] for s in placed) if rest > 0: lines.append(f'↳ 나머지 {rest:,}주는 스톱 없음 (계속 보유)') for s in failed: lines.append(f'⚠️ {s["n"]}단계 −{_num(s["pct"])}% {s["qty"]:,}주 접수 실패 ' f'({s.get("reason")}) {s.get("error", "")}'.rstrip()) if info.get('register_error'): lines.append('⚠️ 주문은 접수됐지만 트레일링 등록 실패 — 손절선이 따라 올라가지 않습니다. ' f'({info["register_error"]})') elif info.get('res_id'): lines.append(f'감시 시작 ({info["res_id"]}) — 고점이 오르면 계단이 함께 올라갑니다') return '\n'.join(lines) def format_expired(card_id: str, side: str) -> str: side_word = '매수' if side == 'BUY' else '매도' return f'⏱️ [#{card_id}] 승인 만료. {side_word}가 취소되었습니다.' def format_canceled(card_id: str, side: str) -> str: side_word = '매수' if side == 'BUY' else '매도' return f'❌ [#{card_id}] {side_word} 취소되었습니다.' def format_pin_mismatch(card_id: str) -> str: return f'❌ [#{card_id}] PIN 불일치. 카드 무효. 다시 신호 보내주세요.' def format_partial(card_id: str, side: str, symbol_name: str, cntr_qty: int, order_qty: int, fill_price: int, ord_no: str) -> str: side_word = '매수' if side == 'BUY' else '매도' return (f'🟡 [#{card_id}] {side_word} 부분체결: {symbol_name} ' f'{cntr_qty:,}/{order_qty:,}주 @ {_money(fill_price)}\n주문번호: {ord_no}') def format_broker_post_reject(card_id: str, side: str, symbol_name: str, ord_no: str, mdfy_cncl: str) -> str: side_word = '매수' if side == 'BUY' else '매도' return (f'❌ [#{card_id}] {side_word} 사후거절({mdfy_cncl}): {symbol_name}\n주문번호: {ord_no}') def format_unfilled_timeout(card_id: str, side: str, symbol_name: str, cntr_qty: int, order_qty: int, ord_no: str) -> str: side_word = '매수' if side == 'BUY' else '매도' if cntr_qty == 0: head = f'⏱️ [#{card_id}] {side_word} 30분째 미체결: {symbol_name} 0/{order_qty:,}주' else: head = (f'⏱️ [#{card_id}] {side_word} 30분 추적 종료 (부분체결): ' f'{symbol_name} {cntr_qty:,}/{order_qty:,}주') return (f'{head}\n주문번호: {ord_no}\n' f'필요 시 키움 직접 정정/취소 또는 추가 추적 명령') def format_cancel_confirmed(side: str, symbol_name: str, orig_ord_no: str, new_ord_no: str, cancel_qty: int) -> str: side_word = '매수' if side == 'BUY' else ('매도' if side == 'SELL' else side) return (f'✅ {side_word} 취소 확인: {symbol_name} {cancel_qty:,}주 취소됨\n' f'원주문: {orig_ord_no} / 취소주문: {new_ord_no}') def format_cancel_unconfirmed_timeout(side: str, symbol_name: str, orig_ord_no: str, new_ord_no: str) -> str: side_word = '매수' if side == 'BUY' else ('매도' if side == 'SELL' else side) return (f'⏱️ {side_word} 취소 30분째 미확인: {symbol_name}\n' f'원주문: {orig_ord_no} / 취소주문: {new_ord_no}\n' f'키움에서 직접 확인 필요') def format_rejected(code: str, message: str) -> str: return f'⛔ 거부 [{code}]: {message}' def format_sidecar_blocked() -> str: return '🚫 매매 비활성화 상태. /orders_on 으로 재개하세요.' def format_card_locked(active: Optional[dict] = None) -> str: header = '⏳ 이전 카드가 아직 활성 — 처리 후 다시 신호 주세요.' if not active: return header side_word = '매수' if active.get('side') == 'BUY' else '매도' account = _account_display(active.get('account', '')) name = active.get('symbol_name') or active.get('symbol') or '' qty = active.get('qty') order_type = active.get('order_type') price = active.get('price') if order_type == 'MARKET': price_str = '시장가' elif order_type == 'AGGRESSIVE_LIMIT': price_str = f'공격적 지정가 {_money(price)}' if price else '공격적 지정가' elif order_type == 'TRAILING_STOP': steps = active.get('trail_steps') or [] pcts = '/'.join(f'−{_num(s["pct"])}%' for s in steps) price_str = f'트레일링 {len(steps)}단 {pcts} (1단계 손절 {_money(active.get("stop_price"))})' else: price_str = _money(price) if price is not None else '지정가' qty_str = f'{qty}주' if qty is not None else '' desc = ' · '.join(p for p in [account, side_word, f'{name} {qty_str}'.strip(), price_str] if p) expires_in = int(active.get('expires_in') or 0) return '\n'.join([ header, f' [#{active.get("card_id", "?")}] {desc}', f' 남은 시간: {expires_in}초 (또는 /cancel 로 즉시 폐기)', ]) def format_dryrun(payload: dict) -> str: return '🧪 DRYRUN — 실주문 안 함\n' + json.dumps(payload, ensure_ascii=False, indent=2)