hyowons
fed3526b20
설정·스크립트·스킬·문서·큐레이션 메모리 추적. 시크릿(credentials/identity)·런타임 상태(state/logs/sessions/sqlite)· 백업(clobbered/bak)·dream 캐시는 .gitignore로 제외. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
297 lines
11 KiB
Python
Executable File
297 lines
11 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""종목별 매매 기록 — ka10170 당일매매일지를 일자별로 적재·조회.
|
||
|
||
키움 REST에는 기간 거래내역 API가 없어 매일 ka10170을 호출해 누적한다.
|
||
- 저장: state/trade_journal.jsonl (한 줄 = (date, account, code) 레코드)
|
||
- 재실행 시 같은 (date, account) 행을 제거 후 재적재 → idempotent
|
||
- 휴장일/주말은 기본 skip (--force로 강제)
|
||
|
||
CLI:
|
||
collect [--date YYYYMMDD] [--force] [--quiet]
|
||
show <code|name>
|
||
query [--from YYYY-MM-DD] [--to YYYY-MM-DD] [--account L] [--code C]
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import sys
|
||
from datetime import datetime, timezone, timedelta
|
||
from pathlib import Path
|
||
|
||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||
import kiwoom_client as kw
|
||
|
||
KST = timezone(timedelta(hours=9))
|
||
WORKSPACE = Path('/Users/snowoyh/.openclaw/agents/stock/workspace')
|
||
JOURNAL = WORKSPACE / 'state' / 'trade_journal.jsonl'
|
||
HOLIDAYS = WORKSPACE / 'state' / 'market_holidays.json'
|
||
|
||
|
||
def is_market_open(d: datetime) -> bool:
|
||
if d.weekday() >= 5:
|
||
return False
|
||
iso = d.strftime('%Y-%m-%d')
|
||
try:
|
||
data = json.loads(HOLIDAYS.read_text())
|
||
return iso not in data.get('holidays', {})
|
||
except FileNotFoundError:
|
||
return True
|
||
|
||
|
||
def _owner(label: str) -> str:
|
||
return '가희' if label.startswith('가희_') else '본인'
|
||
|
||
|
||
def _load_all() -> list[dict]:
|
||
if not JOURNAL.exists():
|
||
return []
|
||
out: list[dict] = []
|
||
for ln in JOURNAL.read_text().splitlines():
|
||
ln = ln.strip()
|
||
if not ln:
|
||
continue
|
||
try:
|
||
out.append(json.loads(ln))
|
||
except json.JSONDecodeError:
|
||
continue
|
||
return out
|
||
|
||
|
||
def _save_all(rows: list[dict]) -> None:
|
||
tmp = JOURNAL.with_suffix('.jsonl.tmp')
|
||
if rows:
|
||
tmp.write_text('\n'.join(json.dumps(r, ensure_ascii=False) for r in rows) + '\n')
|
||
else:
|
||
tmp.write_text('')
|
||
tmp.replace(JOURNAL)
|
||
|
||
|
||
def record_trades(all_trades: dict, *, base_dt: str | None = None, quiet: bool = False, tag: str = 'collect') -> dict:
|
||
"""이미 받은 ka10170 응답({label: [trades]})을 jsonl에 적재 (idempotent).
|
||
`(iso_date, account in all_trades)` 단위 재적재 — 다른 계좌는 영향 X.
|
||
자산웹 페이지 RENDER hook, fill_watcher 등 ka10170 추가 호출 회피 경로에서 사용."""
|
||
now = datetime.now(KST)
|
||
if base_dt is None:
|
||
# 휴장일/주말엔 ka10170이 직전 영업일 데이터를 반환하므로 오늘 날짜로 적재하면 중복.
|
||
# base_dt 명시 호출(collect --date / --force)은 의도적이라 가드 해제.
|
||
if not is_market_open(now):
|
||
if not quiet:
|
||
print(f'[{tag}] {now.strftime("%Y-%m-%d")} 휴장일/주말 — record_trades skip')
|
||
return {'skipped': True, 'date': now.strftime('%Y-%m-%d')}
|
||
base_dt = now.strftime('%Y%m%d')
|
||
iso_date = f'{base_dt[:4]}-{base_dt[4:6]}-{base_dt[6:]}'
|
||
collected_at = now.isoformat()
|
||
|
||
existing = _load_all()
|
||
keep = [
|
||
r for r in existing
|
||
if not (r.get('date') == iso_date and r.get('account') in all_trades)
|
||
]
|
||
new_rows: list[dict] = []
|
||
for label, trades in all_trades.items():
|
||
for t in trades:
|
||
new_rows.append({
|
||
'date': iso_date,
|
||
'account': label,
|
||
'owner': _owner(label),
|
||
**t,
|
||
'collected_at': collected_at,
|
||
})
|
||
_save_all(keep + new_rows)
|
||
counts = {label: len(trades) for label, trades in all_trades.items()}
|
||
if not quiet:
|
||
total = sum(counts.values())
|
||
breakdown = ' '.join(f'{k}={v}' for k, v in counts.items())
|
||
print(f'[{tag}] {iso_date} total={total} {breakdown}')
|
||
return {'date': iso_date, 'counts': counts}
|
||
|
||
|
||
def collect(base_dt: str | None = None, *, force: bool = False, quiet: bool = False) -> dict:
|
||
"""ka10170 4계좌 → jsonl append (idempotent)."""
|
||
now = datetime.now(KST)
|
||
if base_dt is None:
|
||
base_dt = now.strftime('%Y%m%d')
|
||
iso_date = f'{base_dt[:4]}-{base_dt[4:6]}-{base_dt[6:]}'
|
||
|
||
target = datetime.strptime(base_dt, '%Y%m%d').replace(tzinfo=KST)
|
||
if not force and not is_market_open(target):
|
||
if not quiet:
|
||
print(f'[skip] {iso_date} 휴장일/주말 — 수집 안 함 (--force 로 강제)')
|
||
return {'skipped': True, 'date': iso_date}
|
||
|
||
all_trades = kw.get_trade_journal_all(base_dt=base_dt)
|
||
return record_trades(all_trades, base_dt=base_dt, quiet=quiet, tag='collect')
|
||
|
||
|
||
def seed_initial(*, seed_date: str | None = None, quiet: bool = False) -> dict:
|
||
"""현재 4계좌 보유종목 → trade_journal 시드 행 적재.
|
||
|
||
키움이 기간 거래내역 API를 제공하지 않아, jsonl 적재 시작일(2026-05-13) 이전의
|
||
매수 이력은 "지금 평단가 × 보유수량"으로 단일 행 압축한다.
|
||
|
||
시드 수량 = (현재 보유) - (seed_date 이후 jsonl 매수 합) + (seed_date 이후 jsonl 매도 합)
|
||
- (오늘 ka10170 매수) + (오늘 ka10170 매도)
|
||
시드 단가 = avg_price (현재 평단가 — 가중평균이라 과거 단가와 다를 수 있음)
|
||
|
||
seed=true 플래그로 ka10170 일자 행과 구분.
|
||
같은 (date, account, code, seed=true) 행이 이미 있으면 skip → idempotent.
|
||
|
||
seed_date 인자로 적재 날짜 지정 가능 (기본: 오늘 KST). 실거래 행 이전 날짜로
|
||
박으면 모달에서 시각적으로 명확히 분리됨. 이 경우 seed_date 이후 jsonl 거래 효과도
|
||
되돌려 시드 시점 실제 보유수량을 추정한다.
|
||
"""
|
||
now = datetime.now(KST)
|
||
if seed_date is None:
|
||
seed_date = now.strftime('%Y-%m-%d')
|
||
collected_at = now.isoformat()
|
||
|
||
positions = kw.get_positions_all() # {label: [positions]}
|
||
|
||
existing = _load_all()
|
||
seen = {
|
||
(r['date'], r['account'], r['code'])
|
||
for r in existing if r.get('seed') is True
|
||
}
|
||
# seed_date 초과, 오늘 미만의 비-시드 행 효과를 (account, code)별로 집계
|
||
# (오늘 거래는 ka10170 tdy_buyq/tdy_sellq 로 처리 → 중복 방지)
|
||
today_str = now.strftime('%Y-%m-%d')
|
||
post_seed: dict[tuple[str, str], dict[str, int]] = {}
|
||
for r in existing:
|
||
if r.get('seed') is True:
|
||
continue
|
||
if r['date'] <= seed_date or r['date'] >= today_str:
|
||
continue
|
||
key = (r['account'], r['code'])
|
||
agg = post_seed.setdefault(key, {'buy': 0, 'sell': 0})
|
||
agg['buy'] += int(r.get('buy_qty') or 0)
|
||
agg['sell'] += int(r.get('sell_qty') or 0)
|
||
|
||
new_rows: list[dict] = []
|
||
skipped_zero = 0
|
||
skipped_dup = 0
|
||
for label, items in positions.items():
|
||
for p in items:
|
||
agg = post_seed.get((label, p['code']), {'buy': 0, 'sell': 0})
|
||
seed_qty = (
|
||
p['qty']
|
||
- p['tdy_buyq'] + p['tdy_sellq']
|
||
- agg['buy'] + agg['sell']
|
||
)
|
||
if seed_qty <= 0:
|
||
skipped_zero += 1
|
||
continue
|
||
key = (seed_date, label, p['code'])
|
||
if key in seen:
|
||
skipped_dup += 1
|
||
continue
|
||
new_rows.append({
|
||
'date': seed_date,
|
||
'account': label,
|
||
'owner': _owner(label),
|
||
'code': p['code'],
|
||
'name': p['name'],
|
||
'buy_qty': seed_qty,
|
||
'buy_avg': p['avg_price'],
|
||
'buy_amt': seed_qty * p['avg_price'],
|
||
'sell_qty': 0,
|
||
'sell_avg': 0,
|
||
'sell_amt': 0,
|
||
'pl_amt': 0,
|
||
'cmsn_tax': 0,
|
||
'prft_rt': 0.0,
|
||
'seed': True,
|
||
'collected_at': collected_at,
|
||
})
|
||
|
||
if new_rows:
|
||
_save_all(existing + new_rows)
|
||
if not quiet:
|
||
per_label = {l: 0 for l in positions}
|
||
for r in new_rows:
|
||
per_label[r['account']] += 1
|
||
breakdown = ' '.join(f'{k}={v}' for k, v in per_label.items())
|
||
print(f'[seed] {seed_date} added={len(new_rows)} {breakdown}'
|
||
+ (f' (skip_zero={skipped_zero} skip_dup={skipped_dup})' if (skipped_zero or skipped_dup) else ''))
|
||
return {'added': len(new_rows), 'skipped_zero': skipped_zero, 'skipped_dup': skipped_dup}
|
||
|
||
|
||
def _print_rows(rows: list[dict]) -> None:
|
||
print(f'{"date":<12}{"account":<14}{"code":<10}{"name":<16}{"buy":>16}{"sell":>16}{"pl":>12}')
|
||
total_pl = 0
|
||
for r in rows:
|
||
marker = '*' if r.get('seed') else ' '
|
||
buy = f'{r["buy_qty"]}@{r["buy_avg"]:,}' if r['buy_qty'] else '-'
|
||
sell = f'{r["sell_qty"]}@{r["sell_avg"]:,}' if r['sell_qty'] else '-'
|
||
print(f'{r["date"]:<12}{r["account"]:<14}{r["code"]:<10}{marker}{r["name"]:<15}{buy:>16}{sell:>16}{r["pl_amt"]:>12,}')
|
||
total_pl += r['pl_amt']
|
||
print(f'\n실현손익 합계: {total_pl:,}원 ({len(rows)}건, *=시드)')
|
||
|
||
|
||
def show(code_or_name: str) -> None:
|
||
rows = _load_all()
|
||
q = code_or_name.strip()
|
||
matched = [r for r in rows if r.get('code') == q or r.get('name') == q]
|
||
if not matched:
|
||
matched = [r for r in rows if q in (r.get('name') or '')]
|
||
if not matched:
|
||
print(f'[show] 일치 기록 없음: {q}')
|
||
return
|
||
matched.sort(key=lambda r: (r['date'], r['account']))
|
||
print(f'[show] {q} — {len(matched)}건')
|
||
_print_rows(matched)
|
||
|
||
|
||
def query(*, date_from: str | None, date_to: str | None, account: str | None, code: str | None) -> None:
|
||
rows = _load_all()
|
||
if date_from:
|
||
rows = [r for r in rows if r['date'] >= date_from]
|
||
if date_to:
|
||
rows = [r for r in rows if r['date'] <= date_to]
|
||
if account:
|
||
rows = [r for r in rows if r['account'] == account]
|
||
if code:
|
||
rows = [r for r in rows if r['code'] == code or r['name'] == code]
|
||
rows.sort(key=lambda r: (r['date'], r['account'], r['code']))
|
||
if not rows:
|
||
print('[query] 결과 없음')
|
||
return
|
||
_print_rows(rows)
|
||
|
||
|
||
def main(argv: list[str] | None = None) -> int:
|
||
p = argparse.ArgumentParser(description='종목별 매매기록 (ka10170 적재·조회)')
|
||
sub = p.add_subparsers(dest='cmd', required=True)
|
||
|
||
pc = sub.add_parser('collect', help='ka10170 4계좌 수집 → jsonl 적재')
|
||
pc.add_argument('--date', help='YYYYMMDD (기본: 오늘 KST)')
|
||
pc.add_argument('--force', action='store_true', help='휴장일/주말 강제 수집')
|
||
pc.add_argument('--quiet', action='store_true')
|
||
|
||
psd = sub.add_parser('seed', help='4계좌 보유종목 → 현재 평단가로 시드 적재 (최초 1회)')
|
||
psd.add_argument('--date', help='YYYY-MM-DD (기본: 오늘 KST)')
|
||
|
||
ps = sub.add_parser('show', help='종목별 거래내역')
|
||
ps.add_argument('code_or_name')
|
||
|
||
pq = sub.add_parser('query', help='기간/계좌 필터')
|
||
pq.add_argument('--from', dest='date_from')
|
||
pq.add_argument('--to', dest='date_to')
|
||
pq.add_argument('--account')
|
||
pq.add_argument('--code')
|
||
|
||
args = p.parse_args(argv)
|
||
if args.cmd == 'collect':
|
||
collect(base_dt=args.date, force=args.force, quiet=args.quiet)
|
||
elif args.cmd == 'seed':
|
||
seed_initial(seed_date=args.date)
|
||
elif args.cmd == 'show':
|
||
show(args.code_or_name)
|
||
elif args.cmd == 'query':
|
||
query(date_from=args.date_from, date_to=args.date_to, account=args.account, code=args.code)
|
||
return 0
|
||
|
||
|
||
if __name__ == '__main__':
|
||
sys.exit(main())
|