QuantEngineByItz/tools/build_verdict_consistency_lock_v1.py

"""VERDICT_CONSISTENCY_LOCK_V1
Verdict 일관성 잠금 — FINAL_JUDGMENT_GATE_V1 verdict를
operational_report.json의 종목별 서술과 대조한다.

위반(INVALID_VERDICT_OVERRIDE):
  - verdict=BLOCKED/WATCH/SELL 인데 보고서가 해당 종목에 BUY·신규매수·매수 서술
  - verdict=BUY_PILOT 인데 보고서가 해당 종목에 SELL·매도 서술 (과소평가 방지)

정직성 원칙:
  - 사용자 H10 수동 오버라이드(HTS 입력표에 명시)는 예외
  - sell_priority 표의 "보유"는 매도 미진행이므로 SELL verdict와 충돌 아님 (보수적 안전)
  - TRIM verdict와 "보유"는 충돌 — TRIM 지시가 무시된 것
"""
from __future__ import annotations

import argparse
import json
import re
import sys
from pathlib import Path
from typing import Any


def _ensure_utf8_stdio() -> None:
    if sys.stdout.encoding and sys.stdout.encoding.lower() not in ("utf-8", "utf8"):
        sys.stdout = open(sys.stdout.fileno(), mode="w", encoding="utf-8", buffering=1)
    if sys.stderr.encoding and sys.stderr.encoding.lower() not in ("utf-8", "utf8"):
        sys.stderr = open(sys.stderr.fileno(), mode="w", encoding="utf-8", buffering=1)


ROOT = Path(__file__).resolve().parents[1]
DEFAULT_FJ = ROOT / "Temp" / "final_judgment_gate_v1.json"
DEFAULT_REPORT = ROOT / "Temp" / "operational_report.json"
DEFAULT_OUT = ROOT / "Temp" / "verdict_consistency_lock_v1.json"

# BUY 의도를 나타내는 패턴 (한·영 혼용)
# 주의: "신규 매수 수량 50% 축소"는 매수 제한 표현이므로 매치 금지 → 긍정 수식어 필수
BUY_AFFIRM_PATTERNS = [
    r"신규\s*매수\s*(가능|허용|진입|OK|승인)",   # 신규 매수 가능/허용
    r"매수\s*가능",                                # 매수 가능
    r"매수\s*허용",                                # 매수 허용
    r"매수\s*진입\s*(가능|OK|승인)?",              # 매수 진입
    r"신규\s*진입\s*(가능|OK|승인)?",              # 신규 진입
    r"BUY_PILOT\s*가능",                           # BUY_PILOT 가능
    r"(?<!\w)BUY\s+OK(?!\w)",                      # BUY OK (독립 단어)
]


def _load_json(path: Path) -> dict[str, Any]:
    if not path.exists():
        return {}
    try:
        obj = json.loads(path.read_text(encoding="utf-8"))
    except Exception:
        return {}
    return obj if isinstance(obj, dict) else {}


def _sections_text(report: dict[str, Any]) -> dict[str, str]:
    """section name → markdown text."""
    result: dict[str, str] = {}
    for s in (report.get("sections") or []):
        if isinstance(s, dict):
            name = str(s.get("name") or "")
            md = str(s.get("markdown") or "")
            result[name] = md
    return result


def _extract_sell_priority_actions(section_md: str) -> dict[str, str]:
    """sell_priority_decision_table 에서 종목코드 → 최종행동 매핑 추출."""
    result: dict[str, str] = {}
    for line in section_md.splitlines():
        if "|" not in line:
            continue
        cells = [c.strip() for c in line.split("|")]
        # 파이프 아티팩트 제거
        if cells and cells[0] == "":
            cells = cells[1:]
        if cells and cells[-1] == "":
            cells = cells[:-1]
        # 구분선 skip
        if re.match(r"^[-:]+$", cells[0] if cells else ""):
            continue
        # 컬럼: 최종우선순위|원순위|종목|종목명|등급단계|점수|축소방식|현금방식|최종행동|매도사유
        if len(cells) >= 9:
            ticker = cells[2].strip()
            final_action = cells[8].strip()
            if ticker and ticker != "종목":
                result[ticker] = final_action
    return result


def _has_pattern(text: str, patterns: list[str]) -> str | None:
    """패턴 중 하나라도 매치되면 해당 패턴 반환, 없으면 None."""
    for p in patterns:
        if re.search(p, text, re.IGNORECASE):
            return p
    return None


def _check_ticker_in_section(ticker: str, section_md: str, patterns: list[str]) -> str | None:
    """ticker가 등장하는 행에서 patterns 중 하나라도 발견되면 반환."""
    lines = section_md.splitlines()
    for line in lines:
        if ticker in line:
            m = _has_pattern(line, patterns)
            if m:
                return m
    return None


