openclaw/agents/stock/workspace/scripts/send_balance_to_budget.py

#!/usr/bin/env python3
"""매월 1일 04:30 — 본인+가희 계좌 잔액을 골디 inbox에 envelope로 떨어뜨린다.

소유자별(본인 self / 가희 gahee, label prefix '가희_'로 판별)로 평가액(kt00018)·예수금(kt00001)·
총자산을 집계해 payload.by_owner에 담는다. 골디는 self→증권(효원), gahee→증권(가희)로 reconcile.
LLM 불필요 — launchd로 실행. 실패 시 레이 텔레그램으로 자가 알림.
"""
from __future__ import annotations

import json
import sys
import traceback
import urllib.parse
import urllib.request
import uuid
from datetime import datetime
from pathlib import Path
from zoneinfo import ZoneInfo

KST = ZoneInfo('Asia/Seoul')
WORKSPACE = Path('/Users/snowoyh/.openclaw/agents/stock/workspace')
sys.path.insert(0, str(WORKSPACE / 'scripts'))

import kiwoom_client as kw  # noqa: E402

INBOX_DIR = Path('/Users/snowoyh/.openclaw/agents/budget/inbox/incoming')
CONFIG_PATH = Path('/Users/snowoyh/.openclaw/openclaw.json')
TELEGRAM_ACCOUNT = 'stock'
TOPIC = 'securities_balance'
SCHEMA_VERSION = 2
GAHEE_PREFIX = '가희_'


def send_telegram(text: str) -> bool:
    try:
        cfg = json.loads(CONFIG_PATH.read_text())
        acct = cfg['channels']['telegram']['accounts'][TELEGRAM_ACCOUNT]
        token = acct['botToken']
        chat_ids = acct.get('allowFrom') or []
    except Exception as e:
        print(f'telegram cfg load failed: {e}', file=sys.stderr)
        return False
    if not chat_ids:
        return False
    url = f'https://api.telegram.org/bot{token}/sendMessage'
    ok = True
    for chat_id in chat_ids:
        data = urllib.parse.urlencode({
            'chat_id': chat_id,
            'text': text[:4000],
            'disable_web_page_preview': 'true',
        }).encode()
        try:
            req = urllib.request.Request(url, data=data, method='POST')
            with urllib.request.urlopen(req, timeout=15) as r:
                if r.status != 200:
                    ok = False
        except Exception as e:
            print(f'telegram send failed: {e}', file=sys.stderr)
            ok = False
    return ok


def collect_accounts() -> list[dict]:
    accounts = kw.list_accounts()
    out = []
    for a in accounts:
        label = a['label']
        owner = 'gahee' if label.startswith(GAHEE_PREFIX) else 'self'
        balance = kw.get_balance(label)
        positions = kw.get_positions(label)
        eval_amount = sum(p.get('evlt_amt', 0) for p in positions)
        # d2_entra(D+2 정산예수금) — 미결제 매수/매도까지 반영한 실제 가용 예수금.
        # entr은 D+2 결제 전까지 변하지 않아 미결제 매도대금 누락 가능.
        deposit = balance.get('d2_entra', 0)
        out.append({
            'label': label,
            'owner': owner,
            'account_no': a.get('account_no', ''),
            'deposit': deposit,
            'eval_amount': eval_amount,
            'total': deposit + eval_amount,
            'position_count': len(positions),
        })
    return out


def build_message(accounts: list[dict]) -> dict:
    now = datetime.now(KST)
    by_owner: dict[str, dict] = {}
    for a in accounts:
        o = by_owner.setdefault(a['owner'], {'deposit': 0, 'eval_amount': 0, 'total': 0})
        o['deposit'] += a['deposit']
        o['eval_amount'] += a['eval_amount']
        o['total'] += a['total']
    return {
        'message_id': str(uuid.uuid4()),
        'from': 'stock',
        'to': 'budget',
        'topic': TOPIC,
        'created_at': now.isoformat(),
        'schema_version': SCHEMA_VERSION,
        'payload': {
            'as_of': now.strftime('%Y-%m-%d'),
            'owner_scope': 'self_and_gahee',
            'accounts': accounts,
            'by_owner': by_owner,
            'totals': {
                'deposit': sum(a['deposit'] for a in accounts),
                'eval_amount': sum(a['eval_amount'] for a in accounts),
                'total': sum(a['total'] for a in accounts),
            },
        },
    }


def write_message(msg: dict) -> Path:
    INBOX_DIR.mkdir(parents=True, exist_ok=True)
    iso_compact = msg['created_at'].replace(':', '').replace('-', '').split('+')[0]
    filename = f"stock__{TOPIC}__{iso_compact}.json"
    path = INBOX_DIR / filename
    path.write_text(json.dumps(msg, ensure_ascii=False, indent=2))
    return path


def main() -> int:
    try:
        accounts = collect_accounts()
        if not accounts:
            raise RuntimeError('계좌 0개 — 자격증명 점검 필요')
        msg = build_message(accounts)
        path = write_message(msg)
        total = msg['payload']['totals']['total']
        print(f'sent: {path.name} | accounts={len(accounts)} | total={total:,}원')
        return 0
    except Exception as e:
        err = f'⚠️ [send_balance_to_budget] 실패\n{type(e).__name__}: {e}\n\n{traceback.format_exc()[-500:]}'
        print(err, file=sys.stderr)
        send_telegram(err)
        return 1


if __name__ == '__main__':
    sys.exit(main())