hyowons
450 lines
18 KiB
Python
450 lines
18 KiB
Python
#!/usr/bin/env python3
|
||
"""투자원금(순입금) 집계 — 조회 전용.
|
||
|
||
투자원금 = 계좌 개설 이후 누적 입금 − 누적 출금. "내 돈 얼마 넣었나"이며,
|
||
`총 매입금액`(현재 보유종목 매입원가)이나 `예수금`(남은 현금)과 전혀 다른 값이다.
|
||
이 값이 있어야 `순자산 − 투자원금 = 진짜 총수익`(실현+미실현+배당)이 나온다.
|
||
키움 kt00018의 평가손익은 현재 보유분 미실현만 잡아 누적 실현손익·배당을 놓친다.
|
||
|
||
데이터원은 kt00016(일별계좌수익률상세현황)의 기간내총입금·총출금.
|
||
⚠️ kt00016은 **조회기간 1년 제한**이라 연도별 창으로 쪼개 누적한다.
|
||
|
||
캐시 `state/investment_principal.json`:
|
||
- frozen : 지난 연도 = 확정값. 한 번 적재 후 재조회 안 함 (0인 연도는 저장 생략)
|
||
- current: 올해 창(0101~오늘) 누적. 갱신 대상은 이것뿐 → 계좌당 1콜
|
||
- seen : 마지막 갱신 시점의 "당일 입출금" 스냅샷 — 이벤트 감지 기준
|
||
|
||
갱신은 별도 트리거 없이 behive_web 렌더 경로에서 이벤트로만 일어난다.
|
||
당일 입출금(이미 kt00015로 조회 중)이 seen과 다를 때만 올해 창을 재조회하므로
|
||
입출금 없는 평시에는 kt00016 호출이 0이다. 안전망으로 7일 초과 시 강제 갱신.
|
||
|
||
CLI: python3 scripts/investment_principal.py {bootstrap|refresh|show}
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import sys
|
||
import time
|
||
from contextlib import contextmanager
|
||
from datetime import datetime, timedelta, timezone
|
||
from pathlib import Path
|
||
|
||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||
|
||
import kiwoom_client as kc
|
||
|
||
KST = timezone(timedelta(hours=9))
|
||
WORKSPACE = Path('/Users/snowoyh/.openclaw/agents/stock/workspace')
|
||
STATE_FILE = WORKSPACE / 'state' / 'investment_principal.json'
|
||
|
||
# 부트스트랩 하한 연도. 계좌 개설 전 기간은 kt00016이 에러 없이 전 필드 0을 주므로
|
||
# 넉넉히 잡아도 안전하다 (현재 최고(最古) 계좌는 2018년 개설).
|
||
FLOOR_YEAR = 2010
|
||
PACE_SEC = 0.35 # 연속 조회 rate limit 페이싱
|
||
STALE_DAYS = 7 # 이벤트 감지가 못 잡은 경우의 안전망
|
||
|
||
# 적요에 이 중 하나가 들어가면 '내가 넣은/뺀 돈'(원금 반영), 아니면 투자 성과(원금 무관).
|
||
# 원금 반영 : 이체입금(지급결제)·이체오픈뱅킹입금·이체입금·대체입금·소액이체인증입금
|
||
# ·ISA가입인증입금 / 이체출금·대체출금
|
||
# 원금 무관 : 배당금입금·수익분배금입금·예탁금이용료(이자)입금·공모주환불금입금
|
||
# ·쿠폰현금지급(국내)입금·단주매각대금입금·대여수수료입금
|
||
# / 공모불입출금·청약수수료출금
|
||
# ⚠️ 추측이 아니라 실증된 규칙 — 이 분류의 합이 kt00016 기간내총입금/총출금과
|
||
# 4계좌 18개 (계좌,연도) 구간 전부에서 오차 0으로 일치했다(2026-07-29).
|
||
# 새 적요가 등장하면 reconciled=False로 드러나므로 조용히 틀리지 않는다.
|
||
_PRINCIPAL_RMRK_KEYS = ('이체', '대체', '인증')
|
||
|
||
|
||
def is_principal_flow(rmrk: str) -> bool:
|
||
"""적요가 투자원금에 반영되는 외부 입출금인지. 배당·이자 등은 False."""
|
||
return any(k in (rmrk or '') for k in _PRINCIPAL_RMRK_KEYS)
|
||
|
||
|
||
# ---- 캐시 I/O ----
|
||
|
||
@contextmanager
|
||
def _lock():
|
||
"""STATE_FILE 직렬화 — behive_web _stock_notes_lock 과 동일 패턴, 별도 lock 파일."""
|
||
import fcntl
|
||
lock_path = STATE_FILE.with_suffix(STATE_FILE.suffix + '.lock')
|
||
lock_path.parent.mkdir(parents=True, exist_ok=True)
|
||
f = open(lock_path, 'a')
|
||
try:
|
||
fcntl.flock(f.fileno(), fcntl.LOCK_EX)
|
||
yield
|
||
finally:
|
||
try:
|
||
fcntl.flock(f.fileno(), fcntl.LOCK_UN)
|
||
finally:
|
||
f.close()
|
||
|
||
|
||
def _load() -> dict | None:
|
||
"""캐시 로드. 파일 없거나 깨졌으면 None (호출측이 '미집계'로 처리)."""
|
||
if not STATE_FILE.exists():
|
||
return None
|
||
try:
|
||
d = json.loads(STATE_FILE.read_text())
|
||
except Exception:
|
||
return None
|
||
if not isinstance(d, dict) or 'current' not in d:
|
||
return None
|
||
d.setdefault('frozen', {})
|
||
d.setdefault('seen', {})
|
||
d['current'].setdefault('labels', {})
|
||
return d
|
||
|
||
|
||
def _save(d: dict) -> None:
|
||
d['refreshed_at'] = datetime.now(KST).isoformat(timespec='seconds')
|
||
STATE_FILE.parent.mkdir(parents=True, exist_ok=True)
|
||
tmp = STATE_FILE.with_suffix('.json.tmp')
|
||
tmp.write_text(json.dumps(d, ensure_ascii=False, indent=2))
|
||
os.replace(tmp, STATE_FILE)
|
||
|
||
|
||
# ---- 조회 ----
|
||
|
||
def _today_str() -> str:
|
||
return datetime.now(KST).strftime('%Y%m%d')
|
||
|
||
|
||
def _this_year() -> int:
|
||
return datetime.now(KST).year
|
||
|
||
|
||
def _window(year: int) -> tuple[str, str]:
|
||
"""연도별 조회 창. 올해는 오늘까지, 지난 연도는 12/31까지 (1년 제한 내)."""
|
||
if year == _this_year():
|
||
return f'{year}0101', _today_str()
|
||
return f'{year}0101', f'{year}1231'
|
||
|
||
|
||
def _fetch_year(label: str, year: int) -> dict:
|
||
"""한 (계좌, 연도) 창의 합계(kt00016) + 입출금 행(kt00015). 2콜.
|
||
|
||
flows 행의 `pr`은 원금 반영 여부. 분류 합이 kt00016 합계와 맞는지 `rec`에 남겨
|
||
새 적요가 등장해 분류가 어긋나면 화면에서 드러나게 한다.
|
||
"""
|
||
fr, to = _window(year)
|
||
r = kc.get_period_return(label, fr, to)
|
||
time.sleep(PACE_SEC)
|
||
try:
|
||
raw = kc.get_cash_flow(label, strt_dt=fr, end_dt=to)
|
||
except Exception as e:
|
||
sys.stderr.write(f'cash flow {label} {year} 실패: {e}\n')
|
||
raw = None
|
||
flows = None if raw is None else [
|
||
{'d': f['date'], 'io': f['io_tp'], 'amt': f['amount'],
|
||
'rk': f['rmrk'], 'pr': is_principal_flow(f['rmrk'])}
|
||
for f in raw
|
||
]
|
||
rec = None
|
||
if flows is not None:
|
||
pin = sum(f['amt'] for f in flows if f['io'] == 'IN' and f['pr'])
|
||
pout = sum(f['amt'] for f in flows if f['io'] == 'OUT' and f['pr'])
|
||
rec = (pin == r['cash_in'] and pout == r['cash_out'])
|
||
return {
|
||
'in': r['cash_in'], 'out': r['cash_out'],
|
||
'qin': r['qty_in'], 'qout': r['qty_out'],
|
||
'nf': r['net_fr'], 'nt': r['net_to'], 'pl': r['eval_pl'],
|
||
'flows': flows, 'rec': rec,
|
||
}
|
||
|
||
|
||
def _is_empty(y: dict) -> bool:
|
||
"""계좌 개설 전(또는 전면 무활동) 연도 — 저장 생략 대상."""
|
||
return not any(y[k] for k in ('in', 'out', 'qin', 'qout', 'nf', 'nt', 'pl'))
|
||
|
||
|
||
def _labels(labels: list[str] | None = None) -> list[str]:
|
||
all_lb = [a['label'] for a in kc.list_accounts()]
|
||
if labels is None:
|
||
return all_lb
|
||
wanted = set(labels)
|
||
return [lb for lb in all_lb if lb in wanted]
|
||
|
||
|
||
def _flow_totals(cash_flow_by_label: dict) -> dict:
|
||
"""behive_web의 kt00015 당일 입출금 → {label: {'in': n, 'out': n}}."""
|
||
out: dict[str, dict] = {}
|
||
for lb, flows in (cash_flow_by_label or {}).items():
|
||
ci = sum(f['amount'] for f in (flows or []) if f.get('io_tp') == 'IN')
|
||
co = sum(f['amount'] for f in (flows or []) if f.get('io_tp') == 'OUT')
|
||
out[lb] = {'in': ci, 'out': co}
|
||
return out
|
||
|
||
|
||
# ---- 공개 API ----
|
||
|
||
def bootstrap(labels: list[str] | None = None) -> dict:
|
||
"""FLOOR_YEAR ~ 올해 전수 조회로 캐시를 새로 만든다 (계좌×연도, 4×17≈68콜 ≈ 25초).
|
||
|
||
조기 종료 없음 — 잔고가 0으로 비었던 휴면 연도가 중간에 끼어도 그 앞 연도를 놓치지 않는다.
|
||
⚠️ 1회성 CLI 전용. 렌더 경로에서 절대 호출하지 말 것.
|
||
"""
|
||
lbs = _labels(labels)
|
||
this_year = _this_year()
|
||
frozen: dict[str, dict] = {}
|
||
current: dict[str, dict] = {}
|
||
for lb in lbs:
|
||
frozen[lb] = {}
|
||
for year in range(FLOOR_YEAR, this_year + 1):
|
||
y = _fetch_year(lb, year)
|
||
time.sleep(PACE_SEC)
|
||
if year == this_year:
|
||
current[lb] = y
|
||
elif not _is_empty(y):
|
||
frozen[lb][year_key(year)] = y
|
||
with _lock():
|
||
d = _load() or {}
|
||
d['frozen'] = {**(d.get('frozen') or {}), **frozen}
|
||
cur = d.get('current') or {}
|
||
if cur.get('year') != this_year:
|
||
cur = {'year': this_year, 'labels': {}}
|
||
cur['labels'] = {**(cur.get('labels') or {}), **current}
|
||
d['current'] = cur
|
||
d['seen'] = {**(d.get('seen') or {}), **_seen_now(lbs)}
|
||
_save(d)
|
||
return _load()
|
||
|
||
|
||
def year_key(year: int) -> str:
|
||
return str(year)
|
||
|
||
|
||
def _seen_now(labels: list[str]) -> dict:
|
||
"""현재 당일 입출금을 seen 형식으로. (CLI refresh/bootstrap 용 — kt00015 계좌당 1콜)"""
|
||
today = datetime.now(KST).strftime('%Y-%m-%d')
|
||
totals = _flow_totals(kc.get_cash_flow_all(None, labels))
|
||
return {lb: {'date': today, **totals.get(lb, {'in': 0, 'out': 0})} for lb in labels}
|
||
|
||
|
||
def refresh_current(labels: list[str], cash_flow_by_label: dict | None = None) -> None:
|
||
"""올해 창만 재조회 (라벨당 1콜) + 연도 롤오버 시 작년을 확정값으로 이관.
|
||
|
||
cash_flow_by_label: 호출측이 이미 들고 있는 kt00015 당일 입출금.
|
||
None이면 직접 조회. seen 기록에 쓴다 — kt00016 호출 **전** 시점의 값이어야
|
||
그 사이 들어온 입금이 seen에 앞서 반영돼 영구 미갱신되는 일이 없다.
|
||
"""
|
||
lbs = _labels(labels)
|
||
if not lbs:
|
||
return
|
||
if cash_flow_by_label is None:
|
||
seen = _seen_now(lbs)
|
||
else:
|
||
today = datetime.now(KST).strftime('%Y-%m-%d')
|
||
totals = _flow_totals(cash_flow_by_label)
|
||
seen = {lb: {'date': today, **totals.get(lb, {'in': 0, 'out': 0})} for lb in lbs}
|
||
|
||
this_year = _this_year()
|
||
with _lock():
|
||
d = _load()
|
||
if d is None:
|
||
return # 부트스트랩 전 — 렌더 경로를 68콜로 막지 않는다
|
||
cur = d['current']
|
||
prev_year = cur.get('year')
|
||
# 연도 롤오버: current에 담긴 작년치는 to_dt가 12/31이 아니라 미완결이므로
|
||
# 전체 창(0101~1231)으로 다시 받아 확정 이관한다.
|
||
if prev_year and prev_year != this_year:
|
||
for lb in lbs:
|
||
y = _fetch_year(lb, prev_year)
|
||
time.sleep(PACE_SEC)
|
||
if not _is_empty(y):
|
||
d['frozen'].setdefault(lb, {})[year_key(prev_year)] = y
|
||
cur = {'year': this_year, 'labels': {}}
|
||
for lb in lbs:
|
||
cur['labels'][lb] = _fetch_year(lb, this_year)
|
||
time.sleep(PACE_SEC)
|
||
cur['year'] = this_year
|
||
d['current'] = cur
|
||
d['seen'].update(seen)
|
||
_save(d)
|
||
|
||
|
||
def needs_refresh(labels: list[str], cash_flow_by_label: dict) -> bool:
|
||
"""올해 창 재조회가 필요한지 판정 — 순수 캐시 읽기(키움 콜 0).
|
||
|
||
캐시가 없으면 False — 부트스트랩 미실행 상태에서 페이지 로드를 68콜로 막지 않는다.
|
||
"""
|
||
d = _load()
|
||
if d is None:
|
||
return False
|
||
if d['current'].get('year') != _this_year():
|
||
return True
|
||
today = datetime.now(KST).strftime('%Y-%m-%d')
|
||
totals = _flow_totals(cash_flow_by_label)
|
||
for lb in labels:
|
||
if lb not in totals:
|
||
continue # owner 부분 갱신 — 조회 안 된 계좌는 판단 보류
|
||
s = d['seen'].get(lb) or {}
|
||
base = s if s.get('date') == today else {'in': 0, 'out': 0}
|
||
if totals[lb]['in'] != base.get('in', 0) or totals[lb]['out'] != base.get('out', 0):
|
||
return True
|
||
ts = d.get('refreshed_at')
|
||
if not ts:
|
||
return True
|
||
try:
|
||
age = datetime.now(KST) - datetime.fromisoformat(ts)
|
||
except Exception:
|
||
return True
|
||
return age > timedelta(days=STALE_DAYS)
|
||
|
||
|
||
def get_principal(labels: list[str]) -> dict:
|
||
"""캐시만 읽어 누적 원금 산출 (키움 콜 0).
|
||
|
||
반환 principal이 None이면 '미집계' — 요청 라벨 중 부트스트랩 안 된 계좌가 있다는 뜻.
|
||
(0으로 떨어뜨리면 그 계좌 입금이 조용히 빠져 원금이 과소계상된다.)
|
||
qty_flow=True는 주식 입출고(타사대체) 이력 — 현금 기준 원금이 과소계상됨을 알림.
|
||
by_label은 계좌별 순입금 {label: 원} — 자세히보기 토스트용.
|
||
"""
|
||
miss = {'principal': None, 'cash_in': None, 'cash_out': None,
|
||
'qty_flow': False, 'by_label': {}}
|
||
d = _load()
|
||
if d is None:
|
||
return miss
|
||
cur = d['current']['labels']
|
||
total_in = total_out = 0
|
||
qty_flow = False
|
||
by_label: dict[str, int] = {}
|
||
for lb in labels:
|
||
if lb not in cur:
|
||
return miss
|
||
recs = list((d['frozen'].get(lb) or {}).values()) + [cur[lb]]
|
||
lb_in = lb_out = 0
|
||
for y in recs:
|
||
lb_in += y.get('in', 0)
|
||
lb_out += y.get('out', 0)
|
||
qty_flow = qty_flow or bool(y.get('qin') or y.get('qout'))
|
||
by_label[lb] = lb_in - lb_out
|
||
total_in += lb_in
|
||
total_out += lb_out
|
||
return {
|
||
'principal': total_in - total_out,
|
||
'cash_in': total_in,
|
||
'cash_out': total_out,
|
||
'qty_flow': qty_flow,
|
||
'by_label': by_label,
|
||
}
|
||
|
||
|
||
def get_flows(labels: list[str]) -> dict:
|
||
"""입출금 내역 (최신순) — 캐시만 읽는다 (키움 콜 0).
|
||
|
||
반환:
|
||
- rows: [{date, io('IN'|'OUT'), amount, rmrk, principal(bool), label}] 최신순
|
||
- other_in / other_out: 원금 무관 입출금 합계 (배당·이자 등)
|
||
- reconciled: 전 구간에서 적요 분류 합이 kt00016 합계와 일치했는지
|
||
False면 새 적요가 등장해 분류가 어긋났다는 신호
|
||
- available: 내역이 캐시에 있는지 (부트스트랩 전 or 조회 실패 시 False)
|
||
"""
|
||
d = _load()
|
||
if d is None:
|
||
return {'rows': [], 'other_in': 0, 'other_out': 0,
|
||
'reconciled': True, 'available': False}
|
||
cur = d['current']['labels']
|
||
rows: list[dict] = []
|
||
reconciled = True
|
||
available = False
|
||
for lb in labels:
|
||
recs = list((d['frozen'].get(lb) or {}).values())
|
||
if lb in cur:
|
||
recs.append(cur[lb])
|
||
for y in recs:
|
||
flows = y.get('flows')
|
||
if flows is None:
|
||
continue
|
||
available = True
|
||
if y.get('rec') is False:
|
||
reconciled = False
|
||
for f in flows:
|
||
rows.append({'date': f['d'], 'io': f['io'], 'amount': f['amt'],
|
||
'rmrk': f['rk'], 'principal': f['pr'], 'label': lb})
|
||
rows.sort(key=lambda r: (r['date'], r['label']), reverse=True)
|
||
return {
|
||
'rows': rows,
|
||
'other_in': sum(r['amount'] for r in rows if r['io'] == 'IN' and not r['principal']),
|
||
'other_out': sum(r['amount'] for r in rows if r['io'] == 'OUT' and not r['principal']),
|
||
'reconciled': reconciled,
|
||
'available': available,
|
||
}
|
||
|
||
|
||
# ---- CLI ----
|
||
|
||
def _show() -> int:
|
||
d = _load()
|
||
if d is None:
|
||
print(f'캐시 없음: {STATE_FILE}\n → python3 scripts/investment_principal.py bootstrap')
|
||
return 1
|
||
from stock_portfolio_report import derive_owner
|
||
|
||
print(f'갱신: {d.get("refreshed_at")} (올해 창 {d["current"].get("year")})\n')
|
||
lbs = list(d['current']['labels'].keys())
|
||
bad = 0
|
||
for lb in lbs:
|
||
years = sorted((d['frozen'].get(lb) or {}).items())
|
||
years.append((year_key(d['current']['year']), d['current']['labels'][lb]))
|
||
print(f'===== {lb} =====')
|
||
tin = tout = 0
|
||
for yk, y in years:
|
||
# 항등식: 순자산초 + 입금 − 출금 + 평가손익 = 순자산말
|
||
err = y['nf'] + y['in'] - y['out'] + y['pl'] - y['nt']
|
||
bad += 1 if err else 0
|
||
tin += y['in']
|
||
tout += y['out']
|
||
q = f' 입고 {y["qin"]:,} 출고 {y["qout"]:,}' if (y['qin'] or y['qout']) else ''
|
||
nf = len(y.get('flows') or []) if y.get('flows') is not None else None
|
||
rc = {True: '분류OK', False: '분류불일치!', None: '내역없음'}[y.get('rec')]
|
||
print(f' {yk}: 입금 {y["in"]:>12,} 출금 {y["out"]:>11,} '
|
||
f'순자산 {y["nf"]:>12,}→{y["nt"]:>12,} 손익 {y["pl"]:>12,} '
|
||
f'항등식오차 {err:,} 내역 {nf if nf is not None else "-":>3}행 {rc}{q}')
|
||
print(f' >> 누적 입금 {tin:,} − 출금 {tout:,} = 순입금 {tin - tout:,}\n')
|
||
|
||
print('항등식 검증:', '전 행 오차 0 ✅' if not bad else f'⚠️ 오차 있는 행 {bad}개')
|
||
fl = get_flows(lbs)
|
||
print(f'적요 분류 검증: {"전 구간 일치 ✅" if fl["reconciled"] else "⚠️ 불일치 구간 있음 — 새 적요 확인 필요"}'
|
||
f' (내역 {len(fl["rows"])}행, 원금무관 입금 {fl["other_in"]:,}원 / 출금 {fl["other_out"]:,}원)')
|
||
|
||
# 라이브 교차검증 — 원금 대비 총수익 vs 실제 순자산
|
||
print('\n===== 원금 대비 총수익 (라이브 교차검증) =====')
|
||
owners: dict[str, list[str]] = {}
|
||
for lb in lbs:
|
||
owners.setdefault(derive_owner(lb), []).append(lb)
|
||
for owner, olbs in owners.items():
|
||
pr = get_principal(olbs)
|
||
net = 0
|
||
for lb in olbs:
|
||
net += sum(p['evlt_amt'] for p in kc.get_positions(lb)) + kc.get_balance(lb)['d2_entra']
|
||
p = pr['principal']
|
||
ret = net - p
|
||
warn = ' ⚠️ 주식 입출고 있음(원금 과소계상)' if pr['qty_flow'] else ''
|
||
print(f'[{owner}] 투자원금 {p:,}원 (입금 {pr["cash_in"]:,} − 출금 {pr["cash_out"]:,}){warn}')
|
||
print(f' 순자산 {net:,}원 → 총수익 {ret:+,}원 ({ret / p * 100:+.2f}%)' if p else ' 원금 0')
|
||
return 0 if not bad else 1
|
||
|
||
|
||
def main(argv: list[str]) -> int:
|
||
cmd = argv[1] if len(argv) > 1 else 'show'
|
||
if cmd == 'bootstrap':
|
||
lbs = _labels()
|
||
n = len(lbs) * (_this_year() - FLOOR_YEAR + 1)
|
||
print(f'{len(lbs)}계좌 × {_this_year() - FLOOR_YEAR + 1}년 = {n}콜 (약 {n * PACE_SEC:.0f}초)...')
|
||
bootstrap()
|
||
print(f'완료 → {STATE_FILE}\n')
|
||
return _show()
|
||
if cmd == 'refresh':
|
||
refresh_current(_labels())
|
||
print(f'올해 창 갱신 완료 → {STATE_FILE}\n')
|
||
return _show()
|
||
if cmd == 'show':
|
||
return _show()
|
||
print(__doc__)
|
||
return 2
|
||
|
||
|
||
if __name__ == '__main__':
|
||
raise SystemExit(main(sys.argv))
|