245 lines
10 KiB
Python
245 lines
10 KiB
Python
#!/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()
|