hyowons
343 lines
16 KiB
Python
343 lines
16 KiB
Python
"""
|
||
텔레그램 카드 메시지 포맷터.
|
||
|
||
매수/매도, 계좌, 종목명, 가격 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 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':
|
||
# 승인 대상은 "트레일 폭" — 조건단가는 고점이 오르면 이 폭을 유지하며 따라 올라간다.
|
||
stop_p = request.get('stop_price') or 0
|
||
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
|
||
trail_pct_part = f'고점 대비 −{request.get("trail_pct")}%'
|
||
lines.append(f'{marker} 시작 고점: {_md_bold(_money(peak_v))} (등록 시점)')
|
||
lines.append(f'{marker} 트레일 폭: {_md_bold(trail_pct_part)}')
|
||
min_sell = request.get('min_sell_price')
|
||
if min_sell:
|
||
# 최저 매도가가 손절선을 결정했는지(=트레일보다 높았는지) 명시 — 둘 중 어느 게
|
||
# 지배했는지 안 보이면 관리자님이 값을 잘못 읽는다.
|
||
raw_trail = int(peak_v * (1 - float(request['trail_pct']) / 100)) if peak_v else 0
|
||
who = '최저가 적용' if stop_p >= raw_trail else '트레일 적용'
|
||
lines.append(f'{marker} 최저 매도가: {_md_bold(_money(min_sell))} ({who})')
|
||
lines.append(f'{marker} 현재 손절선: {_md_bold(_money(stop_p) + " 도달 시")}{cur_part}')
|
||
lines.append(f'{marker} 매도가격: {_md_bold(_money(request["price"]))} (지정가)')
|
||
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_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':
|
||
price_str = f'트레일링 −{active.get("trail_pct")}% (손절 {_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)
|