auto: 일일 백업 2026-09-11 02:00

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

agents/budget/workspace/skills/whooing-sync/scripts/securities_match.py

@@ -0,0 +1,220 @@
+"""은행 이체 SMS ↔ 증권 계좌 입출금(kt00015) 짝 맞추기.
+
+은행에서 증권으로 이체하면 은행 SMS 에는 **입금자명(=본인 이름)만** 찍혀서
+어디로 갔는지 알 수 없다. 그래서 `방효원` 같은 자체이체 룰에 걸려
+`기초잔액 ← 은행` 이 되고, 자본 분개 금지 가드에 막혀 **미반영**된다.
+반대편(증권 입금)은 SMS 가 아예 없어 월 1회 reconcile 차액으로만 나타나는데,
+그 차액은 전부 `주식평가수익` 으로 분개되므로 **이체가 손익으로 둔갑**한다.
+
+이 모듈은 그 짝을 키움 kt00015 입출금 내역에서 찾아
+`증권(효원) ← 하나은행(효원)` 이체 한 건으로 만들어 준다.
+
+⚠️ **capital-block 직전에만 호출한다** — 평시 키움 콜 0. 이체가 실제로 있었던
+   사이클에만 계좌 수만큼(4콜) 조회한다.
+⚠️ **애매하면 매칭하지 않는다** — 후보가 여럿이면 포기하고 기존 알림 경로로
+   보낸다. 틀린 분개보다 사람이 보는 편이 낫다.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import sys
+from datetime import datetime, timedelta
+from pathlib import Path
+from zoneinfo import ZoneInfo
+
+KST = ZoneInfo("Asia/Seoul")
+ROOT = Path("/Users/snowoyh/.openclaw")
+STOCK_SCRIPTS = ROOT / "agents" / "stock" / "workspace" / "scripts"
+MATCHED_FILE = ROOT / "agents" / "budget" / "workspace" / "state" / "whooing_securities_matched.json"
+
+# 증권 계좌 라벨 → 후잉 자산 계정명. 라벨 prefix 규칙은 send_balance_to_budget.py 와 동일.
+GAHEE_PREFIX = "가희"
+OWNER_ASSET = {"self": "증권(효원)", "gahee": "증권(가희)"}
+
+# 은행 SMS 종류 → 기대되는 증권 쪽 방향.
+OPPOSITE = {"withdrawal": "IN", "deposit": "OUT"}
+
+# 시각까지 일치하는 후보를 우선한다 (오픈뱅킹은 즉시 반영이라 실측 12초 차).
+# 후보가 여럿일 때만 쓰는 tie-breaker 이지 1차 필터가 아니다 —
+# 증권사 처리가 늦어 분 단위로 벌어지는 경우가 있어 창으로 잘라내면 놓친다.
+TIME_TIEBREAK_SECONDS = 600
+
+
+def _owner_of(label: str) -> str:
+    return "gahee" if label.startswith(GAHEE_PREFIX) else "self"
+
+
+def row_key(label: str, row: dict) -> str:
+    """kt00015 행에는 고유 ID 가 없다. 중복 분개 방지용 합성 키.
+    같은 날 같은 금액을 두 번 이체하면 time 으로 갈린다."""
+    return f"{row['date']}|{label}|{row['io_tp']}|{row['amount']}|{row.get('time') or ''}"
+
+
+def _load_matched() -> dict:
+    try:
+        with MATCHED_FILE.open(encoding="utf-8") as f:
+            return json.load(f)
+    except (OSError, ValueError):
+        return {}
+
+
+def mark_matched(key: str, meta: dict) -> None:
+    """분개 성공 후 호출. POST 성공한 뒤에만 기록해야 재시도 여지가 남는다."""
+    data = _load_matched()
+    data[key] = {**meta, "matched_at": datetime.now(KST).isoformat()}
+    # 오래된 항목 정리 (180일) — 무한 증식 방지.
+    cutoff = (datetime.now(KST) - timedelta(days=180)).strftime("%Y%m%d")
+    data = {k: v for k, v in data.items() if k.split("|", 1)[0] >= cutoff}
+    MATCHED_FILE.parent.mkdir(parents=True, exist_ok=True)
+    tmp = MATCHED_FILE.with_suffix(".tmp")
+    with tmp.open("w", encoding="utf-8") as f:
+        json.dump(data, f, ensure_ascii=False, indent=2)
+    os.replace(tmp, MATCHED_FILE)
+
+
+def _fetch_day(base_dt: str) -> dict[str, list[dict]] | None:
+    """전 계좌의 그날 입출금. 조회 실패는 None (= 판단 보류, 매칭 안 함)."""
+    sys.path.insert(0, str(STOCK_SCRIPTS))
+    try:
+        import kiwoom_client as kc
+        import investment_principal as ip
+    except Exception as e:  # noqa: BLE001 — import 실패해도 후잉 동기화는 계속돼야 한다
+        sys.stderr.write(f"securities_match: 키움 모듈 import 실패: {e}\n")
+        return None
+
+    out: dict[str, list[dict]] = {}
+    for acc in kc.list_accounts():
+        label = acc["label"]
+        try:
+            rows = kc.get_cash_flow(label, base_dt=base_dt)
+        except Exception as e:  # noqa: BLE001
+            sys.stderr.write(f"securities_match: {label} kt00015 조회 실패: {e}\n")
+            return None  # 일부만 보고 판단하면 오매칭 — 전부 성공해야 진행
+        out[label] = [r for r in rows if ip.is_principal_flow(r["rmrk"])]
+    return out
+
+
+def _drop_internal_transfers(day: dict[str, list[dict]]) -> dict[str, list[dict]]:
+    """증권 계좌 간 대체(예: 가희_일반 → 가희_ISA)를 후보에서 뺀다.
+    후잉에선 같은 자산이라 분개 대상이 아닌데, 금액이 같으면 은행 SMS 와
+    오매칭될 수 있다. 서로 다른 계좌에 같은 금액의 IN/OUT 이 함께 있으면 제외."""
+    ins = {(lb, r["amount"]) for lb, rows in day.items() for r in rows if r["io_tp"] == "IN"}
+    outs = {(lb, r["amount"]) for lb, rows in day.items() for r in rows if r["io_tp"] == "OUT"}
+    paired = {a for lb, a in ins} & {a for lb, a in outs}
+    if not paired:
+        return day
+    cleaned: dict[str, list[dict]] = {}
+    for lb, rows in day.items():
+        keep = []
+        for r in rows:
+            if r["amount"] in paired and any(
+                other != lb and any(
+                    o["amount"] == r["amount"] and o["io_tp"] != r["io_tp"] for o in day[other]
+                )
+                for other in day
+            ):
+                continue
+            keep.append(r)
+        cleaned[lb] = keep
+    return cleaned
+
+
+def _sms_time(created_at_utc: str | None) -> datetime | None:
+    if not created_at_utc:
+        return None
+    try:
+        return datetime.fromisoformat(created_at_utc.replace("Z", "+00:00")).astimezone(KST)
+    except ValueError:
+        return None
+
+
+def find_match(entry_date: str, amount: int, bank_kind: str,
+               created_at_utc: str | None = None) -> dict | None:
+    """은행 SMS 한 건에 대응하는 증권 입출금 행을 찾는다.
+
+    entry_date:     'YYYYMMDD' (은행 SMS 기준일)
+    amount:         원 (양수)
+    bank_kind:      'withdrawal' (은행→증권) | 'deposit' (증권→은행)
+    created_at_utc: SMS 수신 시각 ISO — 후보가 여럿일 때 tie-break 에만 쓴다
+
+    반환 (매칭 성공 시): {label, asset, row, key}
+    반환 None: 후보 없음 / 후보 여럿 / 이미 처리됨 / 조회 실패
+    """
+    want_io = OPPOSITE.get(bank_kind)
+    if want_io is None:
+        return None
+
+    day = _fetch_day(entry_date)
+    if day is None:
+        return None
+    day = _drop_internal_transfers(day)
+
+    matched = _load_matched()
+    cands = [
+        {"label": lb, "row": r, "key": row_key(lb, r)}
+        for lb, rows in day.items()
+        for r in rows
+        if r["io_tp"] == want_io and r["amount"] == amount and row_key(lb, r) not in matched
+    ]
+    if not cands:
+        return None
+    if len(cands) > 1:
+        # 시각이 가까운 후보가 정확히 하나면 그걸 고른다. 아니면 포기.
+        sms_at = _sms_time(created_at_utc)
+        near = []
+        if sms_at is not None:
+            for c in cands:
+                t = c["row"].get("time")
+                if not t:
+                    continue
+                try:
+                    rt = datetime.strptime(f"{entry_date} {t}", "%Y%m%d %H:%M:%S").replace(tzinfo=KST)
+                except ValueError:
+                    continue
+                if abs((rt - sms_at).total_seconds()) <= TIME_TIEBREAK_SECONDS:
+                    near.append(c)
+        if len(near) != 1:
+            sys.stderr.write(
+                f"securities_match: {entry_date} {amount:,}원 {want_io} 후보 {len(cands)}건 — 매칭 포기\n")
+            return None
+        cands = near
+
+    c = cands[0]
+    c["asset"] = OWNER_ASSET[_owner_of(c["label"])]
+    return c
+
+
+def build_payload(match: dict, bank_asset: str, entry_date: str, amount: int,
+                  bank_kind: str, balance: int | None = None) -> dict:
+    """후잉 이체 payload. 은행→증권이면 left=증권, 증권→은행이면 left=은행."""
+    if bank_kind == "withdrawal":
+        left, right = match["asset"], bank_asset
+    else:
+        left, right = bank_asset, match["asset"]
+    memo = f"{match['label']} {match['row']['rmrk']}"
+    if balance is not None:
+        memo += f" / 잔액 : {balance:,}원"
+    return {
+        "entry_date": entry_date,
+        "money": amount,
+        "item": "이체",
+        "left": left,
+        "right": right,
+        "memo": memo,
+    }
+
+
+if __name__ == "__main__":
+    # 진단용: python3 securities_match.py YYYYMMDD 금액 withdrawal|deposit
+    if len(sys.argv) < 4:
+        print(__doc__)
+        print("usage: securities_match.py YYYYMMDD AMOUNT withdrawal|deposit [SMS_ISO_UTC]")
+        raise SystemExit(2)
+    m = find_match(sys.argv[1], int(sys.argv[2]), sys.argv[3],
+                   sys.argv[4] if len(sys.argv) > 4 else None)
+    if m is None:
+        print("매칭 없음")
+        raise SystemExit(1)
+    print(json.dumps(m, ensure_ascii=False, indent=2))

agents/budget/workspace/skills/whooing-sync/scripts/whooing_sync.py

@@ -25,6 +25,7 @@ import imsg_cli
 import notify
 import whooing_balance
 import gahee_reminder
+import securities_match
 
 # 후잉 웹훅은 HTTP 200을 반환하면서 본문에 "fail"/"Error : ..." 를 줄 수 있어 본문 검증 필수.
 SUCCESS_BODY = "done"
@@ -753,6 +754,32 @@ def detect_pairs(candidates, accounts):
     return pairs, used
 
 
+def try_securities_transfer(parsed, cand, accounts):
+    """자본 분개로 갈 뻔한 은행 SMS 가 실은 증권 이체인지 확인.
+
+    은행 SMS 에는 입금자명(본인 이름)만 찍혀 목적지를 알 수 없지만, 반대편은
+    키움 kt00015 에 남는다. 짝을 찾으면 자본이 아니라 자산↔자산 이체로 분개한다.
+
+    반환: {payload, key, label} | None (짝 없음·애매·조회 실패)
+    """
+    if not parsed or parsed.get("kind") not in ("withdrawal", "deposit"):
+        return None
+    bank_asset = accounts.get("carrier_to_account", {}).get(cand["info"].get("carrier"))
+    if not bank_asset:
+        return None  # 은행 자산 계정을 모르면 이체를 구성할 수 없다
+    entry_date = parsed.get("entry_date")
+    amount = parsed.get("amount")
+    if not entry_date or not amount:
+        return None
+    match = securities_match.find_match(
+        entry_date, amount, parsed["kind"], cand.get("created_at"))
+    if match is None:
+        return None
+    payload = securities_match.build_payload(
+        match, bank_asset, entry_date, amount, parsed["kind"], parsed.get("balance"))
+    return {"payload": payload, "key": match["key"], "label": match["label"]}
+
+
 def post_to_whooing(webhook_url, payload, dry_run=False):
     """payload는 dict. 반환: (ok: bool, status: int, body: str).
     HTTP 2xx + body가 'done' 으로 시작해야 성공. 그 외(HTTP 200 + 'fail' 포함) 모두 실패."""
@@ -879,6 +906,7 @@ def main():
     sent_transfer = 0
     sent_structured = 0
     sent_raw = 0
+    sent_securities = 0
     sent_cancel = 0
     failed = 0
     new_failures = []
@@ -994,24 +1022,45 @@ def main():
         # 처제 대납금 상환이었는데 입금자명이 "방효원"이라 자체이체 룰에 걸렸다).
         # ⚠️ 커서는 latest_skip 으로 진행시킨다 — 재시도해도 같은 룰이 같은 결과를 내므로
         # 여기서 커서를 막으면 이 메시지 뒤의 모든 거래가 영구히 멈춘다.
+        sec_key = None
         if mode == "structured" and (
             structured.get("left") in capital_accounts
             or structured.get("right") in capital_accounts
         ):
-            capital_blocked.append({
-                "label": c["info"]["label"],
-                "created_at": c["created_at"],
-                "payload": payload,
-                "blocked_at": datetime.now(KST).isoformat(),
-            })
-            if c["created_at"] > (latest_skip or ""):
-                latest_skip = c["created_at"]
-            print(f"  ⛔ [capital-block] {c['info']['label']} {c['created_at']} | {display} — 미반영, 알림만")
-            continue
+            # 자본으로 갈 뻔한 건이 실은 증권 이체일 수 있다. 은행 SMS 는 입금자명만
+            # 주지만 반대편이 키움 kt00015 에 남으므로, 짝을 찾으면 자본이 아니라
+            # 자산↔자산 이체로 분개한다. 짝이 없으면 기존대로 미반영+알림.
+            sec = try_securities_transfer(parsed, c, accounts)
+            if sec:
+                payload = {k: str(v) for k, v in sec["payload"].items()}
+                mode = "securities"
+                sec_key = sec["key"]
+                display = (f"{sec['payload']['left']} ← {sec['payload']['right']} "
+                           f"{sec['payload']['money']:,}원 (증권 {sec['label']} 이체)")
+            else:
+                capital_blocked.append({
+                    "label": c["info"]["label"],
+                    "created_at": c["created_at"],
+                    "payload": payload,
+                    "blocked_at": datetime.now(KST).isoformat(),
+                })
+                if c["created_at"] > (latest_skip or ""):
+                    latest_skip = c["created_at"]
+                print(f"  ⛔ [capital-block] {c['info']['label']} {c['created_at']} | {display} — 미반영, 알림만")
+                continue
 
         ok, status, body = post_to_whooing(webhook_url, payload, dry_run=args.dry_run)
         if ok:
-            if mode == "structured":
+            if mode == "securities":
+                sent_securities += 1
+                # POST 성공 후에만 기록한다 — 실패하면 다음에 다시 짝으로 잡혀야 한다.
+                if not args.dry_run:
+                    securities_match.mark_matched(sec_key, {
+                        "sms_at": c["created_at"],
+                        "money": int(payload["money"]),
+                        "entry": display,
+                    })
+            elif mode == "structured":
                 sent_structured += 1
                 carrier_key = c["info"].get("carrier")
                 if carrier_key:
@@ -1353,9 +1402,9 @@ def main():
         save_json(SYNC_STATE_FILE, sync_state)
 
     if blocked_at:
-        print(f"⚠️ 후잉 동기화: transfer {sent_transfer}건, structured {sent_structured}건, raw {sent_raw}건, 취소감액 {sent_cancel}건, 실패 {failed}건 — {blocked_at} 에서 중단(다음 cron 재시도)")
+        print(f"⚠️ 후잉 동기화: transfer {sent_transfer}건, structured {sent_structured}건, 증권이체 {sent_securities}건, raw {sent_raw}건, 취소감액 {sent_cancel}건, 실패 {failed}건 — {blocked_at} 에서 중단(다음 cron 재시도)")
     else:
-        print(f"✅ 후잉 동기화: transfer {sent_transfer}건, structured {sent_structured}건, raw {sent_raw}건, 취소감액 {sent_cancel}건, 실패 {failed}건 (last={final_latest or latest})")
+        print(f"✅ 후잉 동기화: transfer {sent_transfer}건, structured {sent_structured}건, 증권이체 {sent_securities}건, raw {sent_raw}건, 취소감액 {sent_cancel}건, 실패 {failed}건 (last={final_latest or latest})")
 
     # 가희 잔액 리마인더 & 답신 자동 분개 (격리 — 실패해도 결제 sync 결과는 위에서 이미 출력됨)
     run_gahee_reminder()