hyowons
129 lines
5.9 KiB
Python
129 lines
5.9 KiB
Python
#!/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())
|