feat: 공식 문서 근거자료, 브랜치 기능 문서 작성

2026-07-29 18:05:17 +09:00
parent cfd84875bf
commit 58515ab0f3
251 changed files with 31470 additions and 109 deletions
@@ -0,0 +1,244 @@
+#!/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()