#!/usr/bin/env python3 """invest_ledger_check.py — 매매 원장(raw/invest-ledger/ledger.md) 결정론 검사 CLI (stdlib only). 실돈 경로의 LLM 산술·자기신고를 기계 검증으로 대체한다 (하네스 감사 P2-17, OUT-7/OUT-12). 모드: --check /invest-decide 가 행 기록 *후* 호출. 검사: 1) Trade Log 행 스키마 (11열, 날짜/매수매도/수량/단가 형식) 2) 근거(링크) 셀 — 링크 존재 + 타깃 파일 실존 3) 최신 행 staleness — 근거가 invest-daily 면 >24h, invest-research 면 >90d 4) 주간 거래 수 집계 (--weekly-cap N 주면 초과 플래그) exit: 0 clean / 1 flags. --report /invest-review 가 호출. 원장 산술을 기계 재계산해 '손익 요약' 갱신 input 출력 (매수/매도 합·수수료 합·종목별 순수량·매수가중 평균단가·주간 거래 수). 평가금액·환차손익은 현재가 필요 → 본 스크립트 범위 밖(명시 출력). exit 0. 임계값 SSOT: 규칙 자체는 wiki/invest-strategy/strategy.md (①~⑤). 본 스크립트는 구조·산술·날짜만 검사하고, 임계 기본값(24h/90d)은 Spec F C4 의 기록된 기본값이며 플래그로 조정 가능(사용자 위험감내 재량). """ from __future__ import annotations import argparse import datetime as dt import re import sys from pathlib import Path ROOT = Path(__file__).resolve().parents[2] LEDGER_DEFAULT = "raw/invest-ledger/ledger.md" TRADE_COLS = ["날짜", "매수/매도", "종목", "수량", "단가", "수수료", "체결환율", "금액(원)", "계좌", "근거(링크)", "규칙체크"] DATE_RE = re.compile(r"(\d{4}-\d{2}-\d{2})") LINK_RE = re.compile(r"\[\[([^\]|#]+)") NUM_RE = re.compile(r"-?[\d,]+(?:\.\d+)?") def read_text(p: Path) -> str: return p.read_text(encoding="utf-8", errors="replace") def section_lines(text: str, header_prefix: str) -> list[str]: i = text.find(header_prefix) if i == -1: return [] j = text.find("\n## ", i + len(header_prefix)) return text[i: j if j != -1 else len(text)].splitlines() def table_rows(lines: list[str]) -> tuple[list[str], list[list[str]]]: """(헤더 셀, 데이터 행 목록). 구분선(|---|) 은 제외.""" header, rows = [], [] for line in lines: s = line.strip() if not s.startswith("|"): continue cells = [c.strip() for c in s.strip("|").split("|")] if not header: header = cells continue if all(re.fullmatch(r":?-{3,}:?", c) for c in cells if c): continue rows.append(cells) return header, rows def parse_num(cell: str) -> float | None: m = NUM_RE.search(cell.replace(",", "")) return float(m.group(0)) if m else None def doc_date(rel: str, root: Path) -> dt.date | None: """근거 문서의 기준 날짜 — 파일명 YYYY-MM-DD 우선, 없으면 frontmatter date/created.""" m = DATE_RE.search(Path(rel).name) if m: try: return dt.date.fromisoformat(m.group(1)) except ValueError: pass p = root / (rel if rel.endswith(".md") else rel + ".md") if not p.exists(): return None head = read_text(p)[:600] for key in ("date", "created", "last_reviewed"): m = re.search(rf"^{key}:\s*(\d{{4}}-\d{{2}}-\d{{2}})", head, re.M) if m: try: return dt.date.fromisoformat(m.group(1)) except ValueError: continue return None def evidence_targets(cell: str) -> list[str]: """근거 셀에서 위키링크/경로 추출 ([[...]] 또는 bare raw/·wiki/ 경로).""" out = [t.strip() for t in LINK_RE.findall(cell)] out += re.findall(r"(?:raw|wiki)/[\w./-]+", cell) seen, uniq = set(), [] for t in out: t = t[:-3] if t.endswith(".md") else t if t not in seen: seen.add(t) uniq.append(t) return uniq def iso_week(d: dt.date) -> tuple[int, int]: c = d.isocalendar() return (c[0], c[1]) def load_trades(text: str) -> tuple[list[str], list[list[str]], list[str]]: flags = [] lines = section_lines(text, "## 거래 내역") if not lines: flags.append("LEDGER_STRUCTURE: `## 거래 내역` 섹션 부재") return [], [], flags header, rows = table_rows(lines) if [h.strip() for h in header] != TRADE_COLS: flags.append(f"TRADE_SCHEMA: Trade Log 헤더가 템플릿 11열과 불일치 — 기대 {TRADE_COLS}, 실제 {header}") return header, rows, flags def check_rows(rows: list[list[str]], root: Path, daily_max_h: int, research_max_d: int, today: dt.date) -> list[str]: flags = [] for i, cells in enumerate(rows): label = f"행 {i + 1} ({cells[0] if cells else '?'})" if len(cells) != len(TRADE_COLS): flags.append(f"ROW_SCHEMA: {label} — 셀 {len(cells)}개 (기대 {len(TRADE_COLS)})") continue if not DATE_RE.fullmatch(cells[0]): flags.append(f"ROW_DATE: {label} — 날짜가 YYYY-MM-DD 아님: '{cells[0]}'") if cells[1] not in ("매수", "매도"): flags.append(f"ROW_SIDE: {label} — 매수/매도 아님: '{cells[1]}'") for col, idx in (("수량", 3), ("단가", 4)): if parse_num(cells[idx]) is None: flags.append(f"ROW_NUM: {label} — {col} 숫자 아님: '{cells[idx]}'") targets = evidence_targets(cells[9]) if not targets: flags.append(f"NO_EVIDENCE: {label} — 근거(링크) 셀에 문서 링크 없음 (선근거 원칙 위반)") else: for t in targets: if not (root / (t + ".md")).exists() and not (root / t).exists(): flags.append(f"EVIDENCE_MISSING: {label} — 근거 타깃 부재: {t}") # 최신 행(시간 역순 최상단) staleness — 결정 시점 기준. if rows and len(rows[0]) == len(TRADE_COLS): for t in evidence_targets(rows[0][9]): d = doc_date(t, root) if d is None: continue age_d = (today - d).days if t.startswith("raw/invest-daily/") and age_d * 24 > daily_max_h: flags.append(f"STALE_EVIDENCE: 최신 행 근거 {t} — 일일노트 {age_d}d 경과 (기준 {daily_max_h}h). " "조사시점 수치는 실시간 아님 → 재조사/현재가 재확인 먼저") elif t.startswith("raw/invest-research/") and age_d > research_max_d: flags.append(f"STALE_EVIDENCE: 최신 행 근거 {t} — 조사노트 {age_d}d 경과 (기준 {research_max_d}d)") return flags def weekly_count(rows: list[list[str]], today: dt.date) -> int: n = 0 for cells in rows: if cells and DATE_RE.fullmatch(cells[0] if cells else ""): if iso_week(dt.date.fromisoformat(cells[0])) == iso_week(today): n += 1 return n def run_check(text: str, root: Path, daily_max_h: int, research_max_d: int, weekly_cap: int | None, today: dt.date) -> int: header, rows, flags = load_trades(text) flags += check_rows(rows, root, daily_max_h, research_max_d, today) wk = weekly_count(rows, today) print(f"INFO 이번 주(ISO) 거래 수: {wk}" + (f" / 상한 {weekly_cap}" if weekly_cap else " (상한 미지정 — strategy ③ 주간상한과 대조하세요)")) if weekly_cap is not None and wk > weekly_cap: flags.append(f"WEEKLY_CAP: 이번 주 거래 {wk}회 > 상한 {weekly_cap} (strategy ③)") for f in flags: print(f"FLAG {f}") print(f"\n== invest-ledger check: 거래 {len(rows)}행 / 플래그 {len(flags)}건 ==") return 1 if flags else 0 def run_report(text: str, today: dt.date) -> int: _, rows, flags = load_trades(text) buy = sell = fees = 0.0 pos: dict[str, dict] = {} for cells in rows: if len(cells) != len(TRADE_COLS): continue qty, price = parse_num(cells[3]) or 0, parse_num(cells[4]) or 0 amt = parse_num(cells[7]) or 0 fees += parse_num(cells[5]) or 0 p = pos.setdefault(cells[2], {"qty": 0.0, "cost": 0.0}) if cells[1] == "매수": buy += amt p["qty"] += qty p["cost"] += qty * price elif cells[1] == "매도": sell += amt p["qty"] -= qty print("== 손익 요약 갱신 input (기계 재계산 — LLM 산술 금지) ==") print(f"- 거래 수: {len(rows)} (이번 주 {weekly_count(rows, today)})") print(f"- 매수 합(원): {buy:,.0f} / 매도 합(원): {sell:,.0f} / 순투입: {buy - sell:,.0f}") print(f"- 누적 수수료: {fees:,.0f}") for name, p in pos.items(): avg = (p["cost"] / p["qty"]) if p["qty"] else 0 print(f"- {name}: 순수량 {p['qty']:g} / 매수가중 평균단가 {avg:,.2f}") print("- 평가금액·환차손익·세후 추정: 현재가/환율 필요 — 본 스크립트 범위 밖 (출처 있는 시세로 별도 계산)") for f in flags: print(f"FLAG {f}") return 0 def main() -> None: ap = argparse.ArgumentParser(description="매매 원장 결정론 검사") ap.add_argument("--check", action="store_true", help="행 스키마·근거 실존·staleness·주간 거래 수 검사") ap.add_argument("--report", action="store_true", help="손익 요약 갱신 input 기계 재계산") ap.add_argument("--ledger", default=LEDGER_DEFAULT) ap.add_argument("--root", default=str(ROOT)) ap.add_argument("--daily-max-h", type=int, default=24, help="일일노트 근거 staleness 임계 (Spec F C4)") ap.add_argument("--research-max-d", type=int, default=90, help="조사노트 근거 staleness 임계") ap.add_argument("--weekly-cap", type=int, default=None, help="주간 거래 상한 (strategy ③의 N)") ap.add_argument("--today", default=None, help="기준일 YYYY-MM-DD (테스트용; 기본 오늘)") args = ap.parse_args() root = Path(args.root).resolve() today = dt.date.fromisoformat(args.today) if args.today else dt.date.today() p = Path(args.ledger) if not p.is_absolute(): p = root / args.ledger if not p.exists(): print(f"FLAG LEDGER_MISSING: {args.ledger}") sys.exit(1) text = read_text(p) if args.check: sys.exit(run_check(text, root, args.daily_max_h, args.research_max_d, args.weekly_cap, today)) if args.report: sys.exit(run_report(text, today)) ap.error("--check 또는 --report 필요") if __name__ == "__main__": main()