def main() -> int:
    _ensure_utf8_stdio()
    ap = argparse.ArgumentParser(description="VERDICT_CONSISTENCY_LOCK_V1")
    ap.add_argument("--fj", default=str(DEFAULT_FJ))
    ap.add_argument("--report", default=str(DEFAULT_REPORT))
    ap.add_argument("--out", default=str(DEFAULT_OUT))
    args = ap.parse_args()

    def _rp(s: str) -> Path:
        p = Path(s)
        return p if p.is_absolute() else ROOT / p

    fj_path = _rp(args.fj)
    report_path = _rp(args.report)
    out_path = _rp(args.out)

    # ─── 로드 ─────────────────────────────────────────────────────────────
    fj = _load_json(fj_path)
    report = _load_json(report_path)

    if not fj or not fj.get("rows"):
        print("ERROR: final_judgment_gate_v1.json 비어있음 — T2 먼저 실행 필요", file=sys.stderr)
        return 1
    if not report:
        print("ERROR: operational_report.json 비어있음 — render 먼저 실행 필요", file=sys.stderr)
        return 1

    sections = _sections_text(report)
    full_text = "\n".join(sections.values())

    # sell_priority_decision_table 최종행동 추출
    sp_actions = _extract_sell_priority_actions(sections.get("sell_priority_decision_table", ""))

    # ─── 검사 ─────────────────────────────────────────────────────────────
    violations: list[dict[str, Any]] = []
    checked: list[dict[str, Any]] = []

    for row in fj.get("rows") or []:
        ticker = str(row.get("ticker") or "")
        verdict = str(row.get("action_verdict") or "")
        if not ticker:
            continue

        issues: list[str] = []

        # [1] BLOCKED/WATCH/SELL → 보고서에서 긍정적 BUY 서술 금지 (CRITICAL)
        if verdict in {"BLOCKED", "WATCH", "SELL"}:
            buy_hit = _check_ticker_in_section(ticker, full_text, BUY_AFFIRM_PATTERNS)
            if buy_hit:
                issues.append(f"CRITICAL: verdict={verdict} but BUY_AFFIRM pattern found: '{buy_hit}'")

        # [2] TRIM → sell_priority_decision_table 최종행동이 "보유"이면 WARN (soft conflict)
        # 주의: SELL verdict인데 "보유"면 soft conflict이지만 HARD FAIL 아님
        # (보수적 보유 지시 vs 트림 권고 — 안전 방향으로 해석)
        sp_action = sp_actions.get(ticker, "")
        if verdict == "TRIM" and "보유" in sp_action and "매도" not in sp_action:
            issues.append(f"WARN: verdict=TRIM but sell_priority_action='{sp_action}' — trim signal may be ignored")

        entry = {
            "ticker": ticker,
            "action_verdict": verdict,
            "sell_priority_action": sp_actions.get(ticker, "N/A"),
            "violations": issues,
            "status": "INVALID_VERDICT_OVERRIDE" if issues else "OK",
        }
        checked.append(entry)
        if issues:
            violations.append(entry)

    # CRITICAL 위반만 gate=FAIL, WARN은 기록만
    critical_violations = [v for v in violations if any("CRITICAL" in iss for iss in v["violations"])]
    warn_violations = [v for v in violations if all("WARN" in iss for iss in v["violations"])]
    override_count = len(critical_violations)
    gate = "PASS" if override_count == 0 else "FAIL"

    out = {
        "formula_id": "VERDICT_CONSISTENCY_LOCK_V1",
        "gate": gate,
        "override_count": override_count,
        "warn_count": len(warn_violations),
        "ticker_count": len(checked),
        "violations": violations,
        "critical_violations": critical_violations,
        "warn_violations": warn_violations,
        "checked": checked,
    }
    out_path.parent.mkdir(parents=True, exist_ok=True)
    out_path.write_text(json.dumps(out, ensure_ascii=False, indent=2), encoding="utf-8")

    print("VERDICT_CONSISTENCY_LOCK_V1")
    print(f"  tickers_checked: {len(checked)}")
    print(f"  critical_override_count: {override_count}  (반드시 0)")
    print(f"  warn_count: {len(warn_violations)}")
    print(f"  gate: {gate}")
    if critical_violations:
        print("  CRITICAL VIOLATIONS (INVALID_VERDICT_OVERRIDE):")
        for v in critical_violations:
            print(f"    [{v['ticker']}] verdict={v['action_verdict']} | {'; '.join(v['violations'])}")
    else:
        print("  No CRITICAL violations — verdict integrity confirmed.")
    if warn_violations:
        print("  WARN (soft conflicts — not gate failure):")
        for v in warn_violations:
            print(f"    [{v['ticker']}] {'; '.join(v['violations'])}")
    return 0 if gate == "PASS" else 1


if __name__ == "__main__":
    raise SystemExit(main())