자산웹 공모주 탭 추가 + 상장일 관심종목 자동등록

- 자산정보 > 공모주 sub-tab: 오늘청약(증권사별 아이템, 기본 접힘) / 청약완료(상장일까지) - 관리자(청약)·가희(청약) 버튼 — 사람별 종목당 1건 기록 (state/behive_ipo_subscribed.json) - 타이틀 [청약] 배지: 미완료 청약 또는 청약 종목 상장일 - ipo_alert listing(08:30)에서 청약완료 종목을 상장일에 관심종목 등록 - 데이터는 네이버 IPO API, 증거금·수수료는 ipo_alert._manager_blocks 재사용(키움 콜 0) Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-02 12:32:15 +09:00
parent b404af9020
commit 5e47a15547
6 changed files with 761 additions and 13 deletions
@@ -0,0 +1,232 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from datetime import date, datetime
+from decimal import Decimal, ROUND_HALF_UP
+from pathlib import Path
+
+from ipo_calendar_sync import (
+    KST,
+    NAVER_IPO_API_URL,
+    _naver_api_rows,
+    fetch,
+    send_telegram,
+)
+
+
+STATE_FILE = Path('/Users/snowoyh/.openclaw/agents/stock/workspace/state/ipo_alerts.json')
+
+# 온라인 일반 고객 기준. 고객등급·배정 여부에 따라 면제될 수 있다.
+# 확인이 끝난 증권사만 숫자를 두고, 미확인 증권사는 알림에서 추정하지 않는다.
+BROKER_ONLINE_FEES = {
+    '대신증권': 2_000,
+    '하나증권': 2_000,
+    '미래에셋증권': 2_000,
+    '삼성증권': 2_000,
+    'KB증권': 1_500,
+    '유진투자증권': 2_000,
+    '한국투자증권': 2_000,
+    'IBK투자증권': 0,
+}
+
+
+def load_state() -> dict:
+    try:
+        return json.loads(STATE_FILE.read_text())
+    except (FileNotFoundError, json.JSONDecodeError):
+        return {'sent': {}}
+
+
+def save_state(state: dict) -> None:
+    STATE_FILE.parent.mkdir(parents=True, exist_ok=True)
+    STATE_FILE.write_text(json.dumps(state, ensure_ascii=False, indent=2))
+
+
+def _int(value) -> int | None:
+    try:
+        return int(str(value).replace(',', '').strip())
+    except (TypeError, ValueError):
+        return None
+
+
+def _money(value: int | None) -> str:
+    return '미확인' if value is None else f'{value:,}원'
+
+
+def _detail(ipo_code: str) -> dict:
+    url = f'https://stock.naver.com/api/domestic/ipo/{ipo_code}/detail/info'
+    return json.loads(fetch(url, encoding='utf-8'))
+
+
+def _date_range(start: str, end: str) -> str:
+    return start if not end or start == end else f'{start} ~ {end}'
+
+
+def _required_deposit(total: int, ratio) -> int | None:
+    try:
+        pct = Decimal(str(ratio))
+    except Exception:
+        return None
+    return int((Decimal(total) * pct / Decimal(100)).quantize(Decimal('1'), rounding=ROUND_HALF_UP))
+
+
+def _manager_blocks(row: dict, detail: dict) -> list[dict]:
+    info = detail.get('registrationStatement') or {}
+    ratio = info.get('sbscMrgnRatio') or '50'
+    price = _int((detail.get('demandForecastingResult') or {}).get('fixPubPrice'))
+    if price is None:
+        price = _int(row.get('fixPubPrice'))
+
+    managers = detail.get('joinManagers') or []
+    if not managers:
+        names = [x.strip() for x in (row.get('orgNm') or '').split(',') if x.strip()]
+        managers = [{'orgNm': name, 'sbscMinNum': None} for name in names]
+    if not managers:
+        managers = [{'orgNm': '미확인', 'sbscMinNum': None}]
+
+    blocks = []
+    for manager in managers:
+        broker = (manager.get('orgNm') or '미확인').strip()
+        min_qty = _int(manager.get('sbscMinNum'))
+        total = price * min_qty if price is not None and min_qty is not None else None
+        deposit = _required_deposit(total, ratio) if total is not None else None
+        fee = BROKER_ONLINE_FEES.get(broker)
+        blocks.append({
+            'broker': broker,
+            'min_qty': min_qty,
+            'price': price,
+            'total': total,
+            'ratio': ratio,
+            'deposit': deposit,
+            'fee': fee,
+        })
+    return blocks
+
+
+def format_alert(kind: str, row: dict, detail: dict) -> str:
+    name = (row.get('compName') or (detail.get('ipoInfo') or {}).get('compName') or '미확인').strip()
+    info = detail.get('ipoInfo') or {}
+    if kind == 'subscription':
+        heading = '[공모주 청약]'
+        date_label = '청약일'
+        date_value = _date_range(info.get('poStartDate') or row.get('poStartDate') or '', info.get('poEndDate') or row.get('poEndDate') or '')
+    else:
+        heading = '[신규 상장]'
+        date_label = '상장일'
+        date_value = info.get('lcalDate') or row.get('lcalDate') or ''
+
+    sections = []
+    for block in _manager_blocks(row, detail):
+        price = block['price']
+        qty = block['min_qty']
+        if block['deposit'] is not None and qty is not None:
+            cost = f"{block['deposit']:,}원(최소 {qty:,}주)"
+        elif price is not None:
+            cost = '최소청약수량 미확인'
+        else:
+            cost = '미확정'
+        fee = block['fee']
+        cost += f" / 수수료 {_money(fee)}(온라인 일반 기준)"
+        sections.append('\n'.join([
+            heading,
+            f'종목명 : {name}',
+            f"증권사 : {block['broker']}",
+            f'{date_label} : {date_value}',
+            f'증거금 : {cost}',
+        ]))
+    return '\n\n'.join(sections)
+
+
+def collect_for_date(kind: str, target: date) -> list[tuple[dict, dict]]:
+    payload = json.loads(fetch(NAVER_IPO_API_URL, encoding='utf-8'))
+    matches = []
+    field = 'poStartDate' if kind == 'subscription' else 'lcalDate'
+    for row in _naver_api_rows(payload):
+        if row.get(field) != target.isoformat():
+            continue
+        code = (row.get('ipoCode') or row.get('itemCode') or '').strip()
+        if not code:
+            continue
+        matches.append((row, _detail(code)))
+    return matches
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument('kind', choices=('subscription', 'listing'))
+    parser.add_argument('--date', help='YYYY-MM-DD, 기본값은 오늘(KST)')
+    parser.add_argument('--dry-run', action='store_true')
+    args = parser.parse_args()
+
+    target = date.fromisoformat(args.date) if args.date else datetime.now(KST).date()
+    try:
+        rows = collect_for_date(args.kind, target)
+    except Exception as exc:
+        print(f'IPO alert source failed: {type(exc).__name__}: {exc}', file=sys.stderr)
+        rows = None
+    if args.kind == 'listing' and not args.dry_run:
+        # 자산웹에서 청약완료한 종목을 상장일에 관심종목으로 등록. 알림 소스가 죽어도
+        # 청약완료 기록의 상장일로 시도한다(그래서 위에서 바로 return 하지 않는다).
+        listed = {(r.get('ipoCode') or r.get('itemCode') or '').strip() for r, _ in (rows or [])}
+        try:
+            from behive_web import register_listed_ipo_interests
+            added = register_listed_ipo_interests(target.isoformat(), listed)
+            if added:
+                print(f'IPO interests added: {", ".join(added)}')
+        except Exception as exc:
+            print(f'IPO interests add failed: {type(exc).__name__}: {exc}', file=sys.stderr)
+    if rows is None:
+        return 1
+
+    state = load_state()
+    sent = state.setdefault('sent', {})
+    delivered = 0
+    skipped = 0
+    for row, detail in rows:
+        code = (row.get('ipoCode') or row.get('itemCode') or '').strip()
+        key = f'{args.kind}|{target.isoformat()}|{code}'
+        if key in sent and not args.dry_run:
+            skipped += 1
+            continue
+        message = format_alert(args.kind, row, detail)
+        if args.dry_run:
+            print(message)
+            print()
+            continue
+        if not send_telegram(message):
+            print(f'IPO alert send failed: {key}', file=sys.stderr)
+            return 1
+        sent[key] = datetime.now(KST).isoformat()
+        delivered += 1
+
+    if not args.dry_run:
+        # 상태 파일이 무한히 커지지 않도록 최근 120일만 보존한다.
+        cutoff = date.fromordinal(target.toordinal() - 120).isoformat()
+        state['sent'] = {
+            key: value for key, value in sent.items()
+            if len(key.split('|')) < 2 or key.split('|')[1] >= cutoff
+        }
+        state['last_run_at'] = datetime.now(KST).isoformat()
+        state['last_kind'] = args.kind
+        state['last_target_date'] = target.isoformat()
+        state['last_found'] = len(rows)
+        state['last_delivered'] = delivered
+        state['last_skipped'] = skipped
+        save_state(state)
+
+    print(json.dumps({
+        'kind': args.kind,
+        'target_date': target.isoformat(),
+        'found': len(rows),
+        'delivered': delivered,
+        'skipped': skipped,
+        'dry_run': args.dry_run,
+    }, ensure_ascii=False))
+    return 0
+
+
+if __name__ == '__main__':
+    raise SystemExit(main())