auto: 일일 백업 2026-10-03 02:00

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-10-03 02:00:04 +09:00
parent 5e47a15547
commit 5384b1d670
135 changed files with 2190 additions and 1855 deletions
@@ -0,0 +1,128 @@
+#!/usr/bin/env python3
+"""자산웹 손익차트 과거 구간 채우기 — 키움 kt00002(일별추정예탁자산현황) → state/net_worth_history.json.
+
+portfolio_daily_snapshot 은 2026-05-01 부터만 쓸 수 있는 형식이라(4월은 예수금 없는 옛 형식) 그 이전
+차트가 비어 있다. 키움 계좌별 일별 추정예탁자산으로 그 앞 구간을 채운다. 조회 전용, 1회성 CLI.
+
+- 조회기간 한 번에 6개월 미만(`501067`) → 170일 창으로 쪼갠다. 계좌당 3~4콜.
+- kt00002 는 주말·휴장일도 행이 온다 → KOSPI 일별종가가 있는 날(sim index_history)만 남긴다.
+  market_holidays.json 은 2026년부터라 2025 휴장일을 못 거른다.
+- 값 기준이 스냅샷과 다르다(KRX 종가 − 추정 매도비용 ≈0.18%, 스냅샷은 20:10 NXT 가격) →
+  겹치는 첫날(스냅샷 첫 날짜)의 비율로 owner 별 스케일을 맞춰 경계일 단차를 없앤다.
+- 입출금은 investment_principal 캐시(kt00015, 키움 콜 0)에서 — 스냅샷 cash_in 과 같은 의미
+  (배당·이자 포함 전체 입출금).
+
+CLI: python3 net_worth_backfill.py [--from YYYY-MM-DD] [--dry-run]
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import sys
+import time
+from datetime import date, datetime, timedelta
+from pathlib import Path
+from zoneinfo import ZoneInfo
+
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+import investment_principal as ip  # noqa: E402
+import kiwoom_client as kc  # noqa: E402
+
+KST = ZoneInfo('Asia/Seoul')
+STATE = Path(__file__).resolve().parent.parent / 'state'
+SNAPSHOT_FILE = STATE / 'portfolio_daily_snapshot.json'
+INDEX_FILE = STATE / 'sim' / 'index_history.json'
+OUT_FILE = STATE / 'net_worth_history.json'
+OWNERS = {'본인': ['일반', 'ISA'], '가희': ['가희_일반', '가희_ISA']}
+WINDOW_DAYS = 170
+
+
+def _first_snapshot_date(snap: dict, owner: str) -> str | None:
+    """스냅샷에서 차트가 쓰는 형식(holdings+deposit)이 시작되는 날짜."""
+    return next((d for d in sorted(snap) if 'holdings' in ((snap.get(d) or {}).get(owner) or {})), None)
+
+
+def _fetch_account(label: str, start: date, end: date) -> dict[str, int]:
+    """{YYYY-MM-DD: 추정예탁자산}. 계좌 개설 전 구간은 '내역 없음' 에러라 빈 창으로 친다."""
+    out: dict[str, int] = {}
+    cur = start
+    while cur <= end:
+        w_end = min(cur + timedelta(days=WINDOW_DAYS), end)
+        try:
+            r = kc._call_paginated(label, 'kt00002',
+                                   {'start_dt': cur.strftime('%Y%m%d'), 'end_dt': w_end.strftime('%Y%m%d')},
+                                   'daly_prsm_dpst_aset_amt_prst')
+            for x in r.get('daly_prsm_dpst_aset_amt_prst') or []:
+                v = kc._to_int(x.get('prsm_dpst_aset_amt'))
+                if v:
+                    d = x['dt']
+                    out[f'{d[:4]}-{d[4:6]}-{d[6:]}'] = v
+        except RuntimeError as e:
+            if '506476' not in str(e):  # 506476 = 내역 없음(개설 전)
+                raise
+        time.sleep(0.4)
+        cur = w_end + timedelta(days=1)
+    return out
+
+
+def main() -> int:
+    ap = argparse.ArgumentParser()
+    ap.add_argument('--from', dest='frm', default='2025-01-01')
+    ap.add_argument('--dry-run', action='store_true')
+    a = ap.parse_args()
+
+    snap = json.loads(SNAPSHOT_FILE.read_text())
+    idx = json.loads(INDEX_FILE.read_text()).get('KOSPI') or {}
+    trading = {f'{d[:4]}-{d[4:6]}-{d[6:]}' for d in idx}
+    start = date.fromisoformat(a.frm)
+    if f'{start:%Y-%m-%d}' < min(trading):
+        print(f'⚠️ 거래일 달력(KOSPI)이 {min(trading)} 부터라 그 이전은 거를 수 없다', file=sys.stderr)
+        return 1
+
+    days: dict[str, dict] = {}
+    scale: dict[str, float] = {}
+    for owner, labels in OWNERS.items():
+        first = _first_snapshot_date(snap, owner)
+        if not first:
+            print(f'{owner}: 스냅샷 없음 — 건너뜀', file=sys.stderr)
+            continue
+        end = date.fromisoformat(first)  # 겹치는 첫날까지 받아 스케일 기준으로 쓴다
+        per_acct = {lb: _fetch_account(lb, start, end) for lb in labels}
+        # 기준일(스냅샷 첫날)은 휴장일일 수 있어(2026-05-01 근로자의 날) 거래일 필터는 저장 행에만 건다.
+        total = {d: sum(per_acct[lb].get(d, 0) for lb in labels) for d in set().union(*per_acct.values())}
+        o_snap = snap[first][owner]
+        snap_net = sum(int(h.get('eval_value', 0) or 0) for h in o_snap['holdings'].values()) + int(o_snap.get('deposit', 0) or 0)
+        if not total.get(first):
+            print(f'{owner}: 기준일 {first} kt00002 값 없음 — 중단', file=sys.stderr)
+            return 1
+        scale[owner] = snap_net / total[first]
+        flows: dict[str, dict] = {}
+        for f in ip.get_flows(labels)['rows']:
+            d = f'{f["date"][:4]}-{f["date"][4:6]}-{f["date"][6:]}'
+            k = 'cash_in' if f['io'] == 'IN' else 'cash_out'
+            flows.setdefault(d, {'cash_in': 0, 'cash_out': 0})[k] += int(f['amount'])
+        n = 0
+        for d in sorted(total):
+            if d >= first or d not in trading:
+                continue
+            rec = {'net': round(total[d] * scale[owner])}
+            rec.update(flows.get(d) or {})
+            days.setdefault(d, {})[owner] = rec
+            n += 1
+        print(f'{owner}: {n}일 ({min(days)} ~ {first} 직전), 스케일 {scale[owner]:.5f} (기준 {first} 스냅샷 {snap_net:,} / 키움 {total[first]:,})')
+
+    doc = {'source': 'kiwoom kt00002', 'fetched_at': datetime.now(KST).isoformat(timespec='seconds'),
+           'scale': scale, 'days': dict(sorted(days.items()))}
+    if a.dry_run:
+        print('dry-run — 저장 안 함')
+        return 0
+    tmp = OUT_FILE.with_suffix('.tmp')
+    tmp.write_text(json.dumps(doc, ensure_ascii=False, indent=1))
+    os.replace(tmp, OUT_FILE)
+    print(f'저장: {OUT_FILE} ({len(days)}일)')
+    return 0
+
+
+if __name__ == '__main__':
+    sys.exit(main())