hyowons
5e47a15547
- 자산정보 > 공모주 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>
233 lines
8.0 KiB
Python
233 lines
8.0 KiB
Python
#!/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())
